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
10 #include <lttng/trigger/trigger.h>
11 #include <common/error.h>
12 #include <common/config/session-config.h>
13 #include <common/defaults.h>
14 #include <common/utils.h>
15 #include <common/futex.h>
16 #include <common/align.h>
17 #include <common/time.h>
18 #include <common/hashtable/utils.h>
22 #include <sys/eventfd.h>
28 #include <common/kernel-ctl/kernel-ctl.h>
29 #include <lttng/notification/channel-internal.h>
30 #include <lttng/rotate-internal.h>
31 #include <lttng/location-internal.h>
32 #include <lttng/condition/condition-internal.h>
33 #include <lttng/notification/notification-internal.h>
35 #include "rotation-thread.h"
36 #include "lttng-sessiond.h"
37 #include "health-sessiond.h"
42 #include "notification-thread-commands.h"
47 #include <urcu/list.h>
49 struct lttng_notification_channel
*rotate_notification_channel
= NULL
;
52 * This eventfd is used to wake-up the rotation thread whenever a command
53 * completes on the notification channel. This ensures that any notification
54 * that was queued while waiting for a reply to the command is eventually
57 int rotate_notification_channel_subscription_change_eventfd
= -1;
59 struct rotation_thread
{
60 struct lttng_poll_event events
;
63 struct rotation_thread_job
{
64 enum rotation_thread_job_type type
;
65 struct ltt_session
*session
;
66 /* List member in struct rotation_thread_timer_queue. */
67 struct cds_list_head head
;
71 * The timer thread enqueues jobs and wakes up the rotation thread.
72 * When the rotation thread wakes up, it empties the queue.
74 struct rotation_thread_timer_queue
{
75 struct lttng_pipe
*event_pipe
;
76 struct cds_list_head list
;
80 struct rotation_thread_handle
{
81 struct rotation_thread_timer_queue
*rotation_timer_queue
;
82 /* Access to the notification thread cmd_queue */
83 struct notification_thread_handle
*notification_thread_handle
;
84 /* Thread-specific quit pipe. */
85 struct lttng_pipe
*quit_pipe
;
89 const char *get_job_type_str(enum rotation_thread_job_type job_type
)
92 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
93 return "CHECK_PENDING_ROTATION";
94 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
95 return "SCHEDULED_ROTATION";
101 struct rotation_thread_timer_queue
*rotation_thread_timer_queue_create(void)
103 struct rotation_thread_timer_queue
*queue
= NULL
;
105 queue
= zmalloc(sizeof(*queue
));
107 PERROR("Failed to allocate timer rotate queue");
111 queue
->event_pipe
= lttng_pipe_open(FD_CLOEXEC
| O_NONBLOCK
);
112 CDS_INIT_LIST_HEAD(&queue
->list
);
113 pthread_mutex_init(&queue
->lock
, NULL
);
118 void rotation_thread_timer_queue_destroy(
119 struct rotation_thread_timer_queue
*queue
)
125 lttng_pipe_destroy(queue
->event_pipe
);
127 pthread_mutex_lock(&queue
->lock
);
128 assert(cds_list_empty(&queue
->list
));
129 pthread_mutex_unlock(&queue
->lock
);
130 pthread_mutex_destroy(&queue
->lock
);
135 * Destroy the thread data previously created by the init function.
137 void rotation_thread_handle_destroy(
138 struct rotation_thread_handle
*handle
)
140 lttng_pipe_destroy(handle
->quit_pipe
);
144 struct rotation_thread_handle
*rotation_thread_handle_create(
145 struct rotation_thread_timer_queue
*rotation_timer_queue
,
146 struct notification_thread_handle
*notification_thread_handle
)
148 struct rotation_thread_handle
*handle
;
150 handle
= zmalloc(sizeof(*handle
));
155 handle
->rotation_timer_queue
= rotation_timer_queue
;
156 handle
->notification_thread_handle
= notification_thread_handle
;
157 handle
->quit_pipe
= lttng_pipe_open(FD_CLOEXEC
);
158 if (!handle
->quit_pipe
) {
165 rotation_thread_handle_destroy(handle
);
170 * Called with the rotation_thread_timer_queue lock held.
171 * Return true if the same timer job already exists in the queue, false if not.
174 bool timer_job_exists(const struct rotation_thread_timer_queue
*queue
,
175 enum rotation_thread_job_type job_type
,
176 struct ltt_session
*session
)
179 struct rotation_thread_job
*job
;
181 cds_list_for_each_entry(job
, &queue
->list
, head
) {
182 if (job
->session
== session
&& job
->type
== job_type
) {
191 void rotation_thread_enqueue_job(struct rotation_thread_timer_queue
*queue
,
192 enum rotation_thread_job_type job_type
,
193 struct ltt_session
*session
)
196 const char dummy
= '!';
197 struct rotation_thread_job
*job
= NULL
;
198 const char *job_type_str
= get_job_type_str(job_type
);
200 pthread_mutex_lock(&queue
->lock
);
201 if (timer_job_exists(queue
, job_type
, session
)) {
203 * This timer job is already pending, we don't need to add
209 job
= zmalloc(sizeof(struct rotation_thread_job
));
211 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
212 job_type_str
, session
->name
);
215 /* No reason for this to fail as the caller must hold a reference. */
216 (void) session_get(session
);
218 job
->session
= session
;
219 job
->type
= job_type
;
220 cds_list_add_tail(&job
->head
, &queue
->list
);
222 ret
= lttng_write(lttng_pipe_get_writefd(queue
->event_pipe
), &dummy
,
226 * We do not want to block in the timer handler, the job has
227 * been enqueued in the list, the wakeup pipe is probably full,
228 * the job will be processed when the rotation_thread catches
231 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
233 * Not an error, but would be surprising and indicate
234 * that the rotation thread can't keep up with the
237 DBG("Wake-up pipe of rotation thread job queue is full");
240 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
241 job_type_str
, session
->name
);
246 pthread_mutex_unlock(&queue
->lock
);
250 int init_poll_set(struct lttng_poll_event
*poll_set
,
251 struct rotation_thread_handle
*handle
)
256 * Create pollset with size 3:
257 * - rotation thread quit pipe,
258 * - rotation thread timer queue pipe,
259 * - notification channel sock,
261 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
266 ret
= lttng_poll_add(poll_set
,
267 lttng_pipe_get_readfd(handle
->quit_pipe
),
270 ERR("Failed to add quit pipe read fd to poll set");
274 ret
= lttng_poll_add(poll_set
,
275 lttng_pipe_get_readfd(handle
->rotation_timer_queue
->event_pipe
),
278 ERR("Failed to add rotate_pending fd to poll set");
284 lttng_poll_clean(poll_set
);
289 void fini_thread_state(struct rotation_thread
*state
)
291 lttng_poll_clean(&state
->events
);
292 if (rotate_notification_channel
) {
293 lttng_notification_channel_destroy(rotate_notification_channel
);
296 if (rotate_notification_channel_subscription_change_eventfd
>= 0) {
297 const int close_ret
= close(rotate_notification_channel_subscription_change_eventfd
);
300 PERROR("Failed to close rotation thread notification channel subscription change eventfd");
306 int init_thread_state(struct rotation_thread_handle
*handle
,
307 struct rotation_thread
*state
)
311 memset(state
, 0, sizeof(*state
));
312 lttng_poll_init(&state
->events
);
314 ret
= init_poll_set(&state
->events
, handle
);
316 ERR("Failed to initialize rotation thread poll set");
320 rotate_notification_channel
= lttng_notification_channel_create(
321 lttng_session_daemon_notification_endpoint
);
322 if (!rotate_notification_channel
) {
323 ERR("Could not create notification channel");
327 ret
= lttng_poll_add(&state
->events
, rotate_notification_channel
->socket
,
330 ERR("Failed to add notification fd to pollset");
334 rotate_notification_channel_subscription_change_eventfd
=
335 eventfd(0, EFD_CLOEXEC
| EFD_SEMAPHORE
);
336 if (rotate_notification_channel_subscription_change_eventfd
< 0) {
337 PERROR("Failed to create rotation thread notification channel subscription change eventfd");
341 ret
= lttng_poll_add(
342 &state
->events
, rotate_notification_channel_subscription_change_eventfd
, LPOLLIN
);
344 ERR("Failed to add rotation thread notification channel subscription change eventfd to pollset");
353 void check_session_rotation_pending_on_consumers(struct ltt_session
*session
,
354 bool *_rotation_completed
)
357 struct consumer_socket
*socket
;
358 struct cds_lfht_iter iter
;
359 enum consumer_trace_chunk_exists_status exists_status
;
361 bool chunk_exists_on_peer
= false;
362 enum lttng_trace_chunk_status chunk_status
;
364 assert(session
->chunk_being_archived
);
367 * Check for a local pending rotation on all consumers (32-bit
368 * user space, 64-bit user space, and kernel).
371 if (!session
->ust_session
) {
374 cds_lfht_for_each_entry(session
->ust_session
->consumer
->socks
->ht
,
375 &iter
, socket
, node
.node
) {
376 relayd_id
= session
->ust_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
378 session
->ust_session
->consumer
->net_seq_index
;
380 pthread_mutex_lock(socket
->lock
);
381 ret
= consumer_trace_chunk_exists(socket
,
383 session
->id
, session
->chunk_being_archived
,
386 pthread_mutex_unlock(socket
->lock
);
387 ERR("Error occurred while checking rotation status on consumer daemon");
391 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
392 pthread_mutex_unlock(socket
->lock
);
393 chunk_exists_on_peer
= true;
396 pthread_mutex_unlock(socket
->lock
);
400 if (!session
->kernel_session
) {
403 cds_lfht_for_each_entry(session
->kernel_session
->consumer
->socks
->ht
,
404 &iter
, socket
, node
.node
) {
405 pthread_mutex_lock(socket
->lock
);
406 relayd_id
= session
->kernel_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
408 session
->kernel_session
->consumer
->net_seq_index
;
410 ret
= consumer_trace_chunk_exists(socket
,
412 session
->id
, session
->chunk_being_archived
,
415 pthread_mutex_unlock(socket
->lock
);
416 ERR("Error occurred while checking rotation status on consumer daemon");
420 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
421 pthread_mutex_unlock(socket
->lock
);
422 chunk_exists_on_peer
= true;
425 pthread_mutex_unlock(socket
->lock
);
431 if (!chunk_exists_on_peer
) {
432 uint64_t chunk_being_archived_id
;
434 chunk_status
= lttng_trace_chunk_get_id(
435 session
->chunk_being_archived
,
436 &chunk_being_archived_id
);
437 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
438 DBG("Rotation of trace archive %" PRIu64
" of session \"%s\" is complete on all consumers",
439 chunk_being_archived_id
,
442 *_rotation_completed
= !chunk_exists_on_peer
;
444 ret
= session_reset_rotation_state(session
,
445 LTTNG_ROTATION_STATE_ERROR
);
447 ERR("Failed to reset rotation state of session \"%s\"",
454 * Check if the last rotation was completed, called with session lock held.
455 * Should only return non-zero in the event of a fatal error. Doing so will
456 * shutdown the thread.
459 int check_session_rotation_pending(struct ltt_session
*session
,
460 struct notification_thread_handle
*notification_thread_handle
)
463 struct lttng_trace_archive_location
*location
;
464 enum lttng_trace_chunk_status chunk_status
;
465 bool rotation_completed
= false;
466 const char *archived_chunk_name
;
467 uint64_t chunk_being_archived_id
;
469 if (!session
->chunk_being_archived
) {
474 chunk_status
= lttng_trace_chunk_get_id(session
->chunk_being_archived
,
475 &chunk_being_archived_id
);
476 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
478 DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64
,
479 session
->name
, chunk_being_archived_id
);
482 * The rotation-pending check timer of a session is launched in
483 * one-shot mode. If the rotation is incomplete, the rotation
484 * thread will re-enable the pending-check timer.
486 * The timer thread can't stop the timer itself since it is involved
487 * in the check for the timer's quiescence.
489 ret
= timer_session_rotation_pending_check_stop(session
);
491 goto check_ongoing_rotation
;
494 check_session_rotation_pending_on_consumers(session
,
495 &rotation_completed
);
496 if (!rotation_completed
||
497 session
->rotation_state
== LTTNG_ROTATION_STATE_ERROR
) {
498 goto check_ongoing_rotation
;
502 * Now we can clear the "ONGOING" state in the session. New
503 * rotations can start now.
505 chunk_status
= lttng_trace_chunk_get_name(session
->chunk_being_archived
,
506 &archived_chunk_name
, NULL
);
507 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
508 free(session
->last_archived_chunk_name
);
509 session
->last_archived_chunk_name
= strdup(archived_chunk_name
);
510 if (!session
->last_archived_chunk_name
) {
511 PERROR("Failed to duplicate archived chunk name");
513 session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_COMPLETED
);
515 if (!session
->quiet_rotation
) {
516 location
= session_get_trace_archive_location(session
);
517 ret
= notification_thread_command_session_rotation_completed(
518 notification_thread_handle
,
522 session
->last_archived_chunk_id
.value
,
524 lttng_trace_archive_location_put(location
);
525 if (ret
!= LTTNG_OK
) {
526 ERR("Failed to notify notification thread of completed rotation for session %s",
532 check_ongoing_rotation
:
533 if (session
->rotation_state
== LTTNG_ROTATION_STATE_ONGOING
) {
534 chunk_status
= lttng_trace_chunk_get_id(
535 session
->chunk_being_archived
,
536 &chunk_being_archived_id
);
537 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
539 DBG("Rotation of trace archive %" PRIu64
" is still pending for session %s",
540 chunk_being_archived_id
, session
->name
);
541 ret
= timer_session_rotation_pending_check_start(session
,
542 DEFAULT_ROTATE_PENDING_TIMER
);
544 ERR("Failed to re-enable rotation pending timer");
554 /* Call with the session and session_list locks held. */
556 int launch_session_rotation(struct ltt_session
*session
)
559 struct lttng_rotate_session_return rotation_return
;
561 DBG("Launching scheduled time-based rotation on session \"%s\"",
564 ret
= cmd_rotate_session(session
, &rotation_return
, false,
565 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
566 if (ret
== LTTNG_OK
) {
567 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
570 /* Don't consider errors as fatal. */
571 DBG("Scheduled time-based rotation aborted for session %s: %s",
572 session
->name
, lttng_strerror(ret
));
578 int run_job(struct rotation_thread_job
*job
, struct ltt_session
*session
,
579 struct notification_thread_handle
*notification_thread_handle
)
584 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
585 ret
= launch_session_rotation(session
);
587 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
588 ret
= check_session_rotation_pending(session
,
589 notification_thread_handle
);
598 int handle_job_queue(struct rotation_thread_handle
*handle
,
599 struct rotation_thread
*state
,
600 struct rotation_thread_timer_queue
*queue
)
605 struct ltt_session
*session
;
606 struct rotation_thread_job
*job
;
608 /* Take the queue lock only to pop an element from the list. */
609 pthread_mutex_lock(&queue
->lock
);
610 if (cds_list_empty(&queue
->list
)) {
611 pthread_mutex_unlock(&queue
->lock
);
614 job
= cds_list_first_entry(&queue
->list
,
616 cds_list_del(&job
->head
);
617 pthread_mutex_unlock(&queue
->lock
);
620 session
= job
->session
;
623 * This is a non-fatal error, and we cannot report it to
624 * the user (timer), so just print the error and
625 * continue the processing.
627 * While the timer thread will purge pending signals for
628 * a session on the session's destruction, it is
629 * possible for a job targeting that session to have
630 * already been queued before it was destroyed.
633 session_put(session
);
634 session_unlock_list();
638 session_lock(session
);
639 ret
= run_job(job
, session
, handle
->notification_thread_handle
);
640 session_unlock(session
);
641 /* Release reference held by the job. */
642 session_put(session
);
643 session_unlock_list();
657 int handle_condition(const struct lttng_notification
*notification
,
658 struct notification_thread_handle
*notification_thread_handle
)
661 const char *condition_session_name
= NULL
;
662 enum lttng_condition_type condition_type
;
663 enum lttng_condition_status condition_status
;
664 enum lttng_evaluation_status evaluation_status
;
666 struct ltt_session
*session
;
667 const struct lttng_condition
*condition
=
668 lttng_notification_get_const_condition(notification
);
669 const struct lttng_evaluation
*evaluation
=
670 lttng_notification_get_const_evaluation(notification
);
672 condition_type
= lttng_condition_get_type(condition
);
674 if (condition_type
!= LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
) {
676 ERR("Condition type and session usage type are not the same");
680 /* Fetch info to test */
681 condition_status
= lttng_condition_session_consumed_size_get_session_name(
682 condition
, &condition_session_name
);
683 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
684 ERR("Session name could not be fetched");
688 evaluation_status
= lttng_evaluation_session_consumed_size_get_consumed_size(evaluation
,
690 if (evaluation_status
!= LTTNG_EVALUATION_STATUS_OK
) {
691 ERR("Failed to get evaluation");
697 session
= session_find_by_name(condition_session_name
);
699 DBG("Failed to find session while handling notification: notification type = %s, session name = `%s`",
700 lttng_condition_type_str(condition_type
),
701 condition_session_name
);
703 * Not a fatal error: a session can be destroyed before we get
704 * the chance to handle the notification.
707 session_unlock_list();
710 session_lock(session
);
712 if (!lttng_trigger_is_equal(session
->rotate_trigger
,
713 lttng_notification_get_const_trigger(notification
))) {
714 /* Notification does not originate from our rotation trigger. */
719 ret
= unsubscribe_session_consumed_size_rotation(session
,
720 notification_thread_handle
);
725 ret
= cmd_rotate_session(
726 session
, NULL
, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
730 case -LTTNG_ERR_ROTATION_PENDING
:
731 DBG("Rotate already pending, subscribe to the next threshold value");
733 case -LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
734 DBG("Rotation already happened since last stop, subscribe to the next threshold value");
736 case -LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
737 DBG("Rotation already happened since last stop and clear, subscribe to the next threshold value");
740 ERR("Failed to rotate on size notification with error: %s", lttng_strerror(ret
));
745 ret
= subscribe_session_consumed_size_rotation(
746 session
, consumed
+ session
->rotate_size
, notification_thread_handle
);
748 ERR("Failed to subscribe to session consumed size condition");
754 session_unlock(session
);
755 session_put(session
);
756 session_unlock_list();
762 int handle_notification_channel(int fd
,
763 struct rotation_thread_handle
*handle
,
764 struct rotation_thread
*state
)
767 bool notification_pending
= true;
768 struct lttng_notification
*notification
= NULL
;
769 enum lttng_notification_channel_status status
;
772 * A notification channel may have multiple notifications queued-up internally in
773 * its buffers. This is because a notification channel multiplexes command replies
774 * and notifications. The current protocol specifies that multiple notifications can be
775 * received before the reply to a command.
777 * In such cases, the notification channel client implementation internally queues them and
778 * provides them on the next calls to lttng_notification_channel_get_next_notification().
779 * This is correct with respect to the public API, which is intended to be used in "blocking
782 * However, this internal user relies on poll/epoll to wake-up when data is available
783 * on the notification channel's socket. As such, it can't assume that a wake-up means only
784 * one notification is available for consumption since many of them may have been queued in
785 * the channel's internal buffers.
787 while (notification_pending
) {
788 status
= lttng_notification_channel_has_pending_notification(
789 rotate_notification_channel
, ¬ification_pending
);
790 if (status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
791 ERR("Error occurred while checking for pending notification");
796 if (!notification_pending
) {
801 /* Receive the next notification. */
802 status
= lttng_notification_channel_get_next_notification(
803 rotate_notification_channel
, ¬ification
);
805 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
:
807 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
:
808 WARN("Dropped notification detected on notification channel used by the rotation management thread.");
811 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED
:
812 ERR("Notification channel was closed");
816 /* Unhandled conditions / errors. */
817 ERR("Unknown notification channel status");
822 ret
= handle_condition(notification
, handle
->notification_thread_handle
);
823 lttng_notification_destroy(notification
);
833 void *thread_rotation(void *data
)
836 struct rotation_thread_handle
*handle
= data
;
837 struct rotation_thread thread
;
840 DBG("Started rotation thread");
841 rcu_register_thread();
843 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
844 health_code_update();
847 ERR("Invalid thread context provided");
851 queue_pipe_fd
= lttng_pipe_get_readfd(
852 handle
->rotation_timer_queue
->event_pipe
);
855 ret
= init_thread_state(handle
, &thread
);
864 DBG("Entering poll wait");
865 ret
= lttng_poll_wait(&thread
.events
, -1);
866 DBG("Poll wait returned (%i)", ret
);
870 * Restart interrupted system call.
872 if (errno
== EINTR
) {
875 ERR("Error encountered during lttng_poll_wait (%i)", ret
);
880 for (i
= 0; i
< fd_count
; i
++) {
881 int fd
= LTTNG_POLL_GETFD(&thread
.events
, i
);
882 uint32_t revents
= LTTNG_POLL_GETEV(&thread
.events
, i
);
884 DBG("Handling fd (%i) activity (%u)",
887 if (revents
& LPOLLERR
) {
888 ERR("Polling returned an error on fd %i", fd
);
892 if (fd
== rotate_notification_channel
->socket
||
893 fd
== rotate_notification_channel_subscription_change_eventfd
) {
894 ret
= handle_notification_channel(fd
, handle
, &thread
);
896 ERR("Error occurred while handling activity on notification channel socket");
900 if (fd
== rotate_notification_channel_subscription_change_eventfd
) {
901 uint64_t eventfd_value
;
902 const int read_ret
= lttng_read(fd
, &eventfd_value
, sizeof(eventfd_value
));
904 if (read_ret
!= sizeof(eventfd_value
)) {
905 PERROR("Failed to read value from rotation thread as writing to the rotation thread notification channel subscription change eventfd");
910 /* Job queue or quit pipe activity. */
913 * The job queue is serviced if there is
914 * activity on the quit pipe to ensure it is
915 * flushed and all references held in the queue
918 ret
= handle_job_queue(handle
, &thread
,
919 handle
->rotation_timer_queue
);
921 ERR("Failed to handle rotation timer pipe event");
925 if (fd
== queue_pipe_fd
) {
928 ret
= lttng_read(fd
, &buf
, 1);
930 ERR("Failed to read from wakeup pipe (fd = %i)", fd
);
934 DBG("Quit pipe activity");
943 fini_thread_state(&thread
);
945 health_unregister(the_health_sessiond
);
946 rcu_thread_offline();
947 rcu_unregister_thread();
952 bool shutdown_rotation_thread(void *thread_data
)
954 struct rotation_thread_handle
*handle
= thread_data
;
955 const int write_fd
= lttng_pipe_get_writefd(handle
->quit_pipe
);
957 return notify_thread_pipe(write_fd
) == 1;
960 bool launch_rotation_thread(struct rotation_thread_handle
*handle
)
962 struct lttng_thread
*thread
;
964 thread
= lttng_thread_create("Rotation",
966 shutdown_rotation_thread
,
972 lttng_thread_put(thread
);