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