consumerd: move rotation logic to domain-agnostic read path
[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>
51230d70
DG
22
23#include "consumer-stream.h"
24
25/*
26 * RCU call to free stream. MUST only be used with call_rcu().
27 */
28static void free_stream_rcu(struct rcu_head *head)
29{
30 struct lttng_ht_node_u64 *node =
31 caa_container_of(head, struct lttng_ht_node_u64, head);
32 struct lttng_consumer_stream *stream =
33 caa_container_of(node, struct lttng_consumer_stream, node);
34
35 pthread_mutex_destroy(&stream->lock);
36 free(stream);
37}
38
39/*
40 * Close stream on the relayd side. This call can destroy a relayd if the
41 * conditions are met.
42 *
43 * A RCU read side lock MUST be acquired if the relayd object was looked up in
44 * a hash table before calling this.
45 */
46void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
47 struct consumer_relayd_sock_pair *relayd)
48{
49 int ret;
50
51 assert(stream);
52 assert(relayd);
53
d01178b6
DG
54 if (stream->sent_to_relayd) {
55 uatomic_dec(&relayd->refcount);
56 assert(uatomic_read(&relayd->refcount) >= 0);
57 }
51230d70
DG
58
59 /* Closing streams requires to lock the control socket. */
60 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
61 ret = relayd_send_close_stream(&relayd->control_sock,
62 stream->relayd_stream_id,
63 stream->next_net_seq_num - 1);
64 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
65 if (ret < 0) {
9276e5c8
JR
66 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
67 lttng_consumer_cleanup_relayd(relayd);
51230d70
DG
68 }
69
70 /* Both conditions are met, we destroy the relayd. */
71 if (uatomic_read(&relayd->refcount) == 0 &&
72 uatomic_read(&relayd->destroy_flag)) {
73 consumer_destroy_relayd(relayd);
74 }
10a50311 75 stream->net_seq_idx = (uint64_t) -1ULL;
d01178b6 76 stream->sent_to_relayd = 0;
51230d70
DG
77}
78
79/*
80 * Close stream's file descriptors and, if needed, close stream also on the
81 * relayd side.
82 *
83 * The consumer data lock MUST be acquired.
84 * The stream lock MUST be acquired.
85 */
86void consumer_stream_close(struct lttng_consumer_stream *stream)
87{
88 int ret;
89 struct consumer_relayd_sock_pair *relayd;
90
91 assert(stream);
92
93 switch (consumer_data.type) {
94 case LTTNG_CONSUMER_KERNEL:
95 if (stream->mmap_base != NULL) {
96 ret = munmap(stream->mmap_base, stream->mmap_len);
97 if (ret != 0) {
98 PERROR("munmap");
99 }
100 }
101
102 if (stream->wait_fd >= 0) {
103 ret = close(stream->wait_fd);
104 if (ret) {
105 PERROR("close");
106 }
10a50311 107 stream->wait_fd = -1;
51230d70 108 }
a2361a61
JD
109 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
110 utils_close_pipe(stream->splice_pipe);
111 }
51230d70
DG
112 break;
113 case LTTNG_CONSUMER32_UST:
114 case LTTNG_CONSUMER64_UST:
6d574024
DG
115 {
116 /*
117 * Special case for the metadata since the wait fd is an internal pipe
118 * polled in the metadata thread.
119 */
120 if (stream->metadata_flag && stream->chan->monitor) {
121 int rpipe = stream->ust_metadata_poll_pipe[0];
122
123 /*
124 * This will stop the channel timer if one and close the write side
125 * of the metadata poll pipe.
126 */
127 lttng_ustconsumer_close_metadata(stream->chan);
128 if (rpipe >= 0) {
129 ret = close(rpipe);
130 if (ret < 0) {
b4a650f3 131 PERROR("closing metadata pipe read side");
6d574024
DG
132 }
133 stream->ust_metadata_poll_pipe[0] = -1;
134 }
135 }
51230d70 136 break;
6d574024 137 }
51230d70
DG
138 default:
139 ERR("Unknown consumer_data type");
140 assert(0);
141 }
142
143 /* Close output fd. Could be a socket or local file at this point. */
144 if (stream->out_fd >= 0) {
145 ret = close(stream->out_fd);
146 if (ret) {
147 PERROR("close");
148 }
10a50311 149 stream->out_fd = -1;
51230d70
DG
150 }
151
f8f3885c
MD
152 if (stream->index_file) {
153 lttng_index_file_put(stream->index_file);
154 stream->index_file = NULL;
309167d2
JD
155 }
156
d2956687
JG
157 lttng_trace_chunk_put(stream->trace_chunk);
158 stream->trace_chunk = NULL;
159
51230d70
DG
160 /* Check and cleanup relayd if needed. */
161 rcu_read_lock();
162 relayd = consumer_find_relayd(stream->net_seq_idx);
163 if (relayd != NULL) {
164 consumer_stream_relayd_close(stream, relayd);
165 }
166 rcu_read_unlock();
167}
168
169/*
170 * Delete the stream from all possible hash tables.
171 *
172 * The consumer data lock MUST be acquired.
173 * The stream lock MUST be acquired.
174 */
175void consumer_stream_delete(struct lttng_consumer_stream *stream,
176 struct lttng_ht *ht)
177{
178 int ret;
179 struct lttng_ht_iter iter;
180
181 assert(stream);
10a50311
JD
182 /* Should NEVER be called not in monitor mode. */
183 assert(stream->chan->monitor);
51230d70
DG
184
185 rcu_read_lock();
186
187 if (ht) {
188 iter.iter.node = &stream->node.node;
189 ret = lttng_ht_del(ht, &iter);
190 assert(!ret);
191 }
192
193 /* Delete from stream per channel ID hash table. */
194 iter.iter.node = &stream->node_channel_id.node;
195 /*
196 * The returned value is of no importance. Even if the node is NOT in the
197 * hash table, we continue since we may have been called by a code path
198 * that did not add the stream to a (all) hash table. Same goes for the
199 * next call ht del call.
200 */
201 (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
202
203 /* Delete from the global stream list. */
204 iter.iter.node = &stream->node_session_id.node;
205 /* See the previous ht del on why we ignore the returned value. */
206 (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
207
208 rcu_read_unlock();
209
6d574024
DG
210 if (!stream->metadata_flag) {
211 /* Decrement the stream count of the global consumer data. */
212 assert(consumer_data.stream_count > 0);
213 consumer_data.stream_count--;
214 }
51230d70
DG
215}
216
217/*
218 * Free the given stream within a RCU call.
219 */
220void consumer_stream_free(struct lttng_consumer_stream *stream)
221{
222 assert(stream);
223
224 call_rcu(&stream->node.head, free_stream_rcu);
225}
226
227/*
10a50311 228 * Destroy the stream's buffers of the tracer.
51230d70 229 */
10a50311 230void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
51230d70 231{
10a50311
JD
232 assert(stream);
233
234 switch (consumer_data.type) {
235 case LTTNG_CONSUMER_KERNEL:
236 break;
237 case LTTNG_CONSUMER32_UST:
238 case LTTNG_CONSUMER64_UST:
239 lttng_ustconsumer_del_stream(stream);
240 break;
241 default:
242 ERR("Unknown consumer_data type");
243 assert(0);
244 }
245}
51230d70 246
10a50311 247/*
4891ece8 248 * Destroy and close a already created stream.
10a50311 249 */
4891ece8 250static void destroy_close_stream(struct lttng_consumer_stream *stream)
10a50311 251{
51230d70
DG
252 assert(stream);
253
4891ece8 254 DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
10a50311
JD
255
256 /* Destroy tracer buffers of the stream. */
257 consumer_stream_destroy_buffers(stream);
258 /* Close down everything including the relayd if one. */
259 consumer_stream_close(stream);
260}
51230d70 261
10a50311 262/*
4891ece8
DG
263 * Decrement the stream's channel refcount and if down to 0, return the channel
264 * pointer so it can be destroyed by the caller or NULL if not.
10a50311 265 */
4891ece8
DG
266static struct lttng_consumer_channel *unref_channel(
267 struct lttng_consumer_stream *stream)
10a50311 268{
4891ece8
DG
269 struct lttng_consumer_channel *free_chan = NULL;
270
10a50311 271 assert(stream);
4891ece8 272 assert(stream->chan);
10a50311 273
4891ece8
DG
274 /* Update refcount of channel and see if we need to destroy it. */
275 if (!uatomic_sub_return(&stream->chan->refcount, 1)
276 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
277 free_chan = stream->chan;
278 }
51230d70 279
4891ece8 280 return free_chan;
10a50311 281}
51230d70 282
10a50311
JD
283/*
284 * Destroy a stream completely. This will delete, close and free the stream.
285 * Once return, the stream is NO longer usable. Its channel may get destroyed
286 * if conditions are met for a monitored stream.
287 *
288 * This MUST be called WITHOUT the consumer data and stream lock acquired if
289 * the stream is in _monitor_ mode else it does not matter.
290 */
291void consumer_stream_destroy(struct lttng_consumer_stream *stream,
292 struct lttng_ht *ht)
293{
294 assert(stream);
295
296 /* Stream is in monitor mode. */
4891ece8 297 if (stream->monitor) {
10a50311 298 struct lttng_consumer_channel *free_chan = NULL;
51230d70 299
4891ece8
DG
300 /*
301 * This means that the stream was successfully removed from the streams
302 * list of the channel and sent to the right thread managing this
303 * stream thus being globally visible.
304 */
305 if (stream->globally_visible) {
306 pthread_mutex_lock(&consumer_data.lock);
a9838785 307 pthread_mutex_lock(&stream->chan->lock);
4891ece8
DG
308 pthread_mutex_lock(&stream->lock);
309 /* Remove every reference of the stream in the consumer. */
310 consumer_stream_delete(stream, ht);
311
312 destroy_close_stream(stream);
313
314 /* Update channel's refcount of the stream. */
315 free_chan = unref_channel(stream);
316
317 /* Indicates that the consumer data state MUST be updated after this. */
318 consumer_data.need_update = 1;
319
320 pthread_mutex_unlock(&stream->lock);
a9838785 321 pthread_mutex_unlock(&stream->chan->lock);
4891ece8
DG
322 pthread_mutex_unlock(&consumer_data.lock);
323 } else {
324 /*
325 * If the stream is not visible globally, this needs to be done
326 * outside of the consumer data lock section.
327 */
328 free_chan = unref_channel(stream);
10a50311
JD
329 }
330
10a50311
JD
331 if (free_chan) {
332 consumer_del_channel(free_chan);
333 }
334 } else {
4891ece8 335 destroy_close_stream(stream);
51230d70
DG
336 }
337
338 /* Free stream within a RCU call. */
d2956687
JG
339 lttng_trace_chunk_put(stream->trace_chunk);
340 stream->trace_chunk = NULL;
51230d70
DG
341 consumer_stream_free(stream);
342}
1c20f0e2
JD
343
344/*
345 * Write index of a specific stream either on the relayd or local disk.
346 *
347 * Return 0 on success or else a negative value.
348 */
349int consumer_stream_write_index(struct lttng_consumer_stream *stream,
f8f3885c 350 struct ctf_packet_index *element)
1c20f0e2
JD
351{
352 int ret;
1c20f0e2
JD
353
354 assert(stream);
f8f3885c 355 assert(element);
1c20f0e2
JD
356
357 rcu_read_lock();
23c910e5
JR
358 if (stream->net_seq_idx != (uint64_t) -1ULL) {
359 struct consumer_relayd_sock_pair *relayd;
360 relayd = consumer_find_relayd(stream->net_seq_idx);
361 if (relayd) {
362 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
363 ret = relayd_send_index(&relayd->control_sock, element,
1c20f0e2 364 stream->relayd_stream_id, stream->next_net_seq_num - 1);
9276e5c8
JR
365 if (ret < 0) {
366 /*
367 * Communication error with lttng-relayd,
368 * perform cleanup now
369 */
370 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
371 lttng_consumer_cleanup_relayd(relayd);
372 ret = -1;
373 }
23c910e5
JR
374 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
375 } else {
376 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
377 stream->key, stream->net_seq_idx);
378 ret = -1;
379 }
1c20f0e2 380 } else {
f8f3885c 381 if (lttng_index_file_write(stream->index_file, element)) {
6cd525e8
MD
382 ret = -1;
383 } else {
384 ret = 0;
385 }
1c20f0e2
JD
386 }
387 if (ret < 0) {
388 goto error;
389 }
390
391error:
392 rcu_read_unlock();
393 return ret;
394}
94d49140
JD
395
396/*
e5ca40ee 397 * Actually do the metadata sync using the given metadata stream.
94d49140 398 *
e5ca40ee
DG
399 * Return 0 on success else a negative value. ENODATA can be returned also
400 * indicating that there is no metadata available for that stream.
94d49140 401 */
e5ca40ee
DG
402static int do_sync_metadata(struct lttng_consumer_stream *metadata,
403 struct lttng_consumer_local_data *ctx)
94d49140
JD
404{
405 int ret;
94d49140 406
e5ca40ee
DG
407 assert(metadata);
408 assert(metadata->metadata_flag);
94d49140
JD
409 assert(ctx);
410
94d49140
JD
411 /*
412 * In UST, since we have to write the metadata from the cache packet
413 * by packet, we might need to start this procedure multiple times
414 * until all the metadata from the cache has been extracted.
415 */
416 do {
417 /*
418 * Steps :
419 * - Lock the metadata stream
420 * - Check if metadata stream node was deleted before locking.
421 * - if yes, release and return success
422 * - Check if new metadata is ready (flush + snapshot pos)
423 * - If nothing : release and return.
424 * - Lock the metadata_rdv_lock
425 * - Unlock the metadata stream
426 * - cond_wait on metadata_rdv to wait the wakeup from the
427 * metadata thread
428 * - Unlock the metadata_rdv_lock
429 */
430 pthread_mutex_lock(&metadata->lock);
431
432 /*
433 * There is a possibility that we were able to acquire a reference on the
434 * stream from the RCU hash table but between then and now, the node might
435 * have been deleted just before the lock is acquired. Thus, after locking,
436 * we make sure the metadata node has not been deleted which means that the
437 * buffers are closed.
438 *
439 * In that case, there is no need to sync the metadata hence returning a
440 * success return code.
441 */
442 ret = cds_lfht_is_node_deleted(&metadata->node.node);
443 if (ret) {
444 ret = 0;
445 goto end_unlock_mutex;
446 }
447
448 switch (ctx->type) {
449 case LTTNG_CONSUMER_KERNEL:
450 /*
451 * Empty the metadata cache and flush the current stream.
452 */
453 ret = lttng_kconsumer_sync_metadata(metadata);
454 break;
455 case LTTNG_CONSUMER32_UST:
456 case LTTNG_CONSUMER64_UST:
457 /*
458 * Ask the sessiond if we have new metadata waiting and update the
459 * consumer metadata cache.
460 */
461 ret = lttng_ustconsumer_sync_metadata(ctx, metadata);
462 break;
463 default:
464 assert(0);
465 ret = -1;
466 break;
467 }
468 /*
469 * Error or no new metadata, we exit here.
470 */
471 if (ret <= 0 || ret == ENODATA) {
472 goto end_unlock_mutex;
473 }
474
475 /*
476 * At this point, new metadata have been flushed, so we wait on the
477 * rendez-vous point for the metadata thread to wake us up when it
478 * finishes consuming the metadata and continue execution.
479 */
480
481 pthread_mutex_lock(&metadata->metadata_rdv_lock);
482
483 /*
484 * Release metadata stream lock so the metadata thread can process it.
485 */
486 pthread_mutex_unlock(&metadata->lock);
487
488 /*
489 * Wait on the rendez-vous point. Once woken up, it means the metadata was
490 * consumed and thus synchronization is achieved.
491 */
492 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
493 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
494 } while (ret == EAGAIN);
495
e5ca40ee
DG
496 /* Success */
497 return 0;
94d49140
JD
498
499end_unlock_mutex:
500 pthread_mutex_unlock(&metadata->lock);
e5ca40ee
DG
501 return ret;
502}
503
504/*
505 * Synchronize the metadata using a given session ID. A successful acquisition
506 * of a metadata stream will trigger a request to the session daemon and a
507 * snapshot so the metadata thread can consume it.
508 *
509 * This function call is a rendez-vous point between the metadata thread and
510 * the data thread.
511 *
512 * Return 0 on success or else a negative value.
513 */
514int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
515 uint64_t session_id)
516{
517 int ret;
518 struct lttng_consumer_stream *stream = NULL;
519 struct lttng_ht_iter iter;
520 struct lttng_ht *ht;
521
522 assert(ctx);
523
524 /* Ease our life a bit. */
525 ht = consumer_data.stream_list_ht;
526
527 rcu_read_lock();
528
529 /* Search the metadata associated with the session id of the given stream. */
530
531 cds_lfht_for_each_entry_duplicate(ht->ht,
532 ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
533 &session_id, &iter.iter, stream, node_session_id.node) {
534 if (!stream->metadata_flag) {
535 continue;
536 }
537
538 ret = do_sync_metadata(stream, ctx);
539 if (ret < 0) {
540 goto end;
541 }
542 }
543
544 /*
545 * Force return code to 0 (success) since ret might be ENODATA for instance
546 * which is not an error but rather that we should come back.
547 */
548 ret = 0;
549
550end:
94d49140
JD
551 rcu_read_unlock();
552 return ret;
553}
d2956687
JG
554
555int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
556 bool create_index)
557{
558 int ret;
559 enum lttng_trace_chunk_status chunk_status;
560 const int flags = O_WRONLY | O_CREAT | O_TRUNC;
561 const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
562 char stream_path[LTTNG_PATH_MAX];
563
564 ASSERT_LOCKED(stream->lock);
565 assert(stream->trace_chunk);
566
567 ret = utils_stream_file_path(stream->chan->pathname, stream->name,
568 stream->chan->tracefile_size,
3b16476a 569 stream->tracefile_count_current, NULL,
d2956687
JG
570 stream_path, sizeof(stream_path));
571 if (ret < 0) {
572 goto end;
573 }
574
575 if (stream->out_fd >= 0) {
576 ret = close(stream->out_fd);
577 if (ret < 0) {
578 PERROR("Failed to close stream file \"%s\"",
579 stream->name);
580 goto end;
581 }
582 stream->out_fd = -1;
583 }
584
585 DBG("Opening stream output file \"%s\"", stream_path);
586 chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
3ff5c5db 587 flags, mode, &stream->out_fd, false);
d2956687
JG
588 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
589 ERR("Failed to open stream file \"%s\"", stream->name);
590 ret = -1;
591 goto end;
592 }
593
594 if (!stream->metadata_flag && (create_index || stream->index_file)) {
595 if (stream->index_file) {
596 lttng_index_file_put(stream->index_file);
597 }
3ff5c5db 598 chunk_status = lttng_index_file_create_from_trace_chunk(
d2956687
JG
599 stream->trace_chunk,
600 stream->chan->pathname,
601 stream->name,
602 stream->chan->tracefile_size,
603 stream->tracefile_count_current,
604 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
3ff5c5db
MD
605 false, &stream->index_file);
606 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
d2956687
JG
607 ret = -1;
608 goto end;
609 }
610 }
611
612 /* Reset current size because we just perform a rotation. */
613 stream->tracefile_size_current = 0;
614 stream->out_fd_offset = 0;
615end:
616 return ret;
617}
618
619int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
620{
621 int ret;
622
623 stream->tracefile_count_current++;
624 if (stream->chan->tracefile_count > 0) {
625 stream->tracefile_count_current %=
626 stream->chan->tracefile_count;
627 }
628
629 DBG("Rotating output files of stream \"%s\"", stream->name);
630 ret = consumer_stream_create_output_files(stream, true);
631 if (ret) {
632 goto end;
633 }
634
635end:
636 return ret;
637}
cdb72e4e
JG
638
639bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
640{
641 /*
642 * This function does not take a const stream since
643 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
644 */
645 assert(stream);
646 return cds_lfht_is_node_deleted(&stream->node.node);
647}
This page took 0.077314 seconds and 4 git commands to generate.