Fix: invalid discarded events on start/stop without event production
[lttng-tools.git] / src / common / consumer / consumer-stream.c
... / ...
CommitLineData
1/*
2 * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
5 *
6 * SPDX-License-Identifier: GPL-2.0-only
7 *
8 */
9
10#define _LGPL_SOURCE
11#include <assert.h>
12#include <inttypes.h>
13#include <sys/mman.h>
14#include <unistd.h>
15
16#include <common/common.h>
17#include <common/index/index.h>
18#include <common/kernel-consumer/kernel-consumer.h>
19#include <common/relayd/relayd.h>
20#include <common/ust-consumer/ust-consumer.h>
21#include <common/utils.h>
22#include <common/consumer/consumer.h>
23#include <common/consumer/consumer-timer.h>
24#include <common/consumer/metadata-bucket.h>
25
26#include "consumer-stream.h"
27
28/*
29 * RCU call to free stream. MUST only be used with call_rcu().
30 */
31static void free_stream_rcu(struct rcu_head *head)
32{
33 struct lttng_ht_node_u64 *node =
34 caa_container_of(head, struct lttng_ht_node_u64, head);
35 struct lttng_consumer_stream *stream =
36 caa_container_of(node, struct lttng_consumer_stream, node);
37
38 pthread_mutex_destroy(&stream->lock);
39 free(stream);
40}
41
42static void consumer_stream_data_lock_all(struct lttng_consumer_stream *stream)
43{
44 pthread_mutex_lock(&stream->chan->lock);
45 pthread_mutex_lock(&stream->lock);
46}
47
48static void consumer_stream_data_unlock_all(struct lttng_consumer_stream *stream)
49{
50 pthread_mutex_unlock(&stream->lock);
51 pthread_mutex_unlock(&stream->chan->lock);
52}
53
54static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream *stream)
55{
56 consumer_stream_data_lock_all(stream);
57 pthread_mutex_lock(&stream->metadata_rdv_lock);
58}
59
60static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream *stream)
61{
62 pthread_mutex_unlock(&stream->metadata_rdv_lock);
63 consumer_stream_data_unlock_all(stream);
64}
65
66/* Only used for data streams. */
67static int consumer_stream_update_stats(struct lttng_consumer_stream *stream,
68 const struct stream_subbuffer *subbuf)
69{
70 int ret = 0;
71 uint64_t sequence_number;
72 const uint64_t discarded_events = subbuf->info.data.events_discarded;
73
74 if (!subbuf->info.data.sequence_number.is_set) {
75 /* Command not supported by the tracer. */
76 sequence_number = -1ULL;
77 stream->sequence_number_unavailable = true;
78 } else {
79 sequence_number = subbuf->info.data.sequence_number.value;
80 }
81
82 /*
83 * Start the sequence when we extract the first packet in case we don't
84 * start at 0 (for example if a consumer is not connected to the
85 * session immediately after the beginning).
86 */
87 if (stream->last_sequence_number == -1ULL) {
88 stream->last_sequence_number = sequence_number;
89 } else if (sequence_number > stream->last_sequence_number) {
90 stream->chan->lost_packets += sequence_number -
91 stream->last_sequence_number - 1;
92 } else {
93 /* seq <= last_sequence_number */
94 ERR("Sequence number inconsistent : prev = %" PRIu64
95 ", current = %" PRIu64,
96 stream->last_sequence_number, sequence_number);
97 ret = -1;
98 goto end;
99 }
100 stream->last_sequence_number = sequence_number;
101
102 if (discarded_events < stream->last_discarded_events) {
103 /*
104 * Overflow has occurred. We assume only one wrap-around
105 * has occurred.
106 */
107 stream->chan->discarded_events +=
108 (1ULL << (CAA_BITS_PER_LONG - 1)) -
109 stream->last_discarded_events +
110 discarded_events;
111 } else {
112 stream->chan->discarded_events += discarded_events -
113 stream->last_discarded_events;
114 }
115 stream->last_discarded_events = discarded_events;
116 ret = 0;
117
118end:
119 return ret;
120}
121
122static
123void ctf_packet_index_populate(struct ctf_packet_index *index,
124 off_t offset, const struct stream_subbuffer *subbuffer)
125{
126 *index = (typeof(*index)){
127 .offset = htobe64(offset),
128 .packet_size = htobe64(subbuffer->info.data.packet_size),
129 .content_size = htobe64(subbuffer->info.data.content_size),
130 .timestamp_begin = htobe64(
131 subbuffer->info.data.timestamp_begin),
132 .timestamp_end = htobe64(
133 subbuffer->info.data.timestamp_end),
134 .events_discarded = htobe64(
135 subbuffer->info.data.events_discarded),
136 .stream_id = htobe64(subbuffer->info.data.stream_id),
137 .stream_instance_id = htobe64(
138 subbuffer->info.data.stream_instance_id.is_set ?
139 subbuffer->info.data.stream_instance_id.value : -1ULL),
140 .packet_seq_num = htobe64(
141 subbuffer->info.data.sequence_number.is_set ?
142 subbuffer->info.data.sequence_number.value : -1ULL),
143 };
144}
145
146static ssize_t consumer_stream_consume_mmap(
147 struct lttng_consumer_local_data *ctx,
148 struct lttng_consumer_stream *stream,
149 const struct stream_subbuffer *subbuffer)
150{
151 const unsigned long padding_size =
152 subbuffer->info.data.padded_subbuf_size -
153 subbuffer->info.data.subbuf_size;
154
155 return lttng_consumer_on_read_subbuffer_mmap(
156 stream, &subbuffer->buffer.buffer, padding_size);
157}
158
159static ssize_t consumer_stream_consume_splice(
160 struct lttng_consumer_local_data *ctx,
161 struct lttng_consumer_stream *stream,
162 const struct stream_subbuffer *subbuffer)
163{
164 return lttng_consumer_on_read_subbuffer_splice(ctx, stream,
165 subbuffer->info.data.padded_subbuf_size, 0);
166}
167
168static int consumer_stream_send_index(
169 struct lttng_consumer_stream *stream,
170 const struct stream_subbuffer *subbuffer,
171 struct lttng_consumer_local_data *ctx)
172{
173 off_t packet_offset = 0;
174 struct ctf_packet_index index = {};
175
176 /*
177 * This is called after consuming the sub-buffer; substract the
178 * effect this sub-buffer from the offset.
179 */
180 if (stream->net_seq_idx == (uint64_t) -1ULL) {
181 packet_offset = stream->out_fd_offset -
182 subbuffer->info.data.padded_subbuf_size;
183 }
184
185 ctf_packet_index_populate(&index, packet_offset, subbuffer);
186 return consumer_stream_write_index(stream, &index);
187}
188
189/*
190 * Actually do the metadata sync using the given metadata stream.
191 *
192 * Return 0 on success else a negative value. ENODATA can be returned also
193 * indicating that there is no metadata available for that stream.
194 */
195static int do_sync_metadata(struct lttng_consumer_stream *metadata,
196 struct lttng_consumer_local_data *ctx)
197{
198 int ret;
199
200 assert(metadata);
201 assert(metadata->metadata_flag);
202 assert(ctx);
203
204 /*
205 * In UST, since we have to write the metadata from the cache packet
206 * by packet, we might need to start this procedure multiple times
207 * until all the metadata from the cache has been extracted.
208 */
209 do {
210 /*
211 * Steps :
212 * - Lock the metadata stream
213 * - Check if metadata stream node was deleted before locking.
214 * - if yes, release and return success
215 * - Check if new metadata is ready (flush + snapshot pos)
216 * - If nothing : release and return.
217 * - Lock the metadata_rdv_lock
218 * - Unlock the metadata stream
219 * - cond_wait on metadata_rdv to wait the wakeup from the
220 * metadata thread
221 * - Unlock the metadata_rdv_lock
222 */
223 pthread_mutex_lock(&metadata->lock);
224
225 /*
226 * There is a possibility that we were able to acquire a reference on the
227 * stream from the RCU hash table but between then and now, the node might
228 * have been deleted just before the lock is acquired. Thus, after locking,
229 * we make sure the metadata node has not been deleted which means that the
230 * buffers are closed.
231 *
232 * In that case, there is no need to sync the metadata hence returning a
233 * success return code.
234 */
235 ret = cds_lfht_is_node_deleted(&metadata->node.node);
236 if (ret) {
237 ret = 0;
238 goto end_unlock_mutex;
239 }
240
241 switch (ctx->type) {
242 case LTTNG_CONSUMER_KERNEL:
243 /*
244 * Empty the metadata cache and flush the current stream.
245 */
246 ret = lttng_kconsumer_sync_metadata(metadata);
247 break;
248 case LTTNG_CONSUMER32_UST:
249 case LTTNG_CONSUMER64_UST:
250 /*
251 * Ask the sessiond if we have new metadata waiting and update the
252 * consumer metadata cache.
253 */
254 ret = lttng_ustconsumer_sync_metadata(ctx, metadata);
255 break;
256 default:
257 assert(0);
258 ret = -1;
259 break;
260 }
261 /*
262 * Error or no new metadata, we exit here.
263 */
264 if (ret <= 0 || ret == ENODATA) {
265 goto end_unlock_mutex;
266 }
267
268 /*
269 * At this point, new metadata have been flushed, so we wait on the
270 * rendez-vous point for the metadata thread to wake us up when it
271 * finishes consuming the metadata and continue execution.
272 */
273
274 pthread_mutex_lock(&metadata->metadata_rdv_lock);
275
276 /*
277 * Release metadata stream lock so the metadata thread can process it.
278 */
279 pthread_mutex_unlock(&metadata->lock);
280
281 /*
282 * Wait on the rendez-vous point. Once woken up, it means the metadata was
283 * consumed and thus synchronization is achieved.
284 */
285 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
286 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
287 } while (ret == EAGAIN);
288
289 /* Success */
290 return 0;
291
292end_unlock_mutex:
293 pthread_mutex_unlock(&metadata->lock);
294 return ret;
295}
296
297/*
298 * Synchronize the metadata using a given session ID. A successful acquisition
299 * of a metadata stream will trigger a request to the session daemon and a
300 * snapshot so the metadata thread can consume it.
301 *
302 * This function call is a rendez-vous point between the metadata thread and
303 * the data thread.
304 *
305 * Return 0 on success or else a negative value.
306 */
307int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
308 uint64_t session_id)
309{
310 int ret;
311 struct lttng_consumer_stream *stream = NULL;
312 struct lttng_ht_iter iter;
313 struct lttng_ht *ht;
314
315 assert(ctx);
316
317 /* Ease our life a bit. */
318 ht = consumer_data.stream_list_ht;
319
320 rcu_read_lock();
321
322 /* Search the metadata associated with the session id of the given stream. */
323
324 cds_lfht_for_each_entry_duplicate(ht->ht,
325 ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
326 &session_id, &iter.iter, stream, node_session_id.node) {
327 if (!stream->metadata_flag) {
328 continue;
329 }
330
331 ret = do_sync_metadata(stream, ctx);
332 if (ret < 0) {
333 goto end;
334 }
335 }
336
337 /*
338 * Force return code to 0 (success) since ret might be ENODATA for instance
339 * which is not an error but rather that we should come back.
340 */
341 ret = 0;
342
343end:
344 rcu_read_unlock();
345 return ret;
346}
347
348static int consumer_stream_sync_metadata_index(
349 struct lttng_consumer_stream *stream,
350 const struct stream_subbuffer *subbuffer,
351 struct lttng_consumer_local_data *ctx)
352{
353 int ret;
354
355 /* Block until all the metadata is sent. */
356 pthread_mutex_lock(&stream->metadata_timer_lock);
357 assert(!stream->missed_metadata_flush);
358 stream->waiting_on_metadata = true;
359 pthread_mutex_unlock(&stream->metadata_timer_lock);
360
361 ret = consumer_stream_sync_metadata(ctx, stream->session_id);
362
363 pthread_mutex_lock(&stream->metadata_timer_lock);
364 stream->waiting_on_metadata = false;
365 if (stream->missed_metadata_flush) {
366 stream->missed_metadata_flush = false;
367 pthread_mutex_unlock(&stream->metadata_timer_lock);
368 (void) stream->read_subbuffer_ops.send_live_beacon(stream);
369 } else {
370 pthread_mutex_unlock(&stream->metadata_timer_lock);
371 }
372 if (ret < 0) {
373 goto end;
374 }
375
376 ret = consumer_stream_send_index(stream, subbuffer, ctx);
377end:
378 return ret;
379}
380
381/*
382 * Check if the local version of the metadata stream matches with the version
383 * of the metadata stream in the kernel. If it was updated, set the reset flag
384 * on the stream.
385 */
386static
387int metadata_stream_check_version(struct lttng_consumer_stream *stream,
388 const struct stream_subbuffer *subbuffer)
389{
390 if (stream->metadata_version == subbuffer->info.metadata.version) {
391 goto end;
392 }
393
394 DBG("New metadata version detected");
395 consumer_stream_metadata_set_version(stream,
396 subbuffer->info.metadata.version);
397
398 if (stream->read_subbuffer_ops.reset_metadata) {
399 stream->read_subbuffer_ops.reset_metadata(stream);
400 }
401
402end:
403 return 0;
404}
405
406struct lttng_consumer_stream *consumer_stream_create(
407 struct lttng_consumer_channel *channel,
408 uint64_t channel_key,
409 uint64_t stream_key,
410 const char *channel_name,
411 uint64_t relayd_id,
412 uint64_t session_id,
413 struct lttng_trace_chunk *trace_chunk,
414 int cpu,
415 int *alloc_ret,
416 enum consumer_channel_type type,
417 unsigned int monitor)
418{
419 int ret;
420 struct lttng_consumer_stream *stream;
421
422 stream = zmalloc(sizeof(*stream));
423 if (stream == NULL) {
424 PERROR("malloc struct lttng_consumer_stream");
425 ret = -ENOMEM;
426 goto end;
427 }
428
429 if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
430 ERR("Failed to acquire trace chunk reference during the creation of a stream");
431 ret = -1;
432 goto error;
433 }
434
435 rcu_read_lock();
436 stream->chan = channel;
437 stream->key = stream_key;
438 stream->trace_chunk = trace_chunk;
439 stream->out_fd = -1;
440 stream->out_fd_offset = 0;
441 stream->output_written = 0;
442 stream->net_seq_idx = relayd_id;
443 stream->session_id = session_id;
444 stream->monitor = monitor;
445 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
446 stream->index_file = NULL;
447 stream->last_sequence_number = -1ULL;
448 stream->rotate_position = -1ULL;
449 pthread_mutex_init(&stream->lock, NULL);
450 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
451
452 /* If channel is the metadata, flag this stream as metadata. */
453 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
454 stream->metadata_flag = 1;
455 /* Metadata is flat out. */
456 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
457 /* Live rendez-vous point. */
458 pthread_cond_init(&stream->metadata_rdv, NULL);
459 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
460 } else {
461 /* Format stream name to <channel_name>_<cpu_number> */
462 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
463 channel_name, cpu);
464 if (ret < 0) {
465 PERROR("snprintf stream name");
466 goto error;
467 }
468 }
469
470 switch (channel->output) {
471 case CONSUMER_CHANNEL_SPLICE:
472 stream->output = LTTNG_EVENT_SPLICE;
473 ret = utils_create_pipe(stream->splice_pipe);
474 if (ret < 0) {
475 goto error;
476 }
477 break;
478 case CONSUMER_CHANNEL_MMAP:
479 stream->output = LTTNG_EVENT_MMAP;
480 break;
481 default:
482 abort();
483 }
484
485 /* Key is always the wait_fd for streams. */
486 lttng_ht_node_init_u64(&stream->node, stream->key);
487
488 /* Init node per channel id key */
489 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
490
491 /* Init session id node with the stream session id */
492 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
493
494 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
495 " relayd_id %" PRIu64 ", session_id %" PRIu64,
496 stream->name, stream->key, channel_key,
497 stream->net_seq_idx, stream->session_id);
498
499 rcu_read_unlock();
500
501 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
502 stream->read_subbuffer_ops.lock =
503 consumer_stream_metadata_lock_all;
504 stream->read_subbuffer_ops.unlock =
505 consumer_stream_metadata_unlock_all;
506 stream->read_subbuffer_ops.pre_consume_subbuffer =
507 metadata_stream_check_version;
508 } else {
509 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
510 stream->read_subbuffer_ops.unlock =
511 consumer_stream_data_unlock_all;
512 stream->read_subbuffer_ops.pre_consume_subbuffer =
513 consumer_stream_update_stats;
514 if (channel->is_live) {
515 stream->read_subbuffer_ops.post_consume =
516 consumer_stream_sync_metadata_index;
517 } else {
518 stream->read_subbuffer_ops.post_consume =
519 consumer_stream_send_index;
520 }
521 }
522
523 if (channel->output == CONSUMER_CHANNEL_MMAP) {
524 stream->read_subbuffer_ops.consume_subbuffer =
525 consumer_stream_consume_mmap;
526 } else {
527 stream->read_subbuffer_ops.consume_subbuffer =
528 consumer_stream_consume_splice;
529 }
530
531 return stream;
532
533error:
534 rcu_read_unlock();
535 lttng_trace_chunk_put(stream->trace_chunk);
536 free(stream);
537end:
538 if (alloc_ret) {
539 *alloc_ret = ret;
540 }
541 return NULL;
542}
543
544/*
545 * Close stream on the relayd side. This call can destroy a relayd if the
546 * conditions are met.
547 *
548 * A RCU read side lock MUST be acquired if the relayd object was looked up in
549 * a hash table before calling this.
550 */
551void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
552 struct consumer_relayd_sock_pair *relayd)
553{
554 int ret;
555
556 assert(stream);
557 assert(relayd);
558
559 if (stream->sent_to_relayd) {
560 uatomic_dec(&relayd->refcount);
561 assert(uatomic_read(&relayd->refcount) >= 0);
562 }
563
564 /* Closing streams requires to lock the control socket. */
565 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
566 ret = relayd_send_close_stream(&relayd->control_sock,
567 stream->relayd_stream_id,
568 stream->next_net_seq_num - 1);
569 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
570 if (ret < 0) {
571 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
572 lttng_consumer_cleanup_relayd(relayd);
573 }
574
575 /* Both conditions are met, we destroy the relayd. */
576 if (uatomic_read(&relayd->refcount) == 0 &&
577 uatomic_read(&relayd->destroy_flag)) {
578 consumer_destroy_relayd(relayd);
579 }
580 stream->net_seq_idx = (uint64_t) -1ULL;
581 stream->sent_to_relayd = 0;
582}
583
584/*
585 * Close stream's file descriptors and, if needed, close stream also on the
586 * relayd side.
587 *
588 * The consumer data lock MUST be acquired.
589 * The stream lock MUST be acquired.
590 */
591void consumer_stream_close(struct lttng_consumer_stream *stream)
592{
593 int ret;
594 struct consumer_relayd_sock_pair *relayd;
595
596 assert(stream);
597
598 switch (consumer_data.type) {
599 case LTTNG_CONSUMER_KERNEL:
600 if (stream->mmap_base != NULL) {
601 ret = munmap(stream->mmap_base, stream->mmap_len);
602 if (ret != 0) {
603 PERROR("munmap");
604 }
605 }
606
607 if (stream->wait_fd >= 0) {
608 ret = close(stream->wait_fd);
609 if (ret) {
610 PERROR("close");
611 }
612 stream->wait_fd = -1;
613 }
614 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
615 utils_close_pipe(stream->splice_pipe);
616 }
617 break;
618 case LTTNG_CONSUMER32_UST:
619 case LTTNG_CONSUMER64_UST:
620 {
621 /*
622 * Special case for the metadata since the wait fd is an internal pipe
623 * polled in the metadata thread.
624 */
625 if (stream->metadata_flag && stream->chan->monitor) {
626 int rpipe = stream->ust_metadata_poll_pipe[0];
627
628 /*
629 * This will stop the channel timer if one and close the write side
630 * of the metadata poll pipe.
631 */
632 lttng_ustconsumer_close_metadata(stream->chan);
633 if (rpipe >= 0) {
634 ret = close(rpipe);
635 if (ret < 0) {
636 PERROR("closing metadata pipe read side");
637 }
638 stream->ust_metadata_poll_pipe[0] = -1;
639 }
640 }
641 break;
642 }
643 default:
644 ERR("Unknown consumer_data type");
645 assert(0);
646 }
647
648 /* Close output fd. Could be a socket or local file at this point. */
649 if (stream->out_fd >= 0) {
650 ret = close(stream->out_fd);
651 if (ret) {
652 PERROR("close");
653 }
654 stream->out_fd = -1;
655 }
656
657 if (stream->index_file) {
658 lttng_index_file_put(stream->index_file);
659 stream->index_file = NULL;
660 }
661
662 lttng_trace_chunk_put(stream->trace_chunk);
663 stream->trace_chunk = NULL;
664
665 /* Check and cleanup relayd if needed. */
666 rcu_read_lock();
667 relayd = consumer_find_relayd(stream->net_seq_idx);
668 if (relayd != NULL) {
669 consumer_stream_relayd_close(stream, relayd);
670 }
671 rcu_read_unlock();
672}
673
674/*
675 * Delete the stream from all possible hash tables.
676 *
677 * The consumer data lock MUST be acquired.
678 * The stream lock MUST be acquired.
679 */
680void consumer_stream_delete(struct lttng_consumer_stream *stream,
681 struct lttng_ht *ht)
682{
683 int ret;
684 struct lttng_ht_iter iter;
685
686 assert(stream);
687 /* Should NEVER be called not in monitor mode. */
688 assert(stream->chan->monitor);
689
690 rcu_read_lock();
691
692 if (ht) {
693 iter.iter.node = &stream->node.node;
694 ret = lttng_ht_del(ht, &iter);
695 assert(!ret);
696 }
697
698 /* Delete from stream per channel ID hash table. */
699 iter.iter.node = &stream->node_channel_id.node;
700 /*
701 * The returned value is of no importance. Even if the node is NOT in the
702 * hash table, we continue since we may have been called by a code path
703 * that did not add the stream to a (all) hash table. Same goes for the
704 * next call ht del call.
705 */
706 (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
707
708 /* Delete from the global stream list. */
709 iter.iter.node = &stream->node_session_id.node;
710 /* See the previous ht del on why we ignore the returned value. */
711 (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
712
713 rcu_read_unlock();
714
715 if (!stream->metadata_flag) {
716 /* Decrement the stream count of the global consumer data. */
717 assert(consumer_data.stream_count > 0);
718 consumer_data.stream_count--;
719 }
720}
721
722/*
723 * Free the given stream within a RCU call.
724 */
725void consumer_stream_free(struct lttng_consumer_stream *stream)
726{
727 assert(stream);
728
729 metadata_bucket_destroy(stream->metadata_bucket);
730 call_rcu(&stream->node.head, free_stream_rcu);
731}
732
733/*
734 * Destroy the stream's buffers of the tracer.
735 */
736void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
737{
738 assert(stream);
739
740 switch (consumer_data.type) {
741 case LTTNG_CONSUMER_KERNEL:
742 break;
743 case LTTNG_CONSUMER32_UST:
744 case LTTNG_CONSUMER64_UST:
745 lttng_ustconsumer_del_stream(stream);
746 break;
747 default:
748 ERR("Unknown consumer_data type");
749 assert(0);
750 }
751}
752
753/*
754 * Destroy and close a already created stream.
755 */
756static void destroy_close_stream(struct lttng_consumer_stream *stream)
757{
758 assert(stream);
759
760 DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
761
762 /* Destroy tracer buffers of the stream. */
763 consumer_stream_destroy_buffers(stream);
764 /* Close down everything including the relayd if one. */
765 consumer_stream_close(stream);
766}
767
768/*
769 * Decrement the stream's channel refcount and if down to 0, return the channel
770 * pointer so it can be destroyed by the caller or NULL if not.
771 */
772static struct lttng_consumer_channel *unref_channel(
773 struct lttng_consumer_stream *stream)
774{
775 struct lttng_consumer_channel *free_chan = NULL;
776
777 assert(stream);
778 assert(stream->chan);
779
780 /* Update refcount of channel and see if we need to destroy it. */
781 if (!uatomic_sub_return(&stream->chan->refcount, 1)
782 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
783 free_chan = stream->chan;
784 }
785
786 return free_chan;
787}
788
789/*
790 * Destroy a stream completely. This will delete, close and free the stream.
791 * Once return, the stream is NO longer usable. Its channel may get destroyed
792 * if conditions are met for a monitored stream.
793 *
794 * This MUST be called WITHOUT the consumer data and stream lock acquired if
795 * the stream is in _monitor_ mode else it does not matter.
796 */
797void consumer_stream_destroy(struct lttng_consumer_stream *stream,
798 struct lttng_ht *ht)
799{
800 assert(stream);
801
802 /* Stream is in monitor mode. */
803 if (stream->monitor) {
804 struct lttng_consumer_channel *free_chan = NULL;
805
806 /*
807 * This means that the stream was successfully removed from the streams
808 * list of the channel and sent to the right thread managing this
809 * stream thus being globally visible.
810 */
811 if (stream->globally_visible) {
812 pthread_mutex_lock(&consumer_data.lock);
813 pthread_mutex_lock(&stream->chan->lock);
814 pthread_mutex_lock(&stream->lock);
815 /* Remove every reference of the stream in the consumer. */
816 consumer_stream_delete(stream, ht);
817
818 destroy_close_stream(stream);
819
820 /* Update channel's refcount of the stream. */
821 free_chan = unref_channel(stream);
822
823 /* Indicates that the consumer data state MUST be updated after this. */
824 consumer_data.need_update = 1;
825
826 pthread_mutex_unlock(&stream->lock);
827 pthread_mutex_unlock(&stream->chan->lock);
828 pthread_mutex_unlock(&consumer_data.lock);
829 } else {
830 /*
831 * If the stream is not visible globally, this needs to be done
832 * outside of the consumer data lock section.
833 */
834 free_chan = unref_channel(stream);
835 }
836
837 if (free_chan) {
838 consumer_del_channel(free_chan);
839 }
840 } else {
841 destroy_close_stream(stream);
842 }
843
844 /* Free stream within a RCU call. */
845 lttng_trace_chunk_put(stream->trace_chunk);
846 stream->trace_chunk = NULL;
847 consumer_stream_free(stream);
848}
849
850/*
851 * Write index of a specific stream either on the relayd or local disk.
852 *
853 * Return 0 on success or else a negative value.
854 */
855int consumer_stream_write_index(struct lttng_consumer_stream *stream,
856 struct ctf_packet_index *element)
857{
858 int ret;
859
860 assert(stream);
861 assert(element);
862
863 rcu_read_lock();
864 if (stream->net_seq_idx != (uint64_t) -1ULL) {
865 struct consumer_relayd_sock_pair *relayd;
866 relayd = consumer_find_relayd(stream->net_seq_idx);
867 if (relayd) {
868 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
869 ret = relayd_send_index(&relayd->control_sock, element,
870 stream->relayd_stream_id, stream->next_net_seq_num - 1);
871 if (ret < 0) {
872 /*
873 * Communication error with lttng-relayd,
874 * perform cleanup now
875 */
876 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
877 lttng_consumer_cleanup_relayd(relayd);
878 ret = -1;
879 }
880 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
881 } else {
882 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
883 stream->key, stream->net_seq_idx);
884 ret = -1;
885 }
886 } else {
887 if (lttng_index_file_write(stream->index_file, element)) {
888 ret = -1;
889 } else {
890 ret = 0;
891 }
892 }
893 if (ret < 0) {
894 goto error;
895 }
896
897error:
898 rcu_read_unlock();
899 return ret;
900}
901
902int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
903 bool create_index)
904{
905 int ret;
906 enum lttng_trace_chunk_status chunk_status;
907 const int flags = O_WRONLY | O_CREAT | O_TRUNC;
908 const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
909 char stream_path[LTTNG_PATH_MAX];
910
911 ASSERT_LOCKED(stream->lock);
912 assert(stream->trace_chunk);
913
914 ret = utils_stream_file_path(stream->chan->pathname, stream->name,
915 stream->chan->tracefile_size,
916 stream->tracefile_count_current, NULL,
917 stream_path, sizeof(stream_path));
918 if (ret < 0) {
919 goto end;
920 }
921
922 if (stream->out_fd >= 0) {
923 ret = close(stream->out_fd);
924 if (ret < 0) {
925 PERROR("Failed to close stream file \"%s\"",
926 stream->name);
927 goto end;
928 }
929 stream->out_fd = -1;
930 }
931
932 DBG("Opening stream output file \"%s\"", stream_path);
933 chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
934 flags, mode, &stream->out_fd, false);
935 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
936 ERR("Failed to open stream file \"%s\"", stream->name);
937 ret = -1;
938 goto end;
939 }
940
941 if (!stream->metadata_flag && (create_index || stream->index_file)) {
942 if (stream->index_file) {
943 lttng_index_file_put(stream->index_file);
944 }
945 chunk_status = lttng_index_file_create_from_trace_chunk(
946 stream->trace_chunk,
947 stream->chan->pathname,
948 stream->name,
949 stream->chan->tracefile_size,
950 stream->tracefile_count_current,
951 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
952 false, &stream->index_file);
953 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
954 ret = -1;
955 goto end;
956 }
957 }
958
959 /* Reset current size because we just perform a rotation. */
960 stream->tracefile_size_current = 0;
961 stream->out_fd_offset = 0;
962end:
963 return ret;
964}
965
966int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
967{
968 int ret;
969
970 stream->tracefile_count_current++;
971 if (stream->chan->tracefile_count > 0) {
972 stream->tracefile_count_current %=
973 stream->chan->tracefile_count;
974 }
975
976 DBG("Rotating output files of stream \"%s\"", stream->name);
977 ret = consumer_stream_create_output_files(stream, true);
978 if (ret) {
979 goto end;
980 }
981
982end:
983 return ret;
984}
985
986bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
987{
988 /*
989 * This function does not take a const stream since
990 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
991 */
992 assert(stream);
993 return cds_lfht_is_node_deleted(&stream->node.node);
994}
995
996static ssize_t metadata_bucket_flush(
997 const struct stream_subbuffer *buffer, void *data)
998{
999 ssize_t ret;
1000 struct lttng_consumer_stream *stream = data;
1001
1002 ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1003 if (ret < 0) {
1004 goto end;
1005 }
1006end:
1007 return ret;
1008}
1009
1010static ssize_t metadata_bucket_consume(
1011 struct lttng_consumer_local_data *unused,
1012 struct lttng_consumer_stream *stream,
1013 const struct stream_subbuffer *subbuffer)
1014{
1015 ssize_t ret;
1016 enum metadata_bucket_status status;
1017
1018 status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1019 switch (status) {
1020 case METADATA_BUCKET_STATUS_OK:
1021 /* Return consumed size. */
1022 ret = subbuffer->buffer.buffer.size;
1023 break;
1024 default:
1025 ret = -1;
1026 }
1027
1028 return ret;
1029}
1030
1031int consumer_stream_enable_metadata_bucketization(
1032 struct lttng_consumer_stream *stream)
1033{
1034 int ret = 0;
1035
1036 assert(stream->metadata_flag);
1037 assert(!stream->metadata_bucket);
1038 assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1039
1040 stream->metadata_bucket = metadata_bucket_create(
1041 metadata_bucket_flush, stream);
1042 if (!stream->metadata_bucket) {
1043 ret = -1;
1044 goto end;
1045 }
1046
1047 stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1048end:
1049 return ret;
1050}
1051
1052void consumer_stream_metadata_set_version(
1053 struct lttng_consumer_stream *stream, uint64_t new_version)
1054{
1055 assert(new_version > stream->metadata_version);
1056 stream->metadata_version = new_version;
1057 stream->reset_metadata_flag = 1;
1058
1059 if (stream->metadata_bucket) {
1060 metadata_bucket_reset(stream->metadata_bucket);
1061 }
1062}
This page took 0.026555 seconds and 4 git commands to generate.