Fix: ust-consumerd: leak of stream control structure
[lttng-tools.git] / src / common / consumer / consumer-stream.cpp
CommitLineData
51230d70 1/*
21cf9b6b 2 * Copyright (C) 2011 EfficiOS Inc.
ab5be9fa
MJ
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
10a50311 11#include <inttypes.h>
51230d70
DG
12#include <sys/mman.h>
13#include <unistd.h>
14
15#include <common/common.h>
947bd097
JR
16#include <common/consumer/consumer-timer.h>
17#include <common/consumer/consumer-timer.h>
18#include <common/consumer/consumer.h>
19#include <common/consumer/consumer.h>
20#include <common/consumer/metadata-bucket.h>
21#include <common/consumer/metadata-bucket.h>
1c20f0e2 22#include <common/index/index.h>
94d49140 23#include <common/kernel-consumer/kernel-consumer.h>
947bd097
JR
24#include <common/kernel-ctl/kernel-ctl.h>
25#include <common/macros.h>
51230d70
DG
26#include <common/relayd/relayd.h>
27#include <common/ust-consumer/ust-consumer.h>
a2361a61 28#include <common/utils.h>
51230d70
DG
29
30#include "consumer-stream.h"
31
32/*
33 * RCU call to free stream. MUST only be used with call_rcu().
34 */
35static void free_stream_rcu(struct rcu_head *head)
36{
37 struct lttng_ht_node_u64 *node =
38 caa_container_of(head, struct lttng_ht_node_u64, head);
39 struct lttng_consumer_stream *stream =
40 caa_container_of(node, struct lttng_consumer_stream, node);
41
42 pthread_mutex_destroy(&stream->lock);
43 free(stream);
44}
45
6f9449c2
JG
46static 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
52static 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
947bd097
JR
58static 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
6f9449c2
JG
64static 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
70static 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
947bd097
JR
76static 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
6f9449c2
JG
82/* Only used for data streams. */
83static 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;
d9b063d7 88 const uint64_t discarded_events = subbuf->info.data.events_discarded;
6f9449c2
JG
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
134end:
135 return ret;
136}
137
138static
139void 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
162static ssize_t consumer_stream_consume_mmap(
163 struct lttng_consumer_local_data *ctx,
164 struct lttng_consumer_stream *stream,
165 const struct stream_subbuffer *subbuffer)
166{
167 const unsigned long padding_size =
168 subbuffer->info.data.padded_subbuf_size -
169 subbuffer->info.data.subbuf_size;
514775d9 170 const ssize_t written_bytes = lttng_consumer_on_read_subbuffer_mmap(
f5ba75b4 171 stream, &subbuffer->buffer.buffer, padding_size);
514775d9
FD
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;
6f9449c2
JG
204}
205
206static 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{
514775d9
FD
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;
6f9449c2
JG
229}
230
231static int consumer_stream_send_index(
232 struct lttng_consumer_stream *stream,
233 const struct stream_subbuffer *subbuffer,
234 struct lttng_consumer_local_data *ctx)
235{
236 off_t packet_offset = 0;
237 struct ctf_packet_index index = {};
238
239 /*
240 * This is called after consuming the sub-buffer; substract the
241 * effect this sub-buffer from the offset.
242 */
243 if (stream->net_seq_idx == (uint64_t) -1ULL) {
244 packet_offset = stream->out_fd_offset -
245 subbuffer->info.data.padded_subbuf_size;
246 }
247
248 ctf_packet_index_populate(&index, packet_offset, subbuffer);
249 return consumer_stream_write_index(stream, &index);
250}
251
252/*
253 * Actually do the metadata sync using the given metadata stream.
254 *
255 * Return 0 on success else a negative value. ENODATA can be returned also
256 * indicating that there is no metadata available for that stream.
257 */
258static int do_sync_metadata(struct lttng_consumer_stream *metadata,
259 struct lttng_consumer_local_data *ctx)
260{
261 int ret;
577eea73 262 enum sync_metadata_status status;
6f9449c2 263
a0377dfe
FD
264 LTTNG_ASSERT(metadata);
265 LTTNG_ASSERT(metadata->metadata_flag);
266 LTTNG_ASSERT(ctx);
6f9449c2
JG
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 */
577eea73 310 status = lttng_kconsumer_sync_metadata(metadata);
6f9449c2
JG
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 */
577eea73 318 status = lttng_ustconsumer_sync_metadata(ctx, metadata);
6f9449c2
JG
319 break;
320 default:
577eea73 321 abort();
6f9449c2 322 }
577eea73
JG
323
324 switch (status) {
325 case SYNC_METADATA_STATUS_NEW_DATA:
326 break;
327 case SYNC_METADATA_STATUS_NO_DATA:
328 ret = 0;
6f9449c2 329 goto end_unlock_mutex;
577eea73
JG
330 case SYNC_METADATA_STATUS_ERROR:
331 ret = -1;
332 goto end_unlock_mutex;
333 default:
334 abort();
6f9449c2
JG
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);
577eea73 356 } while (status == SYNC_METADATA_STATUS_NEW_DATA);
6f9449c2
JG
357
358 /* Success */
359 return 0;
360
361end_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 */
376int 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
a0377dfe 384 LTTNG_ASSERT(ctx);
6f9449c2
JG
385
386 /* Ease our life a bit. */
fa29bfbf 387 ht = the_consumer_data.stream_list_ht;
6f9449c2
JG
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
412end:
413 rcu_read_unlock();
414 return ret;
415}
416
417static 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{
6653fca6 422 bool missed_metadata_flush;
6f9449c2
JG
423 int ret;
424
425 /* Block until all the metadata is sent. */
426 pthread_mutex_lock(&stream->metadata_timer_lock);
a0377dfe 427 LTTNG_ASSERT(!stream->missed_metadata_flush);
6f9449c2
JG
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;
6653fca6
MD
435 missed_metadata_flush = stream->missed_metadata_flush;
436 if (missed_metadata_flush) {
6f9449c2 437 stream->missed_metadata_flush = false;
6f9449c2 438 }
6653fca6 439 pthread_mutex_unlock(&stream->metadata_timer_lock);
6f9449c2
JG
440 if (ret < 0) {
441 goto end;
442 }
443
444 ret = consumer_stream_send_index(stream, subbuffer, ctx);
6653fca6
MD
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 }
6f9449c2
JG
463end:
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 */
472static
473int 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");
55954e07
JG
481 consumer_stream_metadata_set_version(stream,
482 subbuffer->info.metadata.version);
f5ba75b4 483
6f9449c2
JG
484 if (stream->read_subbuffer_ops.reset_metadata) {
485 stream->read_subbuffer_ops.reset_metadata(stream);
486 }
487
488end:
489 return 0;
490}
491
503fefca
JG
492static
493bool 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 }
507end:
508 return rotating_to_null_chunk;
509}
510
511enum 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
580end:
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 */
592static
593int post_consume_open_new_packet(struct lttng_consumer_stream *stream,
594 const struct stream_subbuffer *subbuffer,
595 struct lttng_consumer_local_data *ctx)
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
637end:
638 return ret;
639}
640
6f9449c2
JG
641struct 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
97535efa 657 stream = (lttng_consumer_stream *) zmalloc(sizeof(*stream));
6f9449c2
JG
658 if (stream == NULL) {
659 PERROR("malloc struct lttng_consumer_stream");
660 ret = -ENOMEM;
661 goto end;
662 }
663
63e9caa2
MD
664 rcu_read_lock();
665
6f9449c2
JG
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
6f9449c2
JG
672 stream->chan = channel;
673 stream->key = stream_key;
674 stream->trace_chunk = trace_chunk;
675 stream->out_fd = -1;
676 stream->out_fd_offset = 0;
677 stream->output_written = 0;
678 stream->net_seq_idx = relayd_id;
679 stream->session_id = session_id;
680 stream->monitor = monitor;
681 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
682 stream->index_file = NULL;
683 stream->last_sequence_number = -1ULL;
684 stream->rotate_position = -1ULL;
f96af312
JG
685 /* Buffer is created with an open packet. */
686 stream->opened_packet_in_current_trace_chunk = true;
6f9449c2
JG
687 pthread_mutex_init(&stream->lock, NULL);
688 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
689
690 /* If channel is the metadata, flag this stream as metadata. */
691 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
692 stream->metadata_flag = 1;
693 /* Metadata is flat out. */
694 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
695 /* Live rendez-vous point. */
696 pthread_cond_init(&stream->metadata_rdv, NULL);
697 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
698 } else {
699 /* Format stream name to <channel_name>_<cpu_number> */
700 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
701 channel_name, cpu);
702 if (ret < 0) {
703 PERROR("snprintf stream name");
704 goto error;
705 }
706 }
707
708 switch (channel->output) {
709 case CONSUMER_CHANNEL_SPLICE:
710 stream->output = LTTNG_EVENT_SPLICE;
711 ret = utils_create_pipe(stream->splice_pipe);
712 if (ret < 0) {
713 goto error;
714 }
715 break;
716 case CONSUMER_CHANNEL_MMAP:
717 stream->output = LTTNG_EVENT_MMAP;
718 break;
719 default:
720 abort();
721 }
722
723 /* Key is always the wait_fd for streams. */
724 lttng_ht_node_init_u64(&stream->node, stream->key);
725
726 /* Init node per channel id key */
727 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
728
729 /* Init session id node with the stream session id */
730 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
731
732 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
733 " relayd_id %" PRIu64 ", session_id %" PRIu64,
734 stream->name, stream->key, channel_key,
735 stream->net_seq_idx, stream->session_id);
736
737 rcu_read_unlock();
738
503fefca
JG
739 lttng_dynamic_array_init(&stream->read_subbuffer_ops.post_consume_cbs,
740 sizeof(post_consume_cb), NULL);
741
6f9449c2
JG
742 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
743 stream->read_subbuffer_ops.lock =
744 consumer_stream_metadata_lock_all;
745 stream->read_subbuffer_ops.unlock =
746 consumer_stream_metadata_unlock_all;
947bd097
JR
747 stream->read_subbuffer_ops.assert_locked =
748 consumer_stream_metadata_assert_locked_all;
6f9449c2
JG
749 stream->read_subbuffer_ops.pre_consume_subbuffer =
750 metadata_stream_check_version;
751 } else {
503fefca
JG
752 const post_consume_cb post_consume_index_op = channel->is_live ?
753 consumer_stream_sync_metadata_index :
754 consumer_stream_send_index;
97535efa
SM
755 const post_consume_cb post_consume_open_new_packet_ =
756 post_consume_open_new_packet;
503fefca
JG
757
758 ret = lttng_dynamic_array_add_element(
759 &stream->read_subbuffer_ops.post_consume_cbs,
760 &post_consume_index_op);
761 if (ret) {
762 PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
763 goto error;
764 }
765
766 ret = lttng_dynamic_array_add_element(
767 &stream->read_subbuffer_ops.post_consume_cbs,
97535efa 768 &post_consume_open_new_packet_);
503fefca
JG
769 if (ret) {
770 PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
771 goto error;
772 }
773
6f9449c2
JG
774 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
775 stream->read_subbuffer_ops.unlock =
776 consumer_stream_data_unlock_all;
947bd097
JR
777 stream->read_subbuffer_ops.assert_locked =
778 consumer_stream_data_assert_locked_all;
6f9449c2
JG
779 stream->read_subbuffer_ops.pre_consume_subbuffer =
780 consumer_stream_update_stats;
6f9449c2
JG
781 }
782
783 if (channel->output == CONSUMER_CHANNEL_MMAP) {
784 stream->read_subbuffer_ops.consume_subbuffer =
785 consumer_stream_consume_mmap;
786 } else {
787 stream->read_subbuffer_ops.consume_subbuffer =
788 consumer_stream_consume_splice;
789 }
790
791 return stream;
792
793error:
794 rcu_read_unlock();
795 lttng_trace_chunk_put(stream->trace_chunk);
503fefca 796 lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
6f9449c2
JG
797 free(stream);
798end:
799 if (alloc_ret) {
800 *alloc_ret = ret;
801 }
802 return NULL;
803}
804
51230d70
DG
805/*
806 * Close stream on the relayd side. This call can destroy a relayd if the
807 * conditions are met.
808 *
809 * A RCU read side lock MUST be acquired if the relayd object was looked up in
810 * a hash table before calling this.
811 */
812void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
813 struct consumer_relayd_sock_pair *relayd)
814{
815 int ret;
816
a0377dfe
FD
817 LTTNG_ASSERT(stream);
818 LTTNG_ASSERT(relayd);
51230d70 819
d01178b6
DG
820 if (stream->sent_to_relayd) {
821 uatomic_dec(&relayd->refcount);
a0377dfe 822 LTTNG_ASSERT(uatomic_read(&relayd->refcount) >= 0);
d01178b6 823 }
51230d70
DG
824
825 /* Closing streams requires to lock the control socket. */
826 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
827 ret = relayd_send_close_stream(&relayd->control_sock,
828 stream->relayd_stream_id,
829 stream->next_net_seq_num - 1);
830 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
831 if (ret < 0) {
9276e5c8
JR
832 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
833 lttng_consumer_cleanup_relayd(relayd);
51230d70
DG
834 }
835
836 /* Both conditions are met, we destroy the relayd. */
837 if (uatomic_read(&relayd->refcount) == 0 &&
838 uatomic_read(&relayd->destroy_flag)) {
839 consumer_destroy_relayd(relayd);
840 }
10a50311 841 stream->net_seq_idx = (uint64_t) -1ULL;
d01178b6 842 stream->sent_to_relayd = 0;
51230d70
DG
843}
844
845/*
846 * Close stream's file descriptors and, if needed, close stream also on the
847 * relayd side.
848 *
849 * The consumer data lock MUST be acquired.
850 * The stream lock MUST be acquired.
851 */
852void consumer_stream_close(struct lttng_consumer_stream *stream)
853{
854 int ret;
855 struct consumer_relayd_sock_pair *relayd;
856
a0377dfe 857 LTTNG_ASSERT(stream);
51230d70 858
fa29bfbf 859 switch (the_consumer_data.type) {
51230d70
DG
860 case LTTNG_CONSUMER_KERNEL:
861 if (stream->mmap_base != NULL) {
862 ret = munmap(stream->mmap_base, stream->mmap_len);
863 if (ret != 0) {
864 PERROR("munmap");
865 }
866 }
867
868 if (stream->wait_fd >= 0) {
869 ret = close(stream->wait_fd);
870 if (ret) {
871 PERROR("close");
872 }
10a50311 873 stream->wait_fd = -1;
51230d70 874 }
a2361a61
JD
875 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
876 utils_close_pipe(stream->splice_pipe);
877 }
51230d70
DG
878 break;
879 case LTTNG_CONSUMER32_UST:
880 case LTTNG_CONSUMER64_UST:
6d574024
DG
881 {
882 /*
883 * Special case for the metadata since the wait fd is an internal pipe
884 * polled in the metadata thread.
885 */
886 if (stream->metadata_flag && stream->chan->monitor) {
887 int rpipe = stream->ust_metadata_poll_pipe[0];
888
889 /*
890 * This will stop the channel timer if one and close the write side
891 * of the metadata poll pipe.
892 */
893 lttng_ustconsumer_close_metadata(stream->chan);
894 if (rpipe >= 0) {
895 ret = close(rpipe);
896 if (ret < 0) {
b4a650f3 897 PERROR("closing metadata pipe read side");
6d574024
DG
898 }
899 stream->ust_metadata_poll_pipe[0] = -1;
900 }
901 }
51230d70 902 break;
6d574024 903 }
51230d70
DG
904 default:
905 ERR("Unknown consumer_data type");
a0377dfe 906 abort();
51230d70
DG
907 }
908
909 /* Close output fd. Could be a socket or local file at this point. */
910 if (stream->out_fd >= 0) {
911 ret = close(stream->out_fd);
912 if (ret) {
913 PERROR("close");
914 }
10a50311 915 stream->out_fd = -1;
51230d70
DG
916 }
917
f8f3885c
MD
918 if (stream->index_file) {
919 lttng_index_file_put(stream->index_file);
920 stream->index_file = NULL;
309167d2
JD
921 }
922
d2956687
JG
923 lttng_trace_chunk_put(stream->trace_chunk);
924 stream->trace_chunk = NULL;
925
51230d70
DG
926 /* Check and cleanup relayd if needed. */
927 rcu_read_lock();
928 relayd = consumer_find_relayd(stream->net_seq_idx);
929 if (relayd != NULL) {
930 consumer_stream_relayd_close(stream, relayd);
931 }
932 rcu_read_unlock();
933}
934
935/*
936 * Delete the stream from all possible hash tables.
937 *
938 * The consumer data lock MUST be acquired.
939 * The stream lock MUST be acquired.
940 */
941void consumer_stream_delete(struct lttng_consumer_stream *stream,
942 struct lttng_ht *ht)
943{
944 int ret;
945 struct lttng_ht_iter iter;
946
a0377dfe 947 LTTNG_ASSERT(stream);
10a50311 948 /* Should NEVER be called not in monitor mode. */
a0377dfe 949 LTTNG_ASSERT(stream->chan->monitor);
51230d70
DG
950
951 rcu_read_lock();
952
953 if (ht) {
954 iter.iter.node = &stream->node.node;
955 ret = lttng_ht_del(ht, &iter);
a0377dfe 956 LTTNG_ASSERT(!ret);
51230d70
DG
957 }
958
959 /* Delete from stream per channel ID hash table. */
960 iter.iter.node = &stream->node_channel_id.node;
961 /*
962 * The returned value is of no importance. Even if the node is NOT in the
963 * hash table, we continue since we may have been called by a code path
964 * that did not add the stream to a (all) hash table. Same goes for the
965 * next call ht del call.
966 */
fa29bfbf 967 (void) lttng_ht_del(the_consumer_data.stream_per_chan_id_ht, &iter);
51230d70
DG
968
969 /* Delete from the global stream list. */
970 iter.iter.node = &stream->node_session_id.node;
971 /* See the previous ht del on why we ignore the returned value. */
fa29bfbf 972 (void) lttng_ht_del(the_consumer_data.stream_list_ht, &iter);
51230d70
DG
973
974 rcu_read_unlock();
975
6d574024
DG
976 if (!stream->metadata_flag) {
977 /* Decrement the stream count of the global consumer data. */
a0377dfe 978 LTTNG_ASSERT(the_consumer_data.stream_count > 0);
fa29bfbf 979 the_consumer_data.stream_count--;
6d574024 980 }
51230d70
DG
981}
982
983/*
984 * Free the given stream within a RCU call.
985 */
986void consumer_stream_free(struct lttng_consumer_stream *stream)
987{
a0377dfe 988 LTTNG_ASSERT(stream);
51230d70 989
f5ba75b4 990 metadata_bucket_destroy(stream->metadata_bucket);
51230d70
DG
991 call_rcu(&stream->node.head, free_stream_rcu);
992}
993
994/*
10a50311 995 * Destroy the stream's buffers of the tracer.
51230d70 996 */
10a50311 997void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
51230d70 998{
a0377dfe 999 LTTNG_ASSERT(stream);
10a50311 1000
fa29bfbf 1001 switch (the_consumer_data.type) {
10a50311
JD
1002 case LTTNG_CONSUMER_KERNEL:
1003 break;
1004 case LTTNG_CONSUMER32_UST:
1005 case LTTNG_CONSUMER64_UST:
1006 lttng_ustconsumer_del_stream(stream);
1007 break;
1008 default:
1009 ERR("Unknown consumer_data type");
a0377dfe 1010 abort();
10a50311
JD
1011 }
1012}
51230d70 1013
10a50311 1014/*
4891ece8 1015 * Destroy and close a already created stream.
10a50311 1016 */
4891ece8 1017static void destroy_close_stream(struct lttng_consumer_stream *stream)
10a50311 1018{
a0377dfe 1019 LTTNG_ASSERT(stream);
51230d70 1020
4891ece8 1021 DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
10a50311
JD
1022
1023 /* Destroy tracer buffers of the stream. */
1024 consumer_stream_destroy_buffers(stream);
1025 /* Close down everything including the relayd if one. */
1026 consumer_stream_close(stream);
1027}
51230d70 1028
10a50311 1029/*
4891ece8
DG
1030 * Decrement the stream's channel refcount and if down to 0, return the channel
1031 * pointer so it can be destroyed by the caller or NULL if not.
10a50311 1032 */
4891ece8
DG
1033static struct lttng_consumer_channel *unref_channel(
1034 struct lttng_consumer_stream *stream)
10a50311 1035{
4891ece8
DG
1036 struct lttng_consumer_channel *free_chan = NULL;
1037
a0377dfe
FD
1038 LTTNG_ASSERT(stream);
1039 LTTNG_ASSERT(stream->chan);
10a50311 1040
4891ece8
DG
1041 /* Update refcount of channel and see if we need to destroy it. */
1042 if (!uatomic_sub_return(&stream->chan->refcount, 1)
1043 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
1044 free_chan = stream->chan;
1045 }
51230d70 1046
4891ece8 1047 return free_chan;
10a50311 1048}
51230d70 1049
10a50311
JD
1050/*
1051 * Destroy a stream completely. This will delete, close and free the stream.
1052 * Once return, the stream is NO longer usable. Its channel may get destroyed
1053 * if conditions are met for a monitored stream.
1054 *
1055 * This MUST be called WITHOUT the consumer data and stream lock acquired if
1056 * the stream is in _monitor_ mode else it does not matter.
1057 */
1058void consumer_stream_destroy(struct lttng_consumer_stream *stream,
1059 struct lttng_ht *ht)
1060{
a0377dfe 1061 LTTNG_ASSERT(stream);
10a50311
JD
1062
1063 /* Stream is in monitor mode. */
4891ece8 1064 if (stream->monitor) {
10a50311 1065 struct lttng_consumer_channel *free_chan = NULL;
51230d70 1066
4891ece8
DG
1067 /*
1068 * This means that the stream was successfully removed from the streams
1069 * list of the channel and sent to the right thread managing this
1070 * stream thus being globally visible.
1071 */
1072 if (stream->globally_visible) {
fa29bfbf 1073 pthread_mutex_lock(&the_consumer_data.lock);
a9838785 1074 pthread_mutex_lock(&stream->chan->lock);
4891ece8
DG
1075 pthread_mutex_lock(&stream->lock);
1076 /* Remove every reference of the stream in the consumer. */
1077 consumer_stream_delete(stream, ht);
1078
1079 destroy_close_stream(stream);
1080
1081 /* Update channel's refcount of the stream. */
1082 free_chan = unref_channel(stream);
1083
1084 /* Indicates that the consumer data state MUST be updated after this. */
fa29bfbf 1085 the_consumer_data.need_update = 1;
4891ece8
DG
1086
1087 pthread_mutex_unlock(&stream->lock);
a9838785 1088 pthread_mutex_unlock(&stream->chan->lock);
fa29bfbf 1089 pthread_mutex_unlock(&the_consumer_data.lock);
4891ece8
DG
1090 } else {
1091 /*
1092 * If the stream is not visible globally, this needs to be done
1093 * outside of the consumer data lock section.
1094 */
38aea171 1095 destroy_close_stream(stream);
4891ece8 1096 free_chan = unref_channel(stream);
10a50311
JD
1097 }
1098
10a50311
JD
1099 if (free_chan) {
1100 consumer_del_channel(free_chan);
1101 }
1102 } else {
4891ece8 1103 destroy_close_stream(stream);
51230d70
DG
1104 }
1105
1106 /* Free stream within a RCU call. */
d2956687
JG
1107 lttng_trace_chunk_put(stream->trace_chunk);
1108 stream->trace_chunk = NULL;
503fefca 1109 lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
51230d70
DG
1110 consumer_stream_free(stream);
1111}
1c20f0e2
JD
1112
1113/*
1114 * Write index of a specific stream either on the relayd or local disk.
1115 *
1116 * Return 0 on success or else a negative value.
1117 */
1118int consumer_stream_write_index(struct lttng_consumer_stream *stream,
f8f3885c 1119 struct ctf_packet_index *element)
1c20f0e2
JD
1120{
1121 int ret;
1c20f0e2 1122
a0377dfe
FD
1123 LTTNG_ASSERT(stream);
1124 LTTNG_ASSERT(element);
1c20f0e2
JD
1125
1126 rcu_read_lock();
23c910e5
JR
1127 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1128 struct consumer_relayd_sock_pair *relayd;
1129 relayd = consumer_find_relayd(stream->net_seq_idx);
1130 if (relayd) {
1131 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1132 ret = relayd_send_index(&relayd->control_sock, element,
1c20f0e2 1133 stream->relayd_stream_id, stream->next_net_seq_num - 1);
9276e5c8
JR
1134 if (ret < 0) {
1135 /*
1136 * Communication error with lttng-relayd,
1137 * perform cleanup now
1138 */
1139 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
1140 lttng_consumer_cleanup_relayd(relayd);
1141 ret = -1;
1142 }
23c910e5
JR
1143 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1144 } else {
1145 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
1146 stream->key, stream->net_seq_idx);
1147 ret = -1;
1148 }
1c20f0e2 1149 } else {
f8f3885c 1150 if (lttng_index_file_write(stream->index_file, element)) {
6cd525e8
MD
1151 ret = -1;
1152 } else {
1153 ret = 0;
1154 }
1c20f0e2
JD
1155 }
1156 if (ret < 0) {
1157 goto error;
1158 }
1159
1160error:
1161 rcu_read_unlock();
1162 return ret;
1163}
94d49140 1164
d2956687
JG
1165int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
1166 bool create_index)
1167{
1168 int ret;
1169 enum lttng_trace_chunk_status chunk_status;
1170 const int flags = O_WRONLY | O_CREAT | O_TRUNC;
1171 const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
1172 char stream_path[LTTNG_PATH_MAX];
1173
1174 ASSERT_LOCKED(stream->lock);
a0377dfe 1175 LTTNG_ASSERT(stream->trace_chunk);
d2956687
JG
1176
1177 ret = utils_stream_file_path(stream->chan->pathname, stream->name,
1178 stream->chan->tracefile_size,
3b16476a 1179 stream->tracefile_count_current, NULL,
d2956687
JG
1180 stream_path, sizeof(stream_path));
1181 if (ret < 0) {
1182 goto end;
1183 }
1184
1185 if (stream->out_fd >= 0) {
1186 ret = close(stream->out_fd);
1187 if (ret < 0) {
1188 PERROR("Failed to close stream file \"%s\"",
1189 stream->name);
1190 goto end;
1191 }
1192 stream->out_fd = -1;
2932b393 1193 }
d2956687
JG
1194
1195 DBG("Opening stream output file \"%s\"", stream_path);
1196 chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
3ff5c5db 1197 flags, mode, &stream->out_fd, false);
2932b393 1198 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
d2956687
JG
1199 ERR("Failed to open stream file \"%s\"", stream->name);
1200 ret = -1;
1201 goto end;
2932b393 1202 }
d2956687
JG
1203
1204 if (!stream->metadata_flag && (create_index || stream->index_file)) {
1205 if (stream->index_file) {
1206 lttng_index_file_put(stream->index_file);
1207 }
3ff5c5db 1208 chunk_status = lttng_index_file_create_from_trace_chunk(
d2956687
JG
1209 stream->trace_chunk,
1210 stream->chan->pathname,
1211 stream->name,
1212 stream->chan->tracefile_size,
1213 stream->tracefile_count_current,
1214 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
3ff5c5db
MD
1215 false, &stream->index_file);
1216 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
d2956687
JG
1217 ret = -1;
1218 goto end;
1219 }
1220 }
1221
1222 /* Reset current size because we just perform a rotation. */
1223 stream->tracefile_size_current = 0;
1224 stream->out_fd_offset = 0;
1225end:
1226 return ret;
1227}
1228
1229int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
1230{
1231 int ret;
1232
1233 stream->tracefile_count_current++;
1234 if (stream->chan->tracefile_count > 0) {
1235 stream->tracefile_count_current %=
1236 stream->chan->tracefile_count;
1237 }
1238
1239 DBG("Rotating output files of stream \"%s\"", stream->name);
1240 ret = consumer_stream_create_output_files(stream, true);
1241 if (ret) {
1242 goto end;
1243 }
1244
1245end:
1246 return ret;
1247}
cdb72e4e
JG
1248
1249bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
1250{
1251 /*
1252 * This function does not take a const stream since
1253 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1254 */
a0377dfe 1255 LTTNG_ASSERT(stream);
cdb72e4e
JG
1256 return cds_lfht_is_node_deleted(&stream->node.node);
1257}
f5ba75b4
JG
1258
1259static ssize_t metadata_bucket_flush(
1260 const struct stream_subbuffer *buffer, void *data)
1261{
1262 ssize_t ret;
97535efa 1263 struct lttng_consumer_stream *stream = (lttng_consumer_stream *) data;
f5ba75b4
JG
1264
1265 ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1266 if (ret < 0) {
1267 goto end;
1268 }
1269end:
1270 return ret;
1271}
1272
1273static ssize_t metadata_bucket_consume(
1274 struct lttng_consumer_local_data *unused,
1275 struct lttng_consumer_stream *stream,
1276 const struct stream_subbuffer *subbuffer)
1277{
1278 ssize_t ret;
1279 enum metadata_bucket_status status;
1280
1281 status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1282 switch (status) {
1283 case METADATA_BUCKET_STATUS_OK:
1284 /* Return consumed size. */
1285 ret = subbuffer->buffer.buffer.size;
1286 break;
1287 default:
1288 ret = -1;
1289 }
1290
1291 return ret;
1292}
1293
1294int consumer_stream_enable_metadata_bucketization(
1295 struct lttng_consumer_stream *stream)
1296{
1297 int ret = 0;
1298
a0377dfe
FD
1299 LTTNG_ASSERT(stream->metadata_flag);
1300 LTTNG_ASSERT(!stream->metadata_bucket);
1301 LTTNG_ASSERT(stream->chan->output == CONSUMER_CHANNEL_MMAP);
f5ba75b4
JG
1302
1303 stream->metadata_bucket = metadata_bucket_create(
1304 metadata_bucket_flush, stream);
1305 if (!stream->metadata_bucket) {
1306 ret = -1;
1307 goto end;
1308 }
1309
1310 stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1311end:
1312 return ret;
1313}
55954e07
JG
1314
1315void consumer_stream_metadata_set_version(
1316 struct lttng_consumer_stream *stream, uint64_t new_version)
1317{
a0377dfe 1318 LTTNG_ASSERT(new_version > stream->metadata_version);
55954e07
JG
1319 stream->metadata_version = new_version;
1320 stream->reset_metadata_flag = 1;
1321
1322 if (stream->metadata_bucket) {
1323 metadata_bucket_reset(stream->metadata_bucket);
1324 }
1325}
503fefca
JG
1326
1327int consumer_stream_flush_buffer(struct lttng_consumer_stream *stream,
1328 bool producer_active)
1329{
1330 int ret = 0;
1331
fa29bfbf 1332 switch (the_consumer_data.type) {
503fefca
JG
1333 case LTTNG_CONSUMER_KERNEL:
1334 if (producer_active) {
1335 ret = kernctl_buffer_flush(stream->wait_fd);
1336 if (ret < 0) {
1337 ERR("Failed to flush kernel stream");
1338 goto end;
1339 }
1340 } else {
1341 ret = kernctl_buffer_flush_empty(stream->wait_fd);
1342 if (ret < 0) {
1343 /*
1344 * Doing a buffer flush which does not take into
1345 * account empty packets. This is not perfect,
1346 * but required as a fall-back when
1347 * "flush_empty" is not implemented by
1348 * lttng-modules.
1349 */
1350 ret = kernctl_buffer_flush(stream->wait_fd);
1351 if (ret < 0) {
1352 ERR("Failed to flush kernel stream");
1353 goto end;
1354 }
1355 }
1356 }
1357 break;
1358 case LTTNG_CONSUMER32_UST:
1359 case LTTNG_CONSUMER64_UST:
881fc67f 1360 ret = lttng_ustconsumer_flush_buffer(stream, (int) producer_active);
503fefca
JG
1361 break;
1362 default:
1363 ERR("Unknown consumer_data type");
1364 abort();
1365 }
1366
1367end:
1368 return ret;
1369}
This page took 0.119685 seconds and 4 git commands to generate.