2 * Copyright (C) 2017 Julien Desfossez <jdesfossez@efficios.com>
3 * Copyright (C) 2018 Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 * SPDX-License-Identifier: GPL-2.0-only
11 #include "health-sessiond.hpp"
12 #include "lttng-sessiond.hpp"
13 #include "notification-thread-commands.hpp"
15 #include "rotation-thread.hpp"
16 #include "session.hpp"
21 #include <common/align.hpp>
22 #include <common/config/session-config.hpp>
23 #include <common/defaults.hpp>
24 #include <common/error.hpp>
25 #include <common/futex.hpp>
26 #include <common/hashtable/utils.hpp>
27 #include <common/kernel-ctl/kernel-ctl.hpp>
28 #include <common/time.hpp>
29 #include <common/urcu.hpp>
30 #include <common/utils.hpp>
32 #include <lttng/condition/condition-internal.hpp>
33 #include <lttng/location-internal.hpp>
34 #include <lttng/notification/channel-internal.hpp>
35 #include <lttng/notification/notification-internal.hpp>
36 #include <lttng/rotate-internal.hpp>
37 #include <lttng/trigger/trigger.h>
41 #include <sys/eventfd.h>
45 #include <urcu/list.h>
47 struct lttng_notification_channel
*rotate_notification_channel
= nullptr;
49 * This eventfd is used to wake-up the rotation thread whenever a command
50 * completes on the notification channel. This ensures that any notification
51 * that was queued while waiting for a reply to the command is eventually
54 int rotate_notification_channel_subscription_change_eventfd
= -1;
56 struct rotation_thread
{
57 struct lttng_poll_event events
;
61 * The timer thread enqueues jobs and wakes up the rotation thread.
62 * When the rotation thread wakes up, it empties the queue.
64 struct rotation_thread_timer_queue
{
65 struct lttng_pipe
*event_pipe
;
66 struct cds_list_head list
;
70 struct rotation_thread_handle
{
71 struct rotation_thread_timer_queue
*rotation_timer_queue
;
72 /* Access to the notification thread cmd_queue */
73 struct notification_thread_handle
*notification_thread_handle
;
74 /* Thread-specific quit pipe. */
75 struct lttng_pipe
*quit_pipe
;
79 struct rotation_thread_job
{
80 enum rotation_thread_job_type type
;
81 struct ltt_session
*session
;
82 /* List member in struct rotation_thread_timer_queue. */
83 struct cds_list_head head
;
87 static const char *get_job_type_str(enum rotation_thread_job_type job_type
)
90 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
91 return "CHECK_PENDING_ROTATION";
92 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
93 return "SCHEDULED_ROTATION";
99 struct rotation_thread_timer_queue
*rotation_thread_timer_queue_create()
101 struct rotation_thread_timer_queue
*queue
= nullptr;
103 queue
= zmalloc
<rotation_thread_timer_queue
>();
105 PERROR("Failed to allocate timer rotate queue");
109 queue
->event_pipe
= lttng_pipe_open(FD_CLOEXEC
| O_NONBLOCK
);
110 CDS_INIT_LIST_HEAD(&queue
->list
);
111 pthread_mutex_init(&queue
->lock
, nullptr);
116 void rotation_thread_timer_queue_destroy(struct rotation_thread_timer_queue
*queue
)
122 lttng_pipe_destroy(queue
->event_pipe
);
124 pthread_mutex_lock(&queue
->lock
);
125 LTTNG_ASSERT(cds_list_empty(&queue
->list
));
126 pthread_mutex_unlock(&queue
->lock
);
127 pthread_mutex_destroy(&queue
->lock
);
132 * Destroy the thread data previously created by the init function.
134 void rotation_thread_handle_destroy(struct rotation_thread_handle
*handle
)
136 lttng_pipe_destroy(handle
->quit_pipe
);
140 struct rotation_thread_handle
*
141 rotation_thread_handle_create(struct rotation_thread_timer_queue
*rotation_timer_queue
,
142 struct notification_thread_handle
*notification_thread_handle
)
144 struct rotation_thread_handle
*handle
;
146 handle
= zmalloc
<rotation_thread_handle
>();
151 handle
->rotation_timer_queue
= rotation_timer_queue
;
152 handle
->notification_thread_handle
= notification_thread_handle
;
153 handle
->quit_pipe
= lttng_pipe_open(FD_CLOEXEC
);
154 if (!handle
->quit_pipe
) {
161 rotation_thread_handle_destroy(handle
);
166 * Called with the rotation_thread_timer_queue lock held.
167 * Return true if the same timer job already exists in the queue, false if not.
169 static bool timer_job_exists(const struct rotation_thread_timer_queue
*queue
,
170 enum rotation_thread_job_type job_type
,
171 struct ltt_session
*session
)
174 struct rotation_thread_job
*job
;
176 cds_list_for_each_entry (job
, &queue
->list
, head
) {
177 if (job
->session
== session
&& job
->type
== job_type
) {
186 void rotation_thread_enqueue_job(struct rotation_thread_timer_queue
*queue
,
187 enum rotation_thread_job_type job_type
,
188 struct ltt_session
*session
)
191 const char dummy
= '!';
192 struct rotation_thread_job
*job
= nullptr;
193 const char *job_type_str
= get_job_type_str(job_type
);
195 pthread_mutex_lock(&queue
->lock
);
196 if (timer_job_exists(queue
, job_type
, session
)) {
198 * This timer job is already pending, we don't need to add
204 job
= zmalloc
<rotation_thread_job
>();
206 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
211 /* No reason for this to fail as the caller must hold a reference. */
212 (void) session_get(session
);
214 job
->session
= session
;
215 job
->type
= job_type
;
216 cds_list_add_tail(&job
->head
, &queue
->list
);
218 ret
= lttng_write(lttng_pipe_get_writefd(queue
->event_pipe
), &dummy
, sizeof(dummy
));
221 * We do not want to block in the timer handler, the job has
222 * been enqueued in the list, the wakeup pipe is probably full,
223 * the job will be processed when the rotation_thread catches
227 DIAGNOSTIC_IGNORE_LOGICAL_OP
228 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
231 * Not an error, but would be surprising and indicate
232 * that the rotation thread can't keep up with the
235 DBG("Wake-up pipe of rotation thread job queue is full");
238 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
245 pthread_mutex_unlock(&queue
->lock
);
248 static int init_poll_set(struct lttng_poll_event
*poll_set
, struct rotation_thread_handle
*handle
)
253 * Create pollset with size 3:
254 * - rotation thread quit pipe,
255 * - rotation thread timer queue pipe,
256 * - notification channel sock,
258 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
263 ret
= lttng_poll_add(poll_set
, lttng_pipe_get_readfd(handle
->quit_pipe
), LPOLLIN
);
265 ERR("Failed to add quit pipe read fd to poll set");
269 ret
= lttng_poll_add(
270 poll_set
, lttng_pipe_get_readfd(handle
->rotation_timer_queue
->event_pipe
), LPOLLIN
);
272 ERR("Failed to add rotate_pending fd to poll set");
278 lttng_poll_clean(poll_set
);
282 static void fini_thread_state(struct rotation_thread
*state
)
284 lttng_poll_clean(&state
->events
);
285 if (rotate_notification_channel
) {
286 lttng_notification_channel_destroy(rotate_notification_channel
);
289 if (rotate_notification_channel_subscription_change_eventfd
>= 0) {
290 const int close_ret
= close(rotate_notification_channel_subscription_change_eventfd
);
293 PERROR("Failed to close rotation thread notification channel subscription change eventfd");
298 static int init_thread_state(struct rotation_thread_handle
*handle
, struct rotation_thread
*state
)
302 memset(state
, 0, sizeof(*state
));
303 lttng_poll_init(&state
->events
);
305 ret
= init_poll_set(&state
->events
, handle
);
307 ERR("Failed to initialize rotation thread poll set");
311 rotate_notification_channel
=
312 lttng_notification_channel_create(lttng_session_daemon_notification_endpoint
);
313 if (!rotate_notification_channel
) {
314 ERR("Could not create notification channel");
318 ret
= lttng_poll_add(&state
->events
, rotate_notification_channel
->socket
, LPOLLIN
);
320 ERR("Failed to add notification fd to pollset");
324 rotate_notification_channel_subscription_change_eventfd
=
325 eventfd(0, EFD_CLOEXEC
| EFD_SEMAPHORE
);
326 if (rotate_notification_channel_subscription_change_eventfd
< 0) {
327 PERROR("Failed to create rotation thread notification channel subscription change eventfd");
331 ret
= lttng_poll_add(
332 &state
->events
, rotate_notification_channel_subscription_change_eventfd
, LPOLLIN
);
334 ERR("Failed to add rotation thread notification channel subscription change eventfd to pollset");
342 static void check_session_rotation_pending_on_consumers(struct ltt_session
*session
,
343 bool *_rotation_completed
)
346 struct consumer_socket
*socket
;
347 struct cds_lfht_iter iter
;
348 enum consumer_trace_chunk_exists_status exists_status
;
350 bool chunk_exists_on_peer
= false;
351 enum lttng_trace_chunk_status chunk_status
;
352 lttng::urcu::read_lock_guard read_lock
;
354 LTTNG_ASSERT(session
->chunk_being_archived
);
357 * Check for a local pending rotation on all consumers (32-bit
358 * user space, 64-bit user space, and kernel).
360 if (!session
->ust_session
) {
364 cds_lfht_for_each_entry (
365 session
->ust_session
->consumer
->socks
->ht
, &iter
, socket
, node
.node
) {
366 relayd_id
= session
->ust_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
368 session
->ust_session
->consumer
->net_seq_index
;
370 pthread_mutex_lock(socket
->lock
);
371 ret
= consumer_trace_chunk_exists(socket
,
374 session
->chunk_being_archived
,
377 pthread_mutex_unlock(socket
->lock
);
378 ERR("Error occurred while checking rotation status on consumer daemon");
382 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
383 pthread_mutex_unlock(socket
->lock
);
384 chunk_exists_on_peer
= true;
387 pthread_mutex_unlock(socket
->lock
);
391 if (!session
->kernel_session
) {
394 cds_lfht_for_each_entry (
395 session
->kernel_session
->consumer
->socks
->ht
, &iter
, socket
, node
.node
) {
396 pthread_mutex_lock(socket
->lock
);
397 relayd_id
= session
->kernel_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
399 session
->kernel_session
->consumer
->net_seq_index
;
401 ret
= consumer_trace_chunk_exists(socket
,
404 session
->chunk_being_archived
,
407 pthread_mutex_unlock(socket
->lock
);
408 ERR("Error occurred while checking rotation status on consumer daemon");
412 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
413 pthread_mutex_unlock(socket
->lock
);
414 chunk_exists_on_peer
= true;
417 pthread_mutex_unlock(socket
->lock
);
422 if (!chunk_exists_on_peer
) {
423 uint64_t chunk_being_archived_id
;
425 chunk_status
= lttng_trace_chunk_get_id(session
->chunk_being_archived
,
426 &chunk_being_archived_id
);
427 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
428 DBG("Rotation of trace archive %" PRIu64
429 " of session \"%s\" is complete on all consumers",
430 chunk_being_archived_id
,
433 *_rotation_completed
= !chunk_exists_on_peer
;
435 ret
= session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_ERROR
);
437 ERR("Failed to reset rotation state of session \"%s\"", session
->name
);
443 * Check if the last rotation was completed, called with session lock held.
444 * Should only return non-zero in the event of a fatal error. Doing so will
445 * shutdown the thread.
448 check_session_rotation_pending(struct ltt_session
*session
,
449 struct notification_thread_handle
*notification_thread_handle
)
452 struct lttng_trace_archive_location
*location
;
453 enum lttng_trace_chunk_status chunk_status
;
454 bool rotation_completed
= false;
455 const char *archived_chunk_name
;
456 uint64_t chunk_being_archived_id
;
458 if (!session
->chunk_being_archived
) {
464 lttng_trace_chunk_get_id(session
->chunk_being_archived
, &chunk_being_archived_id
);
465 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
467 DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64
,
469 chunk_being_archived_id
);
472 * The rotation-pending check timer of a session is launched in
473 * one-shot mode. If the rotation is incomplete, the rotation
474 * thread will re-enable the pending-check timer.
476 * The timer thread can't stop the timer itself since it is involved
477 * in the check for the timer's quiescence.
479 ret
= timer_session_rotation_pending_check_stop(session
);
481 goto check_ongoing_rotation
;
484 check_session_rotation_pending_on_consumers(session
, &rotation_completed
);
485 if (!rotation_completed
|| session
->rotation_state
== LTTNG_ROTATION_STATE_ERROR
) {
486 goto check_ongoing_rotation
;
490 * Now we can clear the "ONGOING" state in the session. New
491 * rotations can start now.
493 chunk_status
= lttng_trace_chunk_get_name(
494 session
->chunk_being_archived
, &archived_chunk_name
, nullptr);
495 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
496 free(session
->last_archived_chunk_name
);
497 session
->last_archived_chunk_name
= strdup(archived_chunk_name
);
498 if (!session
->last_archived_chunk_name
) {
499 PERROR("Failed to duplicate archived chunk name");
501 session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_COMPLETED
);
503 if (!session
->quiet_rotation
) {
504 location
= session_get_trace_archive_location(session
);
505 ret
= notification_thread_command_session_rotation_completed(
506 notification_thread_handle
,
508 session
->last_archived_chunk_id
.value
,
510 lttng_trace_archive_location_put(location
);
511 if (ret
!= LTTNG_OK
) {
512 ERR("Failed to notify notification thread of completed rotation for session %s",
518 check_ongoing_rotation
:
519 if (session
->rotation_state
== LTTNG_ROTATION_STATE_ONGOING
) {
520 chunk_status
= lttng_trace_chunk_get_id(session
->chunk_being_archived
,
521 &chunk_being_archived_id
);
522 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
524 DBG("Rotation of trace archive %" PRIu64
" is still pending for session %s",
525 chunk_being_archived_id
,
527 ret
= timer_session_rotation_pending_check_start(session
,
528 DEFAULT_ROTATE_PENDING_TIMER
);
530 ERR("Failed to re-enable rotation pending timer");
540 /* Call with the session and session_list locks held. */
541 static int launch_session_rotation(struct ltt_session
*session
)
544 struct lttng_rotate_session_return rotation_return
;
546 DBG("Launching scheduled time-based rotation on session \"%s\"", session
->name
);
548 ret
= cmd_rotate_session(
549 session
, &rotation_return
, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
550 if (ret
== LTTNG_OK
) {
551 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
554 /* Don't consider errors as fatal. */
555 DBG("Scheduled time-based rotation aborted for session %s: %s",
557 lttng_strerror(ret
));
562 static int run_job(struct rotation_thread_job
*job
,
563 struct ltt_session
*session
,
564 struct notification_thread_handle
*notification_thread_handle
)
569 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
570 ret
= launch_session_rotation(session
);
572 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
573 ret
= check_session_rotation_pending(session
, notification_thread_handle
);
581 static int handle_job_queue(struct rotation_thread_handle
*handle
,
582 struct rotation_thread
*state
__attribute__((unused
)),
583 struct rotation_thread_timer_queue
*queue
)
588 struct ltt_session
*session
;
589 struct rotation_thread_job
*job
;
591 /* Take the queue lock only to pop an element from the list. */
592 pthread_mutex_lock(&queue
->lock
);
593 if (cds_list_empty(&queue
->list
)) {
594 pthread_mutex_unlock(&queue
->lock
);
597 job
= cds_list_first_entry(&queue
->list
, typeof(*job
), head
);
598 cds_list_del(&job
->head
);
599 pthread_mutex_unlock(&queue
->lock
);
602 session
= job
->session
;
604 DBG("Session \"%s\" not found", session
->name
!= NULL
? session
->name
: "");
606 * This is a non-fatal error, and we cannot report it to
607 * the user (timer), so just print the error and
608 * continue the processing.
610 * While the timer thread will purge pending signals for
611 * a session on the session's destruction, it is
612 * possible for a job targeting that session to have
613 * already been queued before it was destroyed.
616 session_put(session
);
617 session_unlock_list();
621 session_lock(session
);
622 ret
= run_job(job
, session
, handle
->notification_thread_handle
);
623 session_unlock(session
);
624 /* Release reference held by the job. */
625 session_put(session
);
626 session_unlock_list();
639 static int handle_condition(const struct lttng_notification
*notification
,
640 struct notification_thread_handle
*notification_thread_handle
)
643 const char *condition_session_name
= nullptr;
644 enum lttng_condition_type condition_type
;
645 enum lttng_condition_status condition_status
;
646 enum lttng_evaluation_status evaluation_status
;
648 struct ltt_session
*session
;
649 const struct lttng_condition
*condition
=
650 lttng_notification_get_const_condition(notification
);
651 const struct lttng_evaluation
*evaluation
=
652 lttng_notification_get_const_evaluation(notification
);
654 condition_type
= lttng_condition_get_type(condition
);
656 if (condition_type
!= LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
) {
658 ERR("Condition type and session usage type are not the same");
662 /* Fetch info to test */
663 condition_status
= lttng_condition_session_consumed_size_get_session_name(
664 condition
, &condition_session_name
);
665 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
666 ERR("Session name could not be fetched");
671 lttng_evaluation_session_consumed_size_get_consumed_size(evaluation
, &consumed
);
672 if (evaluation_status
!= LTTNG_EVALUATION_STATUS_OK
) {
673 ERR("Failed to get evaluation");
679 session
= session_find_by_name(condition_session_name
);
681 DBG("Failed to find session while handling notification: notification type = %s, session name = `%s`",
682 lttng_condition_type_str(condition_type
),
683 condition_session_name
);
685 * Not a fatal error: a session can be destroyed before we get
686 * the chance to handle the notification.
689 session_unlock_list();
692 session_lock(session
);
694 if (!lttng_trigger_is_equal(session
->rotate_trigger
,
695 lttng_notification_get_const_trigger(notification
))) {
696 /* Notification does not originate from our rotation trigger. */
701 ret
= unsubscribe_session_consumed_size_rotation(session
, notification_thread_handle
);
706 ret
= cmd_rotate_session(
707 session
, nullptr, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
711 case -LTTNG_ERR_ROTATION_PENDING
:
712 DBG("Rotate already pending, subscribe to the next threshold value");
714 case -LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
715 DBG("Rotation already happened since last stop, subscribe to the next threshold value");
717 case -LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
718 DBG("Rotation already happened since last stop and clear, subscribe to the next threshold value");
721 ERR("Failed to rotate on size notification with error: %s", lttng_strerror(ret
));
726 ret
= subscribe_session_consumed_size_rotation(
727 session
, consumed
+ session
->rotate_size
, notification_thread_handle
);
729 ERR("Failed to subscribe to session consumed size condition");
735 session_unlock(session
);
736 session_put(session
);
737 session_unlock_list();
742 static int handle_notification_channel(int fd
__attribute__((unused
)),
743 struct rotation_thread_handle
*handle
,
744 struct rotation_thread
*state
__attribute__((unused
)))
747 bool notification_pending
= true;
748 struct lttng_notification
*notification
= nullptr;
749 enum lttng_notification_channel_status status
;
752 * A notification channel may have multiple notifications queued-up internally in
753 * its buffers. This is because a notification channel multiplexes command replies
754 * and notifications. The current protocol specifies that multiple notifications can be
755 * received before the reply to a command.
757 * In such cases, the notification channel client implementation internally queues them and
758 * provides them on the next calls to lttng_notification_channel_get_next_notification().
759 * This is correct with respect to the public API, which is intended to be used in "blocking
762 * However, this internal user relies on poll/epoll to wake-up when data is available
763 * on the notification channel's socket. As such, it can't assume that a wake-up means only
764 * one notification is available for consumption since many of them may have been queued in
765 * the channel's internal buffers.
767 while (notification_pending
) {
768 status
= lttng_notification_channel_has_pending_notification(
769 rotate_notification_channel
, ¬ification_pending
);
770 if (status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
771 ERR("Error occurred while checking for pending notification");
776 if (!notification_pending
) {
781 /* Receive the next notification. */
782 status
= lttng_notification_channel_get_next_notification(
783 rotate_notification_channel
, ¬ification
);
785 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
:
787 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
:
788 WARN("Dropped notification detected on notification channel used by the rotation management thread.");
791 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED
:
792 ERR("Notification channel was closed");
796 /* Unhandled conditions / errors. */
797 ERR("Unknown notification channel status");
802 ret
= handle_condition(notification
, handle
->notification_thread_handle
);
803 lttng_notification_destroy(notification
);
812 static void *thread_rotation(void *data
)
815 struct rotation_thread_handle
*handle
= (rotation_thread_handle
*) data
;
816 struct rotation_thread thread
;
819 DBG("Started rotation thread");
820 rcu_register_thread();
822 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
823 health_code_update();
826 ERR("Invalid thread context provided");
830 queue_pipe_fd
= lttng_pipe_get_readfd(handle
->rotation_timer_queue
->event_pipe
);
832 ret
= init_thread_state(handle
, &thread
);
841 DBG("Entering poll wait");
842 ret
= lttng_poll_wait(&thread
.events
, -1);
843 DBG("Poll wait returned (%i)", ret
);
847 * Restart interrupted system call.
849 if (errno
== EINTR
) {
852 ERR("Error encountered during lttng_poll_wait (%i)", ret
);
857 for (i
= 0; i
< fd_count
; i
++) {
858 int fd
= LTTNG_POLL_GETFD(&thread
.events
, i
);
859 uint32_t revents
= LTTNG_POLL_GETEV(&thread
.events
, i
);
861 DBG("Handling fd (%i) activity (%u)", fd
, revents
);
863 if (revents
& LPOLLERR
) {
864 ERR("Polling returned an error on fd %i", fd
);
868 if (fd
== rotate_notification_channel
->socket
||
869 fd
== rotate_notification_channel_subscription_change_eventfd
) {
870 ret
= handle_notification_channel(fd
, handle
, &thread
);
872 ERR("Error occurred while handling activity on notification channel socket");
876 if (fd
== rotate_notification_channel_subscription_change_eventfd
) {
877 uint64_t eventfd_value
;
878 const int read_ret
= lttng_read(fd
, &eventfd_value
, sizeof(eventfd_value
));
880 if (read_ret
!= sizeof(eventfd_value
)) {
881 PERROR("Failed to read value from rotation thread as writing to the rotation thread notification channel subscription change eventfd");
886 /* Job queue or quit pipe activity. */
889 * The job queue is serviced if there is
890 * activity on the quit pipe to ensure it is
891 * flushed and all references held in the queue
894 ret
= handle_job_queue(
895 handle
, &thread
, handle
->rotation_timer_queue
);
897 ERR("Failed to handle rotation timer pipe event");
901 if (fd
== queue_pipe_fd
) {
904 ret
= lttng_read(fd
, &buf
, 1);
906 ERR("Failed to read from wakeup pipe (fd = %i)",
911 DBG("Quit pipe activity");
920 fini_thread_state(&thread
);
922 health_unregister(the_health_sessiond
);
923 rcu_thread_offline();
924 rcu_unregister_thread();
928 static bool shutdown_rotation_thread(void *thread_data
)
930 struct rotation_thread_handle
*handle
= (rotation_thread_handle
*) thread_data
;
931 const int write_fd
= lttng_pipe_get_writefd(handle
->quit_pipe
);
933 return notify_thread_pipe(write_fd
) == 1;
936 bool launch_rotation_thread(struct rotation_thread_handle
*handle
)
938 struct lttng_thread
*thread
;
940 thread
= lttng_thread_create(
941 "Rotation", thread_rotation
, shutdown_rotation_thread
, nullptr, handle
);
945 lttng_thread_put(thread
);