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
16 #include <common/common.h>
17 #include <common/consumer/consumer-timer.h>
18 #include <common/consumer/consumer.h>
19 #include <common/consumer/metadata-bucket.h>
20 #include <common/index/index.h>
21 #include <common/kernel-consumer/kernel-consumer.h>
22 #include <common/macros.h>
23 #include <common/relayd/relayd.h>
24 #include <common/ust-consumer/ust-consumer.h>
25 #include <common/utils.h>
27 #include "consumer-stream.h"
30 * RCU call to free stream. MUST only be used with call_rcu().
32 static void free_stream_rcu(struct rcu_head
*head
)
34 struct lttng_ht_node_u64
*node
=
35 caa_container_of(head
, struct lttng_ht_node_u64
, head
);
36 struct lttng_consumer_stream
*stream
=
37 caa_container_of(node
, struct lttng_consumer_stream
, node
);
39 pthread_mutex_destroy(&stream
->lock
);
43 static void consumer_stream_data_lock_all(struct lttng_consumer_stream
*stream
)
45 pthread_mutex_lock(&stream
->chan
->lock
);
46 pthread_mutex_lock(&stream
->lock
);
49 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream
*stream
)
51 pthread_mutex_unlock(&stream
->lock
);
52 pthread_mutex_unlock(&stream
->chan
->lock
);
55 static void consumer_stream_data_assert_locked_all(struct lttng_consumer_stream
*stream
)
57 ASSERT_LOCKED(stream
->lock
);
58 ASSERT_LOCKED(stream
->chan
->lock
);
61 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream
*stream
)
63 consumer_stream_data_lock_all(stream
);
64 pthread_mutex_lock(&stream
->metadata_rdv_lock
);
67 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream
*stream
)
69 pthread_mutex_unlock(&stream
->metadata_rdv_lock
);
70 consumer_stream_data_unlock_all(stream
);
73 static void consumer_stream_metadata_assert_locked_all(struct lttng_consumer_stream
*stream
)
75 ASSERT_LOCKED(stream
->metadata_rdv_lock
);
76 consumer_stream_data_assert_locked_all(stream
);
79 /* Only used for data streams. */
80 static int consumer_stream_update_stats(struct lttng_consumer_stream
*stream
,
81 const struct stream_subbuffer
*subbuf
)
84 uint64_t sequence_number
;
85 const uint64_t discarded_events
= subbuf
->info
.data
.events_discarded
;
87 if (!subbuf
->info
.data
.sequence_number
.is_set
) {
88 /* Command not supported by the tracer. */
89 sequence_number
= -1ULL;
90 stream
->sequence_number_unavailable
= true;
92 sequence_number
= subbuf
->info
.data
.sequence_number
.value
;
96 * Start the sequence when we extract the first packet in case we don't
97 * start at 0 (for example if a consumer is not connected to the
98 * session immediately after the beginning).
100 if (stream
->last_sequence_number
== -1ULL) {
101 stream
->last_sequence_number
= sequence_number
;
102 } else if (sequence_number
> stream
->last_sequence_number
) {
103 stream
->chan
->lost_packets
+= sequence_number
-
104 stream
->last_sequence_number
- 1;
106 /* seq <= last_sequence_number */
107 ERR("Sequence number inconsistent : prev = %" PRIu64
108 ", current = %" PRIu64
,
109 stream
->last_sequence_number
, 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
+=
121 (1ULL << (CAA_BITS_PER_LONG
- 1)) -
122 stream
->last_discarded_events
+
125 stream
->chan
->discarded_events
+= discarded_events
-
126 stream
->last_discarded_events
;
128 stream
->last_discarded_events
= discarded_events
;
136 void ctf_packet_index_populate(struct ctf_packet_index
*index
,
137 off_t offset
, const struct stream_subbuffer
*subbuffer
)
139 *index
= (typeof(*index
)){
140 .offset
= htobe64(offset
),
141 .packet_size
= htobe64(subbuffer
->info
.data
.packet_size
),
142 .content_size
= htobe64(subbuffer
->info
.data
.content_size
),
143 .timestamp_begin
= htobe64(
144 subbuffer
->info
.data
.timestamp_begin
),
145 .timestamp_end
= htobe64(
146 subbuffer
->info
.data
.timestamp_end
),
147 .events_discarded
= htobe64(
148 subbuffer
->info
.data
.events_discarded
),
149 .stream_id
= htobe64(subbuffer
->info
.data
.stream_id
),
150 .stream_instance_id
= htobe64(
151 subbuffer
->info
.data
.stream_instance_id
.is_set
?
152 subbuffer
->info
.data
.stream_instance_id
.value
: -1ULL),
153 .packet_seq_num
= htobe64(
154 subbuffer
->info
.data
.sequence_number
.is_set
?
155 subbuffer
->info
.data
.sequence_number
.value
: -1ULL),
159 static ssize_t
consumer_stream_consume_mmap(
160 struct lttng_consumer_local_data
*ctx
,
161 struct lttng_consumer_stream
*stream
,
162 const struct stream_subbuffer
*subbuffer
)
164 const unsigned long padding_size
=
165 subbuffer
->info
.data
.padded_subbuf_size
-
166 subbuffer
->info
.data
.subbuf_size
;
167 const ssize_t written_bytes
= lttng_consumer_on_read_subbuffer_mmap(
168 stream
, &subbuffer
->buffer
.buffer
, padding_size
);
170 if (stream
->net_seq_idx
== -1ULL) {
172 * When writing on disk, check that only the subbuffer (no
173 * padding) was written to disk.
175 if (written_bytes
!= subbuffer
->info
.data
.padded_subbuf_size
) {
176 DBG("Failed to write the entire padded subbuffer on disk (written_bytes: %zd, padded subbuffer size %lu)",
178 subbuffer
->info
.data
.padded_subbuf_size
);
182 * When streaming over the network, check that the entire
183 * subbuffer including padding was successfully written.
185 if (written_bytes
!= subbuffer
->info
.data
.subbuf_size
) {
186 DBG("Failed to write only the subbuffer over the network (written_bytes: %zd, subbuffer size %lu)",
188 subbuffer
->info
.data
.subbuf_size
);
193 * If `lttng_consumer_on_read_subbuffer_mmap()` returned an error, pass
194 * it along to the caller, else return zero.
196 if (written_bytes
< 0) {
197 ERR("Error reading mmap subbuffer: %zd", written_bytes
);
200 return written_bytes
;
203 static ssize_t
consumer_stream_consume_splice(
204 struct lttng_consumer_local_data
*ctx
,
205 struct lttng_consumer_stream
*stream
,
206 const struct stream_subbuffer
*subbuffer
)
208 const ssize_t written_bytes
= lttng_consumer_on_read_subbuffer_splice(
209 ctx
, stream
, subbuffer
->info
.data
.padded_subbuf_size
, 0);
211 if (written_bytes
!= subbuffer
->info
.data
.padded_subbuf_size
) {
212 DBG("Failed to write the entire padded subbuffer (written_bytes: %zd, padded subbuffer size %lu)",
214 subbuffer
->info
.data
.padded_subbuf_size
);
218 * If `lttng_consumer_on_read_subbuffer_splice()` returned an error,
219 * pass it along to the caller, else return zero.
221 if (written_bytes
< 0) {
222 ERR("Error reading splice subbuffer: %zd", written_bytes
);
225 return written_bytes
;
228 static int consumer_stream_send_index(
229 struct lttng_consumer_stream
*stream
,
230 const struct stream_subbuffer
*subbuffer
,
231 struct lttng_consumer_local_data
*ctx
)
233 off_t packet_offset
= 0;
234 struct ctf_packet_index index
= {};
237 * This is called after consuming the sub-buffer; substract the
238 * effect this sub-buffer from the offset.
240 if (stream
->net_seq_idx
== (uint64_t) -1ULL) {
241 packet_offset
= stream
->out_fd_offset
-
242 subbuffer
->info
.data
.padded_subbuf_size
;
245 ctf_packet_index_populate(&index
, packet_offset
, subbuffer
);
246 return consumer_stream_write_index(stream
, &index
);
250 * Actually do the metadata sync using the given metadata stream.
252 * Return 0 on success else a negative value. ENODATA can be returned also
253 * indicating that there is no metadata available for that stream.
255 static int do_sync_metadata(struct lttng_consumer_stream
*metadata
,
256 struct lttng_consumer_local_data
*ctx
)
259 enum sync_metadata_status status
;
262 assert(metadata
->metadata_flag
);
266 * In UST, since we have to write the metadata from the cache packet
267 * by packet, we might need to start this procedure multiple times
268 * until all the metadata from the cache has been extracted.
273 * - Lock the metadata stream
274 * - Check if metadata stream node was deleted before locking.
275 * - if yes, release and return success
276 * - Check if new metadata is ready (flush + snapshot pos)
277 * - If nothing : release and return.
278 * - Lock the metadata_rdv_lock
279 * - Unlock the metadata stream
280 * - cond_wait on metadata_rdv to wait the wakeup from the
282 * - Unlock the metadata_rdv_lock
284 pthread_mutex_lock(&metadata
->lock
);
287 * There is a possibility that we were able to acquire a reference on the
288 * stream from the RCU hash table but between then and now, the node might
289 * have been deleted just before the lock is acquired. Thus, after locking,
290 * we make sure the metadata node has not been deleted which means that the
291 * buffers are closed.
293 * In that case, there is no need to sync the metadata hence returning a
294 * success return code.
296 ret
= cds_lfht_is_node_deleted(&metadata
->node
.node
);
299 goto end_unlock_mutex
;
303 case LTTNG_CONSUMER_KERNEL
:
305 * Empty the metadata cache and flush the current stream.
307 status
= lttng_kconsumer_sync_metadata(metadata
);
309 case LTTNG_CONSUMER32_UST
:
310 case LTTNG_CONSUMER64_UST
:
312 * Ask the sessiond if we have new metadata waiting and update the
313 * consumer metadata cache.
315 status
= lttng_ustconsumer_sync_metadata(ctx
, metadata
);
322 case SYNC_METADATA_STATUS_NEW_DATA
:
324 case SYNC_METADATA_STATUS_NO_DATA
:
326 goto end_unlock_mutex
;
327 case SYNC_METADATA_STATUS_ERROR
:
329 goto end_unlock_mutex
;
335 * At this point, new metadata have been flushed, so we wait on the
336 * rendez-vous point for the metadata thread to wake us up when it
337 * finishes consuming the metadata and continue execution.
340 pthread_mutex_lock(&metadata
->metadata_rdv_lock
);
343 * Release metadata stream lock so the metadata thread can process it.
345 pthread_mutex_unlock(&metadata
->lock
);
348 * Wait on the rendez-vous point. Once woken up, it means the metadata was
349 * consumed and thus synchronization is achieved.
351 pthread_cond_wait(&metadata
->metadata_rdv
, &metadata
->metadata_rdv_lock
);
352 pthread_mutex_unlock(&metadata
->metadata_rdv_lock
);
353 } while (status
== SYNC_METADATA_STATUS_NEW_DATA
);
359 pthread_mutex_unlock(&metadata
->lock
);
364 * Synchronize the metadata using a given session ID. A successful acquisition
365 * of a metadata stream will trigger a request to the session daemon and a
366 * snapshot so the metadata thread can consume it.
368 * This function call is a rendez-vous point between the metadata thread and
371 * Return 0 on success or else a negative value.
373 int consumer_stream_sync_metadata(struct lttng_consumer_local_data
*ctx
,
377 struct lttng_consumer_stream
*stream
= NULL
;
378 struct lttng_ht_iter iter
;
383 /* Ease our life a bit. */
384 ht
= consumer_data
.stream_list_ht
;
388 /* Search the metadata associated with the session id of the given stream. */
390 cds_lfht_for_each_entry_duplicate(ht
->ht
,
391 ht
->hash_fct(&session_id
, lttng_ht_seed
), ht
->match_fct
,
392 &session_id
, &iter
.iter
, stream
, node_session_id
.node
) {
393 if (!stream
->metadata_flag
) {
397 ret
= do_sync_metadata(stream
, ctx
);
404 * Force return code to 0 (success) since ret might be ENODATA for instance
405 * which is not an error but rather that we should come back.
414 static int consumer_stream_sync_metadata_index(
415 struct lttng_consumer_stream
*stream
,
416 const struct stream_subbuffer
*subbuffer
,
417 struct lttng_consumer_local_data
*ctx
)
419 bool missed_metadata_flush
;
422 /* Block until all the metadata is sent. */
423 pthread_mutex_lock(&stream
->metadata_timer_lock
);
424 assert(!stream
->missed_metadata_flush
);
425 stream
->waiting_on_metadata
= true;
426 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
428 ret
= consumer_stream_sync_metadata(ctx
, stream
->session_id
);
430 pthread_mutex_lock(&stream
->metadata_timer_lock
);
431 stream
->waiting_on_metadata
= false;
432 missed_metadata_flush
= stream
->missed_metadata_flush
;
433 if (missed_metadata_flush
) {
434 stream
->missed_metadata_flush
= false;
436 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
441 ret
= consumer_stream_send_index(stream
, subbuffer
, ctx
);
443 * Send the live inactivity beacon to handle the situation where
444 * the live timer is prevented from sampling this stream
445 * because the stream lock was being held while this stream is
446 * waiting on metadata. This ensures live viewer progress in the
447 * unlikely scenario where a live timer would be prevented from
448 * locking a stream lock repeatedly due to a steady flow of
449 * incoming metadata, for a stream which is mostly inactive.
451 * It is important to send the inactivity beacon packet to
452 * relayd _after_ sending the index associated with the data
453 * that was just sent, otherwise this can cause live viewers to
454 * observe timestamps going backwards between an inactivity
455 * beacon and a following trace packet.
457 if (missed_metadata_flush
) {
458 (void) stream
->read_subbuffer_ops
.send_live_beacon(stream
);
465 * Check if the local version of the metadata stream matches with the version
466 * of the metadata stream in the kernel. If it was updated, set the reset flag
470 int metadata_stream_check_version(struct lttng_consumer_stream
*stream
,
471 const struct stream_subbuffer
*subbuffer
)
473 if (stream
->metadata_version
== subbuffer
->info
.metadata
.version
) {
477 DBG("New metadata version detected");
478 consumer_stream_metadata_set_version(stream
,
479 subbuffer
->info
.metadata
.version
);
481 if (stream
->read_subbuffer_ops
.reset_metadata
) {
482 stream
->read_subbuffer_ops
.reset_metadata(stream
);
489 struct lttng_consumer_stream
*consumer_stream_create(
490 struct lttng_consumer_channel
*channel
,
491 uint64_t channel_key
,
493 const char *channel_name
,
496 struct lttng_trace_chunk
*trace_chunk
,
499 enum consumer_channel_type type
,
500 unsigned int monitor
)
503 struct lttng_consumer_stream
*stream
;
505 stream
= zmalloc(sizeof(*stream
));
506 if (stream
== NULL
) {
507 PERROR("malloc struct lttng_consumer_stream");
514 if (trace_chunk
&& !lttng_trace_chunk_get(trace_chunk
)) {
515 ERR("Failed to acquire trace chunk reference during the creation of a stream");
520 stream
->send_node
= (typeof(stream
->send_node
))
521 CDS_LIST_HEAD_INIT(stream
->send_node
);
522 stream
->chan
= channel
;
523 stream
->key
= stream_key
;
524 stream
->trace_chunk
= trace_chunk
;
526 stream
->out_fd_offset
= 0;
527 stream
->output_written
= 0;
528 stream
->net_seq_idx
= relayd_id
;
529 stream
->session_id
= session_id
;
530 stream
->monitor
= monitor
;
531 stream
->endpoint_status
= CONSUMER_ENDPOINT_ACTIVE
;
532 stream
->index_file
= NULL
;
533 stream
->last_sequence_number
= -1ULL;
534 stream
->rotate_position
= -1ULL;
535 /* Buffer is created with an open packet. */
536 stream
->opened_packet_in_current_trace_chunk
= true;
537 pthread_mutex_init(&stream
->lock
, NULL
);
538 pthread_mutex_init(&stream
->metadata_timer_lock
, NULL
);
540 /* If channel is the metadata, flag this stream as metadata. */
541 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
542 stream
->metadata_flag
= 1;
543 /* Metadata is flat out. */
544 strncpy(stream
->name
, DEFAULT_METADATA_NAME
, sizeof(stream
->name
));
545 /* Live rendez-vous point. */
546 pthread_cond_init(&stream
->metadata_rdv
, NULL
);
547 pthread_mutex_init(&stream
->metadata_rdv_lock
, NULL
);
549 /* Format stream name to <channel_name>_<cpu_number> */
550 ret
= snprintf(stream
->name
, sizeof(stream
->name
), "%s_%d",
553 PERROR("snprintf stream name");
558 switch (channel
->output
) {
559 case CONSUMER_CHANNEL_SPLICE
:
560 stream
->output
= LTTNG_EVENT_SPLICE
;
561 ret
= utils_create_pipe(stream
->splice_pipe
);
566 case CONSUMER_CHANNEL_MMAP
:
567 stream
->output
= LTTNG_EVENT_MMAP
;
573 /* Key is always the wait_fd for streams. */
574 lttng_ht_node_init_u64(&stream
->node
, stream
->key
);
576 /* Init node per channel id key */
577 lttng_ht_node_init_u64(&stream
->node_channel_id
, channel_key
);
579 /* Init session id node with the stream session id */
580 lttng_ht_node_init_u64(&stream
->node_session_id
, stream
->session_id
);
582 DBG3("Allocated stream %s (key %" PRIu64
", chan_key %" PRIu64
583 " relayd_id %" PRIu64
", session_id %" PRIu64
,
584 stream
->name
, stream
->key
, channel_key
,
585 stream
->net_seq_idx
, stream
->session_id
);
589 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
590 stream
->read_subbuffer_ops
.lock
=
591 consumer_stream_metadata_lock_all
;
592 stream
->read_subbuffer_ops
.unlock
=
593 consumer_stream_metadata_unlock_all
;
594 stream
->read_subbuffer_ops
.assert_locked
=
595 consumer_stream_metadata_assert_locked_all
;
596 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
597 metadata_stream_check_version
;
599 stream
->read_subbuffer_ops
.lock
= consumer_stream_data_lock_all
;
600 stream
->read_subbuffer_ops
.unlock
=
601 consumer_stream_data_unlock_all
;
602 stream
->read_subbuffer_ops
.assert_locked
=
603 consumer_stream_data_assert_locked_all
;
604 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
605 consumer_stream_update_stats
;
606 if (channel
->is_live
) {
607 stream
->read_subbuffer_ops
.post_consume
=
608 consumer_stream_sync_metadata_index
;
610 stream
->read_subbuffer_ops
.post_consume
=
611 consumer_stream_send_index
;
615 if (channel
->output
== CONSUMER_CHANNEL_MMAP
) {
616 stream
->read_subbuffer_ops
.consume_subbuffer
=
617 consumer_stream_consume_mmap
;
619 stream
->read_subbuffer_ops
.consume_subbuffer
=
620 consumer_stream_consume_splice
;
627 lttng_trace_chunk_put(stream
->trace_chunk
);
637 * Close stream on the relayd side. This call can destroy a relayd if the
638 * conditions are met.
640 * A RCU read side lock MUST be acquired if the relayd object was looked up in
641 * a hash table before calling this.
643 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
644 struct consumer_relayd_sock_pair
*relayd
)
651 if (stream
->sent_to_relayd
) {
652 uatomic_dec(&relayd
->refcount
);
653 assert(uatomic_read(&relayd
->refcount
) >= 0);
656 /* Closing streams requires to lock the control socket. */
657 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
658 ret
= relayd_send_close_stream(&relayd
->control_sock
,
659 stream
->relayd_stream_id
,
660 stream
->next_net_seq_num
- 1);
661 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
663 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
664 lttng_consumer_cleanup_relayd(relayd
);
667 /* Both conditions are met, we destroy the relayd. */
668 if (uatomic_read(&relayd
->refcount
) == 0 &&
669 uatomic_read(&relayd
->destroy_flag
)) {
670 consumer_destroy_relayd(relayd
);
672 stream
->net_seq_idx
= (uint64_t) -1ULL;
673 stream
->sent_to_relayd
= 0;
677 * Close stream's file descriptors and, if needed, close stream also on the
680 * The consumer data lock MUST be acquired.
681 * The stream lock MUST be acquired.
683 void consumer_stream_close(struct lttng_consumer_stream
*stream
)
686 struct consumer_relayd_sock_pair
*relayd
;
690 switch (consumer_data
.type
) {
691 case LTTNG_CONSUMER_KERNEL
:
692 if (stream
->mmap_base
!= NULL
) {
693 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
699 if (stream
->wait_fd
>= 0) {
700 ret
= close(stream
->wait_fd
);
704 stream
->wait_fd
= -1;
706 if (stream
->chan
->output
== CONSUMER_CHANNEL_SPLICE
) {
707 utils_close_pipe(stream
->splice_pipe
);
710 case LTTNG_CONSUMER32_UST
:
711 case LTTNG_CONSUMER64_UST
:
714 * Special case for the metadata since the wait fd is an internal pipe
715 * polled in the metadata thread.
717 if (stream
->metadata_flag
&& stream
->chan
->monitor
) {
718 int rpipe
= stream
->ust_metadata_poll_pipe
[0];
721 * This will stop the channel timer if one and close the write side
722 * of the metadata poll pipe.
724 lttng_ustconsumer_close_metadata(stream
->chan
);
728 PERROR("closing metadata pipe read side");
730 stream
->ust_metadata_poll_pipe
[0] = -1;
736 ERR("Unknown consumer_data type");
740 /* Close output fd. Could be a socket or local file at this point. */
741 if (stream
->out_fd
>= 0) {
742 ret
= close(stream
->out_fd
);
749 if (stream
->index_file
) {
750 lttng_index_file_put(stream
->index_file
);
751 stream
->index_file
= NULL
;
754 lttng_trace_chunk_put(stream
->trace_chunk
);
755 stream
->trace_chunk
= NULL
;
757 /* Check and cleanup relayd if needed. */
759 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
760 if (relayd
!= NULL
) {
761 consumer_stream_relayd_close(stream
, relayd
);
767 * Delete the stream from all possible hash tables.
769 * The consumer data lock MUST be acquired.
770 * The stream lock MUST be acquired.
772 void consumer_stream_delete(struct lttng_consumer_stream
*stream
,
776 struct lttng_ht_iter iter
;
779 /* Should NEVER be called not in monitor mode. */
780 assert(stream
->chan
->monitor
);
785 iter
.iter
.node
= &stream
->node
.node
;
786 ret
= lttng_ht_del(ht
, &iter
);
790 /* Delete from stream per channel ID hash table. */
791 iter
.iter
.node
= &stream
->node_channel_id
.node
;
793 * The returned value is of no importance. Even if the node is NOT in the
794 * hash table, we continue since we may have been called by a code path
795 * that did not add the stream to a (all) hash table. Same goes for the
796 * next call ht del call.
798 (void) lttng_ht_del(consumer_data
.stream_per_chan_id_ht
, &iter
);
800 /* Delete from the global stream list. */
801 iter
.iter
.node
= &stream
->node_session_id
.node
;
802 /* See the previous ht del on why we ignore the returned value. */
803 (void) lttng_ht_del(consumer_data
.stream_list_ht
, &iter
);
807 if (!stream
->metadata_flag
) {
808 /* Decrement the stream count of the global consumer data. */
809 assert(consumer_data
.stream_count
> 0);
810 consumer_data
.stream_count
--;
815 * Free the given stream within a RCU call.
817 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
821 metadata_bucket_destroy(stream
->metadata_bucket
);
822 call_rcu(&stream
->node
.head
, free_stream_rcu
);
826 * Destroy the stream's buffers of the tracer.
828 void consumer_stream_destroy_buffers(struct lttng_consumer_stream
*stream
)
832 switch (consumer_data
.type
) {
833 case LTTNG_CONSUMER_KERNEL
:
835 case LTTNG_CONSUMER32_UST
:
836 case LTTNG_CONSUMER64_UST
:
837 lttng_ustconsumer_del_stream(stream
);
840 ERR("Unknown consumer_data type");
846 * Destroy and close a already created stream.
848 static void destroy_close_stream(struct lttng_consumer_stream
*stream
)
852 DBG("Consumer stream destroy monitored key: %" PRIu64
, stream
->key
);
854 /* Destroy tracer buffers of the stream. */
855 consumer_stream_destroy_buffers(stream
);
856 /* Close down everything including the relayd if one. */
857 consumer_stream_close(stream
);
861 * Decrement the stream's channel refcount and if down to 0, return the channel
862 * pointer so it can be destroyed by the caller or NULL if not.
864 static struct lttng_consumer_channel
*unref_channel(
865 struct lttng_consumer_stream
*stream
)
867 struct lttng_consumer_channel
*free_chan
= NULL
;
870 assert(stream
->chan
);
872 /* Update refcount of channel and see if we need to destroy it. */
873 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1)
874 && !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
875 free_chan
= stream
->chan
;
882 * Destroy a stream completely. This will delete, close and free the stream.
883 * Once return, the stream is NO longer usable. Its channel may get destroyed
884 * if conditions are met for a monitored stream.
886 * This MUST be called WITHOUT the consumer data and stream lock acquired if
887 * the stream is in _monitor_ mode else it does not matter.
889 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
,
894 cds_list_del_init(&stream
->send_node
);
896 /* Stream is in monitor mode. */
897 if (stream
->monitor
) {
898 struct lttng_consumer_channel
*free_chan
= NULL
;
901 * This means that the stream was successfully removed from the streams
902 * list of the channel and sent to the right thread managing this
903 * stream thus being globally visible.
905 if (stream
->globally_visible
) {
906 pthread_mutex_lock(&consumer_data
.lock
);
907 pthread_mutex_lock(&stream
->chan
->lock
);
908 pthread_mutex_lock(&stream
->lock
);
909 /* Remove every reference of the stream in the consumer. */
910 consumer_stream_delete(stream
, ht
);
912 destroy_close_stream(stream
);
914 /* Update channel's refcount of the stream. */
915 free_chan
= unref_channel(stream
);
917 /* Indicates that the consumer data state MUST be updated after this. */
918 consumer_data
.need_update
= 1;
920 pthread_mutex_unlock(&stream
->lock
);
921 pthread_mutex_unlock(&stream
->chan
->lock
);
922 pthread_mutex_unlock(&consumer_data
.lock
);
925 * If the stream is not visible globally, this needs to be done
926 * outside of the consumer data lock section.
928 destroy_close_stream(stream
);
929 free_chan
= unref_channel(stream
);
933 consumer_del_channel(free_chan
);
936 destroy_close_stream(stream
);
939 /* Free stream within a RCU call. */
940 lttng_trace_chunk_put(stream
->trace_chunk
);
941 stream
->trace_chunk
= NULL
;
942 consumer_stream_free(stream
);
946 * Write index of a specific stream either on the relayd or local disk.
948 * Return 0 on success or else a negative value.
950 int consumer_stream_write_index(struct lttng_consumer_stream
*stream
,
951 struct ctf_packet_index
*element
)
959 if (stream
->net_seq_idx
!= (uint64_t) -1ULL) {
960 struct consumer_relayd_sock_pair
*relayd
;
961 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
963 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
964 ret
= relayd_send_index(&relayd
->control_sock
, element
,
965 stream
->relayd_stream_id
, stream
->next_net_seq_num
- 1);
968 * Communication error with lttng-relayd,
969 * perform cleanup now
971 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
972 lttng_consumer_cleanup_relayd(relayd
);
975 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
977 ERR("Stream %" PRIu64
" relayd ID %" PRIu64
" unknown. Can't write index.",
978 stream
->key
, stream
->net_seq_idx
);
982 if (lttng_index_file_write(stream
->index_file
, element
)) {
997 int consumer_stream_create_output_files(struct lttng_consumer_stream
*stream
,
1001 enum lttng_trace_chunk_status chunk_status
;
1002 const int flags
= O_WRONLY
| O_CREAT
| O_TRUNC
;
1003 const mode_t mode
= S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
;
1004 char stream_path
[LTTNG_PATH_MAX
];
1006 ASSERT_LOCKED(stream
->lock
);
1007 assert(stream
->trace_chunk
);
1009 ret
= utils_stream_file_path(stream
->chan
->pathname
, stream
->name
,
1010 stream
->chan
->tracefile_size
,
1011 stream
->tracefile_count_current
, NULL
,
1012 stream_path
, sizeof(stream_path
));
1017 if (stream
->out_fd
>= 0) {
1018 ret
= close(stream
->out_fd
);
1020 PERROR("Failed to close stream file \"%s\"",
1024 stream
->out_fd
= -1;
1027 DBG("Opening stream output file \"%s\"", stream_path
);
1028 chunk_status
= lttng_trace_chunk_open_file(stream
->trace_chunk
, stream_path
,
1029 flags
, mode
, &stream
->out_fd
, false);
1030 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1031 ERR("Failed to open stream file \"%s\"", stream
->name
);
1036 if (!stream
->metadata_flag
&& (create_index
|| stream
->index_file
)) {
1037 if (stream
->index_file
) {
1038 lttng_index_file_put(stream
->index_file
);
1040 chunk_status
= lttng_index_file_create_from_trace_chunk(
1041 stream
->trace_chunk
,
1042 stream
->chan
->pathname
,
1044 stream
->chan
->tracefile_size
,
1045 stream
->tracefile_count_current
,
1046 CTF_INDEX_MAJOR
, CTF_INDEX_MINOR
,
1047 false, &stream
->index_file
);
1048 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
1054 /* Reset current size because we just perform a rotation. */
1055 stream
->tracefile_size_current
= 0;
1056 stream
->out_fd_offset
= 0;
1061 int consumer_stream_rotate_output_files(struct lttng_consumer_stream
*stream
)
1065 stream
->tracefile_count_current
++;
1066 if (stream
->chan
->tracefile_count
> 0) {
1067 stream
->tracefile_count_current
%=
1068 stream
->chan
->tracefile_count
;
1071 DBG("Rotating output files of stream \"%s\"", stream
->name
);
1072 ret
= consumer_stream_create_output_files(stream
, true);
1081 bool consumer_stream_is_deleted(struct lttng_consumer_stream
*stream
)
1084 * This function does not take a const stream since
1085 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1088 return cds_lfht_is_node_deleted(&stream
->node
.node
);
1091 static ssize_t
metadata_bucket_flush(
1092 const struct stream_subbuffer
*buffer
, void *data
)
1095 struct lttng_consumer_stream
*stream
= data
;
1097 ret
= consumer_stream_consume_mmap(NULL
, stream
, buffer
);
1105 static ssize_t
metadata_bucket_consume(
1106 struct lttng_consumer_local_data
*unused
,
1107 struct lttng_consumer_stream
*stream
,
1108 const struct stream_subbuffer
*subbuffer
)
1111 enum metadata_bucket_status status
;
1113 status
= metadata_bucket_fill(stream
->metadata_bucket
, subbuffer
);
1115 case METADATA_BUCKET_STATUS_OK
:
1116 /* Return consumed size. */
1117 ret
= subbuffer
->buffer
.buffer
.size
;
1126 int consumer_stream_enable_metadata_bucketization(
1127 struct lttng_consumer_stream
*stream
)
1131 assert(stream
->metadata_flag
);
1132 assert(!stream
->metadata_bucket
);
1133 assert(stream
->chan
->output
== CONSUMER_CHANNEL_MMAP
);
1135 stream
->metadata_bucket
= metadata_bucket_create(
1136 metadata_bucket_flush
, stream
);
1137 if (!stream
->metadata_bucket
) {
1142 stream
->read_subbuffer_ops
.consume_subbuffer
= metadata_bucket_consume
;
1147 void consumer_stream_metadata_set_version(
1148 struct lttng_consumer_stream
*stream
, uint64_t new_version
)
1150 assert(new_version
> stream
->metadata_version
);
1151 stream
->metadata_version
= new_version
;
1152 stream
->reset_metadata_flag
= 1;
1154 if (stream
->metadata_bucket
) {
1155 metadata_bucket_reset(stream
->metadata_bucket
);