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