Add consumer-stream.c/.h in libconsumer
[lttng-tools.git] / src / common / consumer-stream.c
CommitLineData
51230d70
DG
1/*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * Copyright (C) 2013 - David Goulet <dgoulet@efficios.com>
5 *
6 * This program is free software; you can redistribute it and/or modify it
7 * under the terms of the GNU General Public License, version 2 only, as
8 * published by the Free Software Foundation.
9 *
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
13 * more details.
14 *
15 * You should have received a copy of the GNU General Public License along with
16 * this program; if not, write to the Free Software Foundation, Inc., 51
17 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18 */
19
20#define _GNU_SOURCE
21#include <assert.h>
22#include <sys/mman.h>
23#include <unistd.h>
24
25#include <common/common.h>
26#include <common/relayd/relayd.h>
27#include <common/ust-consumer/ust-consumer.h>
28
29#include "consumer-stream.h"
30
31/*
32 * RCU call to free stream. MUST only be used with call_rcu().
33 */
34static void free_stream_rcu(struct rcu_head *head)
35{
36 struct lttng_ht_node_u64 *node =
37 caa_container_of(head, struct lttng_ht_node_u64, head);
38 struct lttng_consumer_stream *stream =
39 caa_container_of(node, struct lttng_consumer_stream, node);
40
41 pthread_mutex_destroy(&stream->lock);
42 free(stream);
43}
44
45/*
46 * Close stream on the relayd side. This call can destroy a relayd if the
47 * conditions are met.
48 *
49 * A RCU read side lock MUST be acquired if the relayd object was looked up in
50 * a hash table before calling this.
51 */
52void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
53 struct consumer_relayd_sock_pair *relayd)
54{
55 int ret;
56
57 assert(stream);
58 assert(relayd);
59
60 uatomic_dec(&relayd->refcount);
61 assert(uatomic_read(&relayd->refcount) >= 0);
62
63 /* Closing streams requires to lock the control socket. */
64 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
65 ret = relayd_send_close_stream(&relayd->control_sock,
66 stream->relayd_stream_id,
67 stream->next_net_seq_num - 1);
68 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
69 if (ret < 0) {
70 DBG("Unable to close stream on the relayd. Continuing");
71 /*
72 * Continue here. There is nothing we can do for the relayd.
73 * Chances are that the relayd has closed the socket so we just
74 * continue cleaning up.
75 */
76 }
77
78 /* Both conditions are met, we destroy the relayd. */
79 if (uatomic_read(&relayd->refcount) == 0 &&
80 uatomic_read(&relayd->destroy_flag)) {
81 consumer_destroy_relayd(relayd);
82 }
83}
84
85/*
86 * Close stream's file descriptors and, if needed, close stream also on the
87 * relayd side.
88 *
89 * The consumer data lock MUST be acquired.
90 * The stream lock MUST be acquired.
91 */
92void consumer_stream_close(struct lttng_consumer_stream *stream)
93{
94 int ret;
95 struct consumer_relayd_sock_pair *relayd;
96
97 assert(stream);
98
99 switch (consumer_data.type) {
100 case LTTNG_CONSUMER_KERNEL:
101 if (stream->mmap_base != NULL) {
102 ret = munmap(stream->mmap_base, stream->mmap_len);
103 if (ret != 0) {
104 PERROR("munmap");
105 }
106 }
107
108 if (stream->wait_fd >= 0) {
109 ret = close(stream->wait_fd);
110 if (ret) {
111 PERROR("close");
112 }
113 }
114 break;
115 case LTTNG_CONSUMER32_UST:
116 case LTTNG_CONSUMER64_UST:
117 lttng_ustconsumer_del_stream(stream);
118 break;
119 default:
120 ERR("Unknown consumer_data type");
121 assert(0);
122 }
123
124 /* Close output fd. Could be a socket or local file at this point. */
125 if (stream->out_fd >= 0) {
126 ret = close(stream->out_fd);
127 if (ret) {
128 PERROR("close");
129 }
130 }
131
132 /* Check and cleanup relayd if needed. */
133 rcu_read_lock();
134 relayd = consumer_find_relayd(stream->net_seq_idx);
135 if (relayd != NULL) {
136 consumer_stream_relayd_close(stream, relayd);
137 }
138 rcu_read_unlock();
139}
140
141/*
142 * Delete the stream from all possible hash tables.
143 *
144 * The consumer data lock MUST be acquired.
145 * The stream lock MUST be acquired.
146 */
147void consumer_stream_delete(struct lttng_consumer_stream *stream,
148 struct lttng_ht *ht)
149{
150 int ret;
151 struct lttng_ht_iter iter;
152
153 assert(stream);
154
155 rcu_read_lock();
156
157 if (ht) {
158 iter.iter.node = &stream->node.node;
159 ret = lttng_ht_del(ht, &iter);
160 assert(!ret);
161 }
162
163 /* Delete from stream per channel ID hash table. */
164 iter.iter.node = &stream->node_channel_id.node;
165 /*
166 * The returned value is of no importance. Even if the node is NOT in the
167 * hash table, we continue since we may have been called by a code path
168 * that did not add the stream to a (all) hash table. Same goes for the
169 * next call ht del call.
170 */
171 (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
172
173 /* Delete from the global stream list. */
174 iter.iter.node = &stream->node_session_id.node;
175 /* See the previous ht del on why we ignore the returned value. */
176 (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
177
178 rcu_read_unlock();
179
180 /* Decrement the stream count of the global consumer data. */
181 assert(consumer_data.stream_count > 0);
182 consumer_data.stream_count--;
183}
184
185/*
186 * Free the given stream within a RCU call.
187 */
188void consumer_stream_free(struct lttng_consumer_stream *stream)
189{
190 assert(stream);
191
192 call_rcu(&stream->node.head, free_stream_rcu);
193}
194
195/*
196 * Destroy a stream completely. This will delete, close and free the stream.
197 * Once return, the stream is NO longer usable. Its channel may get destroyed
198 * if conditions are met.
199 *
200 * This MUST be called WITHOUT the consumer data and stream lock acquired.
201 */
202void consumer_stream_destroy(struct lttng_consumer_stream *stream,
203 struct lttng_ht *ht)
204{
205 struct lttng_consumer_channel *free_chan = NULL;
206
207 assert(stream);
208
209 DBG("Consumer stream destroy - wait_fd: %d", stream->wait_fd);
210
211 pthread_mutex_lock(&consumer_data.lock);
212 pthread_mutex_lock(&stream->lock);
213
214 /* Remove every reference of the stream in the consumer. */
215 consumer_stream_delete(stream, ht);
216
217 /* Close down everything including the relayd if one. */
218 consumer_stream_close(stream);
219
220 /* Update refcount of channel and see if we need to destroy it. */
221 if (!uatomic_sub_return(&stream->chan->refcount, 1)
222 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
223 free_chan = stream->chan;
224 }
225
226 /* Indicates that the consumer data state MUST be updated after this. */
227 consumer_data.need_update = 1;
228
229 pthread_mutex_unlock(&stream->lock);
230 pthread_mutex_unlock(&consumer_data.lock);
231
232 if (free_chan) {
233 consumer_del_channel(free_chan);
234 }
235
236 /* Free stream within a RCU call. */
237 consumer_stream_free(stream);
238}
This page took 0.047969 seconds and 4 git commands to generate.