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/common.hpp>
16 #include <common/consumer/consumer-timer.hpp>
17 #include <common/consumer/consumer-timer.hpp>
18 #include <common/consumer/consumer.hpp>
19 #include <common/consumer/consumer.hpp>
20 #include <common/consumer/metadata-bucket.hpp>
21 #include <common/consumer/metadata-bucket.hpp>
22 #include <common/index/index.hpp>
23 #include <common/kernel-consumer/kernel-consumer.hpp>
24 #include <common/kernel-ctl/kernel-ctl.hpp>
25 #include <common/macros.hpp>
26 #include <common/relayd/relayd.hpp>
27 #include <common/ust-consumer/ust-consumer.hpp>
28 #include <common/utils.hpp>
30 #include "consumer-stream.hpp"
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 lttng::utils::container_of(head
, <tng_ht_node_u64::head
);
39 struct lttng_consumer_stream
*stream
=
40 lttng::utils::container_of(node
, <tng_consumer_stream::node
);
42 pthread_mutex_destroy(&stream
->lock
);
46 static void consumer_stream_data_lock_all(struct lttng_consumer_stream
*stream
)
48 pthread_mutex_lock(&stream
->chan
->lock
);
49 pthread_mutex_lock(&stream
->lock
);
52 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream
*stream
)
54 pthread_mutex_unlock(&stream
->lock
);
55 pthread_mutex_unlock(&stream
->chan
->lock
);
58 static void consumer_stream_data_assert_locked_all(struct lttng_consumer_stream
*stream
)
60 ASSERT_LOCKED(stream
->lock
);
61 ASSERT_LOCKED(stream
->chan
->lock
);
64 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream
*stream
)
66 consumer_stream_data_lock_all(stream
);
67 pthread_mutex_lock(&stream
->metadata_rdv_lock
);
70 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream
*stream
)
72 pthread_mutex_unlock(&stream
->metadata_rdv_lock
);
73 consumer_stream_data_unlock_all(stream
);
76 static void consumer_stream_metadata_assert_locked_all(struct lttng_consumer_stream
*stream
)
78 ASSERT_LOCKED(stream
->metadata_rdv_lock
);
79 consumer_stream_data_assert_locked_all(stream
);
82 /* Only used for data streams. */
83 static int consumer_stream_update_stats(struct lttng_consumer_stream
*stream
,
84 const struct stream_subbuffer
*subbuf
)
87 uint64_t sequence_number
;
88 const uint64_t discarded_events
= subbuf
->info
.data
.events_discarded
;
90 if (!subbuf
->info
.data
.sequence_number
.is_set
) {
91 /* Command not supported by the tracer. */
92 sequence_number
= -1ULL;
93 stream
->sequence_number_unavailable
= true;
95 sequence_number
= subbuf
->info
.data
.sequence_number
.value
;
99 * Start the sequence when we extract the first packet in case we don't
100 * start at 0 (for example if a consumer is not connected to the
101 * session immediately after the beginning).
103 if (stream
->last_sequence_number
== -1ULL) {
104 stream
->last_sequence_number
= sequence_number
;
105 } else if (sequence_number
> stream
->last_sequence_number
) {
106 stream
->chan
->lost_packets
+= sequence_number
-
107 stream
->last_sequence_number
- 1;
109 /* seq <= last_sequence_number */
110 ERR("Sequence number inconsistent : prev = %" PRIu64
111 ", current = %" PRIu64
,
112 stream
->last_sequence_number
, sequence_number
);
116 stream
->last_sequence_number
= sequence_number
;
118 if (discarded_events
< stream
->last_discarded_events
) {
120 * Overflow has occurred. We assume only one wrap-around
123 stream
->chan
->discarded_events
+=
124 (1ULL << (CAA_BITS_PER_LONG
- 1)) -
125 stream
->last_discarded_events
+
128 stream
->chan
->discarded_events
+= discarded_events
-
129 stream
->last_discarded_events
;
131 stream
->last_discarded_events
= discarded_events
;
139 void ctf_packet_index_populate(struct ctf_packet_index
*index
,
140 off_t offset
, const struct stream_subbuffer
*subbuffer
)
142 *index
= (typeof(*index
)){
143 .offset
= htobe64(offset
),
144 .packet_size
= htobe64(subbuffer
->info
.data
.packet_size
),
145 .content_size
= htobe64(subbuffer
->info
.data
.content_size
),
146 .timestamp_begin
= htobe64(
147 subbuffer
->info
.data
.timestamp_begin
),
148 .timestamp_end
= htobe64(
149 subbuffer
->info
.data
.timestamp_end
),
150 .events_discarded
= htobe64(
151 subbuffer
->info
.data
.events_discarded
),
152 .stream_id
= htobe64(subbuffer
->info
.data
.stream_id
),
153 .stream_instance_id
= htobe64(
154 subbuffer
->info
.data
.stream_instance_id
.is_set
?
155 subbuffer
->info
.data
.stream_instance_id
.value
: -1ULL),
156 .packet_seq_num
= htobe64(
157 subbuffer
->info
.data
.sequence_number
.is_set
?
158 subbuffer
->info
.data
.sequence_number
.value
: -1ULL),
162 static ssize_t
consumer_stream_consume_mmap(
163 struct lttng_consumer_local_data
*ctx
__attribute__((unused
)),
164 struct lttng_consumer_stream
*stream
,
165 const struct stream_subbuffer
*subbuffer
)
167 const unsigned long padding_size
=
168 subbuffer
->info
.data
.padded_subbuf_size
-
169 subbuffer
->info
.data
.subbuf_size
;
170 const ssize_t written_bytes
= lttng_consumer_on_read_subbuffer_mmap(
171 stream
, &subbuffer
->buffer
.buffer
, padding_size
);
173 if (stream
->net_seq_idx
== -1ULL) {
175 * When writing on disk, check that only the subbuffer (no
176 * padding) was written to disk.
178 if (written_bytes
!= subbuffer
->info
.data
.padded_subbuf_size
) {
179 DBG("Failed to write the entire padded subbuffer on disk (written_bytes: %zd, padded subbuffer size %lu)",
181 subbuffer
->info
.data
.padded_subbuf_size
);
185 * When streaming over the network, check that the entire
186 * subbuffer including padding was successfully written.
188 if (written_bytes
!= subbuffer
->info
.data
.subbuf_size
) {
189 DBG("Failed to write only the subbuffer over the network (written_bytes: %zd, subbuffer size %lu)",
191 subbuffer
->info
.data
.subbuf_size
);
196 * If `lttng_consumer_on_read_subbuffer_mmap()` returned an error, pass
197 * it along to the caller, else return zero.
199 if (written_bytes
< 0) {
200 ERR("Error reading mmap subbuffer: %zd", written_bytes
);
203 return written_bytes
;
206 static ssize_t
consumer_stream_consume_splice(
207 struct lttng_consumer_local_data
*ctx
,
208 struct lttng_consumer_stream
*stream
,
209 const struct stream_subbuffer
*subbuffer
)
211 const ssize_t written_bytes
= lttng_consumer_on_read_subbuffer_splice(
212 ctx
, stream
, subbuffer
->info
.data
.padded_subbuf_size
, 0);
214 if (written_bytes
!= subbuffer
->info
.data
.padded_subbuf_size
) {
215 DBG("Failed to write the entire padded subbuffer (written_bytes: %zd, padded subbuffer size %lu)",
217 subbuffer
->info
.data
.padded_subbuf_size
);
221 * If `lttng_consumer_on_read_subbuffer_splice()` returned an error,
222 * pass it along to the caller, else return zero.
224 if (written_bytes
< 0) {
225 ERR("Error reading splice subbuffer: %zd", written_bytes
);
228 return written_bytes
;
231 static int consumer_stream_send_index(
232 struct lttng_consumer_stream
*stream
,
233 const struct stream_subbuffer
*subbuffer
,
234 struct lttng_consumer_local_data
*ctx
__attribute__((unused
)))
236 off_t packet_offset
= 0;
237 struct ctf_packet_index index
= {};
240 * This is called after consuming the sub-buffer; substract the
241 * effect this sub-buffer from the offset.
243 if (stream
->net_seq_idx
== (uint64_t) -1ULL) {
244 packet_offset
= stream
->out_fd_offset
-
245 subbuffer
->info
.data
.padded_subbuf_size
;
248 ctf_packet_index_populate(&index
, packet_offset
, subbuffer
);
249 return consumer_stream_write_index(stream
, &index
);
253 * Actually do the metadata sync using the given metadata stream.
255 * Return 0 on success else a negative value. ENODATA can be returned also
256 * indicating that there is no metadata available for that stream.
258 static int do_sync_metadata(struct lttng_consumer_stream
*metadata
,
259 struct lttng_consumer_local_data
*ctx
)
262 enum sync_metadata_status status
;
264 LTTNG_ASSERT(metadata
);
265 LTTNG_ASSERT(metadata
->metadata_flag
);
269 * In UST, since we have to write the metadata from the cache packet
270 * by packet, we might need to start this procedure multiple times
271 * until all the metadata from the cache has been extracted.
276 * - Lock the metadata stream
277 * - Check if metadata stream node was deleted before locking.
278 * - if yes, release and return success
279 * - Check if new metadata is ready (flush + snapshot pos)
280 * - If nothing : release and return.
281 * - Lock the metadata_rdv_lock
282 * - Unlock the metadata stream
283 * - cond_wait on metadata_rdv to wait the wakeup from the
285 * - Unlock the metadata_rdv_lock
287 pthread_mutex_lock(&metadata
->lock
);
290 * There is a possibility that we were able to acquire a reference on the
291 * stream from the RCU hash table but between then and now, the node might
292 * have been deleted just before the lock is acquired. Thus, after locking,
293 * we make sure the metadata node has not been deleted which means that the
294 * buffers are closed.
296 * In that case, there is no need to sync the metadata hence returning a
297 * success return code.
299 ret
= cds_lfht_is_node_deleted(&metadata
->node
.node
);
302 goto end_unlock_mutex
;
306 case LTTNG_CONSUMER_KERNEL
:
308 * Empty the metadata cache and flush the current stream.
310 status
= lttng_kconsumer_sync_metadata(metadata
);
312 case LTTNG_CONSUMER32_UST
:
313 case LTTNG_CONSUMER64_UST
:
315 * Ask the sessiond if we have new metadata waiting and update the
316 * consumer metadata cache.
318 status
= lttng_ustconsumer_sync_metadata(ctx
, metadata
);
325 case SYNC_METADATA_STATUS_NEW_DATA
:
327 case SYNC_METADATA_STATUS_NO_DATA
:
329 goto end_unlock_mutex
;
330 case SYNC_METADATA_STATUS_ERROR
:
332 goto end_unlock_mutex
;
338 * At this point, new metadata have been flushed, so we wait on the
339 * rendez-vous point for the metadata thread to wake us up when it
340 * finishes consuming the metadata and continue execution.
343 pthread_mutex_lock(&metadata
->metadata_rdv_lock
);
346 * Release metadata stream lock so the metadata thread can process it.
348 pthread_mutex_unlock(&metadata
->lock
);
351 * Wait on the rendez-vous point. Once woken up, it means the metadata was
352 * consumed and thus synchronization is achieved.
354 pthread_cond_wait(&metadata
->metadata_rdv
, &metadata
->metadata_rdv_lock
);
355 pthread_mutex_unlock(&metadata
->metadata_rdv_lock
);
356 } while (status
== SYNC_METADATA_STATUS_NEW_DATA
);
362 pthread_mutex_unlock(&metadata
->lock
);
367 * Synchronize the metadata using a given session ID. A successful acquisition
368 * of a metadata stream will trigger a request to the session daemon and a
369 * snapshot so the metadata thread can consume it.
371 * This function call is a rendez-vous point between the metadata thread and
374 * Return 0 on success or else a negative value.
376 int consumer_stream_sync_metadata(struct lttng_consumer_local_data
*ctx
,
380 struct lttng_consumer_stream
*stream
= NULL
;
381 struct lttng_ht_iter iter
;
386 /* Ease our life a bit. */
387 ht
= the_consumer_data
.stream_list_ht
;
391 /* Search the metadata associated with the session id of the given stream. */
393 cds_lfht_for_each_entry_duplicate(ht
->ht
,
394 ht
->hash_fct(&session_id
, lttng_ht_seed
), ht
->match_fct
,
395 &session_id
, &iter
.iter
, stream
, node_session_id
.node
) {
396 if (!stream
->metadata_flag
) {
400 ret
= do_sync_metadata(stream
, ctx
);
407 * Force return code to 0 (success) since ret might be ENODATA for instance
408 * which is not an error but rather that we should come back.
417 static int consumer_stream_sync_metadata_index(
418 struct lttng_consumer_stream
*stream
,
419 const struct stream_subbuffer
*subbuffer
,
420 struct lttng_consumer_local_data
*ctx
)
422 bool missed_metadata_flush
;
425 /* Block until all the metadata is sent. */
426 pthread_mutex_lock(&stream
->metadata_timer_lock
);
427 LTTNG_ASSERT(!stream
->missed_metadata_flush
);
428 stream
->waiting_on_metadata
= true;
429 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
431 ret
= consumer_stream_sync_metadata(ctx
, stream
->session_id
);
433 pthread_mutex_lock(&stream
->metadata_timer_lock
);
434 stream
->waiting_on_metadata
= false;
435 missed_metadata_flush
= stream
->missed_metadata_flush
;
436 if (missed_metadata_flush
) {
437 stream
->missed_metadata_flush
= false;
439 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
444 ret
= consumer_stream_send_index(stream
, subbuffer
, ctx
);
446 * Send the live inactivity beacon to handle the situation where
447 * the live timer is prevented from sampling this stream
448 * because the stream lock was being held while this stream is
449 * waiting on metadata. This ensures live viewer progress in the
450 * unlikely scenario where a live timer would be prevented from
451 * locking a stream lock repeatedly due to a steady flow of
452 * incoming metadata, for a stream which is mostly inactive.
454 * It is important to send the inactivity beacon packet to
455 * relayd _after_ sending the index associated with the data
456 * that was just sent, otherwise this can cause live viewers to
457 * observe timestamps going backwards between an inactivity
458 * beacon and a following trace packet.
460 if (missed_metadata_flush
) {
461 (void) stream
->read_subbuffer_ops
.send_live_beacon(stream
);
468 * Check if the local version of the metadata stream matches with the version
469 * of the metadata stream in the kernel. If it was updated, set the reset flag
473 int metadata_stream_check_version(struct lttng_consumer_stream
*stream
,
474 const struct stream_subbuffer
*subbuffer
)
476 if (stream
->metadata_version
== subbuffer
->info
.metadata
.version
) {
480 DBG("New metadata version detected");
481 consumer_stream_metadata_set_version(stream
,
482 subbuffer
->info
.metadata
.version
);
484 if (stream
->read_subbuffer_ops
.reset_metadata
) {
485 stream
->read_subbuffer_ops
.reset_metadata(stream
);
493 bool stream_is_rotating_to_null_chunk(
494 const struct lttng_consumer_stream
*stream
)
496 bool rotating_to_null_chunk
= false;
498 if (stream
->rotate_position
== -1ULL) {
499 /* No rotation ongoing. */
503 if (stream
->trace_chunk
== stream
->chan
->trace_chunk
||
504 !stream
->chan
->trace_chunk
) {
505 rotating_to_null_chunk
= true;
508 return rotating_to_null_chunk
;
511 enum consumer_stream_open_packet_status
consumer_stream_open_packet(
512 struct lttng_consumer_stream
*stream
)
515 enum consumer_stream_open_packet_status status
;
516 unsigned long produced_pos_before
, produced_pos_after
;
518 ret
= lttng_consumer_sample_snapshot_positions(stream
);
520 ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
521 ", channel name = %s, session id = %" PRIu64
,
522 stream
->key
, stream
->chan
->name
,
523 stream
->chan
->session_id
);
524 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
528 ret
= lttng_consumer_get_produced_snapshot(
529 stream
, &produced_pos_before
);
531 ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
532 ", channel name = %s, session id = %" PRIu64
,
533 stream
->key
, stream
->chan
->name
,
534 stream
->chan
->session_id
);
535 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
539 ret
= consumer_stream_flush_buffer(stream
, 0);
541 ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
542 ", channel name = %s, session id = %" PRIu64
,
543 stream
->key
, stream
->chan
->name
,
544 stream
->chan
->session_id
);
545 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
549 ret
= lttng_consumer_sample_snapshot_positions(stream
);
551 ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
552 ", channel name = %s, session id = %" PRIu64
,
553 stream
->key
, stream
->chan
->name
,
554 stream
->chan
->session_id
);
555 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
559 ret
= lttng_consumer_get_produced_snapshot(stream
, &produced_pos_after
);
561 ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
562 ", channel name = %s, session id = %" PRIu64
,
563 stream
->key
, stream
->chan
->name
,
564 stream
->chan
->session_id
);
565 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
570 * Determine if the flush had an effect by comparing the produced
571 * positons before and after the flush.
573 status
= produced_pos_before
!= produced_pos_after
?
574 CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
:
575 CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE
;
576 if (status
== CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
) {
577 stream
->opened_packet_in_current_trace_chunk
= true;
585 * An attempt to open a new packet is performed after a rotation completes to
586 * get a begin timestamp as close as possible to the rotation point.
588 * However, that initial attempt at opening a packet can fail due to a full
589 * ring-buffer. In that case, a second attempt is performed after consuming
590 * a packet since that will have freed enough space in the ring-buffer.
593 int post_consume_open_new_packet(struct lttng_consumer_stream
*stream
,
594 const struct stream_subbuffer
*subbuffer
__attribute__((unused
)),
595 struct lttng_consumer_local_data
*ctx
__attribute__((unused
)))
599 if (!stream
->opened_packet_in_current_trace_chunk
&&
600 stream
->trace_chunk
&&
601 !stream_is_rotating_to_null_chunk(stream
)) {
602 const enum consumer_stream_open_packet_status status
=
603 consumer_stream_open_packet(stream
);
606 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
:
607 DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64
608 ", channel name = %s, session id = %" PRIu64
,
609 stream
->key
, stream
->chan
->name
,
610 stream
->chan
->session_id
);
611 stream
->opened_packet_in_current_trace_chunk
= true;
613 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE
:
615 * Can't open a packet as there is no space left.
616 * This means that new events were produced, resulting
617 * in a packet being opened, which is what we want
620 DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
621 ", channel name = %s, session id = %" PRIu64
,
622 stream
->key
, stream
->chan
->name
,
623 stream
->chan
->session_id
);
624 stream
->opened_packet_in_current_trace_chunk
= true;
626 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
:
627 /* Logged by callee. */
634 stream
->opened_packet_in_current_trace_chunk
= true;
641 struct lttng_consumer_stream
*consumer_stream_create(
642 struct lttng_consumer_channel
*channel
,
643 uint64_t channel_key
,
645 const char *channel_name
,
648 struct lttng_trace_chunk
*trace_chunk
,
651 enum consumer_channel_type type
,
652 unsigned int monitor
)
655 struct lttng_consumer_stream
*stream
;
657 stream
= zmalloc
<lttng_consumer_stream
>();
658 if (stream
== NULL
) {
659 PERROR("malloc struct lttng_consumer_stream");
666 if (trace_chunk
&& !lttng_trace_chunk_get(trace_chunk
)) {
667 ERR("Failed to acquire trace chunk reference during the creation of a stream");
672 stream
->send_node
= CDS_LIST_HEAD_INIT(stream
->send_node
);
673 stream
->chan
= channel
;
674 stream
->key
= stream_key
;
675 stream
->trace_chunk
= trace_chunk
;
677 stream
->out_fd_offset
= 0;
678 stream
->output_written
= 0;
679 stream
->net_seq_idx
= relayd_id
;
680 stream
->session_id
= session_id
;
681 stream
->monitor
= monitor
;
682 stream
->endpoint_status
= CONSUMER_ENDPOINT_ACTIVE
;
683 stream
->index_file
= NULL
;
684 stream
->last_sequence_number
= -1ULL;
685 stream
->rotate_position
= -1ULL;
686 /* Buffer is created with an open packet. */
687 stream
->opened_packet_in_current_trace_chunk
= true;
688 pthread_mutex_init(&stream
->lock
, NULL
);
689 pthread_mutex_init(&stream
->metadata_timer_lock
, NULL
);
691 /* If channel is the metadata, flag this stream as metadata. */
692 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
693 stream
->metadata_flag
= 1;
694 /* Metadata is flat out. */
695 strncpy(stream
->name
, DEFAULT_METADATA_NAME
, sizeof(stream
->name
));
696 /* Live rendez-vous point. */
697 pthread_cond_init(&stream
->metadata_rdv
, NULL
);
698 pthread_mutex_init(&stream
->metadata_rdv_lock
, NULL
);
700 /* Format stream name to <channel_name>_<cpu_number> */
701 ret
= snprintf(stream
->name
, sizeof(stream
->name
), "%s_%d",
704 PERROR("snprintf stream name");
709 switch (channel
->output
) {
710 case CONSUMER_CHANNEL_SPLICE
:
711 stream
->output
= LTTNG_EVENT_SPLICE
;
712 ret
= utils_create_pipe(stream
->splice_pipe
);
717 case CONSUMER_CHANNEL_MMAP
:
718 stream
->output
= LTTNG_EVENT_MMAP
;
724 /* Key is always the wait_fd for streams. */
725 lttng_ht_node_init_u64(&stream
->node
, stream
->key
);
727 /* Init node per channel id key */
728 lttng_ht_node_init_u64(&stream
->node_channel_id
, channel_key
);
730 /* Init session id node with the stream session id */
731 lttng_ht_node_init_u64(&stream
->node_session_id
, stream
->session_id
);
733 DBG3("Allocated stream %s (key %" PRIu64
", chan_key %" PRIu64
734 " relayd_id %" PRIu64
", session_id %" PRIu64
,
735 stream
->name
, stream
->key
, channel_key
,
736 stream
->net_seq_idx
, stream
->session_id
);
740 lttng_dynamic_array_init(&stream
->read_subbuffer_ops
.post_consume_cbs
,
741 sizeof(post_consume_cb
), NULL
);
743 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
744 stream
->read_subbuffer_ops
.lock
=
745 consumer_stream_metadata_lock_all
;
746 stream
->read_subbuffer_ops
.unlock
=
747 consumer_stream_metadata_unlock_all
;
748 stream
->read_subbuffer_ops
.assert_locked
=
749 consumer_stream_metadata_assert_locked_all
;
750 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
751 metadata_stream_check_version
;
753 const post_consume_cb post_consume_index_op
= channel
->is_live
?
754 consumer_stream_sync_metadata_index
:
755 consumer_stream_send_index
;
756 const post_consume_cb post_consume_open_new_packet_
=
757 post_consume_open_new_packet
;
759 ret
= lttng_dynamic_array_add_element(
760 &stream
->read_subbuffer_ops
.post_consume_cbs
,
761 &post_consume_index_op
);
763 PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
767 ret
= lttng_dynamic_array_add_element(
768 &stream
->read_subbuffer_ops
.post_consume_cbs
,
769 &post_consume_open_new_packet_
);
771 PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
775 stream
->read_subbuffer_ops
.lock
= consumer_stream_data_lock_all
;
776 stream
->read_subbuffer_ops
.unlock
=
777 consumer_stream_data_unlock_all
;
778 stream
->read_subbuffer_ops
.assert_locked
=
779 consumer_stream_data_assert_locked_all
;
780 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
781 consumer_stream_update_stats
;
784 if (channel
->output
== CONSUMER_CHANNEL_MMAP
) {
785 stream
->read_subbuffer_ops
.consume_subbuffer
=
786 consumer_stream_consume_mmap
;
788 stream
->read_subbuffer_ops
.consume_subbuffer
=
789 consumer_stream_consume_splice
;
796 lttng_trace_chunk_put(stream
->trace_chunk
);
797 lttng_dynamic_array_reset(&stream
->read_subbuffer_ops
.post_consume_cbs
);
807 * Close stream on the relayd side. This call can destroy a relayd if the
808 * conditions are met.
810 * A RCU read side lock MUST be acquired if the relayd object was looked up in
811 * a hash table before calling this.
813 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
814 struct consumer_relayd_sock_pair
*relayd
)
818 LTTNG_ASSERT(stream
);
819 LTTNG_ASSERT(relayd
);
821 if (stream
->sent_to_relayd
) {
822 uatomic_dec(&relayd
->refcount
);
823 LTTNG_ASSERT(uatomic_read(&relayd
->refcount
) >= 0);
826 /* Closing streams requires to lock the control socket. */
827 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
828 ret
= relayd_send_close_stream(&relayd
->control_sock
,
829 stream
->relayd_stream_id
,
830 stream
->next_net_seq_num
- 1);
831 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
833 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
834 lttng_consumer_cleanup_relayd(relayd
);
837 /* Both conditions are met, we destroy the relayd. */
838 if (uatomic_read(&relayd
->refcount
) == 0 &&
839 uatomic_read(&relayd
->destroy_flag
)) {
840 consumer_destroy_relayd(relayd
);
842 stream
->net_seq_idx
= (uint64_t) -1ULL;
843 stream
->sent_to_relayd
= 0;
847 * Close stream's file descriptors and, if needed, close stream also on the
850 * The consumer data lock MUST be acquired.
851 * The stream lock MUST be acquired.
853 void consumer_stream_close_output(struct lttng_consumer_stream
*stream
)
855 struct consumer_relayd_sock_pair
*relayd
;
857 LTTNG_ASSERT(stream
);
859 /* Close output fd. Could be a socket or local file at this point. */
860 if (stream
->out_fd
>= 0) {
861 const auto ret
= close(stream
->out_fd
);
863 PERROR("Failed to close stream output file descriptor");
869 if (stream
->index_file
) {
870 lttng_index_file_put(stream
->index_file
);
871 stream
->index_file
= NULL
;
874 lttng_trace_chunk_put(stream
->trace_chunk
);
875 stream
->trace_chunk
= NULL
;
877 /* Check and cleanup relayd if needed. */
879 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
880 if (relayd
!= NULL
) {
881 consumer_stream_relayd_close(stream
, relayd
);
882 stream
->net_seq_idx
= -1ULL;
889 * Delete the stream from all possible hash tables.
891 * The consumer data lock MUST be acquired.
892 * The stream lock MUST be acquired.
894 void consumer_stream_delete(struct lttng_consumer_stream
*stream
,
898 struct lttng_ht_iter iter
;
900 LTTNG_ASSERT(stream
);
901 /* Should NEVER be called not in monitor mode. */
902 LTTNG_ASSERT(stream
->chan
->monitor
);
907 iter
.iter
.node
= &stream
->node
.node
;
908 ret
= lttng_ht_del(ht
, &iter
);
912 /* Delete from stream per channel ID hash table. */
913 iter
.iter
.node
= &stream
->node_channel_id
.node
;
915 * The returned value is of no importance. Even if the node is NOT in the
916 * hash table, we continue since we may have been called by a code path
917 * that did not add the stream to a (all) hash table. Same goes for the
918 * next call ht del call.
920 (void) lttng_ht_del(the_consumer_data
.stream_per_chan_id_ht
, &iter
);
922 /* Delete from the global stream list. */
923 iter
.iter
.node
= &stream
->node_session_id
.node
;
924 /* See the previous ht del on why we ignore the returned value. */
925 (void) lttng_ht_del(the_consumer_data
.stream_list_ht
, &iter
);
929 if (!stream
->metadata_flag
) {
930 /* Decrement the stream count of the global consumer data. */
931 LTTNG_ASSERT(the_consumer_data
.stream_count
> 0);
932 the_consumer_data
.stream_count
--;
937 * Free the given stream within a RCU call.
939 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
941 LTTNG_ASSERT(stream
);
943 metadata_bucket_destroy(stream
->metadata_bucket
);
944 call_rcu(&stream
->node
.head
, free_stream_rcu
);
948 * Destroy the stream's buffers of the tracer.
950 void consumer_stream_destroy_buffers(struct lttng_consumer_stream
*stream
)
952 LTTNG_ASSERT(stream
);
954 switch (the_consumer_data
.type
) {
955 case LTTNG_CONSUMER_KERNEL
:
956 if (stream
->mmap_base
!= NULL
) {
957 const auto ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
964 if (stream
->wait_fd
>= 0) {
965 const auto ret
= close(stream
->wait_fd
);
971 stream
->wait_fd
= -1;
974 if (stream
->chan
->output
== CONSUMER_CHANNEL_SPLICE
) {
975 utils_close_pipe(stream
->splice_pipe
);
979 case LTTNG_CONSUMER32_UST
:
980 case LTTNG_CONSUMER64_UST
:
982 * Special case for the metadata since the wait fd is an internal pipe
983 * polled in the metadata thread.
985 if (stream
->metadata_flag
&& stream
->chan
->monitor
) {
986 const auto rpipe
= stream
->ust_metadata_poll_pipe
[0];
989 * This will stop the channel timer if one and close the write side
990 * of the metadata poll pipe.
992 lttng_ustconsumer_close_metadata(stream
->chan
);
994 const auto ret
= close(rpipe
);
997 PERROR("closing metadata pipe read side");
1000 stream
->ust_metadata_poll_pipe
[0] = -1;
1004 lttng_ustconsumer_del_stream(stream
);
1007 ERR("Unknown consumer_data type");
1013 * Destroy and close a already created stream.
1015 static void destroy_close_stream(struct lttng_consumer_stream
*stream
)
1017 LTTNG_ASSERT(stream
);
1019 DBG("Consumer stream destroy monitored key: %" PRIu64
, stream
->key
);
1021 /* Destroy tracer buffers of the stream. */
1022 consumer_stream_destroy_buffers(stream
);
1023 /* Close down everything including the relayd if one. */
1024 consumer_stream_close_output(stream
);
1028 * Decrement the stream's channel refcount and if down to 0, return the channel
1029 * pointer so it can be destroyed by the caller or NULL if not.
1031 static struct lttng_consumer_channel
*unref_channel(
1032 struct lttng_consumer_stream
*stream
)
1034 struct lttng_consumer_channel
*free_chan
= NULL
;
1036 LTTNG_ASSERT(stream
);
1037 LTTNG_ASSERT(stream
->chan
);
1039 /* Update refcount of channel and see if we need to destroy it. */
1040 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1)
1041 && !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
1042 free_chan
= stream
->chan
;
1049 * Destroy a stream completely. This will delete, close and free the stream.
1050 * Once return, the stream is NO longer usable. Its channel may get destroyed
1051 * if conditions are met for a monitored stream.
1053 * This MUST be called WITHOUT the consumer data and stream lock acquired if
1054 * the stream is in _monitor_ mode else it does not matter.
1056 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
,
1057 struct lttng_ht
*ht
)
1059 LTTNG_ASSERT(stream
);
1061 cds_list_del_init(&stream
->send_node
);
1063 /* Stream is in monitor mode. */
1064 if (stream
->monitor
) {
1065 struct lttng_consumer_channel
*free_chan
= NULL
;
1068 * This means that the stream was successfully removed from the streams
1069 * list of the channel and sent to the right thread managing this
1070 * stream thus being globally visible.
1072 if (stream
->globally_visible
) {
1073 pthread_mutex_lock(&the_consumer_data
.lock
);
1074 pthread_mutex_lock(&stream
->chan
->lock
);
1076 pthread_mutex_lock(&stream
->lock
);
1077 /* Remove every reference of the stream in the consumer. */
1078 consumer_stream_delete(stream
, ht
);
1081 destroy_close_stream(stream
);
1083 /* Update channel's refcount of the stream. */
1084 free_chan
= unref_channel(stream
);
1086 /* Indicates that the consumer data state MUST be updated after this. */
1087 the_consumer_data
.need_update
= 1;
1089 pthread_mutex_unlock(&stream
->lock
);
1090 pthread_mutex_unlock(&stream
->chan
->lock
);
1091 pthread_mutex_unlock(&the_consumer_data
.lock
);
1094 * If the stream is not visible globally, this needs to be done
1095 * outside of the consumer data lock section.
1097 destroy_close_stream(stream
);
1098 free_chan
= unref_channel(stream
);
1102 consumer_del_channel(free_chan
);
1105 destroy_close_stream(stream
);
1108 /* Free stream within a RCU call. */
1109 lttng_trace_chunk_put(stream
->trace_chunk
);
1110 stream
->trace_chunk
= NULL
;
1111 lttng_dynamic_array_reset(&stream
->read_subbuffer_ops
.post_consume_cbs
);
1112 consumer_stream_free(stream
);
1116 * Write index of a specific stream either on the relayd or local disk.
1118 * Return 0 on success or else a negative value.
1120 int consumer_stream_write_index(struct lttng_consumer_stream
*stream
,
1121 struct ctf_packet_index
*element
)
1125 LTTNG_ASSERT(stream
);
1126 LTTNG_ASSERT(element
);
1129 if (stream
->net_seq_idx
!= (uint64_t) -1ULL) {
1130 struct consumer_relayd_sock_pair
*relayd
;
1131 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
1133 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
1134 ret
= relayd_send_index(&relayd
->control_sock
, element
,
1135 stream
->relayd_stream_id
, stream
->next_net_seq_num
- 1);
1138 * Communication error with lttng-relayd,
1139 * perform cleanup now
1141 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
1142 lttng_consumer_cleanup_relayd(relayd
);
1145 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
1147 ERR("Stream %" PRIu64
" relayd ID %" PRIu64
" unknown. Can't write index.",
1148 stream
->key
, stream
->net_seq_idx
);
1152 if (lttng_index_file_write(stream
->index_file
, element
)) {
1167 int consumer_stream_create_output_files(struct lttng_consumer_stream
*stream
,
1171 enum lttng_trace_chunk_status chunk_status
;
1172 const int flags
= O_WRONLY
| O_CREAT
| O_TRUNC
;
1173 const mode_t mode
= S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
;
1174 char stream_path
[LTTNG_PATH_MAX
];
1176 ASSERT_LOCKED(stream
->lock
);
1177 LTTNG_ASSERT(stream
->trace_chunk
);
1179 ret
= utils_stream_file_path(stream
->chan
->pathname
, stream
->name
,
1180 stream
->chan
->tracefile_size
,
1181 stream
->tracefile_count_current
, NULL
,
1182 stream_path
, sizeof(stream_path
));
1187 if (stream
->out_fd
>= 0) {
1188 ret
= close(stream
->out_fd
);
1190 PERROR("Failed to close stream file \"%s\"",
1194 stream
->out_fd
= -1;
1197 DBG("Opening stream output file \"%s\"", stream_path
);
1198 chunk_status
= lttng_trace_chunk_open_file(stream
->trace_chunk
, stream_path
,
1199 flags
, mode
, &stream
->out_fd
, false);
1200 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1201 ERR("Failed to open stream file \"%s\"", stream
->name
);
1206 if (!stream
->metadata_flag
&& (create_index
|| stream
->index_file
)) {
1207 if (stream
->index_file
) {
1208 lttng_index_file_put(stream
->index_file
);
1210 chunk_status
= lttng_index_file_create_from_trace_chunk(
1211 stream
->trace_chunk
,
1212 stream
->chan
->pathname
,
1214 stream
->chan
->tracefile_size
,
1215 stream
->tracefile_count_current
,
1216 CTF_INDEX_MAJOR
, CTF_INDEX_MINOR
,
1217 false, &stream
->index_file
);
1218 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1224 /* Reset current size because we just perform a rotation. */
1225 stream
->tracefile_size_current
= 0;
1226 stream
->out_fd_offset
= 0;
1231 int consumer_stream_rotate_output_files(struct lttng_consumer_stream
*stream
)
1235 stream
->tracefile_count_current
++;
1236 if (stream
->chan
->tracefile_count
> 0) {
1237 stream
->tracefile_count_current
%=
1238 stream
->chan
->tracefile_count
;
1241 DBG("Rotating output files of stream \"%s\"", stream
->name
);
1242 ret
= consumer_stream_create_output_files(stream
, true);
1251 bool consumer_stream_is_deleted(struct lttng_consumer_stream
*stream
)
1254 * This function does not take a const stream since
1255 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1257 LTTNG_ASSERT(stream
);
1258 return cds_lfht_is_node_deleted(&stream
->node
.node
);
1261 static ssize_t
metadata_bucket_flush(
1262 const struct stream_subbuffer
*buffer
, void *data
)
1265 struct lttng_consumer_stream
*stream
= (lttng_consumer_stream
*) data
;
1267 ret
= consumer_stream_consume_mmap(NULL
, stream
, buffer
);
1275 static ssize_t
metadata_bucket_consume(
1276 struct lttng_consumer_local_data
*unused
__attribute__((unused
)),
1277 struct lttng_consumer_stream
*stream
,
1278 const struct stream_subbuffer
*subbuffer
)
1281 enum metadata_bucket_status status
;
1283 status
= metadata_bucket_fill(stream
->metadata_bucket
, subbuffer
);
1285 case METADATA_BUCKET_STATUS_OK
:
1286 /* Return consumed size. */
1287 ret
= subbuffer
->buffer
.buffer
.size
;
1296 int consumer_stream_enable_metadata_bucketization(
1297 struct lttng_consumer_stream
*stream
)
1301 LTTNG_ASSERT(stream
->metadata_flag
);
1302 LTTNG_ASSERT(!stream
->metadata_bucket
);
1303 LTTNG_ASSERT(stream
->chan
->output
== CONSUMER_CHANNEL_MMAP
);
1305 stream
->metadata_bucket
= metadata_bucket_create(
1306 metadata_bucket_flush
, stream
);
1307 if (!stream
->metadata_bucket
) {
1312 stream
->read_subbuffer_ops
.consume_subbuffer
= metadata_bucket_consume
;
1317 void consumer_stream_metadata_set_version(
1318 struct lttng_consumer_stream
*stream
, uint64_t new_version
)
1320 LTTNG_ASSERT(new_version
> stream
->metadata_version
);
1321 stream
->metadata_version
= new_version
;
1322 stream
->reset_metadata_flag
= 1;
1324 if (stream
->metadata_bucket
) {
1325 metadata_bucket_reset(stream
->metadata_bucket
);
1329 int consumer_stream_flush_buffer(struct lttng_consumer_stream
*stream
,
1330 bool producer_active
)
1334 switch (the_consumer_data
.type
) {
1335 case LTTNG_CONSUMER_KERNEL
:
1336 if (producer_active
) {
1337 ret
= kernctl_buffer_flush(stream
->wait_fd
);
1339 ERR("Failed to flush kernel stream");
1343 ret
= kernctl_buffer_flush_empty(stream
->wait_fd
);
1346 * Doing a buffer flush which does not take into
1347 * account empty packets. This is not perfect,
1348 * but required as a fall-back when
1349 * "flush_empty" is not implemented by
1352 ret
= kernctl_buffer_flush(stream
->wait_fd
);
1354 ERR("Failed to flush kernel stream");
1360 case LTTNG_CONSUMER32_UST
:
1361 case LTTNG_CONSUMER64_UST
:
1362 ret
= lttng_ustconsumer_flush_buffer(stream
, (int) producer_active
);
1365 ERR("Unknown consumer_data type");