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/index/index.h>
28 #include <common/kernel-consumer/kernel-consumer.h>
29 #include <common/relayd/relayd.h>
30 #include <common/ust-consumer/ust-consumer.h>
31 #include <common/utils.h>
33 #include "consumer-stream.h"
36 * RCU call to free stream. MUST only be used with call_rcu().
38 static void free_stream_rcu(struct rcu_head
*head
)
40 struct lttng_ht_node_u64
*node
=
41 caa_container_of(head
, struct lttng_ht_node_u64
, head
);
42 struct lttng_consumer_stream
*stream
=
43 caa_container_of(node
, struct lttng_consumer_stream
, node
);
45 pthread_mutex_destroy(&stream
->lock
);
50 * Close stream on the relayd side. This call can destroy a relayd if the
53 * A RCU read side lock MUST be acquired if the relayd object was looked up in
54 * a hash table before calling this.
56 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
57 struct consumer_relayd_sock_pair
*relayd
)
64 if (stream
->sent_to_relayd
) {
65 uatomic_dec(&relayd
->refcount
);
66 assert(uatomic_read(&relayd
->refcount
) >= 0);
69 /* Closing streams requires to lock the control socket. */
70 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
71 ret
= relayd_send_close_stream(&relayd
->control_sock
,
72 stream
->relayd_stream_id
,
73 stream
->next_net_seq_num
- 1);
74 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
76 DBG("Unable to close stream on the relayd. Continuing");
78 * Continue here. There is nothing we can do for the relayd.
79 * Chances are that the relayd has closed the socket so we just
80 * continue cleaning up.
84 /* Both conditions are met, we destroy the relayd. */
85 if (uatomic_read(&relayd
->refcount
) == 0 &&
86 uatomic_read(&relayd
->destroy_flag
)) {
87 consumer_destroy_relayd(relayd
);
89 stream
->net_seq_idx
= (uint64_t) -1ULL;
90 stream
->sent_to_relayd
= 0;
94 * Close stream's file descriptors and, if needed, close stream also on the
97 * The consumer data lock MUST be acquired.
98 * The stream lock MUST be acquired.
100 void consumer_stream_close(struct lttng_consumer_stream
*stream
)
103 struct consumer_relayd_sock_pair
*relayd
;
107 switch (consumer_data
.type
) {
108 case LTTNG_CONSUMER_KERNEL
:
109 if (stream
->mmap_base
!= NULL
) {
110 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
116 if (stream
->wait_fd
>= 0) {
117 ret
= close(stream
->wait_fd
);
121 stream
->wait_fd
= -1;
123 if (stream
->chan
->output
== CONSUMER_CHANNEL_SPLICE
) {
124 utils_close_pipe(stream
->splice_pipe
);
127 case LTTNG_CONSUMER32_UST
:
128 case LTTNG_CONSUMER64_UST
:
131 * Special case for the metadata since the wait fd is an internal pipe
132 * polled in the metadata thread.
134 if (stream
->metadata_flag
&& stream
->chan
->monitor
) {
135 int rpipe
= stream
->ust_metadata_poll_pipe
[0];
138 * This will stop the channel timer if one and close the write side
139 * of the metadata poll pipe.
141 lttng_ustconsumer_close_metadata(stream
->chan
);
145 PERROR("closing metadata pipe read side");
147 stream
->ust_metadata_poll_pipe
[0] = -1;
153 ERR("Unknown consumer_data type");
157 /* Close output fd. Could be a socket or local file at this point. */
158 if (stream
->out_fd
>= 0) {
159 ret
= close(stream
->out_fd
);
166 if (stream
->index_file
) {
167 lttng_index_file_put(stream
->index_file
);
168 stream
->index_file
= NULL
;
171 /* Check and cleanup relayd if needed. */
173 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
174 if (relayd
!= NULL
) {
175 consumer_stream_relayd_close(stream
, relayd
);
181 * Delete the stream from all possible hash tables.
183 * The consumer data lock MUST be acquired.
184 * The stream lock MUST be acquired.
186 void consumer_stream_delete(struct lttng_consumer_stream
*stream
,
190 struct lttng_ht_iter iter
;
193 /* Should NEVER be called not in monitor mode. */
194 assert(stream
->chan
->monitor
);
199 iter
.iter
.node
= &stream
->node
.node
;
200 ret
= lttng_ht_del(ht
, &iter
);
204 /* Delete from stream per channel ID hash table. */
205 iter
.iter
.node
= &stream
->node_channel_id
.node
;
207 * The returned value is of no importance. Even if the node is NOT in the
208 * hash table, we continue since we may have been called by a code path
209 * that did not add the stream to a (all) hash table. Same goes for the
210 * next call ht del call.
212 (void) lttng_ht_del(consumer_data
.stream_per_chan_id_ht
, &iter
);
214 /* Delete from the global stream list. */
215 iter
.iter
.node
= &stream
->node_session_id
.node
;
216 /* See the previous ht del on why we ignore the returned value. */
217 (void) lttng_ht_del(consumer_data
.stream_list_ht
, &iter
);
221 if (!stream
->metadata_flag
) {
222 /* Decrement the stream count of the global consumer data. */
223 assert(consumer_data
.stream_count
> 0);
224 consumer_data
.stream_count
--;
229 * Free the given stream within a RCU call.
231 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
235 call_rcu(&stream
->node
.head
, free_stream_rcu
);
239 * Destroy the stream's buffers of the tracer.
241 void consumer_stream_destroy_buffers(struct lttng_consumer_stream
*stream
)
245 switch (consumer_data
.type
) {
246 case LTTNG_CONSUMER_KERNEL
:
248 case LTTNG_CONSUMER32_UST
:
249 case LTTNG_CONSUMER64_UST
:
250 lttng_ustconsumer_del_stream(stream
);
253 ERR("Unknown consumer_data type");
259 * Destroy and close a already created stream.
261 static void destroy_close_stream(struct lttng_consumer_stream
*stream
)
265 DBG("Consumer stream destroy monitored key: %" PRIu64
, stream
->key
);
267 /* Destroy tracer buffers of the stream. */
268 consumer_stream_destroy_buffers(stream
);
269 /* Close down everything including the relayd if one. */
270 consumer_stream_close(stream
);
274 * Decrement the stream's channel refcount and if down to 0, return the channel
275 * pointer so it can be destroyed by the caller or NULL if not.
277 static struct lttng_consumer_channel
*unref_channel(
278 struct lttng_consumer_stream
*stream
)
280 struct lttng_consumer_channel
*free_chan
= NULL
;
283 assert(stream
->chan
);
285 /* Update refcount of channel and see if we need to destroy it. */
286 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1)
287 && !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
288 free_chan
= stream
->chan
;
295 * Destroy a stream completely. This will delete, close and free the stream.
296 * Once return, the stream is NO longer usable. Its channel may get destroyed
297 * if conditions are met for a monitored stream.
299 * This MUST be called WITHOUT the consumer data and stream lock acquired if
300 * the stream is in _monitor_ mode else it does not matter.
302 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
,
307 /* Stream is in monitor mode. */
308 if (stream
->monitor
) {
309 struct lttng_consumer_channel
*free_chan
= NULL
;
312 * This means that the stream was successfully removed from the streams
313 * list of the channel and sent to the right thread managing this
314 * stream thus being globally visible.
316 if (stream
->globally_visible
) {
317 pthread_mutex_lock(&consumer_data
.lock
);
318 pthread_mutex_lock(&stream
->chan
->lock
);
319 pthread_mutex_lock(&stream
->lock
);
320 /* Remove every reference of the stream in the consumer. */
321 consumer_stream_delete(stream
, ht
);
323 destroy_close_stream(stream
);
325 /* Update channel's refcount of the stream. */
326 free_chan
= unref_channel(stream
);
328 /* Indicates that the consumer data state MUST be updated after this. */
329 consumer_data
.need_update
= 1;
331 pthread_mutex_unlock(&stream
->lock
);
332 pthread_mutex_unlock(&stream
->chan
->lock
);
333 pthread_mutex_unlock(&consumer_data
.lock
);
336 * If the stream is not visible globally, this needs to be done
337 * outside of the consumer data lock section.
339 free_chan
= unref_channel(stream
);
343 consumer_del_channel(free_chan
);
346 destroy_close_stream(stream
);
349 /* Free stream within a RCU call. */
350 consumer_stream_free(stream
);
354 * Write index of a specific stream either on the relayd or local disk.
356 * Return 0 on success or else a negative value.
358 int consumer_stream_write_index(struct lttng_consumer_stream
*stream
,
359 struct ctf_packet_index
*element
)
367 if (stream
->net_seq_idx
!= (uint64_t) -1ULL) {
368 struct consumer_relayd_sock_pair
*relayd
;
369 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
371 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
372 ret
= relayd_send_index(&relayd
->control_sock
, element
,
373 stream
->relayd_stream_id
, stream
->next_net_seq_num
- 1);
374 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
376 ERR("Stream %" PRIu64
" relayd ID %" PRIu64
" unknown. Can't write index.",
377 stream
->key
, stream
->net_seq_idx
);
381 if (lttng_index_file_write(stream
->index_file
, element
)) {
397 * Actually do the metadata sync using the given metadata stream.
399 * Return 0 on success else a negative value. ENODATA can be returned also
400 * indicating that there is no metadata available for that stream.
402 static int do_sync_metadata(struct lttng_consumer_stream
*metadata
,
403 struct lttng_consumer_local_data
*ctx
)
408 assert(metadata
->metadata_flag
);
412 * In UST, since we have to write the metadata from the cache packet
413 * by packet, we might need to start this procedure multiple times
414 * until all the metadata from the cache has been extracted.
419 * - Lock the metadata stream
420 * - Check if metadata stream node was deleted before locking.
421 * - if yes, release and return success
422 * - Check if new metadata is ready (flush + snapshot pos)
423 * - If nothing : release and return.
424 * - Lock the metadata_rdv_lock
425 * - Unlock the metadata stream
426 * - cond_wait on metadata_rdv to wait the wakeup from the
428 * - Unlock the metadata_rdv_lock
430 pthread_mutex_lock(&metadata
->lock
);
433 * There is a possibility that we were able to acquire a reference on the
434 * stream from the RCU hash table but between then and now, the node might
435 * have been deleted just before the lock is acquired. Thus, after locking,
436 * we make sure the metadata node has not been deleted which means that the
437 * buffers are closed.
439 * In that case, there is no need to sync the metadata hence returning a
440 * success return code.
442 ret
= cds_lfht_is_node_deleted(&metadata
->node
.node
);
445 goto end_unlock_mutex
;
449 case LTTNG_CONSUMER_KERNEL
:
451 * Empty the metadata cache and flush the current stream.
453 ret
= lttng_kconsumer_sync_metadata(metadata
);
455 case LTTNG_CONSUMER32_UST
:
456 case LTTNG_CONSUMER64_UST
:
458 * Ask the sessiond if we have new metadata waiting and update the
459 * consumer metadata cache.
461 ret
= lttng_ustconsumer_sync_metadata(ctx
, metadata
);
469 * Error or no new metadata, we exit here.
471 if (ret
<= 0 || ret
== ENODATA
) {
472 goto end_unlock_mutex
;
476 * At this point, new metadata have been flushed, so we wait on the
477 * rendez-vous point for the metadata thread to wake us up when it
478 * finishes consuming the metadata and continue execution.
481 pthread_mutex_lock(&metadata
->metadata_rdv_lock
);
484 * Release metadata stream lock so the metadata thread can process it.
486 pthread_mutex_unlock(&metadata
->lock
);
489 * Wait on the rendez-vous point. Once woken up, it means the metadata was
490 * consumed and thus synchronization is achieved.
492 pthread_cond_wait(&metadata
->metadata_rdv
, &metadata
->metadata_rdv_lock
);
493 pthread_mutex_unlock(&metadata
->metadata_rdv_lock
);
494 } while (ret
== EAGAIN
);
500 pthread_mutex_unlock(&metadata
->lock
);
505 * Synchronize the metadata using a given session ID. A successful acquisition
506 * of a metadata stream will trigger a request to the session daemon and a
507 * snapshot so the metadata thread can consume it.
509 * This function call is a rendez-vous point between the metadata thread and
512 * Return 0 on success or else a negative value.
514 int consumer_stream_sync_metadata(struct lttng_consumer_local_data
*ctx
,
518 struct lttng_consumer_stream
*stream
= NULL
;
519 struct lttng_ht_iter iter
;
524 /* Ease our life a bit. */
525 ht
= consumer_data
.stream_list_ht
;
529 /* Search the metadata associated with the session id of the given stream. */
531 cds_lfht_for_each_entry_duplicate(ht
->ht
,
532 ht
->hash_fct(&session_id
, lttng_ht_seed
), ht
->match_fct
,
533 &session_id
, &iter
.iter
, stream
, node_session_id
.node
) {
534 if (!stream
->metadata_flag
) {
538 ret
= do_sync_metadata(stream
, ctx
);
545 * Force return code to 0 (success) since ret might be ENODATA for instance
546 * which is not an error but rather that we should come back.