consumerd: refactor: split read_subbuf into sub-operations
[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>
51230d70
DG
24
25#include "consumer-stream.h"
26
27/*
28 * RCU call to free stream. MUST only be used with call_rcu().
29 */
30static void free_stream_rcu(struct rcu_head *head)
31{
32 struct lttng_ht_node_u64 *node =
33 caa_container_of(head, struct lttng_ht_node_u64, head);
34 struct lttng_consumer_stream *stream =
35 caa_container_of(node, struct lttng_consumer_stream, node);
36
37 pthread_mutex_destroy(&stream->lock);
38 free(stream);
39}
40
6f9449c2
JG
41static void consumer_stream_data_lock_all(struct lttng_consumer_stream *stream)
42{
43 pthread_mutex_lock(&stream->chan->lock);
44 pthread_mutex_lock(&stream->lock);
45}
46
47static void consumer_stream_data_unlock_all(struct lttng_consumer_stream *stream)
48{
49 pthread_mutex_unlock(&stream->lock);
50 pthread_mutex_unlock(&stream->chan->lock);
51}
52
53static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream *stream)
54{
55 consumer_stream_data_lock_all(stream);
56 pthread_mutex_lock(&stream->metadata_rdv_lock);
57}
58
59static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream *stream)
60{
61 pthread_mutex_unlock(&stream->metadata_rdv_lock);
62 consumer_stream_data_unlock_all(stream);
63}
64
65/* Only used for data streams. */
66static int consumer_stream_update_stats(struct lttng_consumer_stream *stream,
67 const struct stream_subbuffer *subbuf)
68{
69 int ret = 0;
70 uint64_t sequence_number;
71 const uint64_t discarded_events =
72 LTTNG_OPTIONAL_GET(subbuf->info.data.sequence_number);
73
74 if (!subbuf->info.data.sequence_number.is_set) {
75 /* Command not supported by the tracer. */
76 sequence_number = -1ULL;
77 stream->sequence_number_unavailable = true;
78 } else {
79 sequence_number = subbuf->info.data.sequence_number.value;
80 }
81
82 /*
83 * Start the sequence when we extract the first packet in case we don't
84 * start at 0 (for example if a consumer is not connected to the
85 * session immediately after the beginning).
86 */
87 if (stream->last_sequence_number == -1ULL) {
88 stream->last_sequence_number = sequence_number;
89 } else if (sequence_number > stream->last_sequence_number) {
90 stream->chan->lost_packets += sequence_number -
91 stream->last_sequence_number - 1;
92 } else {
93 /* seq <= last_sequence_number */
94 ERR("Sequence number inconsistent : prev = %" PRIu64
95 ", current = %" PRIu64,
96 stream->last_sequence_number, sequence_number);
97 ret = -1;
98 goto end;
99 }
100 stream->last_sequence_number = sequence_number;
101
102 if (discarded_events < stream->last_discarded_events) {
103 /*
104 * Overflow has occurred. We assume only one wrap-around
105 * has occurred.
106 */
107 stream->chan->discarded_events +=
108 (1ULL << (CAA_BITS_PER_LONG - 1)) -
109 stream->last_discarded_events +
110 discarded_events;
111 } else {
112 stream->chan->discarded_events += discarded_events -
113 stream->last_discarded_events;
114 }
115 stream->last_discarded_events = discarded_events;
116 ret = 0;
117
118end:
119 return ret;
120}
121
122static
123void ctf_packet_index_populate(struct ctf_packet_index *index,
124 off_t offset, const struct stream_subbuffer *subbuffer)
125{
126 *index = (typeof(*index)){
127 .offset = htobe64(offset),
128 .packet_size = htobe64(subbuffer->info.data.packet_size),
129 .content_size = htobe64(subbuffer->info.data.content_size),
130 .timestamp_begin = htobe64(
131 subbuffer->info.data.timestamp_begin),
132 .timestamp_end = htobe64(
133 subbuffer->info.data.timestamp_end),
134 .events_discarded = htobe64(
135 subbuffer->info.data.events_discarded),
136 .stream_id = htobe64(subbuffer->info.data.stream_id),
137 .stream_instance_id = htobe64(
138 subbuffer->info.data.stream_instance_id.is_set ?
139 subbuffer->info.data.stream_instance_id.value : -1ULL),
140 .packet_seq_num = htobe64(
141 subbuffer->info.data.sequence_number.is_set ?
142 subbuffer->info.data.sequence_number.value : -1ULL),
143 };
144}
145
146static ssize_t consumer_stream_consume_mmap(
147 struct lttng_consumer_local_data *ctx,
148 struct lttng_consumer_stream *stream,
149 const struct stream_subbuffer *subbuffer)
150{
151 const unsigned long padding_size =
152 subbuffer->info.data.padded_subbuf_size -
153 subbuffer->info.data.subbuf_size;
154
155 return lttng_consumer_on_read_subbuffer_mmap(
156 ctx, stream, &subbuffer->buffer.buffer, padding_size);
157}
158
159static ssize_t consumer_stream_consume_splice(
160 struct lttng_consumer_local_data *ctx,
161 struct lttng_consumer_stream *stream,
162 const struct stream_subbuffer *subbuffer)
163{
164 return lttng_consumer_on_read_subbuffer_splice(ctx, stream,
165 subbuffer->info.data.padded_subbuf_size, 0);
166}
167
168static int consumer_stream_send_index(
169 struct lttng_consumer_stream *stream,
170 const struct stream_subbuffer *subbuffer,
171 struct lttng_consumer_local_data *ctx)
172{
173 off_t packet_offset = 0;
174 struct ctf_packet_index index = {};
175
176 /*
177 * This is called after consuming the sub-buffer; substract the
178 * effect this sub-buffer from the offset.
179 */
180 if (stream->net_seq_idx == (uint64_t) -1ULL) {
181 packet_offset = stream->out_fd_offset -
182 subbuffer->info.data.padded_subbuf_size;
183 }
184
185 ctf_packet_index_populate(&index, packet_offset, subbuffer);
186 return consumer_stream_write_index(stream, &index);
187}
188
189/*
190 * Actually do the metadata sync using the given metadata stream.
191 *
192 * Return 0 on success else a negative value. ENODATA can be returned also
193 * indicating that there is no metadata available for that stream.
194 */
195static int do_sync_metadata(struct lttng_consumer_stream *metadata,
196 struct lttng_consumer_local_data *ctx)
197{
198 int ret;
199
200 assert(metadata);
201 assert(metadata->metadata_flag);
202 assert(ctx);
203
204 /*
205 * In UST, since we have to write the metadata from the cache packet
206 * by packet, we might need to start this procedure multiple times
207 * until all the metadata from the cache has been extracted.
208 */
209 do {
210 /*
211 * Steps :
212 * - Lock the metadata stream
213 * - Check if metadata stream node was deleted before locking.
214 * - if yes, release and return success
215 * - Check if new metadata is ready (flush + snapshot pos)
216 * - If nothing : release and return.
217 * - Lock the metadata_rdv_lock
218 * - Unlock the metadata stream
219 * - cond_wait on metadata_rdv to wait the wakeup from the
220 * metadata thread
221 * - Unlock the metadata_rdv_lock
222 */
223 pthread_mutex_lock(&metadata->lock);
224
225 /*
226 * There is a possibility that we were able to acquire a reference on the
227 * stream from the RCU hash table but between then and now, the node might
228 * have been deleted just before the lock is acquired. Thus, after locking,
229 * we make sure the metadata node has not been deleted which means that the
230 * buffers are closed.
231 *
232 * In that case, there is no need to sync the metadata hence returning a
233 * success return code.
234 */
235 ret = cds_lfht_is_node_deleted(&metadata->node.node);
236 if (ret) {
237 ret = 0;
238 goto end_unlock_mutex;
239 }
240
241 switch (ctx->type) {
242 case LTTNG_CONSUMER_KERNEL:
243 /*
244 * Empty the metadata cache and flush the current stream.
245 */
246 ret = lttng_kconsumer_sync_metadata(metadata);
247 break;
248 case LTTNG_CONSUMER32_UST:
249 case LTTNG_CONSUMER64_UST:
250 /*
251 * Ask the sessiond if we have new metadata waiting and update the
252 * consumer metadata cache.
253 */
254 ret = lttng_ustconsumer_sync_metadata(ctx, metadata);
255 break;
256 default:
257 assert(0);
258 ret = -1;
259 break;
260 }
261 /*
262 * Error or no new metadata, we exit here.
263 */
264 if (ret <= 0 || ret == ENODATA) {
265 goto end_unlock_mutex;
266 }
267
268 /*
269 * At this point, new metadata have been flushed, so we wait on the
270 * rendez-vous point for the metadata thread to wake us up when it
271 * finishes consuming the metadata and continue execution.
272 */
273
274 pthread_mutex_lock(&metadata->metadata_rdv_lock);
275
276 /*
277 * Release metadata stream lock so the metadata thread can process it.
278 */
279 pthread_mutex_unlock(&metadata->lock);
280
281 /*
282 * Wait on the rendez-vous point. Once woken up, it means the metadata was
283 * consumed and thus synchronization is achieved.
284 */
285 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
286 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
287 } while (ret == EAGAIN);
288
289 /* Success */
290 return 0;
291
292end_unlock_mutex:
293 pthread_mutex_unlock(&metadata->lock);
294 return ret;
295}
296
297/*
298 * Synchronize the metadata using a given session ID. A successful acquisition
299 * of a metadata stream will trigger a request to the session daemon and a
300 * snapshot so the metadata thread can consume it.
301 *
302 * This function call is a rendez-vous point between the metadata thread and
303 * the data thread.
304 *
305 * Return 0 on success or else a negative value.
306 */
307int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
308 uint64_t session_id)
309{
310 int ret;
311 struct lttng_consumer_stream *stream = NULL;
312 struct lttng_ht_iter iter;
313 struct lttng_ht *ht;
314
315 assert(ctx);
316
317 /* Ease our life a bit. */
318 ht = consumer_data.stream_list_ht;
319
320 rcu_read_lock();
321
322 /* Search the metadata associated with the session id of the given stream. */
323
324 cds_lfht_for_each_entry_duplicate(ht->ht,
325 ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
326 &session_id, &iter.iter, stream, node_session_id.node) {
327 if (!stream->metadata_flag) {
328 continue;
329 }
330
331 ret = do_sync_metadata(stream, ctx);
332 if (ret < 0) {
333 goto end;
334 }
335 }
336
337 /*
338 * Force return code to 0 (success) since ret might be ENODATA for instance
339 * which is not an error but rather that we should come back.
340 */
341 ret = 0;
342
343end:
344 rcu_read_unlock();
345 return ret;
346}
347
348static int consumer_stream_sync_metadata_index(
349 struct lttng_consumer_stream *stream,
350 const struct stream_subbuffer *subbuffer,
351 struct lttng_consumer_local_data *ctx)
352{
353 int ret;
354
355 /* Block until all the metadata is sent. */
356 pthread_mutex_lock(&stream->metadata_timer_lock);
357 assert(!stream->missed_metadata_flush);
358 stream->waiting_on_metadata = true;
359 pthread_mutex_unlock(&stream->metadata_timer_lock);
360
361 ret = consumer_stream_sync_metadata(ctx, stream->session_id);
362
363 pthread_mutex_lock(&stream->metadata_timer_lock);
364 stream->waiting_on_metadata = false;
365 if (stream->missed_metadata_flush) {
366 stream->missed_metadata_flush = false;
367 pthread_mutex_unlock(&stream->metadata_timer_lock);
368 (void) stream->read_subbuffer_ops.send_live_beacon(stream);
369 } else {
370 pthread_mutex_unlock(&stream->metadata_timer_lock);
371 }
372 if (ret < 0) {
373 goto end;
374 }
375
376 ret = consumer_stream_send_index(stream, subbuffer, ctx);
377end:
378 return ret;
379}
380
381/*
382 * Check if the local version of the metadata stream matches with the version
383 * of the metadata stream in the kernel. If it was updated, set the reset flag
384 * on the stream.
385 */
386static
387int metadata_stream_check_version(struct lttng_consumer_stream *stream,
388 const struct stream_subbuffer *subbuffer)
389{
390 if (stream->metadata_version == subbuffer->info.metadata.version) {
391 goto end;
392 }
393
394 DBG("New metadata version detected");
395 stream->metadata_version = subbuffer->info.metadata.version;
396 stream->reset_metadata_flag = 1;
397
398 if (stream->read_subbuffer_ops.reset_metadata) {
399 stream->read_subbuffer_ops.reset_metadata(stream);
400 }
401
402end:
403 return 0;
404}
405
406struct lttng_consumer_stream *consumer_stream_create(
407 struct lttng_consumer_channel *channel,
408 uint64_t channel_key,
409 uint64_t stream_key,
410 const char *channel_name,
411 uint64_t relayd_id,
412 uint64_t session_id,
413 struct lttng_trace_chunk *trace_chunk,
414 int cpu,
415 int *alloc_ret,
416 enum consumer_channel_type type,
417 unsigned int monitor)
418{
419 int ret;
420 struct lttng_consumer_stream *stream;
421
422 stream = zmalloc(sizeof(*stream));
423 if (stream == NULL) {
424 PERROR("malloc struct lttng_consumer_stream");
425 ret = -ENOMEM;
426 goto end;
427 }
428
429 if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
430 ERR("Failed to acquire trace chunk reference during the creation of a stream");
431 ret = -1;
432 goto error;
433 }
434
435 rcu_read_lock();
436 stream->chan = channel;
437 stream->key = stream_key;
438 stream->trace_chunk = trace_chunk;
439 stream->out_fd = -1;
440 stream->out_fd_offset = 0;
441 stream->output_written = 0;
442 stream->net_seq_idx = relayd_id;
443 stream->session_id = session_id;
444 stream->monitor = monitor;
445 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
446 stream->index_file = NULL;
447 stream->last_sequence_number = -1ULL;
448 stream->rotate_position = -1ULL;
449 pthread_mutex_init(&stream->lock, NULL);
450 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
451
452 /* If channel is the metadata, flag this stream as metadata. */
453 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
454 stream->metadata_flag = 1;
455 /* Metadata is flat out. */
456 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
457 /* Live rendez-vous point. */
458 pthread_cond_init(&stream->metadata_rdv, NULL);
459 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
460 } else {
461 /* Format stream name to <channel_name>_<cpu_number> */
462 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
463 channel_name, cpu);
464 if (ret < 0) {
465 PERROR("snprintf stream name");
466 goto error;
467 }
468 }
469
470 switch (channel->output) {
471 case CONSUMER_CHANNEL_SPLICE:
472 stream->output = LTTNG_EVENT_SPLICE;
473 ret = utils_create_pipe(stream->splice_pipe);
474 if (ret < 0) {
475 goto error;
476 }
477 break;
478 case CONSUMER_CHANNEL_MMAP:
479 stream->output = LTTNG_EVENT_MMAP;
480 break;
481 default:
482 abort();
483 }
484
485 /* Key is always the wait_fd for streams. */
486 lttng_ht_node_init_u64(&stream->node, stream->key);
487
488 /* Init node per channel id key */
489 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
490
491 /* Init session id node with the stream session id */
492 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
493
494 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
495 " relayd_id %" PRIu64 ", session_id %" PRIu64,
496 stream->name, stream->key, channel_key,
497 stream->net_seq_idx, stream->session_id);
498
499 rcu_read_unlock();
500
501 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
502 stream->read_subbuffer_ops.lock =
503 consumer_stream_metadata_lock_all;
504 stream->read_subbuffer_ops.unlock =
505 consumer_stream_metadata_unlock_all;
506 stream->read_subbuffer_ops.pre_consume_subbuffer =
507 metadata_stream_check_version;
508 } else {
509 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
510 stream->read_subbuffer_ops.unlock =
511 consumer_stream_data_unlock_all;
512 stream->read_subbuffer_ops.pre_consume_subbuffer =
513 consumer_stream_update_stats;
514 if (channel->is_live) {
515 stream->read_subbuffer_ops.post_consume =
516 consumer_stream_sync_metadata_index;
517 } else {
518 stream->read_subbuffer_ops.post_consume =
519 consumer_stream_send_index;
520 }
521 }
522
523 if (channel->output == CONSUMER_CHANNEL_MMAP) {
524 stream->read_subbuffer_ops.consume_subbuffer =
525 consumer_stream_consume_mmap;
526 } else {
527 stream->read_subbuffer_ops.consume_subbuffer =
528 consumer_stream_consume_splice;
529 }
530
531 return stream;
532
533error:
534 rcu_read_unlock();
535 lttng_trace_chunk_put(stream->trace_chunk);
536 free(stream);
537end:
538 if (alloc_ret) {
539 *alloc_ret = ret;
540 }
541 return NULL;
542}
543
51230d70
DG
544/*
545 * Close stream on the relayd side. This call can destroy a relayd if the
546 * conditions are met.
547 *
548 * A RCU read side lock MUST be acquired if the relayd object was looked up in
549 * a hash table before calling this.
550 */
551void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
552 struct consumer_relayd_sock_pair *relayd)
553{
554 int ret;
555
556 assert(stream);
557 assert(relayd);
558
d01178b6
DG
559 if (stream->sent_to_relayd) {
560 uatomic_dec(&relayd->refcount);
561 assert(uatomic_read(&relayd->refcount) >= 0);
562 }
51230d70
DG
563
564 /* Closing streams requires to lock the control socket. */
565 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
566 ret = relayd_send_close_stream(&relayd->control_sock,
567 stream->relayd_stream_id,
568 stream->next_net_seq_num - 1);
569 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
570 if (ret < 0) {
9276e5c8
JR
571 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
572 lttng_consumer_cleanup_relayd(relayd);
51230d70
DG
573 }
574
575 /* Both conditions are met, we destroy the relayd. */
576 if (uatomic_read(&relayd->refcount) == 0 &&
577 uatomic_read(&relayd->destroy_flag)) {
578 consumer_destroy_relayd(relayd);
579 }
10a50311 580 stream->net_seq_idx = (uint64_t) -1ULL;
d01178b6 581 stream->sent_to_relayd = 0;
51230d70
DG
582}
583
584/*
585 * Close stream's file descriptors and, if needed, close stream also on the
586 * relayd side.
587 *
588 * The consumer data lock MUST be acquired.
589 * The stream lock MUST be acquired.
590 */
591void consumer_stream_close(struct lttng_consumer_stream *stream)
592{
593 int ret;
594 struct consumer_relayd_sock_pair *relayd;
595
596 assert(stream);
597
598 switch (consumer_data.type) {
599 case LTTNG_CONSUMER_KERNEL:
600 if (stream->mmap_base != NULL) {
601 ret = munmap(stream->mmap_base, stream->mmap_len);
602 if (ret != 0) {
603 PERROR("munmap");
604 }
605 }
606
607 if (stream->wait_fd >= 0) {
608 ret = close(stream->wait_fd);
609 if (ret) {
610 PERROR("close");
611 }
10a50311 612 stream->wait_fd = -1;
51230d70 613 }
a2361a61
JD
614 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
615 utils_close_pipe(stream->splice_pipe);
616 }
51230d70
DG
617 break;
618 case LTTNG_CONSUMER32_UST:
619 case LTTNG_CONSUMER64_UST:
6d574024
DG
620 {
621 /*
622 * Special case for the metadata since the wait fd is an internal pipe
623 * polled in the metadata thread.
624 */
625 if (stream->metadata_flag && stream->chan->monitor) {
626 int rpipe = stream->ust_metadata_poll_pipe[0];
627
628 /*
629 * This will stop the channel timer if one and close the write side
630 * of the metadata poll pipe.
631 */
632 lttng_ustconsumer_close_metadata(stream->chan);
633 if (rpipe >= 0) {
634 ret = close(rpipe);
635 if (ret < 0) {
b4a650f3 636 PERROR("closing metadata pipe read side");
6d574024
DG
637 }
638 stream->ust_metadata_poll_pipe[0] = -1;
639 }
640 }
51230d70 641 break;
6d574024 642 }
51230d70
DG
643 default:
644 ERR("Unknown consumer_data type");
645 assert(0);
646 }
647
648 /* Close output fd. Could be a socket or local file at this point. */
649 if (stream->out_fd >= 0) {
650 ret = close(stream->out_fd);
651 if (ret) {
652 PERROR("close");
653 }
10a50311 654 stream->out_fd = -1;
51230d70
DG
655 }
656
f8f3885c
MD
657 if (stream->index_file) {
658 lttng_index_file_put(stream->index_file);
659 stream->index_file = NULL;
309167d2
JD
660 }
661
d2956687
JG
662 lttng_trace_chunk_put(stream->trace_chunk);
663 stream->trace_chunk = NULL;
664
51230d70
DG
665 /* Check and cleanup relayd if needed. */
666 rcu_read_lock();
667 relayd = consumer_find_relayd(stream->net_seq_idx);
668 if (relayd != NULL) {
669 consumer_stream_relayd_close(stream, relayd);
670 }
671 rcu_read_unlock();
672}
673
674/*
675 * Delete the stream from all possible hash tables.
676 *
677 * The consumer data lock MUST be acquired.
678 * The stream lock MUST be acquired.
679 */
680void consumer_stream_delete(struct lttng_consumer_stream *stream,
681 struct lttng_ht *ht)
682{
683 int ret;
684 struct lttng_ht_iter iter;
685
686 assert(stream);
10a50311
JD
687 /* Should NEVER be called not in monitor mode. */
688 assert(stream->chan->monitor);
51230d70
DG
689
690 rcu_read_lock();
691
692 if (ht) {
693 iter.iter.node = &stream->node.node;
694 ret = lttng_ht_del(ht, &iter);
695 assert(!ret);
696 }
697
698 /* Delete from stream per channel ID hash table. */
699 iter.iter.node = &stream->node_channel_id.node;
700 /*
701 * The returned value is of no importance. Even if the node is NOT in the
702 * hash table, we continue since we may have been called by a code path
703 * that did not add the stream to a (all) hash table. Same goes for the
704 * next call ht del call.
705 */
706 (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
707
708 /* Delete from the global stream list. */
709 iter.iter.node = &stream->node_session_id.node;
710 /* See the previous ht del on why we ignore the returned value. */
711 (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
712
713 rcu_read_unlock();
714
6d574024
DG
715 if (!stream->metadata_flag) {
716 /* Decrement the stream count of the global consumer data. */
717 assert(consumer_data.stream_count > 0);
718 consumer_data.stream_count--;
719 }
51230d70
DG
720}
721
722/*
723 * Free the given stream within a RCU call.
724 */
725void consumer_stream_free(struct lttng_consumer_stream *stream)
726{
727 assert(stream);
728
729 call_rcu(&stream->node.head, free_stream_rcu);
730}
731
732/*
10a50311 733 * Destroy the stream's buffers of the tracer.
51230d70 734 */
10a50311 735void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
51230d70 736{
10a50311
JD
737 assert(stream);
738
739 switch (consumer_data.type) {
740 case LTTNG_CONSUMER_KERNEL:
741 break;
742 case LTTNG_CONSUMER32_UST:
743 case LTTNG_CONSUMER64_UST:
744 lttng_ustconsumer_del_stream(stream);
745 break;
746 default:
747 ERR("Unknown consumer_data type");
748 assert(0);
749 }
750}
51230d70 751
10a50311 752/*
4891ece8 753 * Destroy and close a already created stream.
10a50311 754 */
4891ece8 755static void destroy_close_stream(struct lttng_consumer_stream *stream)
10a50311 756{
51230d70
DG
757 assert(stream);
758
4891ece8 759 DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
10a50311
JD
760
761 /* Destroy tracer buffers of the stream. */
762 consumer_stream_destroy_buffers(stream);
763 /* Close down everything including the relayd if one. */
764 consumer_stream_close(stream);
765}
51230d70 766
10a50311 767/*
4891ece8
DG
768 * Decrement the stream's channel refcount and if down to 0, return the channel
769 * pointer so it can be destroyed by the caller or NULL if not.
10a50311 770 */
4891ece8
DG
771static struct lttng_consumer_channel *unref_channel(
772 struct lttng_consumer_stream *stream)
10a50311 773{
4891ece8
DG
774 struct lttng_consumer_channel *free_chan = NULL;
775
10a50311 776 assert(stream);
4891ece8 777 assert(stream->chan);
10a50311 778
4891ece8
DG
779 /* Update refcount of channel and see if we need to destroy it. */
780 if (!uatomic_sub_return(&stream->chan->refcount, 1)
781 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
782 free_chan = stream->chan;
783 }
51230d70 784
4891ece8 785 return free_chan;
10a50311 786}
51230d70 787
10a50311
JD
788/*
789 * Destroy a stream completely. This will delete, close and free the stream.
790 * Once return, the stream is NO longer usable. Its channel may get destroyed
791 * if conditions are met for a monitored stream.
792 *
793 * This MUST be called WITHOUT the consumer data and stream lock acquired if
794 * the stream is in _monitor_ mode else it does not matter.
795 */
796void consumer_stream_destroy(struct lttng_consumer_stream *stream,
797 struct lttng_ht *ht)
798{
799 assert(stream);
800
801 /* Stream is in monitor mode. */
4891ece8 802 if (stream->monitor) {
10a50311 803 struct lttng_consumer_channel *free_chan = NULL;
51230d70 804
4891ece8
DG
805 /*
806 * This means that the stream was successfully removed from the streams
807 * list of the channel and sent to the right thread managing this
808 * stream thus being globally visible.
809 */
810 if (stream->globally_visible) {
811 pthread_mutex_lock(&consumer_data.lock);
a9838785 812 pthread_mutex_lock(&stream->chan->lock);
4891ece8
DG
813 pthread_mutex_lock(&stream->lock);
814 /* Remove every reference of the stream in the consumer. */
815 consumer_stream_delete(stream, ht);
816
817 destroy_close_stream(stream);
818
819 /* Update channel's refcount of the stream. */
820 free_chan = unref_channel(stream);
821
822 /* Indicates that the consumer data state MUST be updated after this. */
823 consumer_data.need_update = 1;
824
825 pthread_mutex_unlock(&stream->lock);
a9838785 826 pthread_mutex_unlock(&stream->chan->lock);
4891ece8
DG
827 pthread_mutex_unlock(&consumer_data.lock);
828 } else {
829 /*
830 * If the stream is not visible globally, this needs to be done
831 * outside of the consumer data lock section.
832 */
833 free_chan = unref_channel(stream);
10a50311
JD
834 }
835
10a50311
JD
836 if (free_chan) {
837 consumer_del_channel(free_chan);
838 }
839 } else {
4891ece8 840 destroy_close_stream(stream);
51230d70
DG
841 }
842
843 /* Free stream within a RCU call. */
d2956687
JG
844 lttng_trace_chunk_put(stream->trace_chunk);
845 stream->trace_chunk = NULL;
51230d70
DG
846 consumer_stream_free(stream);
847}
1c20f0e2
JD
848
849/*
850 * Write index of a specific stream either on the relayd or local disk.
851 *
852 * Return 0 on success or else a negative value.
853 */
854int consumer_stream_write_index(struct lttng_consumer_stream *stream,
f8f3885c 855 struct ctf_packet_index *element)
1c20f0e2
JD
856{
857 int ret;
1c20f0e2
JD
858
859 assert(stream);
f8f3885c 860 assert(element);
1c20f0e2
JD
861
862 rcu_read_lock();
23c910e5
JR
863 if (stream->net_seq_idx != (uint64_t) -1ULL) {
864 struct consumer_relayd_sock_pair *relayd;
865 relayd = consumer_find_relayd(stream->net_seq_idx);
866 if (relayd) {
867 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
868 ret = relayd_send_index(&relayd->control_sock, element,
1c20f0e2 869 stream->relayd_stream_id, stream->next_net_seq_num - 1);
9276e5c8
JR
870 if (ret < 0) {
871 /*
872 * Communication error with lttng-relayd,
873 * perform cleanup now
874 */
875 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
876 lttng_consumer_cleanup_relayd(relayd);
877 ret = -1;
878 }
23c910e5
JR
879 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
880 } else {
881 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
882 stream->key, stream->net_seq_idx);
883 ret = -1;
884 }
1c20f0e2 885 } else {
f8f3885c 886 if (lttng_index_file_write(stream->index_file, element)) {
6cd525e8
MD
887 ret = -1;
888 } else {
889 ret = 0;
890 }
1c20f0e2
JD
891 }
892 if (ret < 0) {
893 goto error;
894 }
895
896error:
897 rcu_read_unlock();
898 return ret;
899}
94d49140 900
d2956687
JG
901int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
902 bool create_index)
903{
904 int ret;
905 enum lttng_trace_chunk_status chunk_status;
906 const int flags = O_WRONLY | O_CREAT | O_TRUNC;
907 const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
908 char stream_path[LTTNG_PATH_MAX];
909
910 ASSERT_LOCKED(stream->lock);
911 assert(stream->trace_chunk);
912
913 ret = utils_stream_file_path(stream->chan->pathname, stream->name,
914 stream->chan->tracefile_size,
3b16476a 915 stream->tracefile_count_current, NULL,
d2956687
JG
916 stream_path, sizeof(stream_path));
917 if (ret < 0) {
918 goto end;
919 }
920
921 if (stream->out_fd >= 0) {
922 ret = close(stream->out_fd);
923 if (ret < 0) {
924 PERROR("Failed to close stream file \"%s\"",
925 stream->name);
926 goto end;
927 }
928 stream->out_fd = -1;
929 }
930
931 DBG("Opening stream output file \"%s\"", stream_path);
932 chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
3ff5c5db 933 flags, mode, &stream->out_fd, false);
d2956687
JG
934 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
935 ERR("Failed to open stream file \"%s\"", stream->name);
936 ret = -1;
937 goto end;
938 }
939
940 if (!stream->metadata_flag && (create_index || stream->index_file)) {
941 if (stream->index_file) {
942 lttng_index_file_put(stream->index_file);
943 }
3ff5c5db 944 chunk_status = lttng_index_file_create_from_trace_chunk(
d2956687
JG
945 stream->trace_chunk,
946 stream->chan->pathname,
947 stream->name,
948 stream->chan->tracefile_size,
949 stream->tracefile_count_current,
950 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
3ff5c5db
MD
951 false, &stream->index_file);
952 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
d2956687
JG
953 ret = -1;
954 goto end;
955 }
956 }
957
958 /* Reset current size because we just perform a rotation. */
959 stream->tracefile_size_current = 0;
960 stream->out_fd_offset = 0;
961end:
962 return ret;
963}
964
965int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
966{
967 int ret;
968
969 stream->tracefile_count_current++;
970 if (stream->chan->tracefile_count > 0) {
971 stream->tracefile_count_current %=
972 stream->chan->tracefile_count;
973 }
974
975 DBG("Rotating output files of stream \"%s\"", stream->name);
976 ret = consumer_stream_create_output_files(stream, true);
977 if (ret) {
978 goto end;
979 }
980
981end:
982 return ret;
983}
cdb72e4e
JG
984
985bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
986{
987 /*
988 * This function does not take a const stream since
989 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
990 */
991 assert(stream);
992 return cds_lfht_is_node_deleted(&stream->node.node);
993}
This page took 0.136366 seconds and 4 git commands to generate.