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>
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.
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
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.
26 #include <common/common.h>
27 #include <common/relayd/relayd.h>
28 #include <common/ust-consumer/ust-consumer.h>
30 #include "consumer-stream.h"
33 * RCU call to free stream. MUST only be used with call_rcu().
35 static void free_stream_rcu(struct rcu_head
*head
)
37 struct lttng_ht_node_u64
*node
=
38 caa_container_of(head
, struct lttng_ht_node_u64
, head
);
39 struct lttng_consumer_stream
*stream
=
40 caa_container_of(node
, struct lttng_consumer_stream
, node
);
42 pthread_mutex_destroy(&stream
->lock
);
47 * Close stream on the relayd side. This call can destroy a relayd if the
50 * A RCU read side lock MUST be acquired if the relayd object was looked up in
51 * a hash table before calling this.
53 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
54 struct consumer_relayd_sock_pair
*relayd
)
61 uatomic_dec(&relayd
->refcount
);
62 assert(uatomic_read(&relayd
->refcount
) >= 0);
64 /* Closing streams requires to lock the control socket. */
65 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
66 ret
= relayd_send_close_stream(&relayd
->control_sock
,
67 stream
->relayd_stream_id
,
68 stream
->next_net_seq_num
- 1);
69 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
71 DBG("Unable to close stream on the relayd. Continuing");
73 * Continue here. There is nothing we can do for the relayd.
74 * Chances are that the relayd has closed the socket so we just
75 * continue cleaning up.
79 /* Both conditions are met, we destroy the relayd. */
80 if (uatomic_read(&relayd
->refcount
) == 0 &&
81 uatomic_read(&relayd
->destroy_flag
)) {
82 consumer_destroy_relayd(relayd
);
84 stream
->net_seq_idx
= (uint64_t) -1ULL;
88 * Close stream's file descriptors and, if needed, close stream also on the
91 * The consumer data lock MUST be acquired.
92 * The stream lock MUST be acquired.
94 void consumer_stream_close(struct lttng_consumer_stream
*stream
)
97 struct consumer_relayd_sock_pair
*relayd
;
101 switch (consumer_data
.type
) {
102 case LTTNG_CONSUMER_KERNEL
:
103 if (stream
->mmap_base
!= NULL
) {
104 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
110 if (stream
->wait_fd
>= 0) {
111 ret
= close(stream
->wait_fd
);
115 stream
->wait_fd
= -1;
118 case LTTNG_CONSUMER32_UST
:
119 case LTTNG_CONSUMER64_UST
:
122 ERR("Unknown consumer_data type");
126 /* Close output fd. Could be a socket or local file at this point. */
127 if (stream
->out_fd
>= 0) {
128 ret
= close(stream
->out_fd
);
135 /* Check and cleanup relayd if needed. */
137 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
138 if (relayd
!= NULL
) {
139 consumer_stream_relayd_close(stream
, relayd
);
145 * Delete the stream from all possible hash tables.
147 * The consumer data lock MUST be acquired.
148 * The stream lock MUST be acquired.
150 void consumer_stream_delete(struct lttng_consumer_stream
*stream
,
154 struct lttng_ht_iter iter
;
157 /* Should NEVER be called not in monitor mode. */
158 assert(stream
->chan
->monitor
);
163 iter
.iter
.node
= &stream
->node
.node
;
164 ret
= lttng_ht_del(ht
, &iter
);
168 /* Delete from stream per channel ID hash table. */
169 iter
.iter
.node
= &stream
->node_channel_id
.node
;
171 * The returned value is of no importance. Even if the node is NOT in the
172 * hash table, we continue since we may have been called by a code path
173 * that did not add the stream to a (all) hash table. Same goes for the
174 * next call ht del call.
176 (void) lttng_ht_del(consumer_data
.stream_per_chan_id_ht
, &iter
);
178 /* Delete from the global stream list. */
179 iter
.iter
.node
= &stream
->node_session_id
.node
;
180 /* See the previous ht del on why we ignore the returned value. */
181 (void) lttng_ht_del(consumer_data
.stream_list_ht
, &iter
);
185 /* Decrement the stream count of the global consumer data. */
186 assert(consumer_data
.stream_count
> 0);
187 consumer_data
.stream_count
--;
191 * Free the given stream within a RCU call.
193 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
197 call_rcu(&stream
->node
.head
, free_stream_rcu
);
201 * Destroy the stream's buffers of the tracer.
203 void consumer_stream_destroy_buffers(struct lttng_consumer_stream
*stream
)
207 switch (consumer_data
.type
) {
208 case LTTNG_CONSUMER_KERNEL
:
210 case LTTNG_CONSUMER32_UST
:
211 case LTTNG_CONSUMER64_UST
:
212 lttng_ustconsumer_del_stream(stream
);
215 ERR("Unknown consumer_data type");
221 * Destroy and close a already created stream.
223 static void destroy_close_stream(struct lttng_consumer_stream
*stream
)
227 DBG("Consumer stream destroy monitored key: %" PRIu64
, stream
->key
);
229 /* Destroy tracer buffers of the stream. */
230 consumer_stream_destroy_buffers(stream
);
231 /* Close down everything including the relayd if one. */
232 consumer_stream_close(stream
);
236 * Decrement the stream's channel refcount and if down to 0, return the channel
237 * pointer so it can be destroyed by the caller or NULL if not.
239 static struct lttng_consumer_channel
*unref_channel(
240 struct lttng_consumer_stream
*stream
)
242 struct lttng_consumer_channel
*free_chan
= NULL
;
245 assert(stream
->chan
);
247 /* Update refcount of channel and see if we need to destroy it. */
248 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1)
249 && !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
250 free_chan
= stream
->chan
;
257 * Destroy a stream completely. This will delete, close and free the stream.
258 * Once return, the stream is NO longer usable. Its channel may get destroyed
259 * if conditions are met for a monitored stream.
261 * This MUST be called WITHOUT the consumer data and stream lock acquired if
262 * the stream is in _monitor_ mode else it does not matter.
264 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
,
269 /* Stream is in monitor mode. */
270 if (stream
->monitor
) {
271 struct lttng_consumer_channel
*free_chan
= NULL
;
274 * This means that the stream was successfully removed from the streams
275 * list of the channel and sent to the right thread managing this
276 * stream thus being globally visible.
278 if (stream
->globally_visible
) {
279 pthread_mutex_lock(&consumer_data
.lock
);
280 pthread_mutex_lock(&stream
->lock
);
281 /* Remove every reference of the stream in the consumer. */
282 consumer_stream_delete(stream
, ht
);
284 destroy_close_stream(stream
);
286 /* Update channel's refcount of the stream. */
287 free_chan
= unref_channel(stream
);
289 /* Indicates that the consumer data state MUST be updated after this. */
290 consumer_data
.need_update
= 1;
292 pthread_mutex_unlock(&stream
->lock
);
293 pthread_mutex_unlock(&consumer_data
.lock
);
296 * If the stream is not visible globally, this needs to be done
297 * outside of the consumer data lock section.
299 free_chan
= unref_channel(stream
);
303 consumer_del_channel(free_chan
);
306 destroy_close_stream(stream
);
309 /* Free stream within a RCU call. */
310 consumer_stream_free(stream
);