2 * Copyright (C) 2011 EfficiOS Inc.
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
15 #include <common/buffer-view.hpp>
16 #include <common/common.hpp>
17 #include <common/consumer/consumer-timer.hpp>
18 #include <common/consumer/consumer.hpp>
19 #include <common/consumer/metadata-bucket.hpp>
20 #include <common/index/index.hpp>
21 #include <common/kernel-consumer/kernel-consumer.hpp>
22 #include <common/kernel-ctl/kernel-ctl.hpp>
23 #include <common/macros.hpp>
24 #include <common/relayd/relayd.hpp>
25 #include <common/ust-consumer/ust-consumer.hpp>
26 #include <common/utils.hpp>
28 #include "consumer-stream.hpp"
30 struct metadata_packet_header
{
31 uint32_t magic
; /* 0x75D11D57 */
32 uint8_t uuid
[16]; /* Unique Universal Identifier */
33 uint32_t checksum
; /* 0 if unused */
34 uint32_t content_size
; /* in bits */
35 uint32_t packet_size
; /* in bits */
36 uint8_t compression_scheme
; /* 0 if unused */
37 uint8_t encryption_scheme
; /* 0 if unused */
38 uint8_t checksum_scheme
; /* 0 if unused */
39 uint8_t major
; /* CTF spec major version number */
40 uint8_t minor
; /* CTF spec minor version number */
41 uint8_t header_end
[0];
44 static size_t metadata_length(void)
46 return offsetof(struct metadata_packet_header
, header_end
);
50 * RCU call to free stream. MUST only be used with call_rcu().
52 static void free_stream_rcu(struct rcu_head
*head
)
54 struct lttng_ht_node_u64
*node
=
55 lttng::utils::container_of(head
, <tng_ht_node_u64::head
);
56 struct lttng_consumer_stream
*stream
=
57 lttng::utils::container_of(node
, <tng_consumer_stream::node
);
59 pthread_mutex_destroy(&stream
->lock
);
63 static void consumer_stream_data_lock_all(struct lttng_consumer_stream
*stream
)
65 pthread_mutex_lock(&stream
->chan
->lock
);
66 pthread_mutex_lock(&stream
->lock
);
69 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream
*stream
)
71 pthread_mutex_unlock(&stream
->lock
);
72 pthread_mutex_unlock(&stream
->chan
->lock
);
75 static void consumer_stream_data_assert_locked_all(struct lttng_consumer_stream
*stream
)
77 ASSERT_LOCKED(stream
->lock
);
78 ASSERT_LOCKED(stream
->chan
->lock
);
81 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream
*stream
)
83 consumer_stream_data_lock_all(stream
);
84 pthread_mutex_lock(&stream
->metadata_rdv_lock
);
87 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream
*stream
)
89 pthread_mutex_unlock(&stream
->metadata_rdv_lock
);
90 consumer_stream_data_unlock_all(stream
);
93 static void consumer_stream_metadata_assert_locked_all(struct lttng_consumer_stream
*stream
)
95 ASSERT_LOCKED(stream
->metadata_rdv_lock
);
96 consumer_stream_data_assert_locked_all(stream
);
99 /* Only used for data streams. */
100 static int consumer_stream_update_stats(
101 struct lttng_consumer_stream
*stream
, struct stream_subbuffer
*subbuf_
)
104 uint64_t sequence_number
;
105 const stream_subbuffer
*subbuf
= subbuf_
;
106 const uint64_t discarded_events
= subbuf
->info
.data
.events_discarded
;
108 if (!subbuf
->info
.data
.sequence_number
.is_set
) {
109 /* Command not supported by the tracer. */
110 sequence_number
= -1ULL;
111 stream
->sequence_number_unavailable
= true;
113 sequence_number
= subbuf
->info
.data
.sequence_number
.value
;
117 * Start the sequence when we extract the first packet in case we don't
118 * start at 0 (for example if a consumer is not connected to the
119 * session immediately after the beginning).
121 if (stream
->last_sequence_number
== -1ULL) {
122 stream
->last_sequence_number
= sequence_number
;
123 } else if (sequence_number
> stream
->last_sequence_number
) {
124 stream
->chan
->lost_packets
+= sequence_number
-
125 stream
->last_sequence_number
- 1;
127 /* seq <= last_sequence_number */
128 ERR("Sequence number inconsistent : prev = %" PRIu64
129 ", current = %" PRIu64
,
130 stream
->last_sequence_number
, sequence_number
);
134 stream
->last_sequence_number
= sequence_number
;
136 if (discarded_events
< stream
->last_discarded_events
) {
138 * Overflow has occurred. We assume only one wrap-around
141 stream
->chan
->discarded_events
+=
142 (1ULL << (CAA_BITS_PER_LONG
- 1)) -
143 stream
->last_discarded_events
+
146 stream
->chan
->discarded_events
+= discarded_events
-
147 stream
->last_discarded_events
;
149 stream
->last_discarded_events
= discarded_events
;
157 void ctf_packet_index_populate(struct ctf_packet_index
*index
,
158 off_t offset
, const struct stream_subbuffer
*subbuffer
)
160 *index
= (typeof(*index
)){
161 .offset
= htobe64(offset
),
162 .packet_size
= htobe64(subbuffer
->info
.data
.packet_size
),
163 .content_size
= htobe64(subbuffer
->info
.data
.content_size
),
164 .timestamp_begin
= htobe64(
165 subbuffer
->info
.data
.timestamp_begin
),
166 .timestamp_end
= htobe64(
167 subbuffer
->info
.data
.timestamp_end
),
168 .events_discarded
= htobe64(
169 subbuffer
->info
.data
.events_discarded
),
170 .stream_id
= htobe64(subbuffer
->info
.data
.stream_id
),
171 .stream_instance_id
= htobe64(
172 subbuffer
->info
.data
.stream_instance_id
.is_set
?
173 subbuffer
->info
.data
.stream_instance_id
.value
: -1ULL),
174 .packet_seq_num
= htobe64(
175 subbuffer
->info
.data
.sequence_number
.is_set
?
176 subbuffer
->info
.data
.sequence_number
.value
: -1ULL),
180 static ssize_t
consumer_stream_consume_mmap(
181 struct lttng_consumer_local_data
*ctx
__attribute__((unused
)),
182 struct lttng_consumer_stream
*stream
,
183 const struct stream_subbuffer
*subbuffer
)
185 const unsigned long padding_size
=
186 subbuffer
->info
.data
.padded_subbuf_size
-
187 subbuffer
->info
.data
.subbuf_size
;
188 const ssize_t written_bytes
= lttng_consumer_on_read_subbuffer_mmap(
189 stream
, &subbuffer
->buffer
.buffer
, padding_size
);
191 if (stream
->net_seq_idx
== -1ULL) {
193 * When writing on disk, check that only the subbuffer (no
194 * padding) was written to disk.
196 if (written_bytes
!= subbuffer
->info
.data
.padded_subbuf_size
) {
197 DBG("Failed to write the entire padded subbuffer on disk (written_bytes: %zd, padded subbuffer size %lu)",
199 subbuffer
->info
.data
.padded_subbuf_size
);
203 * When streaming over the network, check that the entire
204 * subbuffer including padding was successfully written.
206 if (written_bytes
!= subbuffer
->info
.data
.subbuf_size
) {
207 DBG("Failed to write only the subbuffer over the network (written_bytes: %zd, subbuffer size %lu)",
209 subbuffer
->info
.data
.subbuf_size
);
214 * If `lttng_consumer_on_read_subbuffer_mmap()` returned an error, pass
215 * it along to the caller, else return zero.
217 if (written_bytes
< 0) {
218 ERR("Error reading mmap subbuffer: %zd", written_bytes
);
221 return written_bytes
;
224 static ssize_t
consumer_stream_consume_splice(
225 struct lttng_consumer_local_data
*ctx
,
226 struct lttng_consumer_stream
*stream
,
227 const struct stream_subbuffer
*subbuffer
)
229 const ssize_t written_bytes
= lttng_consumer_on_read_subbuffer_splice(
230 ctx
, stream
, subbuffer
->info
.data
.padded_subbuf_size
, 0);
232 if (written_bytes
!= subbuffer
->info
.data
.padded_subbuf_size
) {
233 DBG("Failed to write the entire padded subbuffer (written_bytes: %zd, padded subbuffer size %lu)",
235 subbuffer
->info
.data
.padded_subbuf_size
);
239 * If `lttng_consumer_on_read_subbuffer_splice()` returned an error,
240 * pass it along to the caller, else return zero.
242 if (written_bytes
< 0) {
243 ERR("Error reading splice subbuffer: %zd", written_bytes
);
246 return written_bytes
;
249 static int consumer_stream_send_index(
250 struct lttng_consumer_stream
*stream
,
251 const struct stream_subbuffer
*subbuffer
,
252 struct lttng_consumer_local_data
*ctx
__attribute__((unused
)))
254 off_t packet_offset
= 0;
255 struct ctf_packet_index index
= {};
258 * This is called after consuming the sub-buffer; substract the
259 * effect this sub-buffer from the offset.
261 if (stream
->net_seq_idx
== (uint64_t) -1ULL) {
262 packet_offset
= stream
->out_fd_offset
-
263 subbuffer
->info
.data
.padded_subbuf_size
;
266 ctf_packet_index_populate(&index
, packet_offset
, subbuffer
);
267 return consumer_stream_write_index(stream
, &index
);
271 * Actually do the metadata sync using the given metadata stream.
273 * Return 0 on success else a negative value. ENODATA can be returned also
274 * indicating that there is no metadata available for that stream.
276 static int do_sync_metadata(struct lttng_consumer_stream
*metadata
,
277 struct lttng_consumer_local_data
*ctx
)
280 enum sync_metadata_status status
;
282 LTTNG_ASSERT(metadata
);
283 LTTNG_ASSERT(metadata
->metadata_flag
);
287 * In UST, since we have to write the metadata from the cache packet
288 * by packet, we might need to start this procedure multiple times
289 * until all the metadata from the cache has been extracted.
294 * - Lock the metadata stream
295 * - Check if metadata stream node was deleted before locking.
296 * - if yes, release and return success
297 * - Check if new metadata is ready (flush + snapshot pos)
298 * - If nothing : release and return.
299 * - Lock the metadata_rdv_lock
300 * - Unlock the metadata stream
301 * - cond_wait on metadata_rdv to wait the wakeup from the
303 * - Unlock the metadata_rdv_lock
305 pthread_mutex_lock(&metadata
->lock
);
308 * There is a possibility that we were able to acquire a reference on the
309 * stream from the RCU hash table but between then and now, the node might
310 * have been deleted just before the lock is acquired. Thus, after locking,
311 * we make sure the metadata node has not been deleted which means that the
312 * buffers are closed.
314 * In that case, there is no need to sync the metadata hence returning a
315 * success return code.
317 ret
= cds_lfht_is_node_deleted(&metadata
->node
.node
);
320 goto end_unlock_mutex
;
324 case LTTNG_CONSUMER_KERNEL
:
326 * Empty the metadata cache and flush the current stream.
328 status
= lttng_kconsumer_sync_metadata(metadata
);
330 case LTTNG_CONSUMER32_UST
:
331 case LTTNG_CONSUMER64_UST
:
333 * Ask the sessiond if we have new metadata waiting and update the
334 * consumer metadata cache.
336 status
= lttng_ustconsumer_sync_metadata(ctx
, metadata
);
343 case SYNC_METADATA_STATUS_NEW_DATA
:
345 case SYNC_METADATA_STATUS_NO_DATA
:
347 goto end_unlock_mutex
;
348 case SYNC_METADATA_STATUS_ERROR
:
350 goto end_unlock_mutex
;
356 * At this point, new metadata have been flushed, so we wait on the
357 * rendez-vous point for the metadata thread to wake us up when it
358 * finishes consuming the metadata and continue execution.
361 pthread_mutex_lock(&metadata
->metadata_rdv_lock
);
364 * Release metadata stream lock so the metadata thread can process it.
366 pthread_mutex_unlock(&metadata
->lock
);
369 * Wait on the rendez-vous point. Once woken up, it means the metadata was
370 * consumed and thus synchronization is achieved.
372 pthread_cond_wait(&metadata
->metadata_rdv
, &metadata
->metadata_rdv_lock
);
373 pthread_mutex_unlock(&metadata
->metadata_rdv_lock
);
374 } while (status
== SYNC_METADATA_STATUS_NEW_DATA
);
380 pthread_mutex_unlock(&metadata
->lock
);
385 * Synchronize the metadata using a given session ID. A successful acquisition
386 * of a metadata stream will trigger a request to the session daemon and a
387 * snapshot so the metadata thread can consume it.
389 * This function call is a rendez-vous point between the metadata thread and
392 * Return 0 on success or else a negative value.
394 int consumer_stream_sync_metadata(struct lttng_consumer_local_data
*ctx
,
398 struct lttng_consumer_stream
*stream
= NULL
;
399 struct lttng_ht_iter iter
;
404 /* Ease our life a bit. */
405 ht
= the_consumer_data
.stream_list_ht
;
409 /* Search the metadata associated with the session id of the given stream. */
411 cds_lfht_for_each_entry_duplicate(ht
->ht
,
412 ht
->hash_fct(&session_id
, lttng_ht_seed
), ht
->match_fct
,
413 &session_id
, &iter
.iter
, stream
, node_session_id
.node
) {
414 if (!stream
->metadata_flag
) {
418 ret
= do_sync_metadata(stream
, ctx
);
425 * Force return code to 0 (success) since ret might be ENODATA for instance
426 * which is not an error but rather that we should come back.
435 static int consumer_stream_sync_metadata_index(
436 struct lttng_consumer_stream
*stream
,
437 const struct stream_subbuffer
*subbuffer
,
438 struct lttng_consumer_local_data
*ctx
)
440 bool missed_metadata_flush
;
443 /* Block until all the metadata is sent. */
444 pthread_mutex_lock(&stream
->metadata_timer_lock
);
445 LTTNG_ASSERT(!stream
->missed_metadata_flush
);
446 stream
->waiting_on_metadata
= true;
447 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
449 ret
= consumer_stream_sync_metadata(ctx
, stream
->session_id
);
451 pthread_mutex_lock(&stream
->metadata_timer_lock
);
452 stream
->waiting_on_metadata
= false;
453 missed_metadata_flush
= stream
->missed_metadata_flush
;
454 if (missed_metadata_flush
) {
455 stream
->missed_metadata_flush
= false;
457 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
462 ret
= consumer_stream_send_index(stream
, subbuffer
, ctx
);
464 * Send the live inactivity beacon to handle the situation where
465 * the live timer is prevented from sampling this stream
466 * because the stream lock was being held while this stream is
467 * waiting on metadata. This ensures live viewer progress in the
468 * unlikely scenario where a live timer would be prevented from
469 * locking a stream lock repeatedly due to a steady flow of
470 * incoming metadata, for a stream which is mostly inactive.
472 * It is important to send the inactivity beacon packet to
473 * relayd _after_ sending the index associated with the data
474 * that was just sent, otherwise this can cause live viewers to
475 * observe timestamps going backwards between an inactivity
476 * beacon and a following trace packet.
478 if (missed_metadata_flush
) {
479 (void) stream
->read_subbuffer_ops
.send_live_beacon(stream
);
486 * Check if the local version of the metadata stream matches with the version
487 * of the metadata stream in the kernel. If it was updated, set the reset flag
490 static void metadata_stream_check_version(
491 struct lttng_consumer_stream
*stream
, const struct stream_subbuffer
*subbuffer
)
493 if (stream
->metadata_version
== subbuffer
->info
.metadata
.version
) {
497 DBG("New metadata version detected");
498 consumer_stream_metadata_set_version(stream
,
499 subbuffer
->info
.metadata
.version
);
501 if (stream
->read_subbuffer_ops
.reset_metadata
) {
502 stream
->read_subbuffer_ops
.reset_metadata(stream
);
506 static void strip_packet_header_from_subbuffer(struct stream_subbuffer
*buffer
)
509 * Change the view and hide the packer header and padding from the view
511 size_t new_subbuf_size
= buffer
->info
.metadata
.subbuf_size
- metadata_length();
513 buffer
->buffer
.buffer
= lttng_buffer_view_from_view(
514 &buffer
->buffer
.buffer
, metadata_length(), new_subbuf_size
);
516 buffer
->info
.metadata
.subbuf_size
= new_subbuf_size
;
517 /* Padding is not present in the view anymore */
518 buffer
->info
.metadata
.padded_subbuf_size
= new_subbuf_size
;
521 static int metadata_stream_pre_consume_ctf1(
522 struct lttng_consumer_stream
*stream
, struct stream_subbuffer
*subbuffer
)
524 (void) metadata_stream_check_version(stream
, subbuffer
);
528 static int metadata_stream_pre_consume_ctf2(
529 struct lttng_consumer_stream
*stream
, struct stream_subbuffer
*subbuffer
)
531 (void) metadata_stream_check_version(stream
, subbuffer
);
532 (void) strip_packet_header_from_subbuffer(subbuffer
);
537 bool stream_is_rotating_to_null_chunk(
538 const struct lttng_consumer_stream
*stream
)
540 bool rotating_to_null_chunk
= false;
542 if (stream
->rotate_position
== -1ULL) {
543 /* No rotation ongoing. */
547 if (stream
->trace_chunk
== stream
->chan
->trace_chunk
||
548 !stream
->chan
->trace_chunk
) {
549 rotating_to_null_chunk
= true;
552 return rotating_to_null_chunk
;
555 enum consumer_stream_open_packet_status
consumer_stream_open_packet(
556 struct lttng_consumer_stream
*stream
)
559 enum consumer_stream_open_packet_status status
;
560 unsigned long produced_pos_before
, produced_pos_after
;
562 ret
= lttng_consumer_sample_snapshot_positions(stream
);
564 ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
565 ", channel name = %s, session id = %" PRIu64
,
566 stream
->key
, stream
->chan
->name
,
567 stream
->chan
->session_id
);
568 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
572 ret
= lttng_consumer_get_produced_snapshot(
573 stream
, &produced_pos_before
);
575 ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
576 ", channel name = %s, session id = %" PRIu64
,
577 stream
->key
, stream
->chan
->name
,
578 stream
->chan
->session_id
);
579 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
583 ret
= consumer_stream_flush_buffer(stream
, 0);
585 ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
586 ", channel name = %s, session id = %" PRIu64
,
587 stream
->key
, stream
->chan
->name
,
588 stream
->chan
->session_id
);
589 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
593 ret
= lttng_consumer_sample_snapshot_positions(stream
);
595 ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
596 ", channel name = %s, session id = %" PRIu64
,
597 stream
->key
, stream
->chan
->name
,
598 stream
->chan
->session_id
);
599 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
603 ret
= lttng_consumer_get_produced_snapshot(stream
, &produced_pos_after
);
605 ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
606 ", channel name = %s, session id = %" PRIu64
,
607 stream
->key
, stream
->chan
->name
,
608 stream
->chan
->session_id
);
609 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
614 * Determine if the flush had an effect by comparing the produced
615 * positons before and after the flush.
617 status
= produced_pos_before
!= produced_pos_after
?
618 CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
:
619 CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE
;
620 if (status
== CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
) {
621 stream
->opened_packet_in_current_trace_chunk
= true;
629 * An attempt to open a new packet is performed after a rotation completes to
630 * get a begin timestamp as close as possible to the rotation point.
632 * However, that initial attempt at opening a packet can fail due to a full
633 * ring-buffer. In that case, a second attempt is performed after consuming
634 * a packet since that will have freed enough space in the ring-buffer.
637 int post_consume_open_new_packet(struct lttng_consumer_stream
*stream
,
638 const struct stream_subbuffer
*subbuffer
__attribute__((unused
)),
639 struct lttng_consumer_local_data
*ctx
__attribute__((unused
)))
643 if (!stream
->opened_packet_in_current_trace_chunk
&&
644 stream
->trace_chunk
&&
645 !stream_is_rotating_to_null_chunk(stream
)) {
646 const enum consumer_stream_open_packet_status status
=
647 consumer_stream_open_packet(stream
);
650 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
:
651 DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64
652 ", channel name = %s, session id = %" PRIu64
,
653 stream
->key
, stream
->chan
->name
,
654 stream
->chan
->session_id
);
655 stream
->opened_packet_in_current_trace_chunk
= true;
657 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE
:
659 * Can't open a packet as there is no space left.
660 * This means that new events were produced, resulting
661 * in a packet being opened, which is what we want
664 DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
665 ", channel name = %s, session id = %" PRIu64
,
666 stream
->key
, stream
->chan
->name
,
667 stream
->chan
->session_id
);
668 stream
->opened_packet_in_current_trace_chunk
= true;
670 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
:
671 /* Logged by callee. */
678 stream
->opened_packet_in_current_trace_chunk
= true;
685 struct lttng_consumer_stream
*consumer_stream_create(struct lttng_consumer_channel
*channel
,
686 uint64_t channel_key
,
688 const char *channel_name
,
691 struct lttng_trace_chunk
*trace_chunk
,
694 enum consumer_channel_type type
,
695 unsigned int monitor
,
699 struct lttng_consumer_stream
*stream
;
701 stream
= zmalloc
<lttng_consumer_stream
>();
702 if (stream
== NULL
) {
703 PERROR("malloc struct lttng_consumer_stream");
710 if (trace_chunk
&& !lttng_trace_chunk_get(trace_chunk
)) {
711 ERR("Failed to acquire trace chunk reference during the creation of a stream");
716 stream
->send_node
= CDS_LIST_HEAD_INIT(stream
->send_node
);
717 stream
->chan
= channel
;
718 stream
->key
= stream_key
;
719 stream
->trace_chunk
= trace_chunk
;
721 stream
->out_fd_offset
= 0;
722 stream
->output_written
= 0;
723 stream
->net_seq_idx
= relayd_id
;
724 stream
->session_id
= session_id
;
725 stream
->monitor
= monitor
;
726 stream
->endpoint_status
= CONSUMER_ENDPOINT_ACTIVE
;
727 stream
->index_file
= NULL
;
728 stream
->last_sequence_number
= -1ULL;
729 stream
->rotate_position
= -1ULL;
730 /* Buffer is created with an open packet. */
731 stream
->opened_packet_in_current_trace_chunk
= true;
732 pthread_mutex_init(&stream
->lock
, NULL
);
733 pthread_mutex_init(&stream
->metadata_timer_lock
, NULL
);
735 /* If channel is the metadata, flag this stream as metadata. */
736 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
737 stream
->metadata_flag
= 1;
738 /* Metadata is flat out. */
739 strncpy(stream
->name
, DEFAULT_METADATA_NAME
, sizeof(stream
->name
));
740 /* Live rendez-vous point. */
741 pthread_cond_init(&stream
->metadata_rdv
, NULL
);
742 pthread_mutex_init(&stream
->metadata_rdv_lock
, NULL
);
744 /* Format stream name to <channel_name>_<cpu_number> */
745 ret
= snprintf(stream
->name
, sizeof(stream
->name
), "%s_%d",
748 PERROR("snprintf stream name");
753 switch (channel
->output
) {
754 case CONSUMER_CHANNEL_SPLICE
:
755 stream
->output
= LTTNG_EVENT_SPLICE
;
756 ret
= utils_create_pipe(stream
->splice_pipe
);
761 case CONSUMER_CHANNEL_MMAP
:
762 stream
->output
= LTTNG_EVENT_MMAP
;
768 /* Key is always the wait_fd for streams. */
769 lttng_ht_node_init_u64(&stream
->node
, stream
->key
);
771 /* Init node per channel id key */
772 lttng_ht_node_init_u64(&stream
->node_channel_id
, channel_key
);
774 /* Init session id node with the stream session id */
775 lttng_ht_node_init_u64(&stream
->node_session_id
, stream
->session_id
);
777 DBG3("Allocated stream %s (key %" PRIu64
", chan_key %" PRIu64
778 " relayd_id %" PRIu64
", session_id %" PRIu64
,
779 stream
->name
, stream
->key
, channel_key
,
780 stream
->net_seq_idx
, stream
->session_id
);
784 lttng_dynamic_array_init(&stream
->read_subbuffer_ops
.post_consume_cbs
,
785 sizeof(post_consume_cb
), NULL
);
787 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
788 stream
->read_subbuffer_ops
.lock
=
789 consumer_stream_metadata_lock_all
;
790 stream
->read_subbuffer_ops
.unlock
=
791 consumer_stream_metadata_unlock_all
;
792 stream
->read_subbuffer_ops
.assert_locked
=
793 consumer_stream_metadata_assert_locked_all
;
794 if (trace_format
== 1) {
795 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
796 metadata_stream_pre_consume_ctf1
;
797 } else if (trace_format
== 2) {
798 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
799 metadata_stream_pre_consume_ctf2
;
804 const post_consume_cb post_consume_index_op
= channel
->is_live
?
805 consumer_stream_sync_metadata_index
:
806 consumer_stream_send_index
;
807 const post_consume_cb post_consume_open_new_packet_
=
808 post_consume_open_new_packet
;
810 ret
= lttng_dynamic_array_add_element(
811 &stream
->read_subbuffer_ops
.post_consume_cbs
,
812 &post_consume_index_op
);
814 PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
818 ret
= lttng_dynamic_array_add_element(
819 &stream
->read_subbuffer_ops
.post_consume_cbs
,
820 &post_consume_open_new_packet_
);
822 PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
826 stream
->read_subbuffer_ops
.lock
= consumer_stream_data_lock_all
;
827 stream
->read_subbuffer_ops
.unlock
=
828 consumer_stream_data_unlock_all
;
829 stream
->read_subbuffer_ops
.assert_locked
=
830 consumer_stream_data_assert_locked_all
;
831 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
832 consumer_stream_update_stats
;
835 if (channel
->output
== CONSUMER_CHANNEL_MMAP
) {
836 stream
->read_subbuffer_ops
.consume_subbuffer
=
837 consumer_stream_consume_mmap
;
839 stream
->read_subbuffer_ops
.consume_subbuffer
=
840 consumer_stream_consume_splice
;
847 lttng_trace_chunk_put(stream
->trace_chunk
);
848 lttng_dynamic_array_reset(&stream
->read_subbuffer_ops
.post_consume_cbs
);
858 * Close stream on the relayd side. This call can destroy a relayd if the
859 * conditions are met.
861 * A RCU read side lock MUST be acquired if the relayd object was looked up in
862 * a hash table before calling this.
864 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
865 struct consumer_relayd_sock_pair
*relayd
)
869 LTTNG_ASSERT(stream
);
870 LTTNG_ASSERT(relayd
);
872 if (stream
->sent_to_relayd
) {
873 uatomic_dec(&relayd
->refcount
);
874 LTTNG_ASSERT(uatomic_read(&relayd
->refcount
) >= 0);
877 /* Closing streams requires to lock the control socket. */
878 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
879 ret
= relayd_send_close_stream(&relayd
->control_sock
,
880 stream
->relayd_stream_id
,
881 stream
->next_net_seq_num
- 1);
882 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
884 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
885 lttng_consumer_cleanup_relayd(relayd
);
888 /* Both conditions are met, we destroy the relayd. */
889 if (uatomic_read(&relayd
->refcount
) == 0 &&
890 uatomic_read(&relayd
->destroy_flag
)) {
891 consumer_destroy_relayd(relayd
);
893 stream
->net_seq_idx
= (uint64_t) -1ULL;
894 stream
->sent_to_relayd
= 0;
898 * Close stream's file descriptors and, if needed, close stream also on the
901 * The consumer data lock MUST be acquired.
902 * The stream lock MUST be acquired.
904 void consumer_stream_close(struct lttng_consumer_stream
*stream
)
907 struct consumer_relayd_sock_pair
*relayd
;
909 LTTNG_ASSERT(stream
);
911 switch (the_consumer_data
.type
) {
912 case LTTNG_CONSUMER_KERNEL
:
913 if (stream
->mmap_base
!= NULL
) {
914 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
920 if (stream
->wait_fd
>= 0) {
921 ret
= close(stream
->wait_fd
);
925 stream
->wait_fd
= -1;
927 if (stream
->chan
->output
== CONSUMER_CHANNEL_SPLICE
) {
928 utils_close_pipe(stream
->splice_pipe
);
931 case LTTNG_CONSUMER32_UST
:
932 case LTTNG_CONSUMER64_UST
:
935 * Special case for the metadata since the wait fd is an internal pipe
936 * polled in the metadata thread.
938 if (stream
->metadata_flag
&& stream
->chan
->monitor
) {
939 int rpipe
= stream
->ust_metadata_poll_pipe
[0];
942 * This will stop the channel timer if one and close the write side
943 * of the metadata poll pipe.
945 lttng_ustconsumer_close_metadata(stream
->chan
);
949 PERROR("closing metadata pipe read side");
951 stream
->ust_metadata_poll_pipe
[0] = -1;
957 ERR("Unknown consumer_data type");
961 /* Close output fd. Could be a socket or local file at this point. */
962 if (stream
->out_fd
>= 0) {
963 ret
= close(stream
->out_fd
);
970 if (stream
->index_file
) {
971 lttng_index_file_put(stream
->index_file
);
972 stream
->index_file
= NULL
;
975 lttng_trace_chunk_put(stream
->trace_chunk
);
976 stream
->trace_chunk
= NULL
;
978 /* Check and cleanup relayd if needed. */
980 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
981 if (relayd
!= NULL
) {
982 consumer_stream_relayd_close(stream
, relayd
);
988 * Delete the stream from all possible hash tables.
990 * The consumer data lock MUST be acquired.
991 * The stream lock MUST be acquired.
993 void consumer_stream_delete(struct lttng_consumer_stream
*stream
,
997 struct lttng_ht_iter iter
;
999 LTTNG_ASSERT(stream
);
1000 /* Should NEVER be called not in monitor mode. */
1001 LTTNG_ASSERT(stream
->chan
->monitor
);
1006 iter
.iter
.node
= &stream
->node
.node
;
1007 ret
= lttng_ht_del(ht
, &iter
);
1011 /* Delete from stream per channel ID hash table. */
1012 iter
.iter
.node
= &stream
->node_channel_id
.node
;
1014 * The returned value is of no importance. Even if the node is NOT in the
1015 * hash table, we continue since we may have been called by a code path
1016 * that did not add the stream to a (all) hash table. Same goes for the
1017 * next call ht del call.
1019 (void) lttng_ht_del(the_consumer_data
.stream_per_chan_id_ht
, &iter
);
1021 /* Delete from the global stream list. */
1022 iter
.iter
.node
= &stream
->node_session_id
.node
;
1023 /* See the previous ht del on why we ignore the returned value. */
1024 (void) lttng_ht_del(the_consumer_data
.stream_list_ht
, &iter
);
1028 if (!stream
->metadata_flag
) {
1029 /* Decrement the stream count of the global consumer data. */
1030 LTTNG_ASSERT(the_consumer_data
.stream_count
> 0);
1031 the_consumer_data
.stream_count
--;
1036 * Free the given stream within a RCU call.
1038 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
1040 LTTNG_ASSERT(stream
);
1042 metadata_bucket_destroy(stream
->metadata_bucket
);
1043 call_rcu(&stream
->node
.head
, free_stream_rcu
);
1047 * Destroy the stream's buffers of the tracer.
1049 void consumer_stream_destroy_buffers(struct lttng_consumer_stream
*stream
)
1051 LTTNG_ASSERT(stream
);
1053 switch (the_consumer_data
.type
) {
1054 case LTTNG_CONSUMER_KERNEL
:
1056 case LTTNG_CONSUMER32_UST
:
1057 case LTTNG_CONSUMER64_UST
:
1058 lttng_ustconsumer_del_stream(stream
);
1061 ERR("Unknown consumer_data type");
1067 * Destroy and close a already created stream.
1069 static void destroy_close_stream(struct lttng_consumer_stream
*stream
)
1071 LTTNG_ASSERT(stream
);
1073 DBG("Consumer stream destroy monitored key: %" PRIu64
, stream
->key
);
1075 /* Destroy tracer buffers of the stream. */
1076 consumer_stream_destroy_buffers(stream
);
1077 /* Close down everything including the relayd if one. */
1078 consumer_stream_close(stream
);
1082 * Decrement the stream's channel refcount and if down to 0, return the channel
1083 * pointer so it can be destroyed by the caller or NULL if not.
1085 static struct lttng_consumer_channel
*unref_channel(
1086 struct lttng_consumer_stream
*stream
)
1088 struct lttng_consumer_channel
*free_chan
= NULL
;
1090 LTTNG_ASSERT(stream
);
1091 LTTNG_ASSERT(stream
->chan
);
1093 /* Update refcount of channel and see if we need to destroy it. */
1094 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1)
1095 && !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
1096 free_chan
= stream
->chan
;
1103 * Destroy a stream completely. This will delete, close and free the stream.
1104 * Once return, the stream is NO longer usable. Its channel may get destroyed
1105 * if conditions are met for a monitored stream.
1107 * This MUST be called WITHOUT the consumer data and stream lock acquired if
1108 * the stream is in _monitor_ mode else it does not matter.
1110 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
,
1111 struct lttng_ht
*ht
)
1113 LTTNG_ASSERT(stream
);
1115 cds_list_del_init(&stream
->send_node
);
1117 /* Stream is in monitor mode. */
1118 if (stream
->monitor
) {
1119 struct lttng_consumer_channel
*free_chan
= NULL
;
1122 * This means that the stream was successfully removed from the streams
1123 * list of the channel and sent to the right thread managing this
1124 * stream thus being globally visible.
1126 if (stream
->globally_visible
) {
1127 pthread_mutex_lock(&the_consumer_data
.lock
);
1128 pthread_mutex_lock(&stream
->chan
->lock
);
1130 pthread_mutex_lock(&stream
->lock
);
1131 /* Remove every reference of the stream in the consumer. */
1132 consumer_stream_delete(stream
, ht
);
1135 destroy_close_stream(stream
);
1137 /* Update channel's refcount of the stream. */
1138 free_chan
= unref_channel(stream
);
1140 /* Indicates that the consumer data state MUST be updated after this. */
1141 the_consumer_data
.need_update
= 1;
1143 pthread_mutex_unlock(&stream
->lock
);
1144 pthread_mutex_unlock(&stream
->chan
->lock
);
1145 pthread_mutex_unlock(&the_consumer_data
.lock
);
1148 * If the stream is not visible globally, this needs to be done
1149 * outside of the consumer data lock section.
1151 destroy_close_stream(stream
);
1152 free_chan
= unref_channel(stream
);
1156 consumer_del_channel(free_chan
);
1159 destroy_close_stream(stream
);
1162 /* Free stream within a RCU call. */
1163 lttng_trace_chunk_put(stream
->trace_chunk
);
1164 stream
->trace_chunk
= NULL
;
1165 lttng_dynamic_array_reset(&stream
->read_subbuffer_ops
.post_consume_cbs
);
1166 consumer_stream_free(stream
);
1170 * Write index of a specific stream either on the relayd or local disk.
1172 * Return 0 on success or else a negative value.
1174 int consumer_stream_write_index(struct lttng_consumer_stream
*stream
,
1175 struct ctf_packet_index
*element
)
1179 LTTNG_ASSERT(stream
);
1180 LTTNG_ASSERT(element
);
1183 if (stream
->net_seq_idx
!= (uint64_t) -1ULL) {
1184 struct consumer_relayd_sock_pair
*relayd
;
1185 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
1187 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
1188 ret
= relayd_send_index(&relayd
->control_sock
, element
,
1189 stream
->relayd_stream_id
, stream
->next_net_seq_num
- 1);
1192 * Communication error with lttng-relayd,
1193 * perform cleanup now
1195 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
1196 lttng_consumer_cleanup_relayd(relayd
);
1199 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
1201 ERR("Stream %" PRIu64
" relayd ID %" PRIu64
" unknown. Can't write index.",
1202 stream
->key
, stream
->net_seq_idx
);
1206 if (lttng_index_file_write(stream
->index_file
, element
)) {
1221 int consumer_stream_create_output_files(struct lttng_consumer_stream
*stream
,
1225 enum lttng_trace_chunk_status chunk_status
;
1226 const int flags
= O_WRONLY
| O_CREAT
| O_TRUNC
;
1227 const mode_t mode
= S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
;
1228 char stream_path
[LTTNG_PATH_MAX
];
1230 ASSERT_LOCKED(stream
->lock
);
1231 LTTNG_ASSERT(stream
->trace_chunk
);
1233 ret
= utils_stream_file_path(stream
->chan
->pathname
, stream
->name
,
1234 stream
->chan
->tracefile_size
,
1235 stream
->tracefile_count_current
, NULL
,
1236 stream_path
, sizeof(stream_path
));
1241 if (stream
->out_fd
>= 0) {
1242 ret
= close(stream
->out_fd
);
1244 PERROR("Failed to close stream file \"%s\"",
1248 stream
->out_fd
= -1;
1251 DBG("Opening stream output file \"%s\"", stream_path
);
1252 chunk_status
= lttng_trace_chunk_open_file(stream
->trace_chunk
, stream_path
,
1253 flags
, mode
, &stream
->out_fd
, false);
1254 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1255 ERR("Failed to open stream file \"%s\"", stream
->name
);
1260 if (!stream
->metadata_flag
&& (create_index
|| stream
->index_file
)) {
1261 if (stream
->index_file
) {
1262 lttng_index_file_put(stream
->index_file
);
1264 chunk_status
= lttng_index_file_create_from_trace_chunk(
1265 stream
->trace_chunk
,
1266 stream
->chan
->pathname
,
1268 stream
->chan
->tracefile_size
,
1269 stream
->tracefile_count_current
,
1270 CTF_INDEX_MAJOR
, CTF_INDEX_MINOR
,
1271 false, &stream
->index_file
);
1272 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1278 /* Reset current size because we just perform a rotation. */
1279 stream
->tracefile_size_current
= 0;
1280 stream
->out_fd_offset
= 0;
1285 int consumer_stream_rotate_output_files(struct lttng_consumer_stream
*stream
)
1289 stream
->tracefile_count_current
++;
1290 if (stream
->chan
->tracefile_count
> 0) {
1291 stream
->tracefile_count_current
%=
1292 stream
->chan
->tracefile_count
;
1295 DBG("Rotating output files of stream \"%s\"", stream
->name
);
1296 ret
= consumer_stream_create_output_files(stream
, true);
1305 bool consumer_stream_is_deleted(struct lttng_consumer_stream
*stream
)
1308 * This function does not take a const stream since
1309 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1311 LTTNG_ASSERT(stream
);
1312 return cds_lfht_is_node_deleted(&stream
->node
.node
);
1315 static ssize_t
metadata_bucket_flush(
1316 const struct stream_subbuffer
*buffer
, void *data
)
1319 struct lttng_consumer_stream
*stream
= (lttng_consumer_stream
*) data
;
1321 ret
= consumer_stream_consume_mmap(NULL
, stream
, buffer
);
1329 static ssize_t
metadata_bucket_consume(
1330 struct lttng_consumer_local_data
*unused
__attribute__((unused
)),
1331 struct lttng_consumer_stream
*stream
,
1332 const struct stream_subbuffer
*subbuffer
)
1335 enum metadata_bucket_status status
;
1337 status
= metadata_bucket_fill(stream
->metadata_bucket
, subbuffer
);
1339 case METADATA_BUCKET_STATUS_OK
:
1340 /* Return consumed size. */
1341 ret
= subbuffer
->buffer
.buffer
.size
;
1350 int consumer_stream_enable_metadata_bucketization(
1351 struct lttng_consumer_stream
*stream
)
1355 LTTNG_ASSERT(stream
->metadata_flag
);
1356 LTTNG_ASSERT(!stream
->metadata_bucket
);
1357 LTTNG_ASSERT(stream
->chan
->output
== CONSUMER_CHANNEL_MMAP
);
1359 stream
->metadata_bucket
= metadata_bucket_create(
1360 metadata_bucket_flush
, stream
);
1361 if (!stream
->metadata_bucket
) {
1366 stream
->read_subbuffer_ops
.consume_subbuffer
= metadata_bucket_consume
;
1371 void consumer_stream_metadata_set_version(
1372 struct lttng_consumer_stream
*stream
, uint64_t new_version
)
1374 LTTNG_ASSERT(new_version
> stream
->metadata_version
);
1375 stream
->metadata_version
= new_version
;
1376 stream
->reset_metadata_flag
= 1;
1378 if (stream
->metadata_bucket
) {
1379 metadata_bucket_reset(stream
->metadata_bucket
);
1383 int consumer_stream_flush_buffer(struct lttng_consumer_stream
*stream
,
1384 bool producer_active
)
1388 switch (the_consumer_data
.type
) {
1389 case LTTNG_CONSUMER_KERNEL
:
1390 if (producer_active
) {
1391 ret
= kernctl_buffer_flush(stream
->wait_fd
);
1393 ERR("Failed to flush kernel stream");
1397 ret
= kernctl_buffer_flush_empty(stream
->wait_fd
);
1400 * Doing a buffer flush which does not take into
1401 * account empty packets. This is not perfect,
1402 * but required as a fall-back when
1403 * "flush_empty" is not implemented by
1406 ret
= kernctl_buffer_flush(stream
->wait_fd
);
1408 ERR("Failed to flush kernel stream");
1414 case LTTNG_CONSUMER32_UST
:
1415 case LTTNG_CONSUMER64_UST
:
1416 ret
= lttng_ustconsumer_flush_buffer(stream
, (int) producer_active
);
1419 ERR("Unknown consumer_data type");