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