2 * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
6 * SPDX-License-Identifier: GPL-2.0-only
16 #include <common/common.h>
17 #include <common/index/index.h>
18 #include <common/kernel-consumer/kernel-consumer.h>
19 #include <common/relayd/relayd.h>
20 #include <common/ust-consumer/ust-consumer.h>
21 #include <common/utils.h>
22 #include <common/consumer/consumer.h>
23 #include <common/consumer/consumer-timer.h>
24 #include <common/consumer/metadata-bucket.h>
25 #include <common/kernel-ctl/kernel-ctl.h>
27 #include "consumer-stream.h"
30 * RCU call to free stream. MUST only be used with call_rcu().
32 static void free_stream_rcu(struct rcu_head
*head
)
34 struct lttng_ht_node_u64
*node
=
35 caa_container_of(head
, struct lttng_ht_node_u64
, head
);
36 struct lttng_consumer_stream
*stream
=
37 caa_container_of(node
, struct lttng_consumer_stream
, node
);
39 pthread_mutex_destroy(&stream
->lock
);
43 static void consumer_stream_data_lock_all(struct lttng_consumer_stream
*stream
)
45 pthread_mutex_lock(&stream
->chan
->lock
);
46 pthread_mutex_lock(&stream
->lock
);
49 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream
*stream
)
51 pthread_mutex_unlock(&stream
->lock
);
52 pthread_mutex_unlock(&stream
->chan
->lock
);
55 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream
*stream
)
57 consumer_stream_data_lock_all(stream
);
58 pthread_mutex_lock(&stream
->metadata_rdv_lock
);
61 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream
*stream
)
63 pthread_mutex_unlock(&stream
->metadata_rdv_lock
);
64 consumer_stream_data_unlock_all(stream
);
67 /* Only used for data streams. */
68 static int consumer_stream_update_stats(struct lttng_consumer_stream
*stream
,
69 const struct stream_subbuffer
*subbuf
)
72 uint64_t sequence_number
;
73 const uint64_t discarded_events
= subbuf
->info
.data
.events_discarded
;
75 if (!subbuf
->info
.data
.sequence_number
.is_set
) {
76 /* Command not supported by the tracer. */
77 sequence_number
= -1ULL;
78 stream
->sequence_number_unavailable
= true;
80 sequence_number
= subbuf
->info
.data
.sequence_number
.value
;
84 * Start the sequence when we extract the first packet in case we don't
85 * start at 0 (for example if a consumer is not connected to the
86 * session immediately after the beginning).
88 if (stream
->last_sequence_number
== -1ULL) {
89 stream
->last_sequence_number
= sequence_number
;
90 } else if (sequence_number
> stream
->last_sequence_number
) {
91 stream
->chan
->lost_packets
+= sequence_number
-
92 stream
->last_sequence_number
- 1;
94 /* seq <= last_sequence_number */
95 ERR("Sequence number inconsistent : prev = %" PRIu64
96 ", current = %" PRIu64
,
97 stream
->last_sequence_number
, sequence_number
);
101 stream
->last_sequence_number
= sequence_number
;
103 if (discarded_events
< stream
->last_discarded_events
) {
105 * Overflow has occurred. We assume only one wrap-around
108 stream
->chan
->discarded_events
+=
109 (1ULL << (CAA_BITS_PER_LONG
- 1)) -
110 stream
->last_discarded_events
+
113 stream
->chan
->discarded_events
+= discarded_events
-
114 stream
->last_discarded_events
;
116 stream
->last_discarded_events
= discarded_events
;
124 void ctf_packet_index_populate(struct ctf_packet_index
*index
,
125 off_t offset
, const struct stream_subbuffer
*subbuffer
)
127 *index
= (typeof(*index
)){
128 .offset
= htobe64(offset
),
129 .packet_size
= htobe64(subbuffer
->info
.data
.packet_size
),
130 .content_size
= htobe64(subbuffer
->info
.data
.content_size
),
131 .timestamp_begin
= htobe64(
132 subbuffer
->info
.data
.timestamp_begin
),
133 .timestamp_end
= htobe64(
134 subbuffer
->info
.data
.timestamp_end
),
135 .events_discarded
= htobe64(
136 subbuffer
->info
.data
.events_discarded
),
137 .stream_id
= htobe64(subbuffer
->info
.data
.stream_id
),
138 .stream_instance_id
= htobe64(
139 subbuffer
->info
.data
.stream_instance_id
.is_set
?
140 subbuffer
->info
.data
.stream_instance_id
.value
: -1ULL),
141 .packet_seq_num
= htobe64(
142 subbuffer
->info
.data
.sequence_number
.is_set
?
143 subbuffer
->info
.data
.sequence_number
.value
: -1ULL),
147 static ssize_t
consumer_stream_consume_mmap(
148 struct lttng_consumer_local_data
*ctx
,
149 struct lttng_consumer_stream
*stream
,
150 const struct stream_subbuffer
*subbuffer
)
152 const unsigned long padding_size
=
153 subbuffer
->info
.data
.padded_subbuf_size
-
154 subbuffer
->info
.data
.subbuf_size
;
156 return lttng_consumer_on_read_subbuffer_mmap(
157 stream
, &subbuffer
->buffer
.buffer
, padding_size
);
160 static ssize_t
consumer_stream_consume_splice(
161 struct lttng_consumer_local_data
*ctx
,
162 struct lttng_consumer_stream
*stream
,
163 const struct stream_subbuffer
*subbuffer
)
165 return lttng_consumer_on_read_subbuffer_splice(ctx
, stream
,
166 subbuffer
->info
.data
.padded_subbuf_size
, 0);
169 static int consumer_stream_send_index(
170 struct lttng_consumer_stream
*stream
,
171 const struct stream_subbuffer
*subbuffer
,
172 struct lttng_consumer_local_data
*ctx
)
174 off_t packet_offset
= 0;
175 struct ctf_packet_index index
= {};
178 * This is called after consuming the sub-buffer; substract the
179 * effect this sub-buffer from the offset.
181 if (stream
->net_seq_idx
== (uint64_t) -1ULL) {
182 packet_offset
= stream
->out_fd_offset
-
183 subbuffer
->info
.data
.padded_subbuf_size
;
186 ctf_packet_index_populate(&index
, packet_offset
, subbuffer
);
187 return consumer_stream_write_index(stream
, &index
);
191 * Actually do the metadata sync using the given metadata stream.
193 * Return 0 on success else a negative value. ENODATA can be returned also
194 * indicating that there is no metadata available for that stream.
196 static int do_sync_metadata(struct lttng_consumer_stream
*metadata
,
197 struct lttng_consumer_local_data
*ctx
)
200 enum sync_metadata_status status
;
203 assert(metadata
->metadata_flag
);
207 * In UST, since we have to write the metadata from the cache packet
208 * by packet, we might need to start this procedure multiple times
209 * until all the metadata from the cache has been extracted.
214 * - Lock the metadata stream
215 * - Check if metadata stream node was deleted before locking.
216 * - if yes, release and return success
217 * - Check if new metadata is ready (flush + snapshot pos)
218 * - If nothing : release and return.
219 * - Lock the metadata_rdv_lock
220 * - Unlock the metadata stream
221 * - cond_wait on metadata_rdv to wait the wakeup from the
223 * - Unlock the metadata_rdv_lock
225 pthread_mutex_lock(&metadata
->lock
);
228 * There is a possibility that we were able to acquire a reference on the
229 * stream from the RCU hash table but between then and now, the node might
230 * have been deleted just before the lock is acquired. Thus, after locking,
231 * we make sure the metadata node has not been deleted which means that the
232 * buffers are closed.
234 * In that case, there is no need to sync the metadata hence returning a
235 * success return code.
237 ret
= cds_lfht_is_node_deleted(&metadata
->node
.node
);
240 goto end_unlock_mutex
;
244 case LTTNG_CONSUMER_KERNEL
:
246 * Empty the metadata cache and flush the current stream.
248 status
= lttng_kconsumer_sync_metadata(metadata
);
250 case LTTNG_CONSUMER32_UST
:
251 case LTTNG_CONSUMER64_UST
:
253 * Ask the sessiond if we have new metadata waiting and update the
254 * consumer metadata cache.
256 status
= lttng_ustconsumer_sync_metadata(ctx
, metadata
);
263 case SYNC_METADATA_STATUS_NEW_DATA
:
265 case SYNC_METADATA_STATUS_NO_DATA
:
267 goto end_unlock_mutex
;
268 case SYNC_METADATA_STATUS_ERROR
:
270 goto end_unlock_mutex
;
276 * At this point, new metadata have been flushed, so we wait on the
277 * rendez-vous point for the metadata thread to wake us up when it
278 * finishes consuming the metadata and continue execution.
281 pthread_mutex_lock(&metadata
->metadata_rdv_lock
);
284 * Release metadata stream lock so the metadata thread can process it.
286 pthread_mutex_unlock(&metadata
->lock
);
289 * Wait on the rendez-vous point. Once woken up, it means the metadata was
290 * consumed and thus synchronization is achieved.
292 pthread_cond_wait(&metadata
->metadata_rdv
, &metadata
->metadata_rdv_lock
);
293 pthread_mutex_unlock(&metadata
->metadata_rdv_lock
);
294 } while (status
== SYNC_METADATA_STATUS_NEW_DATA
);
300 pthread_mutex_unlock(&metadata
->lock
);
305 * Synchronize the metadata using a given session ID. A successful acquisition
306 * of a metadata stream will trigger a request to the session daemon and a
307 * snapshot so the metadata thread can consume it.
309 * This function call is a rendez-vous point between the metadata thread and
312 * Return 0 on success or else a negative value.
314 int consumer_stream_sync_metadata(struct lttng_consumer_local_data
*ctx
,
318 struct lttng_consumer_stream
*stream
= NULL
;
319 struct lttng_ht_iter iter
;
324 /* Ease our life a bit. */
325 ht
= consumer_data
.stream_list_ht
;
329 /* Search the metadata associated with the session id of the given stream. */
331 cds_lfht_for_each_entry_duplicate(ht
->ht
,
332 ht
->hash_fct(&session_id
, lttng_ht_seed
), ht
->match_fct
,
333 &session_id
, &iter
.iter
, stream
, node_session_id
.node
) {
334 if (!stream
->metadata_flag
) {
338 ret
= do_sync_metadata(stream
, ctx
);
345 * Force return code to 0 (success) since ret might be ENODATA for instance
346 * which is not an error but rather that we should come back.
355 static int consumer_stream_sync_metadata_index(
356 struct lttng_consumer_stream
*stream
,
357 const struct stream_subbuffer
*subbuffer
,
358 struct lttng_consumer_local_data
*ctx
)
362 /* Block until all the metadata is sent. */
363 pthread_mutex_lock(&stream
->metadata_timer_lock
);
364 assert(!stream
->missed_metadata_flush
);
365 stream
->waiting_on_metadata
= true;
366 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
368 ret
= consumer_stream_sync_metadata(ctx
, stream
->session_id
);
370 pthread_mutex_lock(&stream
->metadata_timer_lock
);
371 stream
->waiting_on_metadata
= false;
372 if (stream
->missed_metadata_flush
) {
373 stream
->missed_metadata_flush
= false;
374 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
375 (void) stream
->read_subbuffer_ops
.send_live_beacon(stream
);
377 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
383 ret
= consumer_stream_send_index(stream
, subbuffer
, ctx
);
389 * Check if the local version of the metadata stream matches with the version
390 * of the metadata stream in the kernel. If it was updated, set the reset flag
394 int metadata_stream_check_version(struct lttng_consumer_stream
*stream
,
395 const struct stream_subbuffer
*subbuffer
)
397 if (stream
->metadata_version
== subbuffer
->info
.metadata
.version
) {
401 DBG("New metadata version detected");
402 consumer_stream_metadata_set_version(stream
,
403 subbuffer
->info
.metadata
.version
);
405 if (stream
->read_subbuffer_ops
.reset_metadata
) {
406 stream
->read_subbuffer_ops
.reset_metadata(stream
);
414 bool stream_is_rotating_to_null_chunk(
415 const struct lttng_consumer_stream
*stream
)
417 bool rotating_to_null_chunk
= false;
419 if (stream
->rotate_position
== -1ULL) {
420 /* No rotation ongoing. */
424 if (stream
->trace_chunk
== stream
->chan
->trace_chunk
||
425 !stream
->chan
->trace_chunk
) {
426 rotating_to_null_chunk
= true;
429 return rotating_to_null_chunk
;
432 enum consumer_stream_open_packet_status
consumer_stream_open_packet(
433 struct lttng_consumer_stream
*stream
)
436 enum consumer_stream_open_packet_status status
;
437 unsigned long produced_pos_before
, produced_pos_after
;
439 ret
= lttng_consumer_sample_snapshot_positions(stream
);
441 ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
442 ", channel name = %s, session id = %" PRIu64
,
443 stream
->key
, stream
->chan
->name
,
444 stream
->chan
->session_id
);
445 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
449 ret
= lttng_consumer_get_produced_snapshot(
450 stream
, &produced_pos_before
);
452 ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
453 ", channel name = %s, session id = %" PRIu64
,
454 stream
->key
, stream
->chan
->name
,
455 stream
->chan
->session_id
);
456 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
460 ret
= consumer_stream_flush_buffer(stream
, 0);
462 ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
463 ", channel name = %s, session id = %" PRIu64
,
464 stream
->key
, stream
->chan
->name
,
465 stream
->chan
->session_id
);
466 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
470 ret
= lttng_consumer_sample_snapshot_positions(stream
);
472 ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
473 ", channel name = %s, session id = %" PRIu64
,
474 stream
->key
, stream
->chan
->name
,
475 stream
->chan
->session_id
);
476 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
480 ret
= lttng_consumer_get_produced_snapshot(stream
, &produced_pos_after
);
482 ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
483 ", channel name = %s, session id = %" PRIu64
,
484 stream
->key
, stream
->chan
->name
,
485 stream
->chan
->session_id
);
486 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
491 * Determine if the flush had an effect by comparing the produced
492 * positons before and after the flush.
494 status
= produced_pos_before
!= produced_pos_after
?
495 CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
:
496 CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE
;
497 if (status
== CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
) {
498 stream
->opened_packet_in_current_trace_chunk
= true;
506 * An attempt to open a new packet is performed after a rotation completes to
507 * get a begin timestamp as close as possible to the rotation point.
509 * However, that initial attempt at opening a packet can fail due to a full
510 * ring-buffer. In that case, a second attempt is performed after consuming
511 * a packet since that will have freed enough space in the ring-buffer.
514 int post_consume_open_new_packet(struct lttng_consumer_stream
*stream
,
515 const struct stream_subbuffer
*subbuffer
,
516 struct lttng_consumer_local_data
*ctx
)
520 if (!stream
->opened_packet_in_current_trace_chunk
&&
521 stream
->trace_chunk
&&
522 !stream_is_rotating_to_null_chunk(stream
)) {
523 const enum consumer_stream_open_packet_status status
=
524 consumer_stream_open_packet(stream
);
527 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
:
528 DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64
529 ", channel name = %s, session id = %" PRIu64
,
530 stream
->key
, stream
->chan
->name
,
531 stream
->chan
->session_id
);
532 stream
->opened_packet_in_current_trace_chunk
= true;
534 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE
:
536 * Can't open a packet as there is no space left.
537 * This means that new events were produced, resulting
538 * in a packet being opened, which is what we want
541 DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
542 ", channel name = %s, session id = %" PRIu64
,
543 stream
->key
, stream
->chan
->name
,
544 stream
->chan
->session_id
);
545 stream
->opened_packet_in_current_trace_chunk
= true;
547 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
:
548 /* Logged by callee. */
555 stream
->opened_packet_in_current_trace_chunk
= true;
562 struct lttng_consumer_stream
*consumer_stream_create(
563 struct lttng_consumer_channel
*channel
,
564 uint64_t channel_key
,
566 const char *channel_name
,
569 struct lttng_trace_chunk
*trace_chunk
,
572 enum consumer_channel_type type
,
573 unsigned int monitor
)
576 struct lttng_consumer_stream
*stream
;
578 stream
= zmalloc(sizeof(*stream
));
579 if (stream
== NULL
) {
580 PERROR("malloc struct lttng_consumer_stream");
585 if (trace_chunk
&& !lttng_trace_chunk_get(trace_chunk
)) {
586 ERR("Failed to acquire trace chunk reference during the creation of a stream");
592 stream
->chan
= channel
;
593 stream
->key
= stream_key
;
594 stream
->trace_chunk
= trace_chunk
;
596 stream
->out_fd_offset
= 0;
597 stream
->output_written
= 0;
598 stream
->net_seq_idx
= relayd_id
;
599 stream
->session_id
= session_id
;
600 stream
->monitor
= monitor
;
601 stream
->endpoint_status
= CONSUMER_ENDPOINT_ACTIVE
;
602 stream
->index_file
= NULL
;
603 stream
->last_sequence_number
= -1ULL;
604 stream
->rotate_position
= -1ULL;
605 /* Buffer is created with an open packet. */
606 stream
->opened_packet_in_current_trace_chunk
= true;
607 pthread_mutex_init(&stream
->lock
, NULL
);
608 pthread_mutex_init(&stream
->metadata_timer_lock
, NULL
);
610 /* If channel is the metadata, flag this stream as metadata. */
611 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
612 stream
->metadata_flag
= 1;
613 /* Metadata is flat out. */
614 strncpy(stream
->name
, DEFAULT_METADATA_NAME
, sizeof(stream
->name
));
615 /* Live rendez-vous point. */
616 pthread_cond_init(&stream
->metadata_rdv
, NULL
);
617 pthread_mutex_init(&stream
->metadata_rdv_lock
, NULL
);
619 /* Format stream name to <channel_name>_<cpu_number> */
620 ret
= snprintf(stream
->name
, sizeof(stream
->name
), "%s_%d",
623 PERROR("snprintf stream name");
628 switch (channel
->output
) {
629 case CONSUMER_CHANNEL_SPLICE
:
630 stream
->output
= LTTNG_EVENT_SPLICE
;
631 ret
= utils_create_pipe(stream
->splice_pipe
);
636 case CONSUMER_CHANNEL_MMAP
:
637 stream
->output
= LTTNG_EVENT_MMAP
;
643 /* Key is always the wait_fd for streams. */
644 lttng_ht_node_init_u64(&stream
->node
, stream
->key
);
646 /* Init node per channel id key */
647 lttng_ht_node_init_u64(&stream
->node_channel_id
, channel_key
);
649 /* Init session id node with the stream session id */
650 lttng_ht_node_init_u64(&stream
->node_session_id
, stream
->session_id
);
652 DBG3("Allocated stream %s (key %" PRIu64
", chan_key %" PRIu64
653 " relayd_id %" PRIu64
", session_id %" PRIu64
,
654 stream
->name
, stream
->key
, channel_key
,
655 stream
->net_seq_idx
, stream
->session_id
);
659 lttng_dynamic_array_init(&stream
->read_subbuffer_ops
.post_consume_cbs
,
660 sizeof(post_consume_cb
), NULL
);
662 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
663 stream
->read_subbuffer_ops
.lock
=
664 consumer_stream_metadata_lock_all
;
665 stream
->read_subbuffer_ops
.unlock
=
666 consumer_stream_metadata_unlock_all
;
667 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
668 metadata_stream_check_version
;
670 const post_consume_cb post_consume_index_op
= channel
->is_live
?
671 consumer_stream_sync_metadata_index
:
672 consumer_stream_send_index
;
674 ret
= lttng_dynamic_array_add_element(
675 &stream
->read_subbuffer_ops
.post_consume_cbs
,
676 &post_consume_index_op
);
678 PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
682 ret
= lttng_dynamic_array_add_element(
683 &stream
->read_subbuffer_ops
.post_consume_cbs
,
684 &(post_consume_cb
) { post_consume_open_new_packet
});
686 PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
690 stream
->read_subbuffer_ops
.lock
= consumer_stream_data_lock_all
;
691 stream
->read_subbuffer_ops
.unlock
=
692 consumer_stream_data_unlock_all
;
693 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
694 consumer_stream_update_stats
;
697 if (channel
->output
== CONSUMER_CHANNEL_MMAP
) {
698 stream
->read_subbuffer_ops
.consume_subbuffer
=
699 consumer_stream_consume_mmap
;
701 stream
->read_subbuffer_ops
.consume_subbuffer
=
702 consumer_stream_consume_splice
;
709 lttng_trace_chunk_put(stream
->trace_chunk
);
710 lttng_dynamic_array_reset(&stream
->read_subbuffer_ops
.post_consume_cbs
);
720 * Close stream on the relayd side. This call can destroy a relayd if the
721 * conditions are met.
723 * A RCU read side lock MUST be acquired if the relayd object was looked up in
724 * a hash table before calling this.
726 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
727 struct consumer_relayd_sock_pair
*relayd
)
734 if (stream
->sent_to_relayd
) {
735 uatomic_dec(&relayd
->refcount
);
736 assert(uatomic_read(&relayd
->refcount
) >= 0);
739 /* Closing streams requires to lock the control socket. */
740 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
741 ret
= relayd_send_close_stream(&relayd
->control_sock
,
742 stream
->relayd_stream_id
,
743 stream
->next_net_seq_num
- 1);
744 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
746 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
747 lttng_consumer_cleanup_relayd(relayd
);
750 /* Both conditions are met, we destroy the relayd. */
751 if (uatomic_read(&relayd
->refcount
) == 0 &&
752 uatomic_read(&relayd
->destroy_flag
)) {
753 consumer_destroy_relayd(relayd
);
755 stream
->net_seq_idx
= (uint64_t) -1ULL;
756 stream
->sent_to_relayd
= 0;
760 * Close stream's file descriptors and, if needed, close stream also on the
763 * The consumer data lock MUST be acquired.
764 * The stream lock MUST be acquired.
766 void consumer_stream_close(struct lttng_consumer_stream
*stream
)
769 struct consumer_relayd_sock_pair
*relayd
;
773 switch (consumer_data
.type
) {
774 case LTTNG_CONSUMER_KERNEL
:
775 if (stream
->mmap_base
!= NULL
) {
776 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
782 if (stream
->wait_fd
>= 0) {
783 ret
= close(stream
->wait_fd
);
787 stream
->wait_fd
= -1;
789 if (stream
->chan
->output
== CONSUMER_CHANNEL_SPLICE
) {
790 utils_close_pipe(stream
->splice_pipe
);
793 case LTTNG_CONSUMER32_UST
:
794 case LTTNG_CONSUMER64_UST
:
797 * Special case for the metadata since the wait fd is an internal pipe
798 * polled in the metadata thread.
800 if (stream
->metadata_flag
&& stream
->chan
->monitor
) {
801 int rpipe
= stream
->ust_metadata_poll_pipe
[0];
804 * This will stop the channel timer if one and close the write side
805 * of the metadata poll pipe.
807 lttng_ustconsumer_close_metadata(stream
->chan
);
811 PERROR("closing metadata pipe read side");
813 stream
->ust_metadata_poll_pipe
[0] = -1;
819 ERR("Unknown consumer_data type");
823 /* Close output fd. Could be a socket or local file at this point. */
824 if (stream
->out_fd
>= 0) {
825 ret
= close(stream
->out_fd
);
832 if (stream
->index_file
) {
833 lttng_index_file_put(stream
->index_file
);
834 stream
->index_file
= NULL
;
837 lttng_trace_chunk_put(stream
->trace_chunk
);
838 stream
->trace_chunk
= NULL
;
840 /* Check and cleanup relayd if needed. */
842 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
843 if (relayd
!= NULL
) {
844 consumer_stream_relayd_close(stream
, relayd
);
850 * Delete the stream from all possible hash tables.
852 * The consumer data lock MUST be acquired.
853 * The stream lock MUST be acquired.
855 void consumer_stream_delete(struct lttng_consumer_stream
*stream
,
859 struct lttng_ht_iter iter
;
862 /* Should NEVER be called not in monitor mode. */
863 assert(stream
->chan
->monitor
);
868 iter
.iter
.node
= &stream
->node
.node
;
869 ret
= lttng_ht_del(ht
, &iter
);
873 /* Delete from stream per channel ID hash table. */
874 iter
.iter
.node
= &stream
->node_channel_id
.node
;
876 * The returned value is of no importance. Even if the node is NOT in the
877 * hash table, we continue since we may have been called by a code path
878 * that did not add the stream to a (all) hash table. Same goes for the
879 * next call ht del call.
881 (void) lttng_ht_del(consumer_data
.stream_per_chan_id_ht
, &iter
);
883 /* Delete from the global stream list. */
884 iter
.iter
.node
= &stream
->node_session_id
.node
;
885 /* See the previous ht del on why we ignore the returned value. */
886 (void) lttng_ht_del(consumer_data
.stream_list_ht
, &iter
);
890 if (!stream
->metadata_flag
) {
891 /* Decrement the stream count of the global consumer data. */
892 assert(consumer_data
.stream_count
> 0);
893 consumer_data
.stream_count
--;
898 * Free the given stream within a RCU call.
900 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
904 metadata_bucket_destroy(stream
->metadata_bucket
);
905 call_rcu(&stream
->node
.head
, free_stream_rcu
);
909 * Destroy the stream's buffers of the tracer.
911 void consumer_stream_destroy_buffers(struct lttng_consumer_stream
*stream
)
915 switch (consumer_data
.type
) {
916 case LTTNG_CONSUMER_KERNEL
:
918 case LTTNG_CONSUMER32_UST
:
919 case LTTNG_CONSUMER64_UST
:
920 lttng_ustconsumer_del_stream(stream
);
923 ERR("Unknown consumer_data type");
929 * Destroy and close a already created stream.
931 static void destroy_close_stream(struct lttng_consumer_stream
*stream
)
935 DBG("Consumer stream destroy monitored key: %" PRIu64
, stream
->key
);
937 /* Destroy tracer buffers of the stream. */
938 consumer_stream_destroy_buffers(stream
);
939 /* Close down everything including the relayd if one. */
940 consumer_stream_close(stream
);
944 * Decrement the stream's channel refcount and if down to 0, return the channel
945 * pointer so it can be destroyed by the caller or NULL if not.
947 static struct lttng_consumer_channel
*unref_channel(
948 struct lttng_consumer_stream
*stream
)
950 struct lttng_consumer_channel
*free_chan
= NULL
;
953 assert(stream
->chan
);
955 /* Update refcount of channel and see if we need to destroy it. */
956 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1)
957 && !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
958 free_chan
= stream
->chan
;
965 * Destroy a stream completely. This will delete, close and free the stream.
966 * Once return, the stream is NO longer usable. Its channel may get destroyed
967 * if conditions are met for a monitored stream.
969 * This MUST be called WITHOUT the consumer data and stream lock acquired if
970 * the stream is in _monitor_ mode else it does not matter.
972 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
,
977 /* Stream is in monitor mode. */
978 if (stream
->monitor
) {
979 struct lttng_consumer_channel
*free_chan
= NULL
;
982 * This means that the stream was successfully removed from the streams
983 * list of the channel and sent to the right thread managing this
984 * stream thus being globally visible.
986 if (stream
->globally_visible
) {
987 pthread_mutex_lock(&consumer_data
.lock
);
988 pthread_mutex_lock(&stream
->chan
->lock
);
989 pthread_mutex_lock(&stream
->lock
);
990 /* Remove every reference of the stream in the consumer. */
991 consumer_stream_delete(stream
, ht
);
993 destroy_close_stream(stream
);
995 /* Update channel's refcount of the stream. */
996 free_chan
= unref_channel(stream
);
998 /* Indicates that the consumer data state MUST be updated after this. */
999 consumer_data
.need_update
= 1;
1001 pthread_mutex_unlock(&stream
->lock
);
1002 pthread_mutex_unlock(&stream
->chan
->lock
);
1003 pthread_mutex_unlock(&consumer_data
.lock
);
1006 * If the stream is not visible globally, this needs to be done
1007 * outside of the consumer data lock section.
1009 free_chan
= unref_channel(stream
);
1013 consumer_del_channel(free_chan
);
1016 destroy_close_stream(stream
);
1019 /* Free stream within a RCU call. */
1020 lttng_trace_chunk_put(stream
->trace_chunk
);
1021 stream
->trace_chunk
= NULL
;
1022 lttng_dynamic_array_reset(&stream
->read_subbuffer_ops
.post_consume_cbs
);
1023 consumer_stream_free(stream
);
1027 * Write index of a specific stream either on the relayd or local disk.
1029 * Return 0 on success or else a negative value.
1031 int consumer_stream_write_index(struct lttng_consumer_stream
*stream
,
1032 struct ctf_packet_index
*element
)
1040 if (stream
->net_seq_idx
!= (uint64_t) -1ULL) {
1041 struct consumer_relayd_sock_pair
*relayd
;
1042 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
1044 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
1045 ret
= relayd_send_index(&relayd
->control_sock
, element
,
1046 stream
->relayd_stream_id
, stream
->next_net_seq_num
- 1);
1049 * Communication error with lttng-relayd,
1050 * perform cleanup now
1052 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
1053 lttng_consumer_cleanup_relayd(relayd
);
1056 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
1058 ERR("Stream %" PRIu64
" relayd ID %" PRIu64
" unknown. Can't write index.",
1059 stream
->key
, stream
->net_seq_idx
);
1063 if (lttng_index_file_write(stream
->index_file
, element
)) {
1078 int consumer_stream_create_output_files(struct lttng_consumer_stream
*stream
,
1082 enum lttng_trace_chunk_status chunk_status
;
1083 const int flags
= O_WRONLY
| O_CREAT
| O_TRUNC
;
1084 const mode_t mode
= S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
;
1085 char stream_path
[LTTNG_PATH_MAX
];
1087 ASSERT_LOCKED(stream
->lock
);
1088 assert(stream
->trace_chunk
);
1090 ret
= utils_stream_file_path(stream
->chan
->pathname
, stream
->name
,
1091 stream
->chan
->tracefile_size
,
1092 stream
->tracefile_count_current
, NULL
,
1093 stream_path
, sizeof(stream_path
));
1098 if (stream
->out_fd
>= 0) {
1099 ret
= close(stream
->out_fd
);
1101 PERROR("Failed to close stream file \"%s\"",
1105 stream
->out_fd
= -1;
1108 DBG("Opening stream output file \"%s\"", stream_path
);
1109 chunk_status
= lttng_trace_chunk_open_file(stream
->trace_chunk
, stream_path
,
1110 flags
, mode
, &stream
->out_fd
, false);
1111 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1112 ERR("Failed to open stream file \"%s\"", stream
->name
);
1117 if (!stream
->metadata_flag
&& (create_index
|| stream
->index_file
)) {
1118 if (stream
->index_file
) {
1119 lttng_index_file_put(stream
->index_file
);
1121 chunk_status
= lttng_index_file_create_from_trace_chunk(
1122 stream
->trace_chunk
,
1123 stream
->chan
->pathname
,
1125 stream
->chan
->tracefile_size
,
1126 stream
->tracefile_count_current
,
1127 CTF_INDEX_MAJOR
, CTF_INDEX_MINOR
,
1128 false, &stream
->index_file
);
1129 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1135 /* Reset current size because we just perform a rotation. */
1136 stream
->tracefile_size_current
= 0;
1137 stream
->out_fd_offset
= 0;
1142 int consumer_stream_rotate_output_files(struct lttng_consumer_stream
*stream
)
1146 stream
->tracefile_count_current
++;
1147 if (stream
->chan
->tracefile_count
> 0) {
1148 stream
->tracefile_count_current
%=
1149 stream
->chan
->tracefile_count
;
1152 DBG("Rotating output files of stream \"%s\"", stream
->name
);
1153 ret
= consumer_stream_create_output_files(stream
, true);
1162 bool consumer_stream_is_deleted(struct lttng_consumer_stream
*stream
)
1165 * This function does not take a const stream since
1166 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1169 return cds_lfht_is_node_deleted(&stream
->node
.node
);
1172 static ssize_t
metadata_bucket_flush(
1173 const struct stream_subbuffer
*buffer
, void *data
)
1176 struct lttng_consumer_stream
*stream
= data
;
1178 ret
= consumer_stream_consume_mmap(NULL
, stream
, buffer
);
1186 static ssize_t
metadata_bucket_consume(
1187 struct lttng_consumer_local_data
*unused
,
1188 struct lttng_consumer_stream
*stream
,
1189 const struct stream_subbuffer
*subbuffer
)
1192 enum metadata_bucket_status status
;
1194 status
= metadata_bucket_fill(stream
->metadata_bucket
, subbuffer
);
1196 case METADATA_BUCKET_STATUS_OK
:
1197 /* Return consumed size. */
1198 ret
= subbuffer
->buffer
.buffer
.size
;
1207 int consumer_stream_enable_metadata_bucketization(
1208 struct lttng_consumer_stream
*stream
)
1212 assert(stream
->metadata_flag
);
1213 assert(!stream
->metadata_bucket
);
1214 assert(stream
->chan
->output
== CONSUMER_CHANNEL_MMAP
);
1216 stream
->metadata_bucket
= metadata_bucket_create(
1217 metadata_bucket_flush
, stream
);
1218 if (!stream
->metadata_bucket
) {
1223 stream
->read_subbuffer_ops
.consume_subbuffer
= metadata_bucket_consume
;
1228 void consumer_stream_metadata_set_version(
1229 struct lttng_consumer_stream
*stream
, uint64_t new_version
)
1231 assert(new_version
> stream
->metadata_version
);
1232 stream
->metadata_version
= new_version
;
1233 stream
->reset_metadata_flag
= 1;
1235 if (stream
->metadata_bucket
) {
1236 metadata_bucket_reset(stream
->metadata_bucket
);
1240 int consumer_stream_flush_buffer(struct lttng_consumer_stream
*stream
,
1241 bool producer_active
)
1245 switch (consumer_data
.type
) {
1246 case LTTNG_CONSUMER_KERNEL
:
1247 if (producer_active
) {
1248 ret
= kernctl_buffer_flush(stream
->wait_fd
);
1250 ERR("Failed to flush kernel stream");
1254 ret
= kernctl_buffer_flush_empty(stream
->wait_fd
);
1257 * Doing a buffer flush which does not take into
1258 * account empty packets. This is not perfect,
1259 * but required as a fall-back when
1260 * "flush_empty" is not implemented by
1263 ret
= kernctl_buffer_flush(stream
->wait_fd
);
1265 ERR("Failed to flush kernel stream");
1271 case LTTNG_CONSUMER32_UST
:
1272 case LTTNG_CONSUMER64_UST
:
1273 lttng_ustconsumer_flush_buffer(stream
, (int) producer_active
);
1276 ERR("Unknown consumer_data type");