Fix: ust-consumer: segfault on snapshot after regenerate metadata
[lttng-tools.git] / src / common / consumer / consumer-stream.cpp
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 <inttypes.h>
12 #include <sys/mman.h>
13 #include <unistd.h>
14
15 #include <common/common.h>
16 #include <common/consumer/consumer-timer.h>
17 #include <common/consumer/consumer-timer.h>
18 #include <common/consumer/consumer.h>
19 #include <common/consumer/consumer.h>
20 #include <common/consumer/metadata-bucket.h>
21 #include <common/consumer/metadata-bucket.h>
22 #include <common/index/index.h>
23 #include <common/kernel-consumer/kernel-consumer.h>
24 #include <common/kernel-ctl/kernel-ctl.h>
25 #include <common/macros.h>
26 #include <common/relayd/relayd.h>
27 #include <common/ust-consumer/ust-consumer.h>
28 #include <common/utils.h>
29
30 #include "consumer-stream.h"
31
32 /*
33 * RCU call to free stream. MUST only be used with call_rcu().
34 */
35 static void free_stream_rcu(struct rcu_head *head)
36 {
37 struct lttng_ht_node_u64 *node =
38 caa_container_of(head, struct lttng_ht_node_u64, head);
39 struct lttng_consumer_stream *stream =
40 caa_container_of(node, struct lttng_consumer_stream, node);
41
42 pthread_mutex_destroy(&stream->lock);
43 free(stream);
44 }
45
46 static void consumer_stream_data_lock_all(struct lttng_consumer_stream *stream)
47 {
48 pthread_mutex_lock(&stream->chan->lock);
49 pthread_mutex_lock(&stream->lock);
50 }
51
52 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream *stream)
53 {
54 pthread_mutex_unlock(&stream->lock);
55 pthread_mutex_unlock(&stream->chan->lock);
56 }
57
58 static void consumer_stream_data_assert_locked_all(struct lttng_consumer_stream *stream)
59 {
60 ASSERT_LOCKED(stream->lock);
61 ASSERT_LOCKED(stream->chan->lock);
62 }
63
64 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream *stream)
65 {
66 consumer_stream_data_lock_all(stream);
67 pthread_mutex_lock(&stream->metadata_rdv_lock);
68 }
69
70 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream *stream)
71 {
72 pthread_mutex_unlock(&stream->metadata_rdv_lock);
73 consumer_stream_data_unlock_all(stream);
74 }
75
76 static void consumer_stream_metadata_assert_locked_all(struct lttng_consumer_stream *stream)
77 {
78 ASSERT_LOCKED(stream->metadata_rdv_lock);
79 consumer_stream_data_assert_locked_all(stream);
80 }
81
82 /* Only used for data streams. */
83 static int consumer_stream_update_stats(struct lttng_consumer_stream *stream,
84 const struct stream_subbuffer *subbuf)
85 {
86 int ret = 0;
87 uint64_t sequence_number;
88 const uint64_t discarded_events = subbuf->info.data.events_discarded;
89
90 if (!subbuf->info.data.sequence_number.is_set) {
91 /* Command not supported by the tracer. */
92 sequence_number = -1ULL;
93 stream->sequence_number_unavailable = true;
94 } else {
95 sequence_number = subbuf->info.data.sequence_number.value;
96 }
97
98 /*
99 * Start the sequence when we extract the first packet in case we don't
100 * start at 0 (for example if a consumer is not connected to the
101 * session immediately after the beginning).
102 */
103 if (stream->last_sequence_number == -1ULL) {
104 stream->last_sequence_number = sequence_number;
105 } else if (sequence_number > stream->last_sequence_number) {
106 stream->chan->lost_packets += sequence_number -
107 stream->last_sequence_number - 1;
108 } else {
109 /* seq <= last_sequence_number */
110 ERR("Sequence number inconsistent : prev = %" PRIu64
111 ", current = %" PRIu64,
112 stream->last_sequence_number, sequence_number);
113 ret = -1;
114 goto end;
115 }
116 stream->last_sequence_number = sequence_number;
117
118 if (discarded_events < stream->last_discarded_events) {
119 /*
120 * Overflow has occurred. We assume only one wrap-around
121 * has occurred.
122 */
123 stream->chan->discarded_events +=
124 (1ULL << (CAA_BITS_PER_LONG - 1)) -
125 stream->last_discarded_events +
126 discarded_events;
127 } else {
128 stream->chan->discarded_events += discarded_events -
129 stream->last_discarded_events;
130 }
131 stream->last_discarded_events = discarded_events;
132 ret = 0;
133
134 end:
135 return ret;
136 }
137
138 static
139 void ctf_packet_index_populate(struct ctf_packet_index *index,
140 off_t offset, const struct stream_subbuffer *subbuffer)
141 {
142 *index = (typeof(*index)){
143 .offset = htobe64(offset),
144 .packet_size = htobe64(subbuffer->info.data.packet_size),
145 .content_size = htobe64(subbuffer->info.data.content_size),
146 .timestamp_begin = htobe64(
147 subbuffer->info.data.timestamp_begin),
148 .timestamp_end = htobe64(
149 subbuffer->info.data.timestamp_end),
150 .events_discarded = htobe64(
151 subbuffer->info.data.events_discarded),
152 .stream_id = htobe64(subbuffer->info.data.stream_id),
153 .stream_instance_id = htobe64(
154 subbuffer->info.data.stream_instance_id.is_set ?
155 subbuffer->info.data.stream_instance_id.value : -1ULL),
156 .packet_seq_num = htobe64(
157 subbuffer->info.data.sequence_number.is_set ?
158 subbuffer->info.data.sequence_number.value : -1ULL),
159 };
160 }
161
162 static ssize_t consumer_stream_consume_mmap(
163 struct lttng_consumer_local_data *ctx,
164 struct lttng_consumer_stream *stream,
165 const struct stream_subbuffer *subbuffer)
166 {
167 const unsigned long padding_size =
168 subbuffer->info.data.padded_subbuf_size -
169 subbuffer->info.data.subbuf_size;
170 const ssize_t written_bytes = lttng_consumer_on_read_subbuffer_mmap(
171 stream, &subbuffer->buffer.buffer, padding_size);
172
173 if (stream->net_seq_idx == -1ULL) {
174 /*
175 * When writing on disk, check that only the subbuffer (no
176 * padding) was written to disk.
177 */
178 if (written_bytes != subbuffer->info.data.padded_subbuf_size) {
179 DBG("Failed to write the entire padded subbuffer on disk (written_bytes: %zd, padded subbuffer size %lu)",
180 written_bytes,
181 subbuffer->info.data.padded_subbuf_size);
182 }
183 } else {
184 /*
185 * When streaming over the network, check that the entire
186 * subbuffer including padding was successfully written.
187 */
188 if (written_bytes != subbuffer->info.data.subbuf_size) {
189 DBG("Failed to write only the subbuffer over the network (written_bytes: %zd, subbuffer size %lu)",
190 written_bytes,
191 subbuffer->info.data.subbuf_size);
192 }
193 }
194
195 /*
196 * If `lttng_consumer_on_read_subbuffer_mmap()` returned an error, pass
197 * it along to the caller, else return zero.
198 */
199 if (written_bytes < 0) {
200 ERR("Error reading mmap subbuffer: %zd", written_bytes);
201 }
202
203 return written_bytes;
204 }
205
206 static ssize_t consumer_stream_consume_splice(
207 struct lttng_consumer_local_data *ctx,
208 struct lttng_consumer_stream *stream,
209 const struct stream_subbuffer *subbuffer)
210 {
211 const ssize_t written_bytes = lttng_consumer_on_read_subbuffer_splice(
212 ctx, stream, subbuffer->info.data.padded_subbuf_size, 0);
213
214 if (written_bytes != subbuffer->info.data.padded_subbuf_size) {
215 DBG("Failed to write the entire padded subbuffer (written_bytes: %zd, padded subbuffer size %lu)",
216 written_bytes,
217 subbuffer->info.data.padded_subbuf_size);
218 }
219
220 /*
221 * If `lttng_consumer_on_read_subbuffer_splice()` returned an error,
222 * pass it along to the caller, else return zero.
223 */
224 if (written_bytes < 0) {
225 ERR("Error reading splice subbuffer: %zd", written_bytes);
226 }
227
228 return written_bytes;
229 }
230
231 static int consumer_stream_send_index(
232 struct lttng_consumer_stream *stream,
233 const struct stream_subbuffer *subbuffer,
234 struct lttng_consumer_local_data *ctx)
235 {
236 off_t packet_offset = 0;
237 struct ctf_packet_index index = {};
238
239 /*
240 * This is called after consuming the sub-buffer; substract the
241 * effect this sub-buffer from the offset.
242 */
243 if (stream->net_seq_idx == (uint64_t) -1ULL) {
244 packet_offset = stream->out_fd_offset -
245 subbuffer->info.data.padded_subbuf_size;
246 }
247
248 ctf_packet_index_populate(&index, packet_offset, subbuffer);
249 return consumer_stream_write_index(stream, &index);
250 }
251
252 /*
253 * Actually do the metadata sync using the given metadata stream.
254 *
255 * Return 0 on success else a negative value. ENODATA can be returned also
256 * indicating that there is no metadata available for that stream.
257 */
258 static int do_sync_metadata(struct lttng_consumer_stream *metadata,
259 struct lttng_consumer_local_data *ctx)
260 {
261 int ret;
262 enum sync_metadata_status status;
263
264 LTTNG_ASSERT(metadata);
265 LTTNG_ASSERT(metadata->metadata_flag);
266 LTTNG_ASSERT(ctx);
267
268 /*
269 * In UST, since we have to write the metadata from the cache packet
270 * by packet, we might need to start this procedure multiple times
271 * until all the metadata from the cache has been extracted.
272 */
273 do {
274 /*
275 * Steps :
276 * - Lock the metadata stream
277 * - Check if metadata stream node was deleted before locking.
278 * - if yes, release and return success
279 * - Check if new metadata is ready (flush + snapshot pos)
280 * - If nothing : release and return.
281 * - Lock the metadata_rdv_lock
282 * - Unlock the metadata stream
283 * - cond_wait on metadata_rdv to wait the wakeup from the
284 * metadata thread
285 * - Unlock the metadata_rdv_lock
286 */
287 pthread_mutex_lock(&metadata->lock);
288
289 /*
290 * There is a possibility that we were able to acquire a reference on the
291 * stream from the RCU hash table but between then and now, the node might
292 * have been deleted just before the lock is acquired. Thus, after locking,
293 * we make sure the metadata node has not been deleted which means that the
294 * buffers are closed.
295 *
296 * In that case, there is no need to sync the metadata hence returning a
297 * success return code.
298 */
299 ret = cds_lfht_is_node_deleted(&metadata->node.node);
300 if (ret) {
301 ret = 0;
302 goto end_unlock_mutex;
303 }
304
305 switch (ctx->type) {
306 case LTTNG_CONSUMER_KERNEL:
307 /*
308 * Empty the metadata cache and flush the current stream.
309 */
310 status = lttng_kconsumer_sync_metadata(metadata);
311 break;
312 case LTTNG_CONSUMER32_UST:
313 case LTTNG_CONSUMER64_UST:
314 /*
315 * Ask the sessiond if we have new metadata waiting and update the
316 * consumer metadata cache.
317 */
318 status = lttng_ustconsumer_sync_metadata(ctx, metadata);
319 break;
320 default:
321 abort();
322 }
323
324 switch (status) {
325 case SYNC_METADATA_STATUS_NEW_DATA:
326 break;
327 case SYNC_METADATA_STATUS_NO_DATA:
328 ret = 0;
329 goto end_unlock_mutex;
330 case SYNC_METADATA_STATUS_ERROR:
331 ret = -1;
332 goto end_unlock_mutex;
333 default:
334 abort();
335 }
336
337 /*
338 * At this point, new metadata have been flushed, so we wait on the
339 * rendez-vous point for the metadata thread to wake us up when it
340 * finishes consuming the metadata and continue execution.
341 */
342
343 pthread_mutex_lock(&metadata->metadata_rdv_lock);
344
345 /*
346 * Release metadata stream lock so the metadata thread can process it.
347 */
348 pthread_mutex_unlock(&metadata->lock);
349
350 /*
351 * Wait on the rendez-vous point. Once woken up, it means the metadata was
352 * consumed and thus synchronization is achieved.
353 */
354 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
355 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
356 } while (status == SYNC_METADATA_STATUS_NEW_DATA);
357
358 /* Success */
359 return 0;
360
361 end_unlock_mutex:
362 pthread_mutex_unlock(&metadata->lock);
363 return ret;
364 }
365
366 /*
367 * Synchronize the metadata using a given session ID. A successful acquisition
368 * of a metadata stream will trigger a request to the session daemon and a
369 * snapshot so the metadata thread can consume it.
370 *
371 * This function call is a rendez-vous point between the metadata thread and
372 * the data thread.
373 *
374 * Return 0 on success or else a negative value.
375 */
376 int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
377 uint64_t session_id)
378 {
379 int ret;
380 struct lttng_consumer_stream *stream = NULL;
381 struct lttng_ht_iter iter;
382 struct lttng_ht *ht;
383
384 LTTNG_ASSERT(ctx);
385
386 /* Ease our life a bit. */
387 ht = the_consumer_data.stream_list_ht;
388
389 rcu_read_lock();
390
391 /* Search the metadata associated with the session id of the given stream. */
392
393 cds_lfht_for_each_entry_duplicate(ht->ht,
394 ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
395 &session_id, &iter.iter, stream, node_session_id.node) {
396 if (!stream->metadata_flag) {
397 continue;
398 }
399
400 ret = do_sync_metadata(stream, ctx);
401 if (ret < 0) {
402 goto end;
403 }
404 }
405
406 /*
407 * Force return code to 0 (success) since ret might be ENODATA for instance
408 * which is not an error but rather that we should come back.
409 */
410 ret = 0;
411
412 end:
413 rcu_read_unlock();
414 return ret;
415 }
416
417 static int consumer_stream_sync_metadata_index(
418 struct lttng_consumer_stream *stream,
419 const struct stream_subbuffer *subbuffer,
420 struct lttng_consumer_local_data *ctx)
421 {
422 int ret;
423
424 /* Block until all the metadata is sent. */
425 pthread_mutex_lock(&stream->metadata_timer_lock);
426 LTTNG_ASSERT(!stream->missed_metadata_flush);
427 stream->waiting_on_metadata = true;
428 pthread_mutex_unlock(&stream->metadata_timer_lock);
429
430 ret = consumer_stream_sync_metadata(ctx, stream->session_id);
431
432 pthread_mutex_lock(&stream->metadata_timer_lock);
433 stream->waiting_on_metadata = false;
434 if (stream->missed_metadata_flush) {
435 stream->missed_metadata_flush = false;
436 pthread_mutex_unlock(&stream->metadata_timer_lock);
437 (void) stream->read_subbuffer_ops.send_live_beacon(stream);
438 } else {
439 pthread_mutex_unlock(&stream->metadata_timer_lock);
440 }
441 if (ret < 0) {
442 goto end;
443 }
444
445 ret = consumer_stream_send_index(stream, subbuffer, ctx);
446 end:
447 return ret;
448 }
449
450 /*
451 * Check if the local version of the metadata stream matches with the version
452 * of the metadata stream in the kernel. If it was updated, set the reset flag
453 * on the stream.
454 */
455 static
456 int metadata_stream_check_version(struct lttng_consumer_stream *stream,
457 const struct stream_subbuffer *subbuffer)
458 {
459 if (stream->metadata_version == subbuffer->info.metadata.version) {
460 goto end;
461 }
462
463 DBG("New metadata version detected");
464 consumer_stream_metadata_set_version(stream,
465 subbuffer->info.metadata.version);
466
467 if (stream->read_subbuffer_ops.reset_metadata) {
468 stream->read_subbuffer_ops.reset_metadata(stream);
469 }
470
471 end:
472 return 0;
473 }
474
475 static
476 bool stream_is_rotating_to_null_chunk(
477 const struct lttng_consumer_stream *stream)
478 {
479 bool rotating_to_null_chunk = false;
480
481 if (stream->rotate_position == -1ULL) {
482 /* No rotation ongoing. */
483 goto end;
484 }
485
486 if (stream->trace_chunk == stream->chan->trace_chunk ||
487 !stream->chan->trace_chunk) {
488 rotating_to_null_chunk = true;
489 }
490 end:
491 return rotating_to_null_chunk;
492 }
493
494 enum consumer_stream_open_packet_status consumer_stream_open_packet(
495 struct lttng_consumer_stream *stream)
496 {
497 int ret;
498 enum consumer_stream_open_packet_status status;
499 unsigned long produced_pos_before, produced_pos_after;
500
501 ret = lttng_consumer_sample_snapshot_positions(stream);
502 if (ret < 0) {
503 ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
504 ", channel name = %s, session id = %" PRIu64,
505 stream->key, stream->chan->name,
506 stream->chan->session_id);
507 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
508 goto end;
509 }
510
511 ret = lttng_consumer_get_produced_snapshot(
512 stream, &produced_pos_before);
513 if (ret < 0) {
514 ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
515 ", channel name = %s, session id = %" PRIu64,
516 stream->key, stream->chan->name,
517 stream->chan->session_id);
518 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
519 goto end;
520 }
521
522 ret = consumer_stream_flush_buffer(stream, 0);
523 if (ret) {
524 ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
525 ", channel name = %s, session id = %" PRIu64,
526 stream->key, stream->chan->name,
527 stream->chan->session_id);
528 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
529 goto end;
530 }
531
532 ret = lttng_consumer_sample_snapshot_positions(stream);
533 if (ret < 0) {
534 ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
535 ", channel name = %s, session id = %" PRIu64,
536 stream->key, stream->chan->name,
537 stream->chan->session_id);
538 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
539 goto end;
540 }
541
542 ret = lttng_consumer_get_produced_snapshot(stream, &produced_pos_after);
543 if (ret < 0) {
544 ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
545 ", channel name = %s, session id = %" PRIu64,
546 stream->key, stream->chan->name,
547 stream->chan->session_id);
548 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
549 goto end;
550 }
551
552 /*
553 * Determine if the flush had an effect by comparing the produced
554 * positons before and after the flush.
555 */
556 status = produced_pos_before != produced_pos_after ?
557 CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED :
558 CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE;
559 if (status == CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED) {
560 stream->opened_packet_in_current_trace_chunk = true;
561 }
562
563 end:
564 return status;
565 }
566
567 /*
568 * An attempt to open a new packet is performed after a rotation completes to
569 * get a begin timestamp as close as possible to the rotation point.
570 *
571 * However, that initial attempt at opening a packet can fail due to a full
572 * ring-buffer. In that case, a second attempt is performed after consuming
573 * a packet since that will have freed enough space in the ring-buffer.
574 */
575 static
576 int post_consume_open_new_packet(struct lttng_consumer_stream *stream,
577 const struct stream_subbuffer *subbuffer,
578 struct lttng_consumer_local_data *ctx)
579 {
580 int ret = 0;
581
582 if (!stream->opened_packet_in_current_trace_chunk &&
583 stream->trace_chunk &&
584 !stream_is_rotating_to_null_chunk(stream)) {
585 const enum consumer_stream_open_packet_status status =
586 consumer_stream_open_packet(stream);
587
588 switch (status) {
589 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
590 DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64
591 ", channel name = %s, session id = %" PRIu64,
592 stream->key, stream->chan->name,
593 stream->chan->session_id);
594 stream->opened_packet_in_current_trace_chunk = true;
595 break;
596 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
597 /*
598 * Can't open a packet as there is no space left.
599 * This means that new events were produced, resulting
600 * in a packet being opened, which is what we want
601 * anyhow.
602 */
603 DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
604 ", channel name = %s, session id = %" PRIu64,
605 stream->key, stream->chan->name,
606 stream->chan->session_id);
607 stream->opened_packet_in_current_trace_chunk = true;
608 break;
609 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
610 /* Logged by callee. */
611 ret = -1;
612 goto end;
613 default:
614 abort();
615 }
616
617 stream->opened_packet_in_current_trace_chunk = true;
618 }
619
620 end:
621 return ret;
622 }
623
624 struct lttng_consumer_stream *consumer_stream_create(
625 struct lttng_consumer_channel *channel,
626 uint64_t channel_key,
627 uint64_t stream_key,
628 const char *channel_name,
629 uint64_t relayd_id,
630 uint64_t session_id,
631 struct lttng_trace_chunk *trace_chunk,
632 int cpu,
633 int *alloc_ret,
634 enum consumer_channel_type type,
635 unsigned int monitor)
636 {
637 int ret;
638 struct lttng_consumer_stream *stream;
639
640 stream = (lttng_consumer_stream *) zmalloc(sizeof(*stream));
641 if (stream == NULL) {
642 PERROR("malloc struct lttng_consumer_stream");
643 ret = -ENOMEM;
644 goto end;
645 }
646
647 rcu_read_lock();
648
649 if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
650 ERR("Failed to acquire trace chunk reference during the creation of a stream");
651 ret = -1;
652 goto error;
653 }
654
655 stream->chan = channel;
656 stream->key = stream_key;
657 stream->trace_chunk = trace_chunk;
658 stream->out_fd = -1;
659 stream->out_fd_offset = 0;
660 stream->output_written = 0;
661 stream->net_seq_idx = relayd_id;
662 stream->session_id = session_id;
663 stream->monitor = monitor;
664 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
665 stream->index_file = NULL;
666 stream->last_sequence_number = -1ULL;
667 stream->rotate_position = -1ULL;
668 /* Buffer is created with an open packet. */
669 stream->opened_packet_in_current_trace_chunk = true;
670 pthread_mutex_init(&stream->lock, NULL);
671 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
672
673 /* If channel is the metadata, flag this stream as metadata. */
674 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
675 stream->metadata_flag = 1;
676 /* Metadata is flat out. */
677 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
678 /* Live rendez-vous point. */
679 pthread_cond_init(&stream->metadata_rdv, NULL);
680 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
681 } else {
682 /* Format stream name to <channel_name>_<cpu_number> */
683 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
684 channel_name, cpu);
685 if (ret < 0) {
686 PERROR("snprintf stream name");
687 goto error;
688 }
689 }
690
691 switch (channel->output) {
692 case CONSUMER_CHANNEL_SPLICE:
693 stream->output = LTTNG_EVENT_SPLICE;
694 ret = utils_create_pipe(stream->splice_pipe);
695 if (ret < 0) {
696 goto error;
697 }
698 break;
699 case CONSUMER_CHANNEL_MMAP:
700 stream->output = LTTNG_EVENT_MMAP;
701 break;
702 default:
703 abort();
704 }
705
706 /* Key is always the wait_fd for streams. */
707 lttng_ht_node_init_u64(&stream->node, stream->key);
708
709 /* Init node per channel id key */
710 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
711
712 /* Init session id node with the stream session id */
713 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
714
715 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
716 " relayd_id %" PRIu64 ", session_id %" PRIu64,
717 stream->name, stream->key, channel_key,
718 stream->net_seq_idx, stream->session_id);
719
720 rcu_read_unlock();
721
722 lttng_dynamic_array_init(&stream->read_subbuffer_ops.post_consume_cbs,
723 sizeof(post_consume_cb), NULL);
724
725 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
726 stream->read_subbuffer_ops.lock =
727 consumer_stream_metadata_lock_all;
728 stream->read_subbuffer_ops.unlock =
729 consumer_stream_metadata_unlock_all;
730 stream->read_subbuffer_ops.assert_locked =
731 consumer_stream_metadata_assert_locked_all;
732 stream->read_subbuffer_ops.pre_consume_subbuffer =
733 metadata_stream_check_version;
734 } else {
735 const post_consume_cb post_consume_index_op = channel->is_live ?
736 consumer_stream_sync_metadata_index :
737 consumer_stream_send_index;
738 const post_consume_cb post_consume_open_new_packet_ =
739 post_consume_open_new_packet;
740
741 ret = lttng_dynamic_array_add_element(
742 &stream->read_subbuffer_ops.post_consume_cbs,
743 &post_consume_index_op);
744 if (ret) {
745 PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
746 goto error;
747 }
748
749 ret = lttng_dynamic_array_add_element(
750 &stream->read_subbuffer_ops.post_consume_cbs,
751 &post_consume_open_new_packet_);
752 if (ret) {
753 PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
754 goto error;
755 }
756
757 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
758 stream->read_subbuffer_ops.unlock =
759 consumer_stream_data_unlock_all;
760 stream->read_subbuffer_ops.assert_locked =
761 consumer_stream_data_assert_locked_all;
762 stream->read_subbuffer_ops.pre_consume_subbuffer =
763 consumer_stream_update_stats;
764 }
765
766 if (channel->output == CONSUMER_CHANNEL_MMAP) {
767 stream->read_subbuffer_ops.consume_subbuffer =
768 consumer_stream_consume_mmap;
769 } else {
770 stream->read_subbuffer_ops.consume_subbuffer =
771 consumer_stream_consume_splice;
772 }
773
774 return stream;
775
776 error:
777 rcu_read_unlock();
778 lttng_trace_chunk_put(stream->trace_chunk);
779 lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
780 free(stream);
781 end:
782 if (alloc_ret) {
783 *alloc_ret = ret;
784 }
785 return NULL;
786 }
787
788 /*
789 * Close stream on the relayd side. This call can destroy a relayd if the
790 * conditions are met.
791 *
792 * A RCU read side lock MUST be acquired if the relayd object was looked up in
793 * a hash table before calling this.
794 */
795 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
796 struct consumer_relayd_sock_pair *relayd)
797 {
798 int ret;
799
800 LTTNG_ASSERT(stream);
801 LTTNG_ASSERT(relayd);
802
803 if (stream->sent_to_relayd) {
804 uatomic_dec(&relayd->refcount);
805 LTTNG_ASSERT(uatomic_read(&relayd->refcount) >= 0);
806 }
807
808 /* Closing streams requires to lock the control socket. */
809 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
810 ret = relayd_send_close_stream(&relayd->control_sock,
811 stream->relayd_stream_id,
812 stream->next_net_seq_num - 1);
813 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
814 if (ret < 0) {
815 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
816 lttng_consumer_cleanup_relayd(relayd);
817 }
818
819 /* Both conditions are met, we destroy the relayd. */
820 if (uatomic_read(&relayd->refcount) == 0 &&
821 uatomic_read(&relayd->destroy_flag)) {
822 consumer_destroy_relayd(relayd);
823 }
824 stream->net_seq_idx = (uint64_t) -1ULL;
825 stream->sent_to_relayd = 0;
826 }
827
828 /*
829 * Close stream's file descriptors and, if needed, close stream also on the
830 * relayd side.
831 *
832 * The consumer data lock MUST be acquired.
833 * The stream lock MUST be acquired.
834 */
835 void consumer_stream_close(struct lttng_consumer_stream *stream)
836 {
837 int ret;
838 struct consumer_relayd_sock_pair *relayd;
839
840 LTTNG_ASSERT(stream);
841
842 switch (the_consumer_data.type) {
843 case LTTNG_CONSUMER_KERNEL:
844 if (stream->mmap_base != NULL) {
845 ret = munmap(stream->mmap_base, stream->mmap_len);
846 if (ret != 0) {
847 PERROR("munmap");
848 }
849 }
850
851 if (stream->wait_fd >= 0) {
852 ret = close(stream->wait_fd);
853 if (ret) {
854 PERROR("close");
855 }
856 stream->wait_fd = -1;
857 }
858 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
859 utils_close_pipe(stream->splice_pipe);
860 }
861 break;
862 case LTTNG_CONSUMER32_UST:
863 case LTTNG_CONSUMER64_UST:
864 {
865 /*
866 * Special case for the metadata since the wait fd is an internal pipe
867 * polled in the metadata thread.
868 */
869 if (stream->metadata_flag && stream->chan->monitor) {
870 int rpipe = stream->ust_metadata_poll_pipe[0];
871
872 /*
873 * This will stop the channel timer if one and close the write side
874 * of the metadata poll pipe.
875 */
876 lttng_ustconsumer_close_metadata(stream->chan);
877 if (rpipe >= 0) {
878 ret = close(rpipe);
879 if (ret < 0) {
880 PERROR("closing metadata pipe read side");
881 }
882 stream->ust_metadata_poll_pipe[0] = -1;
883 }
884 }
885 break;
886 }
887 default:
888 ERR("Unknown consumer_data type");
889 abort();
890 }
891
892 /* Close output fd. Could be a socket or local file at this point. */
893 if (stream->out_fd >= 0) {
894 ret = close(stream->out_fd);
895 if (ret) {
896 PERROR("close");
897 }
898 stream->out_fd = -1;
899 }
900
901 if (stream->index_file) {
902 lttng_index_file_put(stream->index_file);
903 stream->index_file = NULL;
904 }
905
906 lttng_trace_chunk_put(stream->trace_chunk);
907 stream->trace_chunk = NULL;
908
909 /* Check and cleanup relayd if needed. */
910 rcu_read_lock();
911 relayd = consumer_find_relayd(stream->net_seq_idx);
912 if (relayd != NULL) {
913 consumer_stream_relayd_close(stream, relayd);
914 }
915 rcu_read_unlock();
916 }
917
918 /*
919 * Delete the stream from all possible hash tables.
920 *
921 * The consumer data lock MUST be acquired.
922 * The stream lock MUST be acquired.
923 */
924 void consumer_stream_delete(struct lttng_consumer_stream *stream,
925 struct lttng_ht *ht)
926 {
927 int ret;
928 struct lttng_ht_iter iter;
929
930 LTTNG_ASSERT(stream);
931 /* Should NEVER be called not in monitor mode. */
932 LTTNG_ASSERT(stream->chan->monitor);
933
934 rcu_read_lock();
935
936 if (ht) {
937 iter.iter.node = &stream->node.node;
938 ret = lttng_ht_del(ht, &iter);
939 LTTNG_ASSERT(!ret);
940 }
941
942 /* Delete from stream per channel ID hash table. */
943 iter.iter.node = &stream->node_channel_id.node;
944 /*
945 * The returned value is of no importance. Even if the node is NOT in the
946 * hash table, we continue since we may have been called by a code path
947 * that did not add the stream to a (all) hash table. Same goes for the
948 * next call ht del call.
949 */
950 (void) lttng_ht_del(the_consumer_data.stream_per_chan_id_ht, &iter);
951
952 /* Delete from the global stream list. */
953 iter.iter.node = &stream->node_session_id.node;
954 /* See the previous ht del on why we ignore the returned value. */
955 (void) lttng_ht_del(the_consumer_data.stream_list_ht, &iter);
956
957 rcu_read_unlock();
958
959 if (!stream->metadata_flag) {
960 /* Decrement the stream count of the global consumer data. */
961 LTTNG_ASSERT(the_consumer_data.stream_count > 0);
962 the_consumer_data.stream_count--;
963 }
964 }
965
966 /*
967 * Free the given stream within a RCU call.
968 */
969 void consumer_stream_free(struct lttng_consumer_stream *stream)
970 {
971 LTTNG_ASSERT(stream);
972
973 metadata_bucket_destroy(stream->metadata_bucket);
974 call_rcu(&stream->node.head, free_stream_rcu);
975 }
976
977 /*
978 * Destroy the stream's buffers of the tracer.
979 */
980 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
981 {
982 LTTNG_ASSERT(stream);
983
984 switch (the_consumer_data.type) {
985 case LTTNG_CONSUMER_KERNEL:
986 break;
987 case LTTNG_CONSUMER32_UST:
988 case LTTNG_CONSUMER64_UST:
989 lttng_ustconsumer_del_stream(stream);
990 break;
991 default:
992 ERR("Unknown consumer_data type");
993 abort();
994 }
995 }
996
997 /*
998 * Destroy and close a already created stream.
999 */
1000 static void destroy_close_stream(struct lttng_consumer_stream *stream)
1001 {
1002 LTTNG_ASSERT(stream);
1003
1004 DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
1005
1006 /* Destroy tracer buffers of the stream. */
1007 consumer_stream_destroy_buffers(stream);
1008 /* Close down everything including the relayd if one. */
1009 consumer_stream_close(stream);
1010 }
1011
1012 /*
1013 * Decrement the stream's channel refcount and if down to 0, return the channel
1014 * pointer so it can be destroyed by the caller or NULL if not.
1015 */
1016 static struct lttng_consumer_channel *unref_channel(
1017 struct lttng_consumer_stream *stream)
1018 {
1019 struct lttng_consumer_channel *free_chan = NULL;
1020
1021 LTTNG_ASSERT(stream);
1022 LTTNG_ASSERT(stream->chan);
1023
1024 /* Update refcount of channel and see if we need to destroy it. */
1025 if (!uatomic_sub_return(&stream->chan->refcount, 1)
1026 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
1027 free_chan = stream->chan;
1028 }
1029
1030 return free_chan;
1031 }
1032
1033 /*
1034 * Destroy a stream completely. This will delete, close and free the stream.
1035 * Once return, the stream is NO longer usable. Its channel may get destroyed
1036 * if conditions are met for a monitored stream.
1037 *
1038 * This MUST be called WITHOUT the consumer data and stream lock acquired if
1039 * the stream is in _monitor_ mode else it does not matter.
1040 */
1041 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
1042 struct lttng_ht *ht)
1043 {
1044 LTTNG_ASSERT(stream);
1045
1046 /* Stream is in monitor mode. */
1047 if (stream->monitor) {
1048 struct lttng_consumer_channel *free_chan = NULL;
1049
1050 /*
1051 * This means that the stream was successfully removed from the streams
1052 * list of the channel and sent to the right thread managing this
1053 * stream thus being globally visible.
1054 */
1055 if (stream->globally_visible) {
1056 pthread_mutex_lock(&the_consumer_data.lock);
1057 pthread_mutex_lock(&stream->chan->lock);
1058 pthread_mutex_lock(&stream->lock);
1059 /* Remove every reference of the stream in the consumer. */
1060 consumer_stream_delete(stream, ht);
1061
1062 destroy_close_stream(stream);
1063
1064 /* Update channel's refcount of the stream. */
1065 free_chan = unref_channel(stream);
1066
1067 /* Indicates that the consumer data state MUST be updated after this. */
1068 the_consumer_data.need_update = 1;
1069
1070 pthread_mutex_unlock(&stream->lock);
1071 pthread_mutex_unlock(&stream->chan->lock);
1072 pthread_mutex_unlock(&the_consumer_data.lock);
1073 } else {
1074 /*
1075 * If the stream is not visible globally, this needs to be done
1076 * outside of the consumer data lock section.
1077 */
1078 free_chan = unref_channel(stream);
1079 }
1080
1081 if (free_chan) {
1082 consumer_del_channel(free_chan);
1083 }
1084 } else {
1085 destroy_close_stream(stream);
1086 }
1087
1088 /* Free stream within a RCU call. */
1089 lttng_trace_chunk_put(stream->trace_chunk);
1090 stream->trace_chunk = NULL;
1091 lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
1092 consumer_stream_free(stream);
1093 }
1094
1095 /*
1096 * Write index of a specific stream either on the relayd or local disk.
1097 *
1098 * Return 0 on success or else a negative value.
1099 */
1100 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
1101 struct ctf_packet_index *element)
1102 {
1103 int ret;
1104
1105 LTTNG_ASSERT(stream);
1106 LTTNG_ASSERT(element);
1107
1108 rcu_read_lock();
1109 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1110 struct consumer_relayd_sock_pair *relayd;
1111 relayd = consumer_find_relayd(stream->net_seq_idx);
1112 if (relayd) {
1113 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1114 ret = relayd_send_index(&relayd->control_sock, element,
1115 stream->relayd_stream_id, stream->next_net_seq_num - 1);
1116 if (ret < 0) {
1117 /*
1118 * Communication error with lttng-relayd,
1119 * perform cleanup now
1120 */
1121 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
1122 lttng_consumer_cleanup_relayd(relayd);
1123 ret = -1;
1124 }
1125 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1126 } else {
1127 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
1128 stream->key, stream->net_seq_idx);
1129 ret = -1;
1130 }
1131 } else {
1132 if (lttng_index_file_write(stream->index_file, element)) {
1133 ret = -1;
1134 } else {
1135 ret = 0;
1136 }
1137 }
1138 if (ret < 0) {
1139 goto error;
1140 }
1141
1142 error:
1143 rcu_read_unlock();
1144 return ret;
1145 }
1146
1147 int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
1148 bool create_index)
1149 {
1150 int ret;
1151 enum lttng_trace_chunk_status chunk_status;
1152 const int flags = O_WRONLY | O_CREAT | O_TRUNC;
1153 const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
1154 char stream_path[LTTNG_PATH_MAX];
1155
1156 ASSERT_LOCKED(stream->lock);
1157 LTTNG_ASSERT(stream->trace_chunk);
1158
1159 ret = utils_stream_file_path(stream->chan->pathname, stream->name,
1160 stream->chan->tracefile_size,
1161 stream->tracefile_count_current, NULL,
1162 stream_path, sizeof(stream_path));
1163 if (ret < 0) {
1164 goto end;
1165 }
1166
1167 if (stream->out_fd >= 0) {
1168 ret = close(stream->out_fd);
1169 if (ret < 0) {
1170 PERROR("Failed to close stream file \"%s\"",
1171 stream->name);
1172 goto end;
1173 }
1174 stream->out_fd = -1;
1175 }
1176
1177 DBG("Opening stream output file \"%s\"", stream_path);
1178 chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
1179 flags, mode, &stream->out_fd, false);
1180 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1181 ERR("Failed to open stream file \"%s\"", stream->name);
1182 ret = -1;
1183 goto end;
1184 }
1185
1186 if (!stream->metadata_flag && (create_index || stream->index_file)) {
1187 if (stream->index_file) {
1188 lttng_index_file_put(stream->index_file);
1189 }
1190 chunk_status = lttng_index_file_create_from_trace_chunk(
1191 stream->trace_chunk,
1192 stream->chan->pathname,
1193 stream->name,
1194 stream->chan->tracefile_size,
1195 stream->tracefile_count_current,
1196 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
1197 false, &stream->index_file);
1198 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1199 ret = -1;
1200 goto end;
1201 }
1202 }
1203
1204 /* Reset current size because we just perform a rotation. */
1205 stream->tracefile_size_current = 0;
1206 stream->out_fd_offset = 0;
1207 end:
1208 return ret;
1209 }
1210
1211 int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
1212 {
1213 int ret;
1214
1215 stream->tracefile_count_current++;
1216 if (stream->chan->tracefile_count > 0) {
1217 stream->tracefile_count_current %=
1218 stream->chan->tracefile_count;
1219 }
1220
1221 DBG("Rotating output files of stream \"%s\"", stream->name);
1222 ret = consumer_stream_create_output_files(stream, true);
1223 if (ret) {
1224 goto end;
1225 }
1226
1227 end:
1228 return ret;
1229 }
1230
1231 bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
1232 {
1233 /*
1234 * This function does not take a const stream since
1235 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1236 */
1237 LTTNG_ASSERT(stream);
1238 return cds_lfht_is_node_deleted(&stream->node.node);
1239 }
1240
1241 static ssize_t metadata_bucket_flush(
1242 const struct stream_subbuffer *buffer, void *data)
1243 {
1244 ssize_t ret;
1245 struct lttng_consumer_stream *stream = (lttng_consumer_stream *) data;
1246
1247 ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1248 if (ret < 0) {
1249 goto end;
1250 }
1251 end:
1252 return ret;
1253 }
1254
1255 static ssize_t metadata_bucket_consume(
1256 struct lttng_consumer_local_data *unused,
1257 struct lttng_consumer_stream *stream,
1258 const struct stream_subbuffer *subbuffer)
1259 {
1260 ssize_t ret;
1261 enum metadata_bucket_status status;
1262
1263 status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1264 switch (status) {
1265 case METADATA_BUCKET_STATUS_OK:
1266 /* Return consumed size. */
1267 ret = subbuffer->buffer.buffer.size;
1268 break;
1269 default:
1270 ret = -1;
1271 }
1272
1273 return ret;
1274 }
1275
1276 int consumer_stream_enable_metadata_bucketization(
1277 struct lttng_consumer_stream *stream)
1278 {
1279 int ret = 0;
1280
1281 LTTNG_ASSERT(stream->metadata_flag);
1282 LTTNG_ASSERT(!stream->metadata_bucket);
1283 LTTNG_ASSERT(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1284
1285 stream->metadata_bucket = metadata_bucket_create(
1286 metadata_bucket_flush, stream);
1287 if (!stream->metadata_bucket) {
1288 ret = -1;
1289 goto end;
1290 }
1291
1292 stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1293 end:
1294 return ret;
1295 }
1296
1297 void consumer_stream_metadata_set_version(
1298 struct lttng_consumer_stream *stream, uint64_t new_version)
1299 {
1300 LTTNG_ASSERT(new_version > stream->metadata_version);
1301 stream->metadata_version = new_version;
1302 stream->reset_metadata_flag = 1;
1303
1304 if (stream->metadata_bucket) {
1305 metadata_bucket_reset(stream->metadata_bucket);
1306 }
1307 }
1308
1309 int consumer_stream_flush_buffer(struct lttng_consumer_stream *stream,
1310 bool producer_active)
1311 {
1312 int ret = 0;
1313
1314 switch (the_consumer_data.type) {
1315 case LTTNG_CONSUMER_KERNEL:
1316 if (producer_active) {
1317 ret = kernctl_buffer_flush(stream->wait_fd);
1318 if (ret < 0) {
1319 ERR("Failed to flush kernel stream");
1320 goto end;
1321 }
1322 } else {
1323 ret = kernctl_buffer_flush_empty(stream->wait_fd);
1324 if (ret < 0) {
1325 /*
1326 * Doing a buffer flush which does not take into
1327 * account empty packets. This is not perfect,
1328 * but required as a fall-back when
1329 * "flush_empty" is not implemented by
1330 * lttng-modules.
1331 */
1332 ret = kernctl_buffer_flush(stream->wait_fd);
1333 if (ret < 0) {
1334 ERR("Failed to flush kernel stream");
1335 goto end;
1336 }
1337 }
1338 }
1339 break;
1340 case LTTNG_CONSUMER32_UST:
1341 case LTTNG_CONSUMER64_UST:
1342 ret = lttng_ustconsumer_flush_buffer(stream, (int) producer_active);
1343 break;
1344 default:
1345 ERR("Unknown consumer_data type");
1346 abort();
1347 }
1348
1349 end:
1350 return ret;
1351 }
This page took 0.056753 seconds and 5 git commands to generate.