Fix: incorrect specifier %lu used with size_t argument
[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 stream->metadata_version = subbuffer->info.metadata.version;
397 stream->reset_metadata_flag = 1;
398
399 if (stream->metadata_bucket) {
400 metadata_bucket_reset(stream->metadata_bucket);
401 }
402
403 if (stream->read_subbuffer_ops.reset_metadata) {
404 stream->read_subbuffer_ops.reset_metadata(stream);
405 }
406
407 end:
408 return 0;
409 }
410
411 struct lttng_consumer_stream *consumer_stream_create(
412 struct lttng_consumer_channel *channel,
413 uint64_t channel_key,
414 uint64_t stream_key,
415 const char *channel_name,
416 uint64_t relayd_id,
417 uint64_t session_id,
418 struct lttng_trace_chunk *trace_chunk,
419 int cpu,
420 int *alloc_ret,
421 enum consumer_channel_type type,
422 unsigned int monitor)
423 {
424 int ret;
425 struct lttng_consumer_stream *stream;
426
427 stream = zmalloc(sizeof(*stream));
428 if (stream == NULL) {
429 PERROR("malloc struct lttng_consumer_stream");
430 ret = -ENOMEM;
431 goto end;
432 }
433
434 if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
435 ERR("Failed to acquire trace chunk reference during the creation of a stream");
436 ret = -1;
437 goto error;
438 }
439
440 rcu_read_lock();
441 stream->chan = channel;
442 stream->key = stream_key;
443 stream->trace_chunk = trace_chunk;
444 stream->out_fd = -1;
445 stream->out_fd_offset = 0;
446 stream->output_written = 0;
447 stream->net_seq_idx = relayd_id;
448 stream->session_id = session_id;
449 stream->monitor = monitor;
450 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
451 stream->index_file = NULL;
452 stream->last_sequence_number = -1ULL;
453 stream->rotate_position = -1ULL;
454 pthread_mutex_init(&stream->lock, NULL);
455 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
456
457 /* If channel is the metadata, flag this stream as metadata. */
458 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
459 stream->metadata_flag = 1;
460 /* Metadata is flat out. */
461 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
462 /* Live rendez-vous point. */
463 pthread_cond_init(&stream->metadata_rdv, NULL);
464 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
465 } else {
466 /* Format stream name to <channel_name>_<cpu_number> */
467 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
468 channel_name, cpu);
469 if (ret < 0) {
470 PERROR("snprintf stream name");
471 goto error;
472 }
473 }
474
475 switch (channel->output) {
476 case CONSUMER_CHANNEL_SPLICE:
477 stream->output = LTTNG_EVENT_SPLICE;
478 ret = utils_create_pipe(stream->splice_pipe);
479 if (ret < 0) {
480 goto error;
481 }
482 break;
483 case CONSUMER_CHANNEL_MMAP:
484 stream->output = LTTNG_EVENT_MMAP;
485 break;
486 default:
487 abort();
488 }
489
490 /* Key is always the wait_fd for streams. */
491 lttng_ht_node_init_u64(&stream->node, stream->key);
492
493 /* Init node per channel id key */
494 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
495
496 /* Init session id node with the stream session id */
497 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
498
499 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
500 " relayd_id %" PRIu64 ", session_id %" PRIu64,
501 stream->name, stream->key, channel_key,
502 stream->net_seq_idx, stream->session_id);
503
504 rcu_read_unlock();
505
506 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
507 stream->read_subbuffer_ops.lock =
508 consumer_stream_metadata_lock_all;
509 stream->read_subbuffer_ops.unlock =
510 consumer_stream_metadata_unlock_all;
511 stream->read_subbuffer_ops.pre_consume_subbuffer =
512 metadata_stream_check_version;
513 } else {
514 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
515 stream->read_subbuffer_ops.unlock =
516 consumer_stream_data_unlock_all;
517 stream->read_subbuffer_ops.pre_consume_subbuffer =
518 consumer_stream_update_stats;
519 if (channel->is_live) {
520 stream->read_subbuffer_ops.post_consume =
521 consumer_stream_sync_metadata_index;
522 } else {
523 stream->read_subbuffer_ops.post_consume =
524 consumer_stream_send_index;
525 }
526 }
527
528 if (channel->output == CONSUMER_CHANNEL_MMAP) {
529 stream->read_subbuffer_ops.consume_subbuffer =
530 consumer_stream_consume_mmap;
531 } else {
532 stream->read_subbuffer_ops.consume_subbuffer =
533 consumer_stream_consume_splice;
534 }
535
536 return stream;
537
538 error:
539 rcu_read_unlock();
540 lttng_trace_chunk_put(stream->trace_chunk);
541 free(stream);
542 end:
543 if (alloc_ret) {
544 *alloc_ret = ret;
545 }
546 return NULL;
547 }
548
549 /*
550 * Close stream on the relayd side. This call can destroy a relayd if the
551 * conditions are met.
552 *
553 * A RCU read side lock MUST be acquired if the relayd object was looked up in
554 * a hash table before calling this.
555 */
556 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
557 struct consumer_relayd_sock_pair *relayd)
558 {
559 int ret;
560
561 assert(stream);
562 assert(relayd);
563
564 if (stream->sent_to_relayd) {
565 uatomic_dec(&relayd->refcount);
566 assert(uatomic_read(&relayd->refcount) >= 0);
567 }
568
569 /* Closing streams requires to lock the control socket. */
570 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
571 ret = relayd_send_close_stream(&relayd->control_sock,
572 stream->relayd_stream_id,
573 stream->next_net_seq_num - 1);
574 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
575 if (ret < 0) {
576 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
577 lttng_consumer_cleanup_relayd(relayd);
578 }
579
580 /* Both conditions are met, we destroy the relayd. */
581 if (uatomic_read(&relayd->refcount) == 0 &&
582 uatomic_read(&relayd->destroy_flag)) {
583 consumer_destroy_relayd(relayd);
584 }
585 stream->net_seq_idx = (uint64_t) -1ULL;
586 stream->sent_to_relayd = 0;
587 }
588
589 /*
590 * Close stream's file descriptors and, if needed, close stream also on the
591 * relayd side.
592 *
593 * The consumer data lock MUST be acquired.
594 * The stream lock MUST be acquired.
595 */
596 void consumer_stream_close(struct lttng_consumer_stream *stream)
597 {
598 int ret;
599 struct consumer_relayd_sock_pair *relayd;
600
601 assert(stream);
602
603 switch (consumer_data.type) {
604 case LTTNG_CONSUMER_KERNEL:
605 if (stream->mmap_base != NULL) {
606 ret = munmap(stream->mmap_base, stream->mmap_len);
607 if (ret != 0) {
608 PERROR("munmap");
609 }
610 }
611
612 if (stream->wait_fd >= 0) {
613 ret = close(stream->wait_fd);
614 if (ret) {
615 PERROR("close");
616 }
617 stream->wait_fd = -1;
618 }
619 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
620 utils_close_pipe(stream->splice_pipe);
621 }
622 break;
623 case LTTNG_CONSUMER32_UST:
624 case LTTNG_CONSUMER64_UST:
625 {
626 /*
627 * Special case for the metadata since the wait fd is an internal pipe
628 * polled in the metadata thread.
629 */
630 if (stream->metadata_flag && stream->chan->monitor) {
631 int rpipe = stream->ust_metadata_poll_pipe[0];
632
633 /*
634 * This will stop the channel timer if one and close the write side
635 * of the metadata poll pipe.
636 */
637 lttng_ustconsumer_close_metadata(stream->chan);
638 if (rpipe >= 0) {
639 ret = close(rpipe);
640 if (ret < 0) {
641 PERROR("closing metadata pipe read side");
642 }
643 stream->ust_metadata_poll_pipe[0] = -1;
644 }
645 }
646 break;
647 }
648 default:
649 ERR("Unknown consumer_data type");
650 assert(0);
651 }
652
653 /* Close output fd. Could be a socket or local file at this point. */
654 if (stream->out_fd >= 0) {
655 ret = close(stream->out_fd);
656 if (ret) {
657 PERROR("close");
658 }
659 stream->out_fd = -1;
660 }
661
662 if (stream->index_file) {
663 lttng_index_file_put(stream->index_file);
664 stream->index_file = NULL;
665 }
666
667 lttng_trace_chunk_put(stream->trace_chunk);
668 stream->trace_chunk = NULL;
669
670 /* Check and cleanup relayd if needed. */
671 rcu_read_lock();
672 relayd = consumer_find_relayd(stream->net_seq_idx);
673 if (relayd != NULL) {
674 consumer_stream_relayd_close(stream, relayd);
675 }
676 rcu_read_unlock();
677 }
678
679 /*
680 * Delete the stream from all possible hash tables.
681 *
682 * The consumer data lock MUST be acquired.
683 * The stream lock MUST be acquired.
684 */
685 void consumer_stream_delete(struct lttng_consumer_stream *stream,
686 struct lttng_ht *ht)
687 {
688 int ret;
689 struct lttng_ht_iter iter;
690
691 assert(stream);
692 /* Should NEVER be called not in monitor mode. */
693 assert(stream->chan->monitor);
694
695 rcu_read_lock();
696
697 if (ht) {
698 iter.iter.node = &stream->node.node;
699 ret = lttng_ht_del(ht, &iter);
700 assert(!ret);
701 }
702
703 /* Delete from stream per channel ID hash table. */
704 iter.iter.node = &stream->node_channel_id.node;
705 /*
706 * The returned value is of no importance. Even if the node is NOT in the
707 * hash table, we continue since we may have been called by a code path
708 * that did not add the stream to a (all) hash table. Same goes for the
709 * next call ht del call.
710 */
711 (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
712
713 /* Delete from the global stream list. */
714 iter.iter.node = &stream->node_session_id.node;
715 /* See the previous ht del on why we ignore the returned value. */
716 (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
717
718 rcu_read_unlock();
719
720 if (!stream->metadata_flag) {
721 /* Decrement the stream count of the global consumer data. */
722 assert(consumer_data.stream_count > 0);
723 consumer_data.stream_count--;
724 }
725 }
726
727 /*
728 * Free the given stream within a RCU call.
729 */
730 void consumer_stream_free(struct lttng_consumer_stream *stream)
731 {
732 assert(stream);
733
734 metadata_bucket_destroy(stream->metadata_bucket);
735 call_rcu(&stream->node.head, free_stream_rcu);
736 }
737
738 /*
739 * Destroy the stream's buffers of the tracer.
740 */
741 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
742 {
743 assert(stream);
744
745 switch (consumer_data.type) {
746 case LTTNG_CONSUMER_KERNEL:
747 break;
748 case LTTNG_CONSUMER32_UST:
749 case LTTNG_CONSUMER64_UST:
750 lttng_ustconsumer_del_stream(stream);
751 break;
752 default:
753 ERR("Unknown consumer_data type");
754 assert(0);
755 }
756 }
757
758 /*
759 * Destroy and close a already created stream.
760 */
761 static void destroy_close_stream(struct lttng_consumer_stream *stream)
762 {
763 assert(stream);
764
765 DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
766
767 /* Destroy tracer buffers of the stream. */
768 consumer_stream_destroy_buffers(stream);
769 /* Close down everything including the relayd if one. */
770 consumer_stream_close(stream);
771 }
772
773 /*
774 * Decrement the stream's channel refcount and if down to 0, return the channel
775 * pointer so it can be destroyed by the caller or NULL if not.
776 */
777 static struct lttng_consumer_channel *unref_channel(
778 struct lttng_consumer_stream *stream)
779 {
780 struct lttng_consumer_channel *free_chan = NULL;
781
782 assert(stream);
783 assert(stream->chan);
784
785 /* Update refcount of channel and see if we need to destroy it. */
786 if (!uatomic_sub_return(&stream->chan->refcount, 1)
787 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
788 free_chan = stream->chan;
789 }
790
791 return free_chan;
792 }
793
794 /*
795 * Destroy a stream completely. This will delete, close and free the stream.
796 * Once return, the stream is NO longer usable. Its channel may get destroyed
797 * if conditions are met for a monitored stream.
798 *
799 * This MUST be called WITHOUT the consumer data and stream lock acquired if
800 * the stream is in _monitor_ mode else it does not matter.
801 */
802 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
803 struct lttng_ht *ht)
804 {
805 assert(stream);
806
807 /* Stream is in monitor mode. */
808 if (stream->monitor) {
809 struct lttng_consumer_channel *free_chan = NULL;
810
811 /*
812 * This means that the stream was successfully removed from the streams
813 * list of the channel and sent to the right thread managing this
814 * stream thus being globally visible.
815 */
816 if (stream->globally_visible) {
817 pthread_mutex_lock(&consumer_data.lock);
818 pthread_mutex_lock(&stream->chan->lock);
819 pthread_mutex_lock(&stream->lock);
820 /* Remove every reference of the stream in the consumer. */
821 consumer_stream_delete(stream, ht);
822
823 destroy_close_stream(stream);
824
825 /* Update channel's refcount of the stream. */
826 free_chan = unref_channel(stream);
827
828 /* Indicates that the consumer data state MUST be updated after this. */
829 consumer_data.need_update = 1;
830
831 pthread_mutex_unlock(&stream->lock);
832 pthread_mutex_unlock(&stream->chan->lock);
833 pthread_mutex_unlock(&consumer_data.lock);
834 } else {
835 /*
836 * If the stream is not visible globally, this needs to be done
837 * outside of the consumer data lock section.
838 */
839 free_chan = unref_channel(stream);
840 }
841
842 if (free_chan) {
843 consumer_del_channel(free_chan);
844 }
845 } else {
846 destroy_close_stream(stream);
847 }
848
849 /* Free stream within a RCU call. */
850 lttng_trace_chunk_put(stream->trace_chunk);
851 stream->trace_chunk = NULL;
852 consumer_stream_free(stream);
853 }
854
855 /*
856 * Write index of a specific stream either on the relayd or local disk.
857 *
858 * Return 0 on success or else a negative value.
859 */
860 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
861 struct ctf_packet_index *element)
862 {
863 int ret;
864
865 assert(stream);
866 assert(element);
867
868 rcu_read_lock();
869 if (stream->net_seq_idx != (uint64_t) -1ULL) {
870 struct consumer_relayd_sock_pair *relayd;
871 relayd = consumer_find_relayd(stream->net_seq_idx);
872 if (relayd) {
873 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
874 ret = relayd_send_index(&relayd->control_sock, element,
875 stream->relayd_stream_id, stream->next_net_seq_num - 1);
876 if (ret < 0) {
877 /*
878 * Communication error with lttng-relayd,
879 * perform cleanup now
880 */
881 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
882 lttng_consumer_cleanup_relayd(relayd);
883 ret = -1;
884 }
885 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
886 } else {
887 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
888 stream->key, stream->net_seq_idx);
889 ret = -1;
890 }
891 } else {
892 if (lttng_index_file_write(stream->index_file, element)) {
893 ret = -1;
894 } else {
895 ret = 0;
896 }
897 }
898 if (ret < 0) {
899 goto error;
900 }
901
902 error:
903 rcu_read_unlock();
904 return ret;
905 }
906
907 int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
908 bool create_index)
909 {
910 int ret;
911 enum lttng_trace_chunk_status chunk_status;
912 const int flags = O_WRONLY | O_CREAT | O_TRUNC;
913 const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
914 char stream_path[LTTNG_PATH_MAX];
915
916 ASSERT_LOCKED(stream->lock);
917 assert(stream->trace_chunk);
918
919 ret = utils_stream_file_path(stream->chan->pathname, stream->name,
920 stream->chan->tracefile_size,
921 stream->tracefile_count_current, NULL,
922 stream_path, sizeof(stream_path));
923 if (ret < 0) {
924 goto end;
925 }
926
927 if (stream->out_fd >= 0) {
928 ret = close(stream->out_fd);
929 if (ret < 0) {
930 PERROR("Failed to close stream file \"%s\"",
931 stream->name);
932 goto end;
933 }
934 stream->out_fd = -1;
935 }
936
937 DBG("Opening stream output file \"%s\"", stream_path);
938 chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
939 flags, mode, &stream->out_fd, false);
940 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
941 ERR("Failed to open stream file \"%s\"", stream->name);
942 ret = -1;
943 goto end;
944 }
945
946 if (!stream->metadata_flag && (create_index || stream->index_file)) {
947 if (stream->index_file) {
948 lttng_index_file_put(stream->index_file);
949 }
950 chunk_status = lttng_index_file_create_from_trace_chunk(
951 stream->trace_chunk,
952 stream->chan->pathname,
953 stream->name,
954 stream->chan->tracefile_size,
955 stream->tracefile_count_current,
956 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
957 false, &stream->index_file);
958 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
959 ret = -1;
960 goto end;
961 }
962 }
963
964 /* Reset current size because we just perform a rotation. */
965 stream->tracefile_size_current = 0;
966 stream->out_fd_offset = 0;
967 end:
968 return ret;
969 }
970
971 int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
972 {
973 int ret;
974
975 stream->tracefile_count_current++;
976 if (stream->chan->tracefile_count > 0) {
977 stream->tracefile_count_current %=
978 stream->chan->tracefile_count;
979 }
980
981 DBG("Rotating output files of stream \"%s\"", stream->name);
982 ret = consumer_stream_create_output_files(stream, true);
983 if (ret) {
984 goto end;
985 }
986
987 end:
988 return ret;
989 }
990
991 bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
992 {
993 /*
994 * This function does not take a const stream since
995 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
996 */
997 assert(stream);
998 return cds_lfht_is_node_deleted(&stream->node.node);
999 }
1000
1001 static ssize_t metadata_bucket_flush(
1002 const struct stream_subbuffer *buffer, void *data)
1003 {
1004 ssize_t ret;
1005 struct lttng_consumer_stream *stream = data;
1006
1007 ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1008 if (ret < 0) {
1009 goto end;
1010 }
1011 end:
1012 return ret;
1013 }
1014
1015 static ssize_t metadata_bucket_consume(
1016 struct lttng_consumer_local_data *unused,
1017 struct lttng_consumer_stream *stream,
1018 const struct stream_subbuffer *subbuffer)
1019 {
1020 ssize_t ret;
1021 enum metadata_bucket_status status;
1022
1023 status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1024 switch (status) {
1025 case METADATA_BUCKET_STATUS_OK:
1026 /* Return consumed size. */
1027 ret = subbuffer->buffer.buffer.size;
1028 break;
1029 default:
1030 ret = -1;
1031 }
1032
1033 return ret;
1034 }
1035
1036 int consumer_stream_enable_metadata_bucketization(
1037 struct lttng_consumer_stream *stream)
1038 {
1039 int ret = 0;
1040
1041 assert(stream->metadata_flag);
1042 assert(!stream->metadata_bucket);
1043 assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1044
1045 stream->metadata_bucket = metadata_bucket_create(
1046 metadata_bucket_flush, stream);
1047 if (!stream->metadata_bucket) {
1048 ret = -1;
1049 goto end;
1050 }
1051
1052 stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1053 end:
1054 return ret;
1055 }
This page took 0.082399 seconds and 4 git commands to generate.