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