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