8c21d6ae1edd3afe71967d6bf5e5cb32cb998675
[lttng-tools.git] / src / common / consumer / consumer-stream.c
1 /*
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>
5 *
6 * SPDX-License-Identifier: GPL-2.0-only
7 *
8 */
9
10 #define _LGPL_SOURCE
11 #include <assert.h>
12 #include <inttypes.h>
13 #include <sys/mman.h>
14 #include <unistd.h>
15
16 #include <common/common.h>
17 #include <common/index/index.h>
18 #include <common/kernel-consumer/kernel-consumer.h>
19 #include <common/relayd/relayd.h>
20 #include <common/ust-consumer/ust-consumer.h>
21 #include <common/utils.h>
22 #include <common/consumer/consumer.h>
23 #include <common/consumer/consumer-timer.h>
24 #include <common/consumer/metadata-bucket.h>
25 #include <common/kernel-ctl/kernel-ctl.h>
26
27 #include "consumer-stream.h"
28
29 /*
30 * RCU call to free stream. MUST only be used with call_rcu().
31 */
32 static 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
43 static 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
49 static 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
55 static 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
61 static 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. */
68 static 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;
73 const uint64_t discarded_events = subbuf->info.data.events_discarded;
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
119 end:
120 return ret;
121 }
122
123 static
124 void 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
147 static 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(
157 stream, &subbuffer->buffer.buffer, padding_size);
158 }
159
160 static 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
169 static 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 */
196 static int do_sync_metadata(struct lttng_consumer_stream *metadata,
197 struct lttng_consumer_local_data *ctx)
198 {
199 int ret;
200 enum sync_metadata_status status;
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 */
248 status = lttng_kconsumer_sync_metadata(metadata);
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 */
256 status = lttng_ustconsumer_sync_metadata(ctx, metadata);
257 break;
258 default:
259 abort();
260 }
261
262 switch (status) {
263 case SYNC_METADATA_STATUS_NEW_DATA:
264 break;
265 case SYNC_METADATA_STATUS_NO_DATA:
266 ret = 0;
267 goto end_unlock_mutex;
268 case SYNC_METADATA_STATUS_ERROR:
269 ret = -1;
270 goto end_unlock_mutex;
271 default:
272 abort();
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);
294 } while (status == SYNC_METADATA_STATUS_NEW_DATA);
295
296 /* Success */
297 return 0;
298
299 end_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 */
314 int 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
350 end:
351 rcu_read_unlock();
352 return ret;
353 }
354
355 static 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);
384 end:
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 */
393 static
394 int 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");
402 consumer_stream_metadata_set_version(stream,
403 subbuffer->info.metadata.version);
404
405 if (stream->read_subbuffer_ops.reset_metadata) {
406 stream->read_subbuffer_ops.reset_metadata(stream);
407 }
408
409 end:
410 return 0;
411 }
412
413 static
414 bool 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 }
428 end:
429 return rotating_to_null_chunk;
430 }
431
432 enum 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
501 end:
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 */
513 static
514 int 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
558 end:
559 return ret;
560 }
561
562 struct 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;
605 /* Buffer is created with an open packet. */
606 stream->opened_packet_in_current_trace_chunk = true;
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
659 lttng_dynamic_array_init(&stream->read_subbuffer_ops.post_consume_cbs,
660 sizeof(post_consume_cb), NULL);
661
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 {
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
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;
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
707 error:
708 rcu_read_unlock();
709 lttng_trace_chunk_put(stream->trace_chunk);
710 lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
711 free(stream);
712 end:
713 if (alloc_ret) {
714 *alloc_ret = ret;
715 }
716 return NULL;
717 }
718
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 */
726 void 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
734 if (stream->sent_to_relayd) {
735 uatomic_dec(&relayd->refcount);
736 assert(uatomic_read(&relayd->refcount) >= 0);
737 }
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) {
746 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
747 lttng_consumer_cleanup_relayd(relayd);
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 }
755 stream->net_seq_idx = (uint64_t) -1ULL;
756 stream->sent_to_relayd = 0;
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 */
766 void 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 }
787 stream->wait_fd = -1;
788 }
789 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
790 utils_close_pipe(stream->splice_pipe);
791 }
792 break;
793 case LTTNG_CONSUMER32_UST:
794 case LTTNG_CONSUMER64_UST:
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) {
811 PERROR("closing metadata pipe read side");
812 }
813 stream->ust_metadata_poll_pipe[0] = -1;
814 }
815 }
816 break;
817 }
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 }
829 stream->out_fd = -1;
830 }
831
832 if (stream->index_file) {
833 lttng_index_file_put(stream->index_file);
834 stream->index_file = NULL;
835 }
836
837 lttng_trace_chunk_put(stream->trace_chunk);
838 stream->trace_chunk = NULL;
839
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 */
855 void 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);
862 /* Should NEVER be called not in monitor mode. */
863 assert(stream->chan->monitor);
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
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 }
895 }
896
897 /*
898 * Free the given stream within a RCU call.
899 */
900 void consumer_stream_free(struct lttng_consumer_stream *stream)
901 {
902 assert(stream);
903
904 metadata_bucket_destroy(stream->metadata_bucket);
905 call_rcu(&stream->node.head, free_stream_rcu);
906 }
907
908 /*
909 * Destroy the stream's buffers of the tracer.
910 */
911 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
912 {
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 }
927
928 /*
929 * Destroy and close a already created stream.
930 */
931 static void destroy_close_stream(struct lttng_consumer_stream *stream)
932 {
933 assert(stream);
934
935 DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
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 }
942
943 /*
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.
946 */
947 static struct lttng_consumer_channel *unref_channel(
948 struct lttng_consumer_stream *stream)
949 {
950 struct lttng_consumer_channel *free_chan = NULL;
951
952 assert(stream);
953 assert(stream->chan);
954
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 }
960
961 return free_chan;
962 }
963
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 */
972 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
973 struct lttng_ht *ht)
974 {
975 assert(stream);
976
977 /* Stream is in monitor mode. */
978 if (stream->monitor) {
979 struct lttng_consumer_channel *free_chan = NULL;
980
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);
988 pthread_mutex_lock(&stream->chan->lock);
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);
1002 pthread_mutex_unlock(&stream->chan->lock);
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);
1010 }
1011
1012 if (free_chan) {
1013 consumer_del_channel(free_chan);
1014 }
1015 } else {
1016 destroy_close_stream(stream);
1017 }
1018
1019 /* Free stream within a RCU call. */
1020 lttng_trace_chunk_put(stream->trace_chunk);
1021 stream->trace_chunk = NULL;
1022 lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
1023 consumer_stream_free(stream);
1024 }
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 */
1031 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
1032 struct ctf_packet_index *element)
1033 {
1034 int ret;
1035
1036 assert(stream);
1037 assert(element);
1038
1039 rcu_read_lock();
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,
1046 stream->relayd_stream_id, stream->next_net_seq_num - 1);
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 }
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 }
1062 } else {
1063 if (lttng_index_file_write(stream->index_file, element)) {
1064 ret = -1;
1065 } else {
1066 ret = 0;
1067 }
1068 }
1069 if (ret < 0) {
1070 goto error;
1071 }
1072
1073 error:
1074 rcu_read_unlock();
1075 return ret;
1076 }
1077
1078 int 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,
1092 stream->tracefile_count_current, NULL,
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,
1110 flags, mode, &stream->out_fd, false);
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 }
1121 chunk_status = lttng_index_file_create_from_trace_chunk(
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,
1128 false, &stream->index_file);
1129 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
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;
1138 end:
1139 return ret;
1140 }
1141
1142 int 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
1158 end:
1159 return ret;
1160 }
1161
1162 bool 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 }
1171
1172 static 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 }
1182 end:
1183 return ret;
1184 }
1185
1186 static 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
1207 int 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;
1224 end:
1225 return ret;
1226 }
1227
1228 void 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 }
1239
1240 int 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
1280 end:
1281 return ret;
1282 }
This page took 0.088444 seconds and 3 git commands to generate.