2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * Copyright (C) 2013 - David Goulet <dgoulet@efficios.com>
6 * This program is free software; you can redistribute it and/or modify it
7 * under the terms of the GNU General Public License, version 2 only, as
8 * published by the Free Software Foundation.
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
15 * You should have received a copy of the GNU General Public License along with
16 * this program; if not, write to the Free Software Foundation, Inc., 51
17 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
26 #include <common/common.h>
27 #include <common/index/index.h>
28 #include <common/kernel-consumer/kernel-consumer.h>
29 #include <common/relayd/relayd.h>
30 #include <common/ust-consumer/ust-consumer.h>
31 #include <common/utils.h>
32 #include <common/consumer/consumer.h>
33 #include <common/consumer/consumer-timer.h>
34 #include <common/consumer/metadata-bucket.h>
36 #include "consumer-stream.h"
39 * RCU call to free stream. MUST only be used with call_rcu().
41 static void free_stream_rcu(struct rcu_head
*head
)
43 struct lttng_ht_node_u64
*node
=
44 caa_container_of(head
, struct lttng_ht_node_u64
, head
);
45 struct lttng_consumer_stream
*stream
=
46 caa_container_of(node
, struct lttng_consumer_stream
, node
);
48 pthread_mutex_destroy(&stream
->lock
);
52 static void consumer_stream_data_lock_all(struct lttng_consumer_stream
*stream
)
54 pthread_mutex_lock(&stream
->chan
->lock
);
55 pthread_mutex_lock(&stream
->lock
);
58 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream
*stream
)
60 pthread_mutex_unlock(&stream
->lock
);
61 pthread_mutex_unlock(&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 /* Only used for data streams. */
77 static int consumer_stream_update_stats(struct lttng_consumer_stream
*stream
,
78 const struct stream_subbuffer
*subbuf
)
81 uint64_t sequence_number
;
82 const uint64_t discarded_events
=
83 LTTNG_OPTIONAL_GET(subbuf
->info
.data
.sequence_number
);
85 if (!subbuf
->info
.data
.sequence_number
.is_set
) {
86 /* Command not supported by the tracer. */
87 sequence_number
= -1ULL;
88 stream
->sequence_number_unavailable
= true;
90 sequence_number
= subbuf
->info
.data
.sequence_number
.value
;
94 * Start the sequence when we extract the first packet in case we don't
95 * start at 0 (for example if a consumer is not connected to the
96 * session immediately after the beginning).
98 if (stream
->last_sequence_number
== -1ULL) {
99 stream
->last_sequence_number
= sequence_number
;
100 } else if (sequence_number
> stream
->last_sequence_number
) {
101 stream
->chan
->lost_packets
+= sequence_number
-
102 stream
->last_sequence_number
- 1;
104 /* seq <= last_sequence_number */
105 ERR("Sequence number inconsistent : prev = %" PRIu64
106 ", current = %" PRIu64
,
107 stream
->last_sequence_number
, sequence_number
);
111 stream
->last_sequence_number
= sequence_number
;
113 if (discarded_events
< stream
->last_discarded_events
) {
115 * Overflow has occurred. We assume only one wrap-around
118 stream
->chan
->discarded_events
+=
119 (1ULL << (CAA_BITS_PER_LONG
- 1)) -
120 stream
->last_discarded_events
+
123 stream
->chan
->discarded_events
+= discarded_events
-
124 stream
->last_discarded_events
;
126 stream
->last_discarded_events
= discarded_events
;
134 void ctf_packet_index_populate(struct ctf_packet_index
*index
,
135 off_t offset
, const struct stream_subbuffer
*subbuffer
)
137 *index
= (typeof(*index
)){
138 .offset
= htobe64(offset
),
139 .packet_size
= htobe64(subbuffer
->info
.data
.packet_size
),
140 .content_size
= htobe64(subbuffer
->info
.data
.content_size
),
141 .timestamp_begin
= htobe64(
142 subbuffer
->info
.data
.timestamp_begin
),
143 .timestamp_end
= htobe64(
144 subbuffer
->info
.data
.timestamp_end
),
145 .events_discarded
= htobe64(
146 subbuffer
->info
.data
.events_discarded
),
147 .stream_id
= htobe64(subbuffer
->info
.data
.stream_id
),
148 .stream_instance_id
= htobe64(
149 subbuffer
->info
.data
.stream_instance_id
.is_set
?
150 subbuffer
->info
.data
.stream_instance_id
.value
: -1ULL),
151 .packet_seq_num
= htobe64(
152 subbuffer
->info
.data
.sequence_number
.is_set
?
153 subbuffer
->info
.data
.sequence_number
.value
: -1ULL),
157 static ssize_t
consumer_stream_consume_mmap(
158 struct lttng_consumer_local_data
*ctx
,
159 struct lttng_consumer_stream
*stream
,
160 const struct stream_subbuffer
*subbuffer
)
162 const unsigned long padding_size
=
163 subbuffer
->info
.data
.padded_subbuf_size
-
164 subbuffer
->info
.data
.subbuf_size
;
166 return lttng_consumer_on_read_subbuffer_mmap(
167 stream
, &subbuffer
->buffer
.buffer
, padding_size
);
170 static ssize_t
consumer_stream_consume_splice(
171 struct lttng_consumer_local_data
*ctx
,
172 struct lttng_consumer_stream
*stream
,
173 const struct stream_subbuffer
*subbuffer
)
175 return lttng_consumer_on_read_subbuffer_splice(ctx
, stream
,
176 subbuffer
->info
.data
.padded_subbuf_size
, 0);
179 static int consumer_stream_send_index(
180 struct lttng_consumer_stream
*stream
,
181 const struct stream_subbuffer
*subbuffer
,
182 struct lttng_consumer_local_data
*ctx
)
184 off_t packet_offset
= 0;
185 struct ctf_packet_index index
= {};
188 * This is called after consuming the sub-buffer; substract the
189 * effect this sub-buffer from the offset.
191 if (stream
->net_seq_idx
== (uint64_t) -1ULL) {
192 packet_offset
= stream
->out_fd_offset
-
193 subbuffer
->info
.data
.padded_subbuf_size
;
196 ctf_packet_index_populate(&index
, packet_offset
, subbuffer
);
197 return consumer_stream_write_index(stream
, &index
);
201 * Actually do the metadata sync using the given metadata stream.
203 * Return 0 on success else a negative value. ENODATA can be returned also
204 * indicating that there is no metadata available for that stream.
206 static int do_sync_metadata(struct lttng_consumer_stream
*metadata
,
207 struct lttng_consumer_local_data
*ctx
)
212 assert(metadata
->metadata_flag
);
216 * In UST, since we have to write the metadata from the cache packet
217 * by packet, we might need to start this procedure multiple times
218 * until all the metadata from the cache has been extracted.
223 * - Lock the metadata stream
224 * - Check if metadata stream node was deleted before locking.
225 * - if yes, release and return success
226 * - Check if new metadata is ready (flush + snapshot pos)
227 * - If nothing : release and return.
228 * - Lock the metadata_rdv_lock
229 * - Unlock the metadata stream
230 * - cond_wait on metadata_rdv to wait the wakeup from the
232 * - Unlock the metadata_rdv_lock
234 pthread_mutex_lock(&metadata
->lock
);
237 * There is a possibility that we were able to acquire a reference on the
238 * stream from the RCU hash table but between then and now, the node might
239 * have been deleted just before the lock is acquired. Thus, after locking,
240 * we make sure the metadata node has not been deleted which means that the
241 * buffers are closed.
243 * In that case, there is no need to sync the metadata hence returning a
244 * success return code.
246 ret
= cds_lfht_is_node_deleted(&metadata
->node
.node
);
249 goto end_unlock_mutex
;
253 case LTTNG_CONSUMER_KERNEL
:
255 * Empty the metadata cache and flush the current stream.
257 ret
= lttng_kconsumer_sync_metadata(metadata
);
259 case LTTNG_CONSUMER32_UST
:
260 case LTTNG_CONSUMER64_UST
:
262 * Ask the sessiond if we have new metadata waiting and update the
263 * consumer metadata cache.
265 ret
= lttng_ustconsumer_sync_metadata(ctx
, metadata
);
273 * Error or no new metadata, we exit here.
275 if (ret
<= 0 || ret
== ENODATA
) {
276 goto end_unlock_mutex
;
280 * At this point, new metadata have been flushed, so we wait on the
281 * rendez-vous point for the metadata thread to wake us up when it
282 * finishes consuming the metadata and continue execution.
285 pthread_mutex_lock(&metadata
->metadata_rdv_lock
);
288 * Release metadata stream lock so the metadata thread can process it.
290 pthread_mutex_unlock(&metadata
->lock
);
293 * Wait on the rendez-vous point. Once woken up, it means the metadata was
294 * consumed and thus synchronization is achieved.
296 pthread_cond_wait(&metadata
->metadata_rdv
, &metadata
->metadata_rdv_lock
);
297 pthread_mutex_unlock(&metadata
->metadata_rdv_lock
);
298 } while (ret
== EAGAIN
);
304 pthread_mutex_unlock(&metadata
->lock
);
309 * Synchronize the metadata using a given session ID. A successful acquisition
310 * of a metadata stream will trigger a request to the session daemon and a
311 * snapshot so the metadata thread can consume it.
313 * This function call is a rendez-vous point between the metadata thread and
316 * Return 0 on success or else a negative value.
318 int consumer_stream_sync_metadata(struct lttng_consumer_local_data
*ctx
,
322 struct lttng_consumer_stream
*stream
= NULL
;
323 struct lttng_ht_iter iter
;
328 /* Ease our life a bit. */
329 ht
= consumer_data
.stream_list_ht
;
333 /* Search the metadata associated with the session id of the given stream. */
335 cds_lfht_for_each_entry_duplicate(ht
->ht
,
336 ht
->hash_fct(&session_id
, lttng_ht_seed
), ht
->match_fct
,
337 &session_id
, &iter
.iter
, stream
, node_session_id
.node
) {
338 if (!stream
->metadata_flag
) {
342 ret
= do_sync_metadata(stream
, ctx
);
349 * Force return code to 0 (success) since ret might be ENODATA for instance
350 * which is not an error but rather that we should come back.
359 static int consumer_stream_sync_metadata_index(
360 struct lttng_consumer_stream
*stream
,
361 const struct stream_subbuffer
*subbuffer
,
362 struct lttng_consumer_local_data
*ctx
)
366 /* Block until all the metadata is sent. */
367 pthread_mutex_lock(&stream
->metadata_timer_lock
);
368 assert(!stream
->missed_metadata_flush
);
369 stream
->waiting_on_metadata
= true;
370 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
372 ret
= consumer_stream_sync_metadata(ctx
, stream
->session_id
);
374 pthread_mutex_lock(&stream
->metadata_timer_lock
);
375 stream
->waiting_on_metadata
= false;
376 if (stream
->missed_metadata_flush
) {
377 stream
->missed_metadata_flush
= false;
378 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
379 (void) stream
->read_subbuffer_ops
.send_live_beacon(stream
);
381 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
387 ret
= consumer_stream_send_index(stream
, subbuffer
, ctx
);
393 * Check if the local version of the metadata stream matches with the version
394 * of the metadata stream in the kernel. If it was updated, set the reset flag
398 int metadata_stream_check_version(struct lttng_consumer_stream
*stream
,
399 const struct stream_subbuffer
*subbuffer
)
401 if (stream
->metadata_version
== subbuffer
->info
.metadata
.version
) {
405 DBG("New metadata version detected");
406 stream
->metadata_version
= subbuffer
->info
.metadata
.version
;
407 stream
->reset_metadata_flag
= 1;
409 if (stream
->metadata_bucket
) {
410 metadata_bucket_reset(stream
->metadata_bucket
);
413 if (stream
->read_subbuffer_ops
.reset_metadata
) {
414 stream
->read_subbuffer_ops
.reset_metadata(stream
);
421 struct lttng_consumer_stream
*consumer_stream_create(
422 struct lttng_consumer_channel
*channel
,
423 uint64_t channel_key
,
425 const char *channel_name
,
428 struct lttng_trace_chunk
*trace_chunk
,
431 enum consumer_channel_type type
,
432 unsigned int monitor
)
435 struct lttng_consumer_stream
*stream
;
437 stream
= zmalloc(sizeof(*stream
));
438 if (stream
== NULL
) {
439 PERROR("malloc struct lttng_consumer_stream");
444 if (trace_chunk
&& !lttng_trace_chunk_get(trace_chunk
)) {
445 ERR("Failed to acquire trace chunk reference during the creation of a stream");
451 stream
->chan
= channel
;
452 stream
->key
= stream_key
;
453 stream
->trace_chunk
= trace_chunk
;
455 stream
->out_fd_offset
= 0;
456 stream
->output_written
= 0;
457 stream
->net_seq_idx
= relayd_id
;
458 stream
->session_id
= session_id
;
459 stream
->monitor
= monitor
;
460 stream
->endpoint_status
= CONSUMER_ENDPOINT_ACTIVE
;
461 stream
->index_file
= NULL
;
462 stream
->last_sequence_number
= -1ULL;
463 stream
->rotate_position
= -1ULL;
464 pthread_mutex_init(&stream
->lock
, NULL
);
465 pthread_mutex_init(&stream
->metadata_timer_lock
, NULL
);
467 /* If channel is the metadata, flag this stream as metadata. */
468 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
469 stream
->metadata_flag
= 1;
470 /* Metadata is flat out. */
471 strncpy(stream
->name
, DEFAULT_METADATA_NAME
, sizeof(stream
->name
));
472 /* Live rendez-vous point. */
473 pthread_cond_init(&stream
->metadata_rdv
, NULL
);
474 pthread_mutex_init(&stream
->metadata_rdv_lock
, NULL
);
476 /* Format stream name to <channel_name>_<cpu_number> */
477 ret
= snprintf(stream
->name
, sizeof(stream
->name
), "%s_%d",
480 PERROR("snprintf stream name");
485 switch (channel
->output
) {
486 case CONSUMER_CHANNEL_SPLICE
:
487 stream
->output
= LTTNG_EVENT_SPLICE
;
488 ret
= utils_create_pipe(stream
->splice_pipe
);
493 case CONSUMER_CHANNEL_MMAP
:
494 stream
->output
= LTTNG_EVENT_MMAP
;
500 /* Key is always the wait_fd for streams. */
501 lttng_ht_node_init_u64(&stream
->node
, stream
->key
);
503 /* Init node per channel id key */
504 lttng_ht_node_init_u64(&stream
->node_channel_id
, channel_key
);
506 /* Init session id node with the stream session id */
507 lttng_ht_node_init_u64(&stream
->node_session_id
, stream
->session_id
);
509 DBG3("Allocated stream %s (key %" PRIu64
", chan_key %" PRIu64
510 " relayd_id %" PRIu64
", session_id %" PRIu64
,
511 stream
->name
, stream
->key
, channel_key
,
512 stream
->net_seq_idx
, stream
->session_id
);
516 if (type
== CONSUMER_CHANNEL_TYPE_METADATA
) {
517 stream
->read_subbuffer_ops
.lock
=
518 consumer_stream_metadata_lock_all
;
519 stream
->read_subbuffer_ops
.unlock
=
520 consumer_stream_metadata_unlock_all
;
521 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
522 metadata_stream_check_version
;
524 stream
->read_subbuffer_ops
.lock
= consumer_stream_data_lock_all
;
525 stream
->read_subbuffer_ops
.unlock
=
526 consumer_stream_data_unlock_all
;
527 stream
->read_subbuffer_ops
.pre_consume_subbuffer
=
528 consumer_stream_update_stats
;
529 if (channel
->is_live
) {
530 stream
->read_subbuffer_ops
.post_consume
=
531 consumer_stream_sync_metadata_index
;
533 stream
->read_subbuffer_ops
.post_consume
=
534 consumer_stream_send_index
;
538 if (channel
->output
== CONSUMER_CHANNEL_MMAP
) {
539 stream
->read_subbuffer_ops
.consume_subbuffer
=
540 consumer_stream_consume_mmap
;
542 stream
->read_subbuffer_ops
.consume_subbuffer
=
543 consumer_stream_consume_splice
;
550 lttng_trace_chunk_put(stream
->trace_chunk
);
560 * Close stream on the relayd side. This call can destroy a relayd if the
561 * conditions are met.
563 * A RCU read side lock MUST be acquired if the relayd object was looked up in
564 * a hash table before calling this.
566 void consumer_stream_relayd_close(struct lttng_consumer_stream
*stream
,
567 struct consumer_relayd_sock_pair
*relayd
)
574 if (stream
->sent_to_relayd
) {
575 uatomic_dec(&relayd
->refcount
);
576 assert(uatomic_read(&relayd
->refcount
) >= 0);
579 /* Closing streams requires to lock the control socket. */
580 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
581 ret
= relayd_send_close_stream(&relayd
->control_sock
,
582 stream
->relayd_stream_id
,
583 stream
->next_net_seq_num
- 1);
584 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
586 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
587 lttng_consumer_cleanup_relayd(relayd
);
590 /* Both conditions are met, we destroy the relayd. */
591 if (uatomic_read(&relayd
->refcount
) == 0 &&
592 uatomic_read(&relayd
->destroy_flag
)) {
593 consumer_destroy_relayd(relayd
);
595 stream
->net_seq_idx
= (uint64_t) -1ULL;
596 stream
->sent_to_relayd
= 0;
600 * Close stream's file descriptors and, if needed, close stream also on the
603 * The consumer data lock MUST be acquired.
604 * The stream lock MUST be acquired.
606 void consumer_stream_close(struct lttng_consumer_stream
*stream
)
609 struct consumer_relayd_sock_pair
*relayd
;
613 switch (consumer_data
.type
) {
614 case LTTNG_CONSUMER_KERNEL
:
615 if (stream
->mmap_base
!= NULL
) {
616 ret
= munmap(stream
->mmap_base
, stream
->mmap_len
);
622 if (stream
->wait_fd
>= 0) {
623 ret
= close(stream
->wait_fd
);
627 stream
->wait_fd
= -1;
629 if (stream
->chan
->output
== CONSUMER_CHANNEL_SPLICE
) {
630 utils_close_pipe(stream
->splice_pipe
);
633 case LTTNG_CONSUMER32_UST
:
634 case LTTNG_CONSUMER64_UST
:
637 * Special case for the metadata since the wait fd is an internal pipe
638 * polled in the metadata thread.
640 if (stream
->metadata_flag
&& stream
->chan
->monitor
) {
641 int rpipe
= stream
->ust_metadata_poll_pipe
[0];
644 * This will stop the channel timer if one and close the write side
645 * of the metadata poll pipe.
647 lttng_ustconsumer_close_metadata(stream
->chan
);
651 PERROR("closing metadata pipe read side");
653 stream
->ust_metadata_poll_pipe
[0] = -1;
659 ERR("Unknown consumer_data type");
663 /* Close output fd. Could be a socket or local file at this point. */
664 if (stream
->out_fd
>= 0) {
665 ret
= close(stream
->out_fd
);
672 if (stream
->index_file
) {
673 lttng_index_file_put(stream
->index_file
);
674 stream
->index_file
= NULL
;
677 lttng_trace_chunk_put(stream
->trace_chunk
);
678 stream
->trace_chunk
= NULL
;
680 /* Check and cleanup relayd if needed. */
682 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
683 if (relayd
!= NULL
) {
684 consumer_stream_relayd_close(stream
, relayd
);
690 * Delete the stream from all possible hash tables.
692 * The consumer data lock MUST be acquired.
693 * The stream lock MUST be acquired.
695 void consumer_stream_delete(struct lttng_consumer_stream
*stream
,
699 struct lttng_ht_iter iter
;
702 /* Should NEVER be called not in monitor mode. */
703 assert(stream
->chan
->monitor
);
708 iter
.iter
.node
= &stream
->node
.node
;
709 ret
= lttng_ht_del(ht
, &iter
);
713 /* Delete from stream per channel ID hash table. */
714 iter
.iter
.node
= &stream
->node_channel_id
.node
;
716 * The returned value is of no importance. Even if the node is NOT in the
717 * hash table, we continue since we may have been called by a code path
718 * that did not add the stream to a (all) hash table. Same goes for the
719 * next call ht del call.
721 (void) lttng_ht_del(consumer_data
.stream_per_chan_id_ht
, &iter
);
723 /* Delete from the global stream list. */
724 iter
.iter
.node
= &stream
->node_session_id
.node
;
725 /* See the previous ht del on why we ignore the returned value. */
726 (void) lttng_ht_del(consumer_data
.stream_list_ht
, &iter
);
730 if (!stream
->metadata_flag
) {
731 /* Decrement the stream count of the global consumer data. */
732 assert(consumer_data
.stream_count
> 0);
733 consumer_data
.stream_count
--;
738 * Free the given stream within a RCU call.
740 void consumer_stream_free(struct lttng_consumer_stream
*stream
)
744 metadata_bucket_destroy(stream
->metadata_bucket
);
745 call_rcu(&stream
->node
.head
, free_stream_rcu
);
749 * Destroy the stream's buffers of the tracer.
751 void consumer_stream_destroy_buffers(struct lttng_consumer_stream
*stream
)
755 switch (consumer_data
.type
) {
756 case LTTNG_CONSUMER_KERNEL
:
758 case LTTNG_CONSUMER32_UST
:
759 case LTTNG_CONSUMER64_UST
:
760 lttng_ustconsumer_del_stream(stream
);
763 ERR("Unknown consumer_data type");
769 * Destroy and close a already created stream.
771 static void destroy_close_stream(struct lttng_consumer_stream
*stream
)
775 DBG("Consumer stream destroy monitored key: %" PRIu64
, stream
->key
);
777 /* Destroy tracer buffers of the stream. */
778 consumer_stream_destroy_buffers(stream
);
779 /* Close down everything including the relayd if one. */
780 consumer_stream_close(stream
);
784 * Decrement the stream's channel refcount and if down to 0, return the channel
785 * pointer so it can be destroyed by the caller or NULL if not.
787 static struct lttng_consumer_channel
*unref_channel(
788 struct lttng_consumer_stream
*stream
)
790 struct lttng_consumer_channel
*free_chan
= NULL
;
793 assert(stream
->chan
);
795 /* Update refcount of channel and see if we need to destroy it. */
796 if (!uatomic_sub_return(&stream
->chan
->refcount
, 1)
797 && !uatomic_read(&stream
->chan
->nb_init_stream_left
)) {
798 free_chan
= stream
->chan
;
805 * Destroy a stream completely. This will delete, close and free the stream.
806 * Once return, the stream is NO longer usable. Its channel may get destroyed
807 * if conditions are met for a monitored stream.
809 * This MUST be called WITHOUT the consumer data and stream lock acquired if
810 * the stream is in _monitor_ mode else it does not matter.
812 void consumer_stream_destroy(struct lttng_consumer_stream
*stream
,
817 /* Stream is in monitor mode. */
818 if (stream
->monitor
) {
819 struct lttng_consumer_channel
*free_chan
= NULL
;
822 * This means that the stream was successfully removed from the streams
823 * list of the channel and sent to the right thread managing this
824 * stream thus being globally visible.
826 if (stream
->globally_visible
) {
827 pthread_mutex_lock(&consumer_data
.lock
);
828 pthread_mutex_lock(&stream
->chan
->lock
);
829 pthread_mutex_lock(&stream
->lock
);
830 /* Remove every reference of the stream in the consumer. */
831 consumer_stream_delete(stream
, ht
);
833 destroy_close_stream(stream
);
835 /* Update channel's refcount of the stream. */
836 free_chan
= unref_channel(stream
);
838 /* Indicates that the consumer data state MUST be updated after this. */
839 consumer_data
.need_update
= 1;
841 pthread_mutex_unlock(&stream
->lock
);
842 pthread_mutex_unlock(&stream
->chan
->lock
);
843 pthread_mutex_unlock(&consumer_data
.lock
);
846 * If the stream is not visible globally, this needs to be done
847 * outside of the consumer data lock section.
849 free_chan
= unref_channel(stream
);
853 consumer_del_channel(free_chan
);
856 destroy_close_stream(stream
);
859 /* Free stream within a RCU call. */
860 lttng_trace_chunk_put(stream
->trace_chunk
);
861 stream
->trace_chunk
= NULL
;
862 consumer_stream_free(stream
);
866 * Write index of a specific stream either on the relayd or local disk.
868 * Return 0 on success or else a negative value.
870 int consumer_stream_write_index(struct lttng_consumer_stream
*stream
,
871 struct ctf_packet_index
*element
)
879 if (stream
->net_seq_idx
!= (uint64_t) -1ULL) {
880 struct consumer_relayd_sock_pair
*relayd
;
881 relayd
= consumer_find_relayd(stream
->net_seq_idx
);
883 pthread_mutex_lock(&relayd
->ctrl_sock_mutex
);
884 ret
= relayd_send_index(&relayd
->control_sock
, element
,
885 stream
->relayd_stream_id
, stream
->next_net_seq_num
- 1);
888 * Communication error with lttng-relayd,
889 * perform cleanup now
891 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64
".", relayd
->net_seq_idx
);
892 lttng_consumer_cleanup_relayd(relayd
);
895 pthread_mutex_unlock(&relayd
->ctrl_sock_mutex
);
897 ERR("Stream %" PRIu64
" relayd ID %" PRIu64
" unknown. Can't write index.",
898 stream
->key
, stream
->net_seq_idx
);
902 if (lttng_index_file_write(stream
->index_file
, element
)) {
917 int consumer_stream_create_output_files(struct lttng_consumer_stream
*stream
,
921 enum lttng_trace_chunk_status chunk_status
;
922 const int flags
= O_WRONLY
| O_CREAT
| O_TRUNC
;
923 const mode_t mode
= S_IRUSR
| S_IWUSR
| S_IRGRP
| S_IWGRP
;
924 char stream_path
[LTTNG_PATH_MAX
];
926 ASSERT_LOCKED(stream
->lock
);
927 assert(stream
->trace_chunk
);
929 ret
= utils_stream_file_path(stream
->chan
->pathname
, stream
->name
,
930 stream
->chan
->tracefile_size
,
931 stream
->tracefile_count_current
, NULL
,
932 stream_path
, sizeof(stream_path
));
937 if (stream
->out_fd
>= 0) {
938 ret
= close(stream
->out_fd
);
940 PERROR("Failed to close stream file \"%s\"",
947 DBG("Opening stream output file \"%s\"", stream_path
);
948 chunk_status
= lttng_trace_chunk_open_file(stream
->trace_chunk
, stream_path
,
949 flags
, mode
, &stream
->out_fd
);
950 if (chunk_status
!= LTTNG_TRACE_CHUNK_STATUS_OK
) {
951 ERR("Failed to open stream file \"%s\"", stream
->name
);
956 if (!stream
->metadata_flag
&& (create_index
|| stream
->index_file
)) {
957 if (stream
->index_file
) {
958 lttng_index_file_put(stream
->index_file
);
960 stream
->index_file
= lttng_index_file_create_from_trace_chunk(
962 stream
->chan
->pathname
,
964 stream
->chan
->tracefile_size
,
965 stream
->tracefile_count_current
,
966 CTF_INDEX_MAJOR
, CTF_INDEX_MINOR
,
968 if (!stream
->index_file
) {
974 /* Reset current size because we just perform a rotation. */
975 stream
->tracefile_size_current
= 0;
976 stream
->out_fd_offset
= 0;
981 int consumer_stream_rotate_output_files(struct lttng_consumer_stream
*stream
)
985 stream
->tracefile_count_current
++;
986 if (stream
->chan
->tracefile_count
> 0) {
987 stream
->tracefile_count_current
%=
988 stream
->chan
->tracefile_count
;
991 DBG("Rotating output files of stream \"%s\"", stream
->name
);
992 ret
= consumer_stream_create_output_files(stream
, true);
1001 bool consumer_stream_is_deleted(struct lttng_consumer_stream
*stream
)
1004 * This function does not take a const stream since
1005 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1008 return cds_lfht_is_node_deleted(&stream
->node
.node
);
1011 static ssize_t
metadata_bucket_flush(
1012 const struct stream_subbuffer
*buffer
, void *data
)
1015 struct lttng_consumer_stream
*stream
= data
;
1017 ret
= consumer_stream_consume_mmap(NULL
, stream
, buffer
);
1025 static ssize_t
metadata_bucket_consume(
1026 struct lttng_consumer_local_data
*unused
,
1027 struct lttng_consumer_stream
*stream
,
1028 const struct stream_subbuffer
*subbuffer
)
1031 enum metadata_bucket_status status
;
1033 status
= metadata_bucket_fill(stream
->metadata_bucket
, subbuffer
);
1035 case METADATA_BUCKET_STATUS_OK
:
1036 /* Return consumed size. */
1037 ret
= subbuffer
->buffer
.buffer
.size
;
1046 int consumer_stream_enable_metadata_bucketization(
1047 struct lttng_consumer_stream
*stream
)
1051 assert(stream
->metadata_flag
);
1052 assert(!stream
->metadata_bucket
);
1053 assert(stream
->chan
->output
== CONSUMER_CHANNEL_MMAP
);
1055 stream
->metadata_bucket
= metadata_bucket_create(
1056 metadata_bucket_flush
, stream
);
1057 if (!stream
->metadata_bucket
) {
1062 stream
->read_subbuffer_ops
.consume_subbuffer
= metadata_bucket_consume
;