consumerd: send a buffer static sample on flush command
[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 <inttypes.h>
12 #include <sys/mman.h>
13 #include <unistd.h>
14
15 #include <common/common.hpp>
16 #include <common/consumer/consumer-timer.hpp>
17 #include <common/consumer/consumer-timer.hpp>
18 #include <common/consumer/consumer.hpp>
19 #include <common/consumer/consumer.hpp>
20 #include <common/consumer/metadata-bucket.hpp>
21 #include <common/consumer/metadata-bucket.hpp>
22 #include <common/index/index.hpp>
23 #include <common/kernel-consumer/kernel-consumer.hpp>
24 #include <common/kernel-ctl/kernel-ctl.hpp>
25 #include <common/macros.hpp>
26 #include <common/relayd/relayd.hpp>
27 #include <common/ust-consumer/ust-consumer.hpp>
28 #include <common/utils.hpp>
29
30 #include "consumer-stream.hpp"
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 lttng::utils::container_of(head, &lttng_ht_node_u64::head);
39 struct lttng_consumer_stream *stream =
40 lttng::utils::container_of(node, &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 __attribute__((unused)),
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 __attribute__((unused)))
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 bool missed_metadata_flush;
423 int ret;
424
425 /* Block until all the metadata is sent. */
426 pthread_mutex_lock(&stream->metadata_timer_lock);
427 LTTNG_ASSERT(!stream->missed_metadata_flush);
428 stream->waiting_on_metadata = true;
429 pthread_mutex_unlock(&stream->metadata_timer_lock);
430
431 ret = consumer_stream_sync_metadata(ctx, stream->session_id);
432
433 pthread_mutex_lock(&stream->metadata_timer_lock);
434 stream->waiting_on_metadata = false;
435 missed_metadata_flush = stream->missed_metadata_flush;
436 if (missed_metadata_flush) {
437 stream->missed_metadata_flush = false;
438 }
439 pthread_mutex_unlock(&stream->metadata_timer_lock);
440 if (ret < 0) {
441 goto end;
442 }
443
444 ret = consumer_stream_send_index(stream, subbuffer, ctx);
445 /*
446 * Send the live inactivity beacon to handle the situation where
447 * the live timer is prevented from sampling this stream
448 * because the stream lock was being held while this stream is
449 * waiting on metadata. This ensures live viewer progress in the
450 * unlikely scenario where a live timer would be prevented from
451 * locking a stream lock repeatedly due to a steady flow of
452 * incoming metadata, for a stream which is mostly inactive.
453 *
454 * It is important to send the inactivity beacon packet to
455 * relayd _after_ sending the index associated with the data
456 * that was just sent, otherwise this can cause live viewers to
457 * observe timestamps going backwards between an inactivity
458 * beacon and a following trace packet.
459 */
460 if (missed_metadata_flush) {
461 (void) stream->read_subbuffer_ops.send_live_beacon(stream);
462 }
463 end:
464 return ret;
465 }
466
467 /*
468 * Check if the local version of the metadata stream matches with the version
469 * of the metadata stream in the kernel. If it was updated, set the reset flag
470 * on the stream.
471 */
472 static
473 int metadata_stream_check_version(struct lttng_consumer_stream *stream,
474 const struct stream_subbuffer *subbuffer)
475 {
476 if (stream->metadata_version == subbuffer->info.metadata.version) {
477 goto end;
478 }
479
480 DBG("New metadata version detected");
481 consumer_stream_metadata_set_version(stream,
482 subbuffer->info.metadata.version);
483
484 if (stream->read_subbuffer_ops.reset_metadata) {
485 stream->read_subbuffer_ops.reset_metadata(stream);
486 }
487
488 end:
489 return 0;
490 }
491
492 static
493 bool stream_is_rotating_to_null_chunk(
494 const struct lttng_consumer_stream *stream)
495 {
496 bool rotating_to_null_chunk = false;
497
498 if (stream->rotate_position == -1ULL) {
499 /* No rotation ongoing. */
500 goto end;
501 }
502
503 if (stream->trace_chunk == stream->chan->trace_chunk ||
504 !stream->chan->trace_chunk) {
505 rotating_to_null_chunk = true;
506 }
507 end:
508 return rotating_to_null_chunk;
509 }
510
511 enum consumer_stream_open_packet_status consumer_stream_open_packet(
512 struct lttng_consumer_stream *stream)
513 {
514 int ret;
515 enum consumer_stream_open_packet_status status;
516 unsigned long produced_pos_before, produced_pos_after;
517
518 ret = lttng_consumer_sample_snapshot_positions(stream);
519 if (ret < 0) {
520 ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
521 ", channel name = %s, session id = %" PRIu64,
522 stream->key, stream->chan->name,
523 stream->chan->session_id);
524 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
525 goto end;
526 }
527
528 ret = lttng_consumer_get_produced_snapshot(
529 stream, &produced_pos_before);
530 if (ret < 0) {
531 ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
532 ", channel name = %s, session id = %" PRIu64,
533 stream->key, stream->chan->name,
534 stream->chan->session_id);
535 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
536 goto end;
537 }
538
539 ret = consumer_stream_flush_buffer(stream, 0);
540 if (ret) {
541 ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
542 ", channel name = %s, session id = %" PRIu64,
543 stream->key, stream->chan->name,
544 stream->chan->session_id);
545 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
546 goto end;
547 }
548
549 ret = lttng_consumer_sample_snapshot_positions(stream);
550 if (ret < 0) {
551 ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
552 ", channel name = %s, session id = %" PRIu64,
553 stream->key, stream->chan->name,
554 stream->chan->session_id);
555 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
556 goto end;
557 }
558
559 ret = lttng_consumer_get_produced_snapshot(stream, &produced_pos_after);
560 if (ret < 0) {
561 ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
562 ", channel name = %s, session id = %" PRIu64,
563 stream->key, stream->chan->name,
564 stream->chan->session_id);
565 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
566 goto end;
567 }
568
569 /*
570 * Determine if the flush had an effect by comparing the produced
571 * positons before and after the flush.
572 */
573 status = produced_pos_before != produced_pos_after ?
574 CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED :
575 CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE;
576 if (status == CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED) {
577 stream->opened_packet_in_current_trace_chunk = true;
578 }
579
580 end:
581 return status;
582 }
583
584 /*
585 * An attempt to open a new packet is performed after a rotation completes to
586 * get a begin timestamp as close as possible to the rotation point.
587 *
588 * However, that initial attempt at opening a packet can fail due to a full
589 * ring-buffer. In that case, a second attempt is performed after consuming
590 * a packet since that will have freed enough space in the ring-buffer.
591 */
592 static
593 int post_consume_open_new_packet(struct lttng_consumer_stream *stream,
594 const struct stream_subbuffer *subbuffer __attribute__((unused)),
595 struct lttng_consumer_local_data *ctx __attribute__((unused)))
596 {
597 int ret = 0;
598
599 if (!stream->opened_packet_in_current_trace_chunk &&
600 stream->trace_chunk &&
601 !stream_is_rotating_to_null_chunk(stream)) {
602 const enum consumer_stream_open_packet_status status =
603 consumer_stream_open_packet(stream);
604
605 switch (status) {
606 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
607 DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64
608 ", channel name = %s, session id = %" PRIu64,
609 stream->key, stream->chan->name,
610 stream->chan->session_id);
611 stream->opened_packet_in_current_trace_chunk = true;
612 break;
613 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
614 /*
615 * Can't open a packet as there is no space left.
616 * This means that new events were produced, resulting
617 * in a packet being opened, which is what we want
618 * anyhow.
619 */
620 DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
621 ", channel name = %s, session id = %" PRIu64,
622 stream->key, stream->chan->name,
623 stream->chan->session_id);
624 stream->opened_packet_in_current_trace_chunk = true;
625 break;
626 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
627 /* Logged by callee. */
628 ret = -1;
629 goto end;
630 default:
631 abort();
632 }
633
634 stream->opened_packet_in_current_trace_chunk = true;
635 }
636
637 end:
638 return ret;
639 }
640
641 struct lttng_consumer_stream *consumer_stream_create(
642 struct lttng_consumer_channel *channel,
643 uint64_t channel_key,
644 uint64_t stream_key,
645 const char *channel_name,
646 uint64_t relayd_id,
647 uint64_t session_id,
648 struct lttng_trace_chunk *trace_chunk,
649 int cpu,
650 int *alloc_ret,
651 enum consumer_channel_type type,
652 unsigned int monitor)
653 {
654 int ret;
655 struct lttng_consumer_stream *stream;
656
657 stream = zmalloc<lttng_consumer_stream>();
658 if (stream == NULL) {
659 PERROR("malloc struct lttng_consumer_stream");
660 ret = -ENOMEM;
661 goto end;
662 }
663
664 rcu_read_lock();
665
666 if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
667 ERR("Failed to acquire trace chunk reference during the creation of a stream");
668 ret = -1;
669 goto error;
670 }
671
672 stream->send_node = CDS_LIST_HEAD_INIT(stream->send_node);
673 stream->chan = channel;
674 stream->key = stream_key;
675 stream->trace_chunk = trace_chunk;
676 stream->out_fd = -1;
677 stream->out_fd_offset = 0;
678 stream->output_written = 0;
679 stream->net_seq_idx = relayd_id;
680 stream->session_id = session_id;
681 stream->monitor = monitor;
682 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
683 stream->index_file = NULL;
684 stream->last_sequence_number = -1ULL;
685 stream->rotate_position = -1ULL;
686 /* Buffer is created with an open packet. */
687 stream->opened_packet_in_current_trace_chunk = true;
688 pthread_mutex_init(&stream->lock, NULL);
689 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
690
691 /* If channel is the metadata, flag this stream as metadata. */
692 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
693 stream->metadata_flag = 1;
694 /* Metadata is flat out. */
695 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
696 /* Live rendez-vous point. */
697 pthread_cond_init(&stream->metadata_rdv, NULL);
698 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
699 } else {
700 /* Format stream name to <channel_name>_<cpu_number> */
701 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
702 channel_name, cpu);
703 if (ret < 0) {
704 PERROR("snprintf stream name");
705 goto error;
706 }
707 }
708
709 switch (channel->output) {
710 case CONSUMER_CHANNEL_SPLICE:
711 stream->output = LTTNG_EVENT_SPLICE;
712 ret = utils_create_pipe(stream->splice_pipe);
713 if (ret < 0) {
714 goto error;
715 }
716 break;
717 case CONSUMER_CHANNEL_MMAP:
718 stream->output = LTTNG_EVENT_MMAP;
719 break;
720 default:
721 abort();
722 }
723
724 /* Key is always the wait_fd for streams. */
725 lttng_ht_node_init_u64(&stream->node, stream->key);
726
727 /* Init node per channel id key */
728 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
729
730 /* Init session id node with the stream session id */
731 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
732
733 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
734 " relayd_id %" PRIu64 ", session_id %" PRIu64,
735 stream->name, stream->key, channel_key,
736 stream->net_seq_idx, stream->session_id);
737
738 rcu_read_unlock();
739
740 lttng_dynamic_array_init(&stream->read_subbuffer_ops.post_consume_cbs,
741 sizeof(post_consume_cb), NULL);
742
743 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
744 stream->read_subbuffer_ops.lock =
745 consumer_stream_metadata_lock_all;
746 stream->read_subbuffer_ops.unlock =
747 consumer_stream_metadata_unlock_all;
748 stream->read_subbuffer_ops.assert_locked =
749 consumer_stream_metadata_assert_locked_all;
750 stream->read_subbuffer_ops.pre_consume_subbuffer =
751 metadata_stream_check_version;
752 } else {
753 const post_consume_cb post_consume_index_op = channel->is_live ?
754 consumer_stream_sync_metadata_index :
755 consumer_stream_send_index;
756 const post_consume_cb post_consume_open_new_packet_ =
757 post_consume_open_new_packet;
758
759 ret = lttng_dynamic_array_add_element(
760 &stream->read_subbuffer_ops.post_consume_cbs,
761 &post_consume_index_op);
762 if (ret) {
763 PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
764 goto error;
765 }
766
767 ret = lttng_dynamic_array_add_element(
768 &stream->read_subbuffer_ops.post_consume_cbs,
769 &post_consume_open_new_packet_);
770 if (ret) {
771 PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
772 goto error;
773 }
774
775 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
776 stream->read_subbuffer_ops.unlock =
777 consumer_stream_data_unlock_all;
778 stream->read_subbuffer_ops.assert_locked =
779 consumer_stream_data_assert_locked_all;
780 stream->read_subbuffer_ops.pre_consume_subbuffer =
781 consumer_stream_update_stats;
782 }
783
784 if (channel->output == CONSUMER_CHANNEL_MMAP) {
785 stream->read_subbuffer_ops.consume_subbuffer =
786 consumer_stream_consume_mmap;
787 } else {
788 stream->read_subbuffer_ops.consume_subbuffer =
789 consumer_stream_consume_splice;
790 }
791
792 return stream;
793
794 error:
795 rcu_read_unlock();
796 lttng_trace_chunk_put(stream->trace_chunk);
797 lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
798 free(stream);
799 end:
800 if (alloc_ret) {
801 *alloc_ret = ret;
802 }
803 return NULL;
804 }
805
806 /*
807 * Close stream on the relayd side. This call can destroy a relayd if the
808 * conditions are met.
809 *
810 * A RCU read side lock MUST be acquired if the relayd object was looked up in
811 * a hash table before calling this.
812 */
813 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
814 struct consumer_relayd_sock_pair *relayd)
815 {
816 int ret;
817
818 LTTNG_ASSERT(stream);
819 LTTNG_ASSERT(relayd);
820
821 if (stream->sent_to_relayd) {
822 uatomic_dec(&relayd->refcount);
823 LTTNG_ASSERT(uatomic_read(&relayd->refcount) >= 0);
824 }
825
826 /* Closing streams requires to lock the control socket. */
827 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
828 ret = relayd_send_close_stream(&relayd->control_sock,
829 stream->relayd_stream_id,
830 stream->next_net_seq_num - 1);
831 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
832 if (ret < 0) {
833 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
834 lttng_consumer_cleanup_relayd(relayd);
835 }
836
837 /* Both conditions are met, we destroy the relayd. */
838 if (uatomic_read(&relayd->refcount) == 0 &&
839 uatomic_read(&relayd->destroy_flag)) {
840 consumer_destroy_relayd(relayd);
841 }
842 stream->net_seq_idx = (uint64_t) -1ULL;
843 stream->sent_to_relayd = 0;
844 }
845
846 /*
847 * Close stream's file descriptors and, if needed, close stream also on the
848 * relayd side.
849 *
850 * The consumer data lock MUST be acquired.
851 * The stream lock MUST be acquired.
852 */
853 void consumer_stream_close(struct lttng_consumer_stream *stream)
854 {
855 int ret;
856 struct consumer_relayd_sock_pair *relayd;
857
858 LTTNG_ASSERT(stream);
859
860 switch (the_consumer_data.type) {
861 case LTTNG_CONSUMER_KERNEL:
862 if (stream->mmap_base != NULL) {
863 ret = munmap(stream->mmap_base, stream->mmap_len);
864 if (ret != 0) {
865 PERROR("munmap");
866 }
867 }
868
869 if (stream->wait_fd >= 0) {
870 ret = close(stream->wait_fd);
871 if (ret) {
872 PERROR("close");
873 }
874 stream->wait_fd = -1;
875 }
876 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
877 utils_close_pipe(stream->splice_pipe);
878 }
879 break;
880 case LTTNG_CONSUMER32_UST:
881 case LTTNG_CONSUMER64_UST:
882 {
883 /*
884 * Special case for the metadata since the wait fd is an internal pipe
885 * polled in the metadata thread.
886 */
887 if (stream->metadata_flag && stream->chan->monitor) {
888 int rpipe = stream->ust_metadata_poll_pipe[0];
889
890 /*
891 * This will stop the channel timer if one and close the write side
892 * of the metadata poll pipe.
893 */
894 lttng_ustconsumer_close_metadata(stream->chan);
895 if (rpipe >= 0) {
896 ret = close(rpipe);
897 if (ret < 0) {
898 PERROR("closing metadata pipe read side");
899 }
900 stream->ust_metadata_poll_pipe[0] = -1;
901 }
902 }
903 break;
904 }
905 default:
906 ERR("Unknown consumer_data type");
907 abort();
908 }
909
910 /* Close output fd. Could be a socket or local file at this point. */
911 if (stream->out_fd >= 0) {
912 ret = close(stream->out_fd);
913 if (ret) {
914 PERROR("close");
915 }
916 stream->out_fd = -1;
917 }
918
919 if (stream->index_file) {
920 lttng_index_file_put(stream->index_file);
921 stream->index_file = NULL;
922 }
923
924 lttng_trace_chunk_put(stream->trace_chunk);
925 stream->trace_chunk = NULL;
926
927 /* Check and cleanup relayd if needed. */
928 rcu_read_lock();
929 relayd = consumer_find_relayd(stream->net_seq_idx);
930 if (relayd != NULL) {
931 consumer_stream_relayd_close(stream, relayd);
932 }
933 rcu_read_unlock();
934 }
935
936 /*
937 * Delete the stream from all possible hash tables.
938 *
939 * The consumer data lock MUST be acquired.
940 * The stream lock MUST be acquired.
941 */
942 void consumer_stream_delete(struct lttng_consumer_stream *stream,
943 struct lttng_ht *ht)
944 {
945 int ret;
946 struct lttng_ht_iter iter;
947
948 LTTNG_ASSERT(stream);
949 /* Should NEVER be called not in monitor mode. */
950 LTTNG_ASSERT(stream->chan->monitor);
951
952 rcu_read_lock();
953
954 if (ht) {
955 iter.iter.node = &stream->node.node;
956 ret = lttng_ht_del(ht, &iter);
957 LTTNG_ASSERT(!ret);
958 }
959
960 /* Delete from stream per channel ID hash table. */
961 iter.iter.node = &stream->node_channel_id.node;
962 /*
963 * The returned value is of no importance. Even if the node is NOT in the
964 * hash table, we continue since we may have been called by a code path
965 * that did not add the stream to a (all) hash table. Same goes for the
966 * next call ht del call.
967 */
968 (void) lttng_ht_del(the_consumer_data.stream_per_chan_id_ht, &iter);
969
970 /* Delete from the global stream list. */
971 iter.iter.node = &stream->node_session_id.node;
972 /* See the previous ht del on why we ignore the returned value. */
973 (void) lttng_ht_del(the_consumer_data.stream_list_ht, &iter);
974
975 rcu_read_unlock();
976
977 if (!stream->metadata_flag) {
978 /* Decrement the stream count of the global consumer data. */
979 LTTNG_ASSERT(the_consumer_data.stream_count > 0);
980 the_consumer_data.stream_count--;
981 }
982 }
983
984 /*
985 * Free the given stream within a RCU call.
986 */
987 void consumer_stream_free(struct lttng_consumer_stream *stream)
988 {
989 LTTNG_ASSERT(stream);
990
991 metadata_bucket_destroy(stream->metadata_bucket);
992 call_rcu(&stream->node.head, free_stream_rcu);
993 }
994
995 /*
996 * Destroy the stream's buffers of the tracer.
997 */
998 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
999 {
1000 LTTNG_ASSERT(stream);
1001
1002 switch (the_consumer_data.type) {
1003 case LTTNG_CONSUMER_KERNEL:
1004 break;
1005 case LTTNG_CONSUMER32_UST:
1006 case LTTNG_CONSUMER64_UST:
1007 lttng_ustconsumer_del_stream(stream);
1008 break;
1009 default:
1010 ERR("Unknown consumer_data type");
1011 abort();
1012 }
1013 }
1014
1015 /*
1016 * Destroy and close a already created stream.
1017 */
1018 static void destroy_close_stream(struct lttng_consumer_stream *stream)
1019 {
1020 LTTNG_ASSERT(stream);
1021
1022 DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
1023
1024 /* Destroy tracer buffers of the stream. */
1025 consumer_stream_destroy_buffers(stream);
1026 /* Close down everything including the relayd if one. */
1027 consumer_stream_close(stream);
1028 }
1029
1030 /*
1031 * Decrement the stream's channel refcount and if down to 0, return the channel
1032 * pointer so it can be destroyed by the caller or NULL if not.
1033 */
1034 static struct lttng_consumer_channel *unref_channel(
1035 struct lttng_consumer_stream *stream)
1036 {
1037 struct lttng_consumer_channel *free_chan = NULL;
1038
1039 LTTNG_ASSERT(stream);
1040 LTTNG_ASSERT(stream->chan);
1041
1042 /* Update refcount of channel and see if we need to destroy it. */
1043 if (!uatomic_sub_return(&stream->chan->refcount, 1)
1044 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
1045 free_chan = stream->chan;
1046 }
1047
1048 return free_chan;
1049 }
1050
1051 /*
1052 * Destroy a stream completely. This will delete, close and free the stream.
1053 * Once return, the stream is NO longer usable. Its channel may get destroyed
1054 * if conditions are met for a monitored stream.
1055 *
1056 * This MUST be called WITHOUT the consumer data and stream lock acquired if
1057 * the stream is in _monitor_ mode else it does not matter.
1058 */
1059 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
1060 struct lttng_ht *ht)
1061 {
1062 LTTNG_ASSERT(stream);
1063
1064 cds_list_del_init(&stream->send_node);
1065
1066 /* Stream is in monitor mode. */
1067 if (stream->monitor) {
1068 struct lttng_consumer_channel *free_chan = NULL;
1069
1070 /*
1071 * This means that the stream was successfully removed from the streams
1072 * list of the channel and sent to the right thread managing this
1073 * stream thus being globally visible.
1074 */
1075 if (stream->globally_visible) {
1076 pthread_mutex_lock(&the_consumer_data.lock);
1077 pthread_mutex_lock(&stream->chan->lock);
1078 pthread_mutex_lock(&stream->lock);
1079 /* Remove every reference of the stream in the consumer. */
1080 consumer_stream_delete(stream, ht);
1081
1082 destroy_close_stream(stream);
1083
1084 /* Update channel's refcount of the stream. */
1085 free_chan = unref_channel(stream);
1086
1087 /* Indicates that the consumer data state MUST be updated after this. */
1088 the_consumer_data.need_update = 1;
1089
1090 pthread_mutex_unlock(&stream->lock);
1091 pthread_mutex_unlock(&stream->chan->lock);
1092 pthread_mutex_unlock(&the_consumer_data.lock);
1093 } else {
1094 /*
1095 * If the stream is not visible globally, this needs to be done
1096 * outside of the consumer data lock section.
1097 */
1098 destroy_close_stream(stream);
1099 free_chan = unref_channel(stream);
1100 }
1101
1102 if (free_chan) {
1103 consumer_del_channel(free_chan);
1104 }
1105 } else {
1106 destroy_close_stream(stream);
1107 }
1108
1109 /* Free stream within a RCU call. */
1110 lttng_trace_chunk_put(stream->trace_chunk);
1111 stream->trace_chunk = NULL;
1112 lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
1113 consumer_stream_free(stream);
1114 }
1115
1116 /*
1117 * Write index of a specific stream either on the relayd or local disk.
1118 *
1119 * Return 0 on success or else a negative value.
1120 */
1121 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
1122 struct ctf_packet_index *element)
1123 {
1124 int ret;
1125
1126 LTTNG_ASSERT(stream);
1127 LTTNG_ASSERT(element);
1128
1129 rcu_read_lock();
1130 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1131 struct consumer_relayd_sock_pair *relayd;
1132 relayd = consumer_find_relayd(stream->net_seq_idx);
1133 if (relayd) {
1134 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1135 ret = relayd_send_index(&relayd->control_sock, element,
1136 stream->relayd_stream_id, stream->next_net_seq_num - 1);
1137 if (ret < 0) {
1138 /*
1139 * Communication error with lttng-relayd,
1140 * perform cleanup now
1141 */
1142 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
1143 lttng_consumer_cleanup_relayd(relayd);
1144 ret = -1;
1145 }
1146 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1147 } else {
1148 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
1149 stream->key, stream->net_seq_idx);
1150 ret = -1;
1151 }
1152 } else {
1153 if (lttng_index_file_write(stream->index_file, element)) {
1154 ret = -1;
1155 } else {
1156 ret = 0;
1157 }
1158 }
1159 if (ret < 0) {
1160 goto error;
1161 }
1162
1163 error:
1164 rcu_read_unlock();
1165 return ret;
1166 }
1167
1168 int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
1169 bool create_index)
1170 {
1171 int ret;
1172 enum lttng_trace_chunk_status chunk_status;
1173 const int flags = O_WRONLY | O_CREAT | O_TRUNC;
1174 const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
1175 char stream_path[LTTNG_PATH_MAX];
1176
1177 ASSERT_LOCKED(stream->lock);
1178 LTTNG_ASSERT(stream->trace_chunk);
1179
1180 ret = utils_stream_file_path(stream->chan->pathname, stream->name,
1181 stream->chan->tracefile_size,
1182 stream->tracefile_count_current, NULL,
1183 stream_path, sizeof(stream_path));
1184 if (ret < 0) {
1185 goto end;
1186 }
1187
1188 if (stream->out_fd >= 0) {
1189 ret = close(stream->out_fd);
1190 if (ret < 0) {
1191 PERROR("Failed to close stream file \"%s\"",
1192 stream->name);
1193 goto end;
1194 }
1195 stream->out_fd = -1;
1196 }
1197
1198 DBG("Opening stream output file \"%s\"", stream_path);
1199 chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
1200 flags, mode, &stream->out_fd, false);
1201 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1202 ERR("Failed to open stream file \"%s\"", stream->name);
1203 ret = -1;
1204 goto end;
1205 }
1206
1207 if (!stream->metadata_flag && (create_index || stream->index_file)) {
1208 if (stream->index_file) {
1209 lttng_index_file_put(stream->index_file);
1210 }
1211 chunk_status = lttng_index_file_create_from_trace_chunk(
1212 stream->trace_chunk,
1213 stream->chan->pathname,
1214 stream->name,
1215 stream->chan->tracefile_size,
1216 stream->tracefile_count_current,
1217 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
1218 false, &stream->index_file);
1219 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1220 ret = -1;
1221 goto end;
1222 }
1223 }
1224
1225 /* Reset current size because we just perform a rotation. */
1226 stream->tracefile_size_current = 0;
1227 stream->out_fd_offset = 0;
1228 end:
1229 return ret;
1230 }
1231
1232 int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
1233 {
1234 int ret;
1235
1236 stream->tracefile_count_current++;
1237 if (stream->chan->tracefile_count > 0) {
1238 stream->tracefile_count_current %=
1239 stream->chan->tracefile_count;
1240 }
1241
1242 DBG("Rotating output files of stream \"%s\"", stream->name);
1243 ret = consumer_stream_create_output_files(stream, true);
1244 if (ret) {
1245 goto end;
1246 }
1247
1248 end:
1249 return ret;
1250 }
1251
1252 bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
1253 {
1254 /*
1255 * This function does not take a const stream since
1256 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1257 */
1258 LTTNG_ASSERT(stream);
1259 return cds_lfht_is_node_deleted(&stream->node.node);
1260 }
1261
1262 static ssize_t metadata_bucket_flush(
1263 const struct stream_subbuffer *buffer, void *data)
1264 {
1265 ssize_t ret;
1266 struct lttng_consumer_stream *stream = (lttng_consumer_stream *) data;
1267
1268 ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1269 if (ret < 0) {
1270 goto end;
1271 }
1272 end:
1273 return ret;
1274 }
1275
1276 static ssize_t metadata_bucket_consume(
1277 struct lttng_consumer_local_data *unused __attribute__((unused)),
1278 struct lttng_consumer_stream *stream,
1279 const struct stream_subbuffer *subbuffer)
1280 {
1281 ssize_t ret;
1282 enum metadata_bucket_status status;
1283
1284 status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1285 switch (status) {
1286 case METADATA_BUCKET_STATUS_OK:
1287 /* Return consumed size. */
1288 ret = subbuffer->buffer.buffer.size;
1289 break;
1290 default:
1291 ret = -1;
1292 }
1293
1294 return ret;
1295 }
1296
1297 int consumer_stream_enable_metadata_bucketization(
1298 struct lttng_consumer_stream *stream)
1299 {
1300 int ret = 0;
1301
1302 LTTNG_ASSERT(stream->metadata_flag);
1303 LTTNG_ASSERT(!stream->metadata_bucket);
1304 LTTNG_ASSERT(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1305
1306 stream->metadata_bucket = metadata_bucket_create(
1307 metadata_bucket_flush, stream);
1308 if (!stream->metadata_bucket) {
1309 ret = -1;
1310 goto end;
1311 }
1312
1313 stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1314 end:
1315 return ret;
1316 }
1317
1318 void consumer_stream_metadata_set_version(
1319 struct lttng_consumer_stream *stream, uint64_t new_version)
1320 {
1321 LTTNG_ASSERT(new_version > stream->metadata_version);
1322 stream->metadata_version = new_version;
1323 stream->reset_metadata_flag = 1;
1324
1325 if (stream->metadata_bucket) {
1326 metadata_bucket_reset(stream->metadata_bucket);
1327 }
1328 }
1329
1330 int consumer_stream_flush_buffer(struct lttng_consumer_stream *stream,
1331 bool producer_active)
1332 {
1333 int ret = 0;
1334
1335 switch (the_consumer_data.type) {
1336 case LTTNG_CONSUMER_KERNEL:
1337 if (producer_active) {
1338 ret = kernctl_buffer_flush(stream->wait_fd);
1339 if (ret < 0) {
1340 ERR("Failed to flush kernel stream");
1341 goto end;
1342 }
1343 } else {
1344 ret = kernctl_buffer_flush_empty(stream->wait_fd);
1345 if (ret < 0) {
1346 /*
1347 * Doing a buffer flush which does not take into
1348 * account empty packets. This is not perfect,
1349 * but required as a fall-back when
1350 * "flush_empty" is not implemented by
1351 * lttng-modules.
1352 */
1353 ret = kernctl_buffer_flush(stream->wait_fd);
1354 if (ret < 0) {
1355 ERR("Failed to flush kernel stream");
1356 goto end;
1357 }
1358 }
1359 }
1360 break;
1361 case LTTNG_CONSUMER32_UST:
1362 case LTTNG_CONSUMER64_UST:
1363 ret = lttng_ustconsumer_flush_buffer(stream, (int) producer_active);
1364 break;
1365 default:
1366 ERR("Unknown consumer_data type");
1367 abort();
1368 }
1369
1370 end:
1371 return ret;
1372 }
This page took 0.075416 seconds and 4 git commands to generate.