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
11 #include "consumer-stream.hpp"
13 #include <common/common.hpp>
14 #include <common/consumer/consumer-timer.hpp>
15 #include <common/consumer/consumer.hpp>
16 #include <common/consumer/metadata-bucket.hpp>
17 #include <common/index/index.hpp>
18 #include <common/kernel-consumer/kernel-consumer.hpp>
19 #include <common/kernel-ctl/kernel-ctl.hpp>
20 #include <common/macros.hpp>
21 #include <common/relayd/relayd.hpp>
22 #include <common/urcu.hpp>
23 #include <common/ust-consumer/ust-consumer.hpp>
24 #include <common/utils.hpp>
32 * RCU call to free stream. MUST only be used with call_rcu().
34 static void free_stream_rcu(struct rcu_head
*head
)
36 struct lttng_ht_node_u64
*node
= lttng::utils::container_of(head
, <tng_ht_node_u64::head
);
37 struct lttng_consumer_stream
*stream
=
38 lttng::utils::container_of(node
, <tng_consumer_stream::node
);
40 pthread_mutex_destroy(&stream
->lock
);
44 static void consumer_stream_data_lock_all(struct lttng_consumer_stream
*stream
)
46 pthread_mutex_lock(&stream
->chan
->lock
);
47 pthread_mutex_lock(&stream
->lock
);
50 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream
*stream
)
52 pthread_mutex_unlock(&stream
->lock
);
53 pthread_mutex_unlock(&stream
->chan
->lock
);
56 static void consumer_stream_data_assert_locked_all(struct lttng_consumer_stream
*stream
)
58 ASSERT_LOCKED(stream
->lock
);
59 ASSERT_LOCKED(stream
->chan
->lock
);
62 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream
*stream
)
64 consumer_stream_data_lock_all(stream
);
65 pthread_mutex_lock(&stream
->metadata_rdv_lock
);
68 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream
*stream
)
70 pthread_mutex_unlock(&stream
->metadata_rdv_lock
);
71 consumer_stream_data_unlock_all(stream
);
74 static void consumer_stream_metadata_assert_locked_all(struct lttng_consumer_stream
*stream
)
76 ASSERT_LOCKED(stream
->metadata_rdv_lock
);
77 consumer_stream_data_assert_locked_all(stream
);
80 /* Only used for data streams. */
81 static int consumer_stream_update_stats(struct lttng_consumer_stream
*stream
,
82 const struct stream_subbuffer
*subbuf
)
85 uint64_t sequence_number
;
86 const uint64_t discarded_events
= subbuf
->info
.data
.events_discarded
;
88 if (!subbuf
->info
.data
.sequence_number
.is_set
) {
89 /* Command not supported by the tracer. */
90 sequence_number
= -1ULL;
91 stream
->sequence_number_unavailable
= true;
93 sequence_number
= subbuf
->info
.data
.sequence_number
.value
;
97 * Start the sequence when we extract the first packet in case we don't
98 * start at 0 (for example if a consumer is not connected to the
99 * session immediately after the beginning).
101 if (stream
->last_sequence_number
== -1ULL) {
102 stream
->last_sequence_number
= sequence_number
;
103 } else if (sequence_number
> stream
->last_sequence_number
) {
104 stream
->chan
->lost_packets
+= sequence_number
- stream
->last_sequence_number
- 1;
106 /* seq <= last_sequence_number */
107 ERR("Sequence number inconsistent : prev = %" PRIu64
", current = %" PRIu64
,
108 stream
->last_sequence_number
,
113 stream
->last_sequence_number
= sequence_number
;
115 if (discarded_events
< stream
->last_discarded_events
) {
117 * Overflow has occurred. We assume only one wrap-around
120 stream
->chan
->discarded_events
+= (1ULL << (CAA_BITS_PER_LONG
- 1)) -
121 stream
->last_discarded_events
+ discarded_events
;
123 stream
->chan
->discarded_events
+= discarded_events
- stream
->last_discarded_events
;
125 stream
->last_discarded_events
= discarded_events
;
132 static void ctf_packet_index_populate(struct ctf_packet_index
*index
,
134 const struct stream_subbuffer
*subbuffer
)
136 *index
= (typeof(*index
)){
137 .offset
= htobe64(offset
),
138 .packet_size
= htobe64(subbuffer
->info
.data
.packet_size
),
139 .content_size
= htobe64(subbuffer
->info
.data
.content_size
),
140 .timestamp_begin
= htobe64(subbuffer
->info
.data
.timestamp_begin
),
141 .timestamp_end
= htobe64(subbuffer
->info
.data
.timestamp_end
),
142 .events_discarded
= htobe64(subbuffer
->info
.data
.events_discarded
),
143 .stream_id
= htobe64(subbuffer
->info
.data
.stream_id
),
144 .stream_instance_id
=
145 htobe64(subbuffer
->info
.data
.stream_instance_id
.is_set
?
146 subbuffer
->info
.data
.stream_instance_id
.value
:
148 .packet_seq_num
= htobe64(subbuffer
->info
.data
.sequence_number
.is_set
?
149 subbuffer
->info
.data
.sequence_number
.value
:
154 static ssize_t
consumer_stream_consume_mmap(struct lttng_consumer_local_data
*ctx
155 __attribute__((unused
)),
156 struct lttng_consumer_stream
*stream
,
157 const struct stream_subbuffer
*subbuffer
)
159 const unsigned long padding_size
=
160 subbuffer
->info
.data
.padded_subbuf_size
- subbuffer
->info
.data
.subbuf_size
;
161 const ssize_t written_bytes
= lttng_consumer_on_read_subbuffer_mmap(
162 stream
, &subbuffer
->buffer
.buffer
, padding_size
);
164 if (stream
->net_seq_idx
== -1ULL) {
166 * When writing on disk, check that only the subbuffer (no
167 * padding) was written to disk.
169 if (written_bytes
!= subbuffer
->info
.data
.padded_subbuf_size
) {
170 DBG("Failed to write the entire padded subbuffer on disk (written_bytes: %zd, padded subbuffer size %lu)",
172 subbuffer
->info
.data
.padded_subbuf_size
);
176 * When streaming over the network, check that the entire
177 * subbuffer including padding was successfully written.
179 if (written_bytes
!= subbuffer
->info
.data
.subbuf_size
) {
180 DBG("Failed to write only the subbuffer over the network (written_bytes: %zd, subbuffer size %lu)",
182 subbuffer
->info
.data
.subbuf_size
);
187 * If `lttng_consumer_on_read_subbuffer_mmap()` returned an error, pass
188 * it along to the caller, else return zero.
190 if (written_bytes
< 0) {
191 ERR("Error reading mmap subbuffer: %zd", written_bytes
);
194 return written_bytes
;
197 static ssize_t
consumer_stream_consume_splice(struct lttng_consumer_local_data
*ctx
,
198 struct lttng_consumer_stream
*stream
,
199 const struct stream_subbuffer
*subbuffer
)
201 const ssize_t written_bytes
= lttng_consumer_on_read_subbuffer_splice(
202 ctx
, stream
, subbuffer
->info
.data
.padded_subbuf_size
, 0);
204 if (written_bytes
!= subbuffer
->info
.data
.padded_subbuf_size
) {
205 DBG("Failed to write the entire padded subbuffer (written_bytes: %zd, padded subbuffer size %lu)",
207 subbuffer
->info
.data
.padded_subbuf_size
);
211 * If `lttng_consumer_on_read_subbuffer_splice()` returned an error,
212 * pass it along to the caller, else return zero.
214 if (written_bytes
< 0) {
215 ERR("Error reading splice subbuffer: %zd", written_bytes
);
218 return written_bytes
;
221 static int consumer_stream_send_index(struct lttng_consumer_stream
*stream
,
222 const struct stream_subbuffer
*subbuffer
,
223 struct lttng_consumer_local_data
*ctx
__attribute__((unused
)))
225 off_t packet_offset
= 0;
226 struct ctf_packet_index index
= {};
229 * This is called after consuming the sub-buffer; substract the
230 * effect this sub-buffer from the offset.
232 if (stream
->net_seq_idx
== (uint64_t) -1ULL) {
233 packet_offset
= stream
->out_fd_offset
- subbuffer
->info
.data
.padded_subbuf_size
;
236 ctf_packet_index_populate(&index
, packet_offset
, subbuffer
);
237 return consumer_stream_write_index(stream
, &index
);
241 * Actually do the metadata sync using the given metadata stream.
243 * Return 0 on success else a negative value. ENODATA can be returned also
244 * indicating that there is no metadata available for that stream.
246 static int do_sync_metadata(struct lttng_consumer_stream
*metadata
,
247 struct lttng_consumer_local_data
*ctx
)
250 enum sync_metadata_status status
;
252 LTTNG_ASSERT(metadata
);
253 LTTNG_ASSERT(metadata
->metadata_flag
);
257 * In UST, since we have to write the metadata from the cache packet
258 * by packet, we might need to start this procedure multiple times
259 * until all the metadata from the cache has been extracted.
264 * - Lock the metadata stream
265 * - Check if metadata stream node was deleted before locking.
266 * - if yes, release and return success
267 * - Check if new metadata is ready (flush + snapshot pos)
268 * - If nothing : release and return.
269 * - Lock the metadata_rdv_lock
270 * - Unlock the metadata stream
271 * - cond_wait on metadata_rdv to wait the wakeup from the
273 * - Unlock the metadata_rdv_lock
275 pthread_mutex_lock(&metadata
->lock
);
278 * There is a possibility that we were able to acquire a reference on the
279 * stream from the RCU hash table but between then and now, the node might
280 * have been deleted just before the lock is acquired. Thus, after locking,
281 * we make sure the metadata node has not been deleted which means that the
282 * buffers are closed.
284 * In that case, there is no need to sync the metadata hence returning a
285 * success return code.
287 ret
= cds_lfht_is_node_deleted(&metadata
->node
.node
);
290 goto end_unlock_mutex
;
294 case LTTNG_CONSUMER_KERNEL
:
296 * Empty the metadata cache and flush the current stream.
298 status
= lttng_kconsumer_sync_metadata(metadata
);
300 case LTTNG_CONSUMER32_UST
:
301 case LTTNG_CONSUMER64_UST
:
303 * Ask the sessiond if we have new metadata waiting and update the
304 * consumer metadata cache.
306 status
= lttng_ustconsumer_sync_metadata(ctx
, metadata
);
313 case SYNC_METADATA_STATUS_NEW_DATA
:
315 case SYNC_METADATA_STATUS_NO_DATA
:
317 goto end_unlock_mutex
;
318 case SYNC_METADATA_STATUS_ERROR
:
320 goto end_unlock_mutex
;
326 * At this point, new metadata have been flushed, so we wait on the
327 * rendez-vous point for the metadata thread to wake us up when it
328 * finishes consuming the metadata and continue execution.
331 pthread_mutex_lock(&metadata
->metadata_rdv_lock
);
334 * Release metadata stream lock so the metadata thread can process it.
336 pthread_mutex_unlock(&metadata
->lock
);
339 * Wait on the rendez-vous point. Once woken up, it means the metadata was
340 * consumed and thus synchronization is achieved.
342 pthread_cond_wait(&metadata
->metadata_rdv
, &metadata
->metadata_rdv_lock
);
343 pthread_mutex_unlock(&metadata
->metadata_rdv_lock
);
344 } while (status
== SYNC_METADATA_STATUS_NEW_DATA
);
350 pthread_mutex_unlock(&metadata
->lock
);
355 * Synchronize the metadata using a given session ID. A successful acquisition
356 * of a metadata stream will trigger a request to the session daemon and a
357 * snapshot so the metadata thread can consume it.
359 * This function call is a rendez-vous point between the metadata thread and
362 * Return 0 on success or else a negative value.
364 int consumer_stream_sync_metadata(struct lttng_consumer_local_data
*ctx
, uint64_t session_id
)
370 /* Search the metadata associated with the session id of the given stream. */
371 for (auto *stream
: lttng::urcu::lfht_filtered_iteration_adapter
<
372 lttng_consumer_stream
,
373 decltype(lttng_consumer_stream::node_session_id
),
374 <tng_consumer_stream::node_session_id
,
375 std::uint64_t>(*the_consumer_data
.stream_list_ht
->ht
,
377 the_consumer_data
.stream_list_ht
->hash_fct(&session_id
,
379 the_consumer_data
.stream_list_ht
->match_fct
)) {
380 if (!stream
->metadata_flag
) {
384 ret
= do_sync_metadata(stream
, ctx
);
391 * Force return code to 0 (success) since ret might be ENODATA for instance
392 * which is not an error but rather that we should come back.
400 static int consumer_stream_sync_metadata_index(struct lttng_consumer_stream
*stream
,
401 const struct stream_subbuffer
*subbuffer
,
402 struct lttng_consumer_local_data
*ctx
)
404 bool missed_metadata_flush
;
407 /* Block until all the metadata is sent. */
408 pthread_mutex_lock(&stream
->metadata_timer_lock
);
409 LTTNG_ASSERT(!stream
->missed_metadata_flush
);
410 stream
->waiting_on_metadata
= true;
411 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
413 ret
= consumer_stream_sync_metadata(ctx
, stream
->session_id
);
415 pthread_mutex_lock(&stream
->metadata_timer_lock
);
416 stream
->waiting_on_metadata
= false;
417 missed_metadata_flush
= stream
->missed_metadata_flush
;
418 if (missed_metadata_flush
) {
419 stream
->missed_metadata_flush
= false;
421 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
426 ret
= consumer_stream_send_index(stream
, subbuffer
, ctx
);
428 * Send the live inactivity beacon to handle the situation where
429 * the live timer is prevented from sampling this stream
430 * because the stream lock was being held while this stream is
431 * waiting on metadata. This ensures live viewer progress in the
432 * unlikely scenario where a live timer would be prevented from
433 * locking a stream lock repeatedly due to a steady flow of
434 * incoming metadata, for a stream which is mostly inactive.
436 * It is important to send the inactivity beacon packet to
437 * relayd _after_ sending the index associated with the data
438 * that was just sent, otherwise this can cause live viewers to
439 * observe timestamps going backwards between an inactivity
440 * beacon and a following trace packet.
442 if (missed_metadata_flush
) {
443 (void) stream
->read_subbuffer_ops
.send_live_beacon(stream
);
450 * Check if the local version of the metadata stream matches with the version
451 * of the metadata stream in the kernel. If it was updated, set the reset flag
454 static int metadata_stream_check_version(struct lttng_consumer_stream
*stream
,
455 const struct stream_subbuffer
*subbuffer
)
457 if (stream
->metadata_version
== subbuffer
->info
.metadata
.version
) {
461 DBG("New metadata version detected");
462 consumer_stream_metadata_set_version(stream
, subbuffer
->info
.metadata
.version
);
464 if (stream
->read_subbuffer_ops
.reset_metadata
) {
465 stream
->read_subbuffer_ops
.reset_metadata(stream
);
472 static bool stream_is_rotating_to_null_chunk(const struct lttng_consumer_stream
*stream
)
474 bool rotating_to_null_chunk
= false;
476 if (stream
->rotate_position
== -1ULL) {
477 /* No rotation ongoing. */
481 if (stream
->trace_chunk
== stream
->chan
->trace_chunk
|| !stream
->chan
->trace_chunk
) {
482 rotating_to_null_chunk
= true;
485 return rotating_to_null_chunk
;
488 enum consumer_stream_open_packet_status
489 consumer_stream_open_packet(struct lttng_consumer_stream
*stream
)
492 enum consumer_stream_open_packet_status status
;
493 unsigned long produced_pos_before
, produced_pos_after
;
495 ret
= lttng_consumer_sample_snapshot_positions(stream
);
497 ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
498 ", channel name = %s, session id = %" PRIu64
,
501 stream
->chan
->session_id
);
502 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
506 ret
= lttng_consumer_get_produced_snapshot(stream
, &produced_pos_before
);
508 ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
509 ", channel name = %s, session id = %" PRIu64
,
512 stream
->chan
->session_id
);
513 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
517 ret
= consumer_stream_flush_buffer(stream
, false);
519 ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
520 ", channel name = %s, session id = %" PRIu64
,
523 stream
->chan
->session_id
);
524 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
528 ret
= lttng_consumer_sample_snapshot_positions(stream
);
530 ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
531 ", channel name = %s, session id = %" PRIu64
,
534 stream
->chan
->session_id
);
535 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
539 ret
= lttng_consumer_get_produced_snapshot(stream
, &produced_pos_after
);
541 ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
542 ", channel name = %s, session id = %" PRIu64
,
545 stream
->chan
->session_id
);
546 status
= CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
;
551 * Determine if the flush had an effect by comparing the produced
552 * positons before and after the flush.
554 status
= produced_pos_before
!= produced_pos_after
?
555 CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
:
556 CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE
;
557 if (status
== CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
) {
558 stream
->opened_packet_in_current_trace_chunk
= true;
566 * An attempt to open a new packet is performed after a rotation completes to
567 * get a begin timestamp as close as possible to the rotation point.
569 * However, that initial attempt at opening a packet can fail due to a full
570 * ring-buffer. In that case, a second attempt is performed after consuming
571 * a packet since that will have freed enough space in the ring-buffer.
573 static int post_consume_open_new_packet(struct lttng_consumer_stream
*stream
,
574 const struct stream_subbuffer
*subbuffer
575 __attribute__((unused
)),
576 struct lttng_consumer_local_data
*ctx
577 __attribute__((unused
)))
581 if (!stream
->opened_packet_in_current_trace_chunk
&& stream
->trace_chunk
&&
582 !stream_is_rotating_to_null_chunk(stream
)) {
583 const enum consumer_stream_open_packet_status status
=
584 consumer_stream_open_packet(stream
);
587 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED
:
588 DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64
589 ", channel name = %s, session id = %" PRIu64
,
592 stream
->chan
->session_id
);
593 stream
->opened_packet_in_current_trace_chunk
= true;
595 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE
:
597 * Can't open a packet as there is no space left.
598 * This means that new events were produced, resulting
599 * in a packet being opened, which is what we want
602 DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
603 ", channel name = %s, session id = %" PRIu64
,
606 stream
->chan
->session_id
);
607 stream
->opened_packet_in_current_trace_chunk
= true;
609 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR
:
610 /* Logged by callee. */
617 stream
->opened_packet_in_current_trace_chunk
= true;
624 struct lttng_consumer_stream
*consumer_stream_create(struct lttng_consumer_channel
*channel
,
625 uint64_t channel_key
,
627 const char *channel_name
,
630 struct lttng_trace_chunk
*trace_chunk
,
633 enum consumer_channel_type type
,
634 unsigned int monitor
)
637 struct lttng_consumer_stream
*stream
;
638 const lttng::urcu::read_lock_guard read_lock
;
640 stream
= zmalloc
<lttng_consumer_stream
>();
641 if (stream
== nullptr) {
642 PERROR("malloc struct lttng_consumer_stream");
647 if (trace_chunk
&& !lttng_trace_chunk_get(trace_chunk
)) {
648 ERR("Failed to acquire trace chunk reference during the creation of a stream");
653 stream
->send_node
= CDS_LIST_HEAD_INIT(stream
->send_node
);
654 stream
->chan
= channel
;
655 stream
->key
= stream_key
;
656 stream
->trace_chunk
= trace_chunk
;
658 stream
->out_fd_offset
= 0;
659 stream
->output_written
= 0;
660 stream
->net_seq_idx
= relayd_id
;
661 stream
->session_id
= session_id
;
662 stream
->monitor
= monitor
;
663 stream
->endpoint_status
= CONSUMER_ENDPOINT_ACTIVE
;
664 stream
->index_file
= nullptr;
665 stream
->last_sequence_number
= -1ULL;
666 stream
->rotate_position
= -1ULL;
667 /* Buffer is created with an open packet. */
668 stream
->opened_packet_in_current_trace_chunk
= true;
669 pthread_mutex_init(&stream
->lock
, nullptr);
670 pthread_mutex_init(&stream
->metadata_timer_lock
, nullptr);
672 /* If channel is the metadata, flag this stream as metadata. */
673 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
674 stream
->metadata_flag
= 1;
675 /* Metadata is flat out. */
676 strncpy(stream
->name
, DEFAULT_METADATA_NAME
, sizeof(stream
->name
));
677 /* Live rendez-vous point. */
678 pthread_cond_init(&stream
->metadata_rdv
, nullptr);
679 pthread_mutex_init(&stream
->metadata_rdv_lock
, nullptr);
681 /* Format stream name to <channel_name>_<cpu_number> */
682 ret
= snprintf(stream
->name
, sizeof(stream
->name
), "%s_%d", channel_name
, cpu
);
684 PERROR("snprintf stream name");
689 switch (channel
->output
) {
690 case CONSUMER_CHANNEL_SPLICE
:
691 stream
->output
= LTTNG_EVENT_SPLICE
;
692 ret
= utils_create_pipe(stream
->splice_pipe
);
697 case CONSUMER_CHANNEL_MMAP
:
698 stream
->output
= LTTNG_EVENT_MMAP
;
704 /* Key is always the wait_fd for streams. */
705 lttng_ht_node_init_u64(&stream
->node
, stream
->key
);
707 /* Init node per channel id key */
708 lttng_ht_node_init_u64(&stream
->node_channel_id
, channel_key
);
710 /* Init session id node with the stream session id */
711 lttng_ht_node_init_u64(&stream
->node_session_id
, stream
->session_id
);
713 DBG3("Allocated stream %s (key %" PRIu64
", chan_key %" PRIu64
" relayd_id %" PRIu64
714 ", session_id %" PRIu64
,
721 lttng_dynamic_array_init(
722 &stream
->read_subbuffer_ops
.post_consume_cbs
, sizeof(post_consume_cb
), nullptr);
724 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
725 stream
->read_subbuffer_ops
.lock
= consumer_stream_metadata_lock_all
;
726 stream
->read_subbuffer_ops
.unlock
= consumer_stream_metadata_unlock_all
;
727 stream
->read_subbuffer_ops
.assert_locked
=
728 consumer_stream_metadata_assert_locked_all
;
729 stream
->read_subbuffer_ops
.pre_consume_subbuffer
= metadata_stream_check_version
;
731 const post_consume_cb post_consume_index_op
= channel
->is_live
?
732 consumer_stream_sync_metadata_index
:
733 consumer_stream_send_index
;
734 const post_consume_cb post_consume_open_new_packet_
= post_consume_open_new_packet
;
736 ret
= lttng_dynamic_array_add_element(&stream
->read_subbuffer_ops
.post_consume_cbs
,
737 &post_consume_index_op
);
739 PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
743 ret
= lttng_dynamic_array_add_element(&stream
->read_subbuffer_ops
.post_consume_cbs
,
744 &post_consume_open_new_packet_
);
746 PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
750 stream
->read_subbuffer_ops
.lock
= consumer_stream_data_lock_all
;
751 stream
->read_subbuffer_ops
.unlock
= consumer_stream_data_unlock_all
;
752 stream
->read_subbuffer_ops
.assert_locked
= consumer_stream_data_assert_locked_all
;
753 stream
->read_subbuffer_ops
.pre_consume_subbuffer
= consumer_stream_update_stats
;
756 if (channel
->output
== CONSUMER_CHANNEL_MMAP
) {
757 stream
->read_subbuffer_ops
.consume_subbuffer
= consumer_stream_consume_mmap
;
759 stream
->read_subbuffer_ops
.consume_subbuffer
= consumer_stream_consume_splice
;
765 lttng_trace_chunk_put(stream
->trace_chunk
);
766 lttng_dynamic_array_reset(&stream
->read_subbuffer_ops
.post_consume_cbs
);
776 * Close stream on the relayd side. This call can destroy a relayd if the
777 * conditions are met.
779 * A RCU read side lock MUST be acquired if the relayd object was looked up in
780 * a hash table before calling this.
782 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
783 struct consumer_relayd_sock_pair
*relayd
)
787 LTTNG_ASSERT(stream
);
788 LTTNG_ASSERT(relayd
);
790 if (stream
->sent_to_relayd
) {
791 uatomic_dec(&relayd
->refcount
);
792 LTTNG_ASSERT(uatomic_read(&relayd
->refcount
) >= 0);
795 /* Closing streams requires to lock the control socket. */
796 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
797 ret
= relayd_send_close_stream(
798 &relayd
->control_sock
, stream
->relayd_stream_id
, stream
->next_net_seq_num
- 1);
799 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
801 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64
".",
802 relayd
->net_seq_idx
);
803 lttng_consumer_cleanup_relayd(relayd
);
806 /* Both conditions are met, we destroy the relayd. */
807 if (uatomic_read(&relayd
->refcount
) == 0 && uatomic_read(&relayd
->destroy_flag
)) {
808 consumer_destroy_relayd(relayd
);
810 stream
->net_seq_idx
= (uint64_t) -1ULL;
811 stream
->sent_to_relayd
= 0;
815 * Close stream's file descriptors and, if needed, close stream also on the
818 * The consumer data lock MUST be acquired.
819 * The stream lock MUST be acquired.
821 void consumer_stream_close_output(struct lttng_consumer_stream
*stream
)
823 struct consumer_relayd_sock_pair
*relayd
;
825 LTTNG_ASSERT(stream
);
827 /* Close output fd. Could be a socket or local file at this point. */
828 if (stream
->out_fd
>= 0) {
829 const auto ret
= close(stream
->out_fd
);
831 PERROR("Failed to close stream output file descriptor");
837 if (stream
->index_file
) {
838 lttng_index_file_put(stream
->index_file
);
839 stream
->index_file
= nullptr;
842 lttng_trace_chunk_put(stream
->trace_chunk
);
843 stream
->trace_chunk
= nullptr;
845 /* Check and cleanup relayd if needed. */
846 const lttng::urcu::read_lock_guard read_lock
;
847 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
848 if (relayd
!= nullptr) {
849 consumer_stream_relayd_close(stream
, relayd
);
850 stream
->net_seq_idx
= -1ULL;
855 * Delete the stream from all possible hash tables.
857 * The consumer data lock MUST be acquired.
858 * The stream lock MUST be acquired.
860 void consumer_stream_delete(struct lttng_consumer_stream
*stream
, struct lttng_ht
*ht
)
863 struct lttng_ht_iter iter
;
865 LTTNG_ASSERT(stream
);
866 /* Should NEVER be called not in monitor mode. */
867 LTTNG_ASSERT(stream
->chan
->monitor
);
869 const lttng::urcu::read_lock_guard read_lock
;
872 iter
.iter
.node
= &stream
->node
.node
;
873 ret
= lttng_ht_del(ht
, &iter
);
877 /* Delete from stream per channel ID hash table. */
878 iter
.iter
.node
= &stream
->node_channel_id
.node
;
880 * The returned value is of no importance. Even if the node is NOT in the
881 * hash table, we continue since we may have been called by a code path
882 * that did not add the stream to a (all) hash table. Same goes for the
883 * next call ht del call.
885 (void) lttng_ht_del(the_consumer_data
.stream_per_chan_id_ht
, &iter
);
887 /* Delete from the global stream list. */
888 iter
.iter
.node
= &stream
->node_session_id
.node
;
889 /* See the previous ht del on why we ignore the returned value. */
890 (void) lttng_ht_del(the_consumer_data
.stream_list_ht
, &iter
);
892 if (!stream
->metadata_flag
) {
893 /* Decrement the stream count of the global consumer data. */
894 LTTNG_ASSERT(the_consumer_data
.stream_count
> 0);
895 the_consumer_data
.stream_count
--;
900 * Free the given stream within a RCU call.
902 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
904 LTTNG_ASSERT(stream
);
906 metadata_bucket_destroy(stream
->metadata_bucket
);
907 call_rcu(&stream
->node
.head
, free_stream_rcu
);
911 * Destroy the stream's buffers of the tracer.
913 void consumer_stream_destroy_buffers(struct lttng_consumer_stream
*stream
)
915 LTTNG_ASSERT(stream
);
917 switch (the_consumer_data
.type
) {
918 case LTTNG_CONSUMER_KERNEL
:
919 if (stream
->mmap_base
!= nullptr) {
920 const auto ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
927 if (stream
->wait_fd
>= 0) {
928 const auto ret
= close(stream
->wait_fd
);
934 stream
->wait_fd
= -1;
937 if (stream
->chan
->output
== CONSUMER_CHANNEL_SPLICE
) {
938 utils_close_pipe(stream
->splice_pipe
);
942 case LTTNG_CONSUMER32_UST
:
943 case LTTNG_CONSUMER64_UST
:
945 * Special case for the metadata since the wait fd is an internal pipe
946 * polled in the metadata thread.
948 if (stream
->metadata_flag
&& stream
->chan
->monitor
) {
949 const auto rpipe
= stream
->ust_metadata_poll_pipe
[0];
952 * This will stop the channel timer if one and close the write side
953 * of the metadata poll pipe.
955 lttng_ustconsumer_close_metadata(stream
->chan
);
957 const auto ret
= close(rpipe
);
960 PERROR("closing metadata pipe read side");
963 stream
->ust_metadata_poll_pipe
[0] = -1;
967 lttng_ustconsumer_del_stream(stream
);
970 ERR("Unknown consumer_data type");
976 * Destroy and close a already created stream.
978 static void destroy_close_stream(struct lttng_consumer_stream
*stream
)
980 LTTNG_ASSERT(stream
);
982 DBG("Consumer stream destroy monitored key: %" PRIu64
, stream
->key
);
984 /* Destroy tracer buffers of the stream. */
985 consumer_stream_destroy_buffers(stream
);
986 /* Close down everything including the relayd if one. */
987 consumer_stream_close_output(stream
);
991 * Decrement the stream's channel refcount and if down to 0, return the channel
992 * pointer so it can be destroyed by the caller or NULL if not.
994 static struct lttng_consumer_channel
*unref_channel(struct lttng_consumer_stream
*stream
)
996 struct lttng_consumer_channel
*free_chan
= nullptr;
998 LTTNG_ASSERT(stream
);
999 LTTNG_ASSERT(stream
->chan
);
1001 /* Update refcount of channel and see if we need to destroy it. */
1002 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1) &&
1003 !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
1004 free_chan
= stream
->chan
;
1011 * Destroy a stream completely. This will delete, close and free the stream.
1012 * Once return, the stream is NO longer usable. Its channel may get destroyed
1013 * if conditions are met for a monitored stream.
1015 * This MUST be called WITHOUT the consumer data and stream lock acquired if
1016 * the stream is in _monitor_ mode else it does not matter.
1018 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
, struct lttng_ht
*ht
)
1020 LTTNG_ASSERT(stream
);
1022 cds_list_del_init(&stream
->send_node
);
1024 /* Stream is in monitor mode. */
1025 if (stream
->monitor
) {
1026 struct lttng_consumer_channel
*free_chan
= nullptr;
1029 * This means that the stream was successfully removed from the streams
1030 * list of the channel and sent to the right thread managing this
1031 * stream thus being globally visible.
1033 if (stream
->globally_visible
) {
1034 pthread_mutex_lock(&the_consumer_data
.lock
);
1035 pthread_mutex_lock(&stream
->chan
->lock
);
1037 pthread_mutex_lock(&stream
->lock
);
1038 /* Remove every reference of the stream in the consumer. */
1039 consumer_stream_delete(stream
, ht
);
1041 destroy_close_stream(stream
);
1043 /* Update channel's refcount of the stream. */
1044 free_chan
= unref_channel(stream
);
1046 /* Indicates that the consumer data state MUST be updated after this. */
1047 the_consumer_data
.need_update
= 1;
1049 pthread_mutex_unlock(&stream
->lock
);
1050 pthread_mutex_unlock(&stream
->chan
->lock
);
1051 pthread_mutex_unlock(&the_consumer_data
.lock
);
1054 * If the stream is not visible globally, this needs to be done
1055 * outside of the consumer data lock section.
1057 destroy_close_stream(stream
);
1058 free_chan
= unref_channel(stream
);
1062 consumer_del_channel(free_chan
);
1065 destroy_close_stream(stream
);
1068 /* Free stream within a RCU call. */
1069 lttng_trace_chunk_put(stream
->trace_chunk
);
1070 stream
->trace_chunk
= nullptr;
1071 lttng_dynamic_array_reset(&stream
->read_subbuffer_ops
.post_consume_cbs
);
1072 consumer_stream_free(stream
);
1076 * Write index of a specific stream either on the relayd or local disk.
1078 * Return 0 on success or else a negative value.
1080 int consumer_stream_write_index(struct lttng_consumer_stream
*stream
,
1081 struct ctf_packet_index
*element
)
1085 LTTNG_ASSERT(stream
);
1086 LTTNG_ASSERT(element
);
1088 const lttng::urcu::read_lock_guard read_lock
;
1089 if (stream
->net_seq_idx
!= (uint64_t) -1ULL) {
1090 struct consumer_relayd_sock_pair
*relayd
;
1091 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
1093 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
1094 ret
= relayd_send_index(&relayd
->control_sock
,
1096 stream
->relayd_stream_id
,
1097 stream
->next_net_seq_num
- 1);
1100 * Communication error with lttng-relayd,
1101 * perform cleanup now
1103 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64
".",
1104 relayd
->net_seq_idx
);
1105 lttng_consumer_cleanup_relayd(relayd
);
1108 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
1110 ERR("Stream %" PRIu64
" relayd ID %" PRIu64
" unknown. Can't write index.",
1112 stream
->net_seq_idx
);
1116 if (lttng_index_file_write(stream
->index_file
, element
)) {
1130 int consumer_stream_create_output_files(struct lttng_consumer_stream
*stream
, bool create_index
)
1133 enum lttng_trace_chunk_status chunk_status
;
1134 const int flags
= O_WRONLY
| O_CREAT
| O_TRUNC
;
1135 const mode_t mode
= S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
;
1136 char stream_path
[LTTNG_PATH_MAX
];
1138 ASSERT_LOCKED(stream
->lock
);
1139 LTTNG_ASSERT(stream
->trace_chunk
);
1141 ret
= utils_stream_file_path(stream
->chan
->pathname
,
1143 stream
->chan
->tracefile_size
,
1144 stream
->tracefile_count_current
,
1147 sizeof(stream_path
));
1152 if (stream
->out_fd
>= 0) {
1153 ret
= close(stream
->out_fd
);
1155 PERROR("Failed to close stream file \"%s\"", stream
->name
);
1158 stream
->out_fd
= -1;
1161 DBG("Opening stream output file \"%s\"", stream_path
);
1162 chunk_status
= lttng_trace_chunk_open_file(
1163 stream
->trace_chunk
, stream_path
, flags
, mode
, &stream
->out_fd
, false);
1164 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1165 ERR("Failed to open stream file \"%s\"", stream
->name
);
1170 if (!stream
->metadata_flag
&& (create_index
|| stream
->index_file
)) {
1171 if (stream
->index_file
) {
1172 lttng_index_file_put(stream
->index_file
);
1175 lttng_index_file_create_from_trace_chunk(stream
->trace_chunk
,
1176 stream
->chan
->pathname
,
1178 stream
->chan
->tracefile_size
,
1179 stream
->tracefile_count_current
,
1183 &stream
->index_file
);
1184 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1190 /* Reset current size because we just perform a rotation. */
1191 stream
->tracefile_size_current
= 0;
1192 stream
->out_fd_offset
= 0;
1197 int consumer_stream_rotate_output_files(struct lttng_consumer_stream
*stream
)
1201 stream
->tracefile_count_current
++;
1202 if (stream
->chan
->tracefile_count
> 0) {
1203 stream
->tracefile_count_current
%= stream
->chan
->tracefile_count
;
1206 DBG("Rotating output files of stream \"%s\"", stream
->name
);
1207 ret
= consumer_stream_create_output_files(stream
, true);
1216 bool consumer_stream_is_deleted(struct lttng_consumer_stream
*stream
)
1219 * This function does not take a const stream since
1220 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1222 LTTNG_ASSERT(stream
);
1223 return cds_lfht_is_node_deleted(&stream
->node
.node
);
1226 static ssize_t
metadata_bucket_flush(const struct stream_subbuffer
*buffer
, void *data
)
1229 struct lttng_consumer_stream
*stream
= (lttng_consumer_stream
*) data
;
1231 ret
= consumer_stream_consume_mmap(nullptr, stream
, buffer
);
1239 static ssize_t
metadata_bucket_consume(struct lttng_consumer_local_data
*unused
1240 __attribute__((unused
)),
1241 struct lttng_consumer_stream
*stream
,
1242 const struct stream_subbuffer
*subbuffer
)
1245 enum metadata_bucket_status status
;
1247 status
= metadata_bucket_fill(stream
->metadata_bucket
, subbuffer
);
1249 case METADATA_BUCKET_STATUS_OK
:
1250 /* Return consumed size. */
1251 ret
= subbuffer
->buffer
.buffer
.size
;
1260 int consumer_stream_enable_metadata_bucketization(struct lttng_consumer_stream
*stream
)
1264 LTTNG_ASSERT(stream
->metadata_flag
);
1265 LTTNG_ASSERT(!stream
->metadata_bucket
);
1266 LTTNG_ASSERT(stream
->chan
->output
== CONSUMER_CHANNEL_MMAP
);
1268 stream
->metadata_bucket
= metadata_bucket_create(metadata_bucket_flush
, stream
);
1269 if (!stream
->metadata_bucket
) {
1274 stream
->read_subbuffer_ops
.consume_subbuffer
= metadata_bucket_consume
;
1279 void consumer_stream_metadata_set_version(struct lttng_consumer_stream
*stream
,
1280 uint64_t new_version
)
1282 LTTNG_ASSERT(new_version
> stream
->metadata_version
);
1283 stream
->metadata_version
= new_version
;
1284 stream
->reset_metadata_flag
= 1;
1286 if (stream
->metadata_bucket
) {
1287 metadata_bucket_reset(stream
->metadata_bucket
);
1291 int consumer_stream_flush_buffer(struct lttng_consumer_stream
*stream
, bool producer_active
)
1295 switch (the_consumer_data
.type
) {
1296 case LTTNG_CONSUMER_KERNEL
:
1297 if (producer_active
) {
1298 ret
= kernctl_buffer_flush(stream
->wait_fd
);
1300 ERR("Failed to flush kernel stream");
1304 ret
= kernctl_buffer_flush_empty(stream
->wait_fd
);
1307 * Doing a buffer flush which does not take into
1308 * account empty packets. This is not perfect,
1309 * but required as a fall-back when
1310 * "flush_empty" is not implemented by
1313 ret
= kernctl_buffer_flush(stream
->wait_fd
);
1315 ERR("Failed to flush kernel stream");
1321 case LTTNG_CONSUMER32_UST
:
1322 case LTTNG_CONSUMER64_UST
:
1323 ret
= lttng_ustconsumer_flush_buffer(stream
, (int) producer_active
);
1326 ERR("Unknown consumer_data type");