2 * Copyright (C) 2012 Julien Desfossez <julien.desfossez@efficios.com>
3 * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
5 * SPDX-License-Identifier: GPL-2.0-only
10 #include <common/common.hpp>
11 #include <common/compat/endian.hpp>
12 #include <common/consumer/consumer-stream.hpp>
13 #include <common/consumer/consumer-testpoint.hpp>
14 #include <common/consumer/consumer-timer.hpp>
15 #include <common/kernel-consumer/kernel-consumer.hpp>
16 #include <common/kernel-ctl/kernel-ctl.hpp>
17 #include <common/ust-consumer/ust-consumer.hpp>
19 #include <bin/lttng-consumerd/health-consumerd.hpp>
23 using sample_positions_cb
= int (*)(struct lttng_consumer_stream
*);
24 using get_consumed_cb
= int (*)(struct lttng_consumer_stream
*, unsigned long *);
25 using get_produced_cb
= int (*)(struct lttng_consumer_stream
*, unsigned long *);
26 using flush_index_cb
= int (*)(struct lttng_consumer_stream
*);
28 static struct timer_signal_data timer_signal
= {
32 .lock
= PTHREAD_MUTEX_INITIALIZER
,
36 * Set custom signal mask to current thread.
38 static void setmask(sigset_t
*mask
)
42 ret
= sigemptyset(mask
);
44 PERROR("sigemptyset");
46 ret
= sigaddset(mask
, LTTNG_CONSUMER_SIG_SWITCH
);
48 PERROR("sigaddset switch");
50 ret
= sigaddset(mask
, LTTNG_CONSUMER_SIG_TEARDOWN
);
52 PERROR("sigaddset teardown");
54 ret
= sigaddset(mask
, LTTNG_CONSUMER_SIG_LIVE
);
56 PERROR("sigaddset live");
58 ret
= sigaddset(mask
, LTTNG_CONSUMER_SIG_MONITOR
);
60 PERROR("sigaddset monitor");
62 ret
= sigaddset(mask
, LTTNG_CONSUMER_SIG_EXIT
);
64 PERROR("sigaddset exit");
68 static int the_channel_monitor_pipe
= -1;
71 * Execute action on a timer switch.
73 * Beware: metadata_switch_timer() should *never* take a mutex also held
74 * while consumer_timer_switch_stop() is called. It would result in
77 static void metadata_switch_timer(struct lttng_consumer_local_data
*ctx
, siginfo_t
*si
)
80 struct lttng_consumer_channel
*channel
;
82 channel
= (lttng_consumer_channel
*) si
->si_value
.sival_ptr
;
83 LTTNG_ASSERT(channel
);
85 if (channel
->switch_timer_error
) {
89 DBG("Switch timer for channel %" PRIu64
, channel
->key
);
91 case LTTNG_CONSUMER32_UST
:
92 case LTTNG_CONSUMER64_UST
:
94 * Locks taken by lttng_ustconsumer_request_metadata():
95 * - metadata_socket_lock
96 * - Calling lttng_ustconsumer_recv_metadata():
97 * - channel->metadata_cache->lock
98 * - Calling consumer_metadata_cache_flushed():
99 * - channel->timer_lock
100 * - channel->metadata_cache->lock
102 * Ensure that neither consumer_data.lock nor
103 * channel->lock are taken within this function, since
104 * they are held while consumer_timer_switch_stop() is
107 ret
= lttng_ustconsumer_request_metadata(ctx
, channel
, 1, 1);
109 channel
->switch_timer_error
= 1;
112 case LTTNG_CONSUMER_KERNEL
:
113 case LTTNG_CONSUMER_UNKNOWN
:
119 static int send_empty_index(struct lttng_consumer_stream
*stream
, uint64_t ts
, uint64_t stream_id
)
122 struct ctf_packet_index index
;
124 memset(&index
, 0, sizeof(index
));
125 index
.stream_id
= htobe64(stream_id
);
126 index
.timestamp_end
= htobe64(ts
);
127 ret
= consumer_stream_write_index(stream
, &index
);
136 int consumer_flush_kernel_index(struct lttng_consumer_stream
*stream
)
138 uint64_t ts
, stream_id
;
141 ret
= kernctl_get_current_timestamp(stream
->wait_fd
, &ts
);
143 ERR("Failed to get the current timestamp");
146 ret
= kernctl_buffer_flush(stream
->wait_fd
);
148 ERR("Failed to flush kernel stream");
151 ret
= kernctl_snapshot(stream
->wait_fd
);
153 if (ret
!= -EAGAIN
&& ret
!= -ENODATA
) {
154 PERROR("live timer kernel snapshot");
158 ret
= kernctl_get_stream_id(stream
->wait_fd
, &stream_id
);
160 PERROR("kernctl_get_stream_id");
163 DBG("Stream %" PRIu64
" empty, sending beacon", stream
->key
);
164 ret
= send_empty_index(stream
, ts
, stream_id
);
174 static int check_stream(struct lttng_consumer_stream
*stream
, flush_index_cb flush_index
)
179 * While holding the stream mutex, try to take a snapshot, if it
180 * succeeds, it means that data is ready to be sent, just let the data
181 * thread handle that. Otherwise, if the snapshot returns EAGAIN, it
182 * means that there is no data to read after the flush, so we can
183 * safely send the empty index.
185 * Doing a trylock and checking if waiting on metadata if
186 * trylock fails. Bail out of the stream is indeed waiting for
187 * metadata to be pushed. Busy wait on trylock otherwise.
190 ret
= pthread_mutex_trylock(&stream
->lock
);
193 break; /* We have the lock. */
195 pthread_mutex_lock(&stream
->metadata_timer_lock
);
196 if (stream
->waiting_on_metadata
) {
198 stream
->missed_metadata_flush
= true;
199 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
200 goto end
; /* Bail out. */
202 pthread_mutex_unlock(&stream
->metadata_timer_lock
);
207 ERR("Unexpected pthread_mutex_trylock error %d", ret
);
213 ret
= flush_index(stream
);
214 pthread_mutex_unlock(&stream
->lock
);
219 int consumer_flush_ust_index(struct lttng_consumer_stream
*stream
)
221 uint64_t ts
, stream_id
;
224 ret
= cds_lfht_is_node_deleted(&stream
->node
.node
);
229 ret
= lttng_ustconsumer_get_current_timestamp(stream
, &ts
);
231 ERR("Failed to get the current timestamp");
234 ret
= lttng_ustconsumer_flush_buffer(stream
, 1);
236 ERR("Failed to flush buffer while flushing index");
239 ret
= lttng_ustconsumer_take_snapshot(stream
);
241 if (ret
!= -EAGAIN
) {
242 ERR("Taking UST snapshot");
246 ret
= lttng_ustconsumer_get_stream_id(stream
, &stream_id
);
248 PERROR("lttng_ust_ctl_get_stream_id");
251 DBG("Stream %" PRIu64
" empty, sending beacon", stream
->key
);
252 ret
= send_empty_index(stream
, ts
, stream_id
);
263 * Execute action on a live timer
265 static void live_timer(struct lttng_consumer_local_data
*ctx
, siginfo_t
*si
)
268 struct lttng_consumer_channel
*channel
;
269 struct lttng_consumer_stream
*stream
;
270 struct lttng_ht_iter iter
;
271 const struct lttng_ht
*ht
= the_consumer_data
.stream_per_chan_id_ht
;
272 const flush_index_cb flush_index
= ctx
->type
== LTTNG_CONSUMER_KERNEL
?
273 consumer_flush_kernel_index
:
274 consumer_flush_ust_index
;
276 channel
= (lttng_consumer_channel
*) si
->si_value
.sival_ptr
;
277 LTTNG_ASSERT(channel
);
279 if (channel
->switch_timer_error
) {
283 DBG("Live timer for channel %" PRIu64
, channel
->key
);
286 cds_lfht_for_each_entry_duplicate(ht
->ht
,
287 ht
->hash_fct(&channel
->key
, lttng_ht_seed
),
292 node_channel_id
.node
)
294 ret
= check_stream(stream
, flush_index
);
307 static void consumer_timer_signal_thread_qs(unsigned int signr
)
309 sigset_t pending_set
;
313 * We need to be the only thread interacting with the thread
314 * that manages signals for teardown synchronization.
316 pthread_mutex_lock(&timer_signal
.lock
);
318 /* Ensure we don't have any signal queued for this channel. */
320 ret
= sigemptyset(&pending_set
);
322 PERROR("sigemptyset");
324 ret
= sigpending(&pending_set
);
326 PERROR("sigpending");
328 if (!sigismember(&pending_set
, signr
)) {
335 * From this point, no new signal handler will be fired that would try to
336 * access "chan". However, we still need to wait for any currently
337 * executing handler to complete.
340 CMM_STORE_SHARED(timer_signal
.qs_done
, 0);
344 * Kill with LTTNG_CONSUMER_SIG_TEARDOWN, so signal management thread wakes
347 kill(getpid(), LTTNG_CONSUMER_SIG_TEARDOWN
);
349 while (!CMM_LOAD_SHARED(timer_signal
.qs_done
)) {
354 pthread_mutex_unlock(&timer_signal
.lock
);
358 * Start a timer channel timer which will fire at a given interval
359 * (timer_interval_us)and fire a given signal (signal).
361 * Returns a negative value on error, 0 if a timer was created, and
362 * a positive value if no timer was created (not an error).
364 static int consumer_channel_timer_start(timer_t
*timer_id
,
365 struct lttng_consumer_channel
*channel
,
366 unsigned int timer_interval_us
,
369 int ret
= 0, delete_ret
;
370 struct sigevent sev
= {};
371 struct itimerspec its
;
373 LTTNG_ASSERT(channel
);
374 LTTNG_ASSERT(channel
->key
);
376 if (timer_interval_us
== 0) {
377 /* No creation needed; not an error. */
382 sev
.sigev_notify
= SIGEV_SIGNAL
;
383 sev
.sigev_signo
= signal
;
384 sev
.sigev_value
.sival_ptr
= channel
;
385 ret
= timer_create(CLOCKID
, &sev
, timer_id
);
387 PERROR("timer_create");
391 its
.it_value
.tv_sec
= timer_interval_us
/ 1000000;
392 its
.it_value
.tv_nsec
= (timer_interval_us
% 1000000) * 1000;
393 its
.it_interval
.tv_sec
= its
.it_value
.tv_sec
;
394 its
.it_interval
.tv_nsec
= its
.it_value
.tv_nsec
;
396 ret
= timer_settime(*timer_id
, 0, &its
, NULL
);
398 PERROR("timer_settime");
399 goto error_destroy_timer
;
404 delete_ret
= timer_delete(*timer_id
);
405 if (delete_ret
== -1) {
406 PERROR("timer_delete");
411 static int consumer_channel_timer_stop(timer_t
*timer_id
, int signal
)
415 ret
= timer_delete(*timer_id
);
417 PERROR("timer_delete");
421 consumer_timer_signal_thread_qs(signal
);
428 * Set the channel's switch timer.
430 void consumer_timer_switch_start(struct lttng_consumer_channel
*channel
,
431 unsigned int switch_timer_interval_us
)
435 LTTNG_ASSERT(channel
);
436 LTTNG_ASSERT(channel
->key
);
438 ret
= consumer_channel_timer_start(&channel
->switch_timer
,
440 switch_timer_interval_us
,
441 LTTNG_CONSUMER_SIG_SWITCH
);
443 channel
->switch_timer_enabled
= !!(ret
== 0);
447 * Stop and delete the channel's switch timer.
449 void consumer_timer_switch_stop(struct lttng_consumer_channel
*channel
)
453 LTTNG_ASSERT(channel
);
455 ret
= consumer_channel_timer_stop(&channel
->switch_timer
, LTTNG_CONSUMER_SIG_SWITCH
);
457 ERR("Failed to stop switch timer");
460 channel
->switch_timer_enabled
= 0;
464 * Set the channel's live timer.
466 void consumer_timer_live_start(struct lttng_consumer_channel
*channel
,
467 unsigned int live_timer_interval_us
)
471 LTTNG_ASSERT(channel
);
472 LTTNG_ASSERT(channel
->key
);
474 ret
= consumer_channel_timer_start(
475 &channel
->live_timer
, channel
, live_timer_interval_us
, LTTNG_CONSUMER_SIG_LIVE
);
477 channel
->live_timer_enabled
= !!(ret
== 0);
481 * Stop and delete the channel's live timer.
483 void consumer_timer_live_stop(struct lttng_consumer_channel
*channel
)
487 LTTNG_ASSERT(channel
);
489 ret
= consumer_channel_timer_stop(&channel
->live_timer
, LTTNG_CONSUMER_SIG_LIVE
);
491 ERR("Failed to stop live timer");
494 channel
->live_timer_enabled
= 0;
498 * Set the channel's monitoring timer.
500 * Returns a negative value on error, 0 if a timer was created, and
501 * a positive value if no timer was created (not an error).
503 int consumer_timer_monitor_start(struct lttng_consumer_channel
*channel
,
504 unsigned int monitor_timer_interval_us
)
508 LTTNG_ASSERT(channel
);
509 LTTNG_ASSERT(channel
->key
);
510 LTTNG_ASSERT(!channel
->monitor_timer_enabled
);
512 ret
= consumer_channel_timer_start(&channel
->monitor_timer
,
514 monitor_timer_interval_us
,
515 LTTNG_CONSUMER_SIG_MONITOR
);
516 channel
->monitor_timer_enabled
= !!(ret
== 0);
521 * Stop and delete the channel's monitoring timer.
523 int consumer_timer_monitor_stop(struct lttng_consumer_channel
*channel
)
527 LTTNG_ASSERT(channel
);
528 LTTNG_ASSERT(channel
->monitor_timer_enabled
);
530 ret
= consumer_channel_timer_stop(&channel
->monitor_timer
, LTTNG_CONSUMER_SIG_MONITOR
);
532 ERR("Failed to stop live timer");
536 channel
->monitor_timer_enabled
= 0;
542 * Block the RT signals for the entire process. It must be called from the
543 * consumer main before creating the threads
545 int consumer_signal_init(void)
550 /* Block signal for entire process, so only our thread processes it. */
552 ret
= pthread_sigmask(SIG_BLOCK
, &mask
, NULL
);
555 PERROR("pthread_sigmask");
561 static int sample_channel_positions(struct lttng_consumer_channel
*channel
,
562 uint64_t *_highest_use
,
563 uint64_t *_lowest_use
,
564 uint64_t *_total_consumed
,
565 sample_positions_cb sample
,
566 get_consumed_cb get_consumed
,
567 get_produced_cb get_produced
)
570 struct lttng_ht_iter iter
;
571 struct lttng_consumer_stream
*stream
;
572 bool empty_channel
= true;
573 uint64_t high
= 0, low
= UINT64_MAX
;
574 struct lttng_ht
*ht
= the_consumer_data
.stream_per_chan_id_ht
;
576 *_total_consumed
= 0;
580 cds_lfht_for_each_entry_duplicate(ht
->ht
,
581 ht
->hash_fct(&channel
->key
, lttng_ht_seed
),
586 node_channel_id
.node
)
588 unsigned long produced
, consumed
, usage
;
590 empty_channel
= false;
592 pthread_mutex_lock(&stream
->lock
);
593 if (cds_lfht_is_node_deleted(&stream
->node
.node
)) {
597 ret
= sample(stream
);
599 ERR("Failed to take buffer position snapshot in monitor timer (ret = %d)",
601 pthread_mutex_unlock(&stream
->lock
);
604 ret
= get_consumed(stream
, &consumed
);
606 ERR("Failed to get buffer consumed position in monitor timer");
607 pthread_mutex_unlock(&stream
->lock
);
610 ret
= get_produced(stream
, &produced
);
612 ERR("Failed to get buffer produced position in monitor timer");
613 pthread_mutex_unlock(&stream
->lock
);
617 usage
= produced
- consumed
;
618 high
= (usage
> high
) ? usage
: high
;
619 low
= (usage
< low
) ? usage
: low
;
622 * We don't use consumed here for 2 reasons:
623 * - output_written takes into account the padding written in the
624 * tracefiles when we stop the session;
625 * - the consumed position is not the accurate representation of what
626 * was extracted from a buffer in overwrite mode.
628 *_total_consumed
+= stream
->output_written
;
630 pthread_mutex_unlock(&stream
->lock
);
633 *_highest_use
= high
;
643 /* Sample and send channel buffering statistics to the session daemon. */
644 void sample_and_send_channel_buffer_stats(struct lttng_consumer_channel
*channel
)
647 int channel_monitor_pipe
= consumer_timer_thread_get_channel_monitor_pipe();
648 struct lttcomm_consumer_channel_monitor_msg msg
= {
650 .session_id
= channel
->session_id
,
653 .consumed_since_last_sample
= 0,
655 sample_positions_cb sample
;
656 get_consumed_cb get_consumed
;
657 get_produced_cb get_produced
;
658 uint64_t lowest
= 0, highest
= 0, total_consumed
= 0;
660 LTTNG_ASSERT(channel
);
662 if (channel_monitor_pipe
< 0) {
666 switch (the_consumer_data
.type
) {
667 case LTTNG_CONSUMER_KERNEL
:
668 sample
= lttng_kconsumer_sample_snapshot_positions
;
669 get_consumed
= lttng_kconsumer_get_consumed_snapshot
;
670 get_produced
= lttng_kconsumer_get_produced_snapshot
;
672 case LTTNG_CONSUMER32_UST
:
673 case LTTNG_CONSUMER64_UST
:
674 sample
= lttng_ustconsumer_sample_snapshot_positions
;
675 get_consumed
= lttng_ustconsumer_get_consumed_snapshot
;
676 get_produced
= lttng_ustconsumer_get_produced_snapshot
;
682 ret
= sample_channel_positions(
683 channel
, &highest
, &lowest
, &total_consumed
, sample
, get_consumed
, get_produced
);
688 msg
.highest
= highest
;
690 msg
.consumed_since_last_sample
= total_consumed
- channel
->last_consumed_size_sample_sent
;
693 * Writes performed here are assumed to be atomic which is only
694 * guaranteed for sizes < than PIPE_BUF.
696 LTTNG_ASSERT(sizeof(msg
) <= PIPE_BUF
);
699 ret
= write(channel_monitor_pipe
, &msg
, sizeof(msg
));
700 } while (ret
== -1 && errno
== EINTR
);
702 if (errno
== EAGAIN
) {
703 /* Not an error, the sample is merely dropped. */
704 DBG("Channel monitor pipe is full; dropping sample for channel key = %" PRIu64
,
707 PERROR("write to the channel monitor pipe");
710 DBG("Sent channel monitoring sample for channel key %" PRIu64
711 ", (highest = %" PRIu64
", lowest = %" PRIu64
")",
715 channel
->last_consumed_size_sample_sent
= msg
.consumed_since_last_sample
;
719 int consumer_timer_thread_get_channel_monitor_pipe(void)
721 return uatomic_read(&the_channel_monitor_pipe
);
724 int consumer_timer_thread_set_channel_monitor_pipe(int fd
)
728 ret
= uatomic_cmpxchg(&the_channel_monitor_pipe
, -1, fd
);
739 * This thread is the sighandler for signals LTTNG_CONSUMER_SIG_SWITCH,
740 * LTTNG_CONSUMER_SIG_TEARDOWN, LTTNG_CONSUMER_SIG_LIVE, and
741 * LTTNG_CONSUMER_SIG_MONITOR, LTTNG_CONSUMER_SIG_EXIT.
743 void *consumer_timer_thread(void *data
)
748 struct lttng_consumer_local_data
*ctx
= (lttng_consumer_local_data
*) data
;
750 rcu_register_thread();
752 health_register(health_consumerd
, HEALTH_CONSUMERD_TYPE_METADATA_TIMER
);
754 if (testpoint(consumerd_thread_metadata_timer
)) {
755 goto error_testpoint
;
758 health_code_update();
760 /* Only self thread will receive signal mask. */
762 CMM_STORE_SHARED(timer_signal
.tid
, pthread_self());
765 health_code_update();
768 signr
= sigwaitinfo(&mask
, &info
);
772 * NOTE: cascading conditions are used instead of a switch case
773 * since the use of SIGRTMIN in the definition of the signals'
774 * values prevents the reduction to an integer constant.
777 if (errno
!= EINTR
) {
778 PERROR("sigwaitinfo");
781 } else if (signr
== LTTNG_CONSUMER_SIG_SWITCH
) {
782 metadata_switch_timer(ctx
, &info
);
783 } else if (signr
== LTTNG_CONSUMER_SIG_TEARDOWN
) {
785 CMM_STORE_SHARED(timer_signal
.qs_done
, 1);
787 DBG("Signal timer metadata thread teardown");
788 } else if (signr
== LTTNG_CONSUMER_SIG_LIVE
) {
789 live_timer(ctx
, &info
);
790 } else if (signr
== LTTNG_CONSUMER_SIG_MONITOR
) {
791 struct lttng_consumer_channel
*channel
;
793 channel
= (lttng_consumer_channel
*) info
.si_value
.sival_ptr
;
794 sample_and_send_channel_buffer_stats(channel
);
795 } else if (signr
== LTTNG_CONSUMER_SIG_EXIT
) {
796 LTTNG_ASSERT(CMM_LOAD_SHARED(consumer_quit
));
799 ERR("Unexpected signal %d\n", info
.si_signo
);
804 /* Only reached in testpoint error */
807 health_unregister(health_consumerd
);
808 rcu_unregister_thread();