2 * Copyright (C) 2017 - Julien Desfossez <jdesfossez@efficios.com>
3 * Copyright (C) 2018 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 * This program is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License, version 2 only, as
7 * published by the Free Software Foundation.
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
14 * You should have received a copy of the GNU General Public License along with
15 * this program; if not, write to the Free Software Foundation, Inc., 51
16 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
20 #include <lttng/trigger/trigger.h>
21 #include <common/error.h>
22 #include <common/config/session-config.h>
23 #include <common/defaults.h>
24 #include <common/utils.h>
25 #include <common/futex.h>
26 #include <common/align.h>
27 #include <common/time.h>
28 #include <common/hashtable/utils.h>
29 #include <sys/eventfd.h>
35 #include <common/kernel-ctl/kernel-ctl.h>
36 #include <lttng/notification/channel-internal.h>
37 #include <lttng/rotate-internal.h>
39 #include "rotation-thread.h"
40 #include "lttng-sessiond.h"
41 #include "health-sessiond.h"
46 #include "notification-thread-commands.h"
49 #include <urcu/list.h>
51 struct lttng_notification_channel
*rotate_notification_channel
= NULL
;
53 struct rotation_thread
{
54 struct lttng_poll_event events
;
57 struct rotation_thread_job
{
58 enum rotation_thread_job_type type
;
59 struct ltt_session
*session
;
60 /* List member in struct rotation_thread_timer_queue. */
61 struct cds_list_head head
;
65 * The timer thread enqueues jobs and wakes up the rotation thread.
66 * When the rotation thread wakes up, it empties the queue.
68 struct rotation_thread_timer_queue
{
69 struct lttng_pipe
*event_pipe
;
70 struct cds_list_head list
;
74 struct rotation_thread_handle
{
75 struct rotation_thread_timer_queue
*rotation_timer_queue
;
76 /* Access to the notification thread cmd_queue */
77 struct notification_thread_handle
*notification_thread_handle
;
81 const char *get_job_type_str(enum rotation_thread_job_type job_type
)
84 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
85 return "CHECK_PENDING_ROTATION";
86 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
87 return "SCHEDULED_ROTATION";
93 struct rotation_thread_timer_queue
*rotation_thread_timer_queue_create(void)
95 struct rotation_thread_timer_queue
*queue
= NULL
;
97 queue
= zmalloc(sizeof(*queue
));
99 PERROR("Failed to allocate timer rotate queue");
103 queue
->event_pipe
= lttng_pipe_open(FD_CLOEXEC
| O_NONBLOCK
);
104 CDS_INIT_LIST_HEAD(&queue
->list
);
105 pthread_mutex_init(&queue
->lock
, NULL
);
110 void log_job_destruction(const struct rotation_thread_job
*job
)
112 enum lttng_error_level log_level
;
113 const char *job_type_str
= get_job_type_str(job
->type
);
116 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
118 * Not a problem, the scheduled rotation is racing with the teardown
119 * of the daemon. In this case, the rotation will not happen, which
120 * is not a problem (or at least, not important enough to delay
121 * the shutdown of the session daemon).
123 log_level
= PRINT_DBG
;
125 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
126 /* This is not expected to happen; warn the user. */
127 log_level
= PRINT_WARN
;
133 LOG(log_level
, "Rotation thread timer queue still contains job of type %s targeting session \"%s\" on destruction",
134 job_type_str
, job
->session
->name
);
137 void rotation_thread_timer_queue_destroy(
138 struct rotation_thread_timer_queue
*queue
)
140 struct rotation_thread_job
*job
, *tmp_job
;
146 lttng_pipe_destroy(queue
->event_pipe
);
148 pthread_mutex_lock(&queue
->lock
);
149 /* Empty wait queue. */
150 cds_list_for_each_entry_safe(job
, tmp_job
, &queue
->list
, head
) {
151 log_job_destruction(job
);
152 cds_list_del(&job
->head
);
155 pthread_mutex_unlock(&queue
->lock
);
156 pthread_mutex_destroy(&queue
->lock
);
161 * Destroy the thread data previously created by the init function.
163 void rotation_thread_handle_destroy(
164 struct rotation_thread_handle
*handle
)
169 struct rotation_thread_handle
*rotation_thread_handle_create(
170 struct rotation_thread_timer_queue
*rotation_timer_queue
,
171 struct notification_thread_handle
*notification_thread_handle
)
173 struct rotation_thread_handle
*handle
;
175 handle
= zmalloc(sizeof(*handle
));
180 handle
->rotation_timer_queue
= rotation_timer_queue
;
181 handle
->notification_thread_handle
= notification_thread_handle
;
188 * Called with the rotation_thread_timer_queue lock held.
189 * Return true if the same timer job already exists in the queue, false if not.
192 bool timer_job_exists(const struct rotation_thread_timer_queue
*queue
,
193 enum rotation_thread_job_type job_type
,
194 struct ltt_session
*session
)
197 struct rotation_thread_job
*job
;
199 cds_list_for_each_entry(job
, &queue
->list
, head
) {
200 if (job
->session
== session
&& job
->type
== job_type
) {
209 void rotation_thread_enqueue_job(struct rotation_thread_timer_queue
*queue
,
210 enum rotation_thread_job_type job_type
,
211 struct ltt_session
*session
)
214 const char * const dummy
= "!";
215 struct rotation_thread_job
*job
= NULL
;
216 const char *job_type_str
= get_job_type_str(job_type
);
218 pthread_mutex_lock(&queue
->lock
);
219 if (timer_job_exists(queue
, job_type
, session
)) {
221 * This timer job is already pending, we don't need to add
227 job
= zmalloc(sizeof(struct rotation_thread_job
));
229 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
230 job_type_str
, session
->name
);
233 /* No reason for this to fail as the caller must hold a reference. */
234 (void) session_get(session
);
236 job
->session
= session
;
237 job
->type
= job_type
;
238 cds_list_add_tail(&job
->head
, &queue
->list
);
240 ret
= lttng_write(lttng_pipe_get_writefd(queue
->event_pipe
), dummy
,
244 * We do not want to block in the timer handler, the job has
245 * been enqueued in the list, the wakeup pipe is probably full,
246 * the job will be processed when the rotation_thread catches
249 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
251 * Not an error, but would be surprising and indicate
252 * that the rotation thread can't keep up with the
255 DBG("Wake-up pipe of rotation thread job queue is full");
258 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
259 job_type_str
, session
->name
);
264 pthread_mutex_unlock(&queue
->lock
);
268 int init_poll_set(struct lttng_poll_event
*poll_set
,
269 struct rotation_thread_handle
*handle
)
274 * Create pollset with size 2:
276 * - rotation thread timer queue pipe,
278 ret
= sessiond_set_thread_pollset(poll_set
, 2);
282 ret
= lttng_poll_add(poll_set
,
283 lttng_pipe_get_readfd(handle
->rotation_timer_queue
->event_pipe
),
286 ERR("[rotation-thread] Failed to add rotate_pending fd to pollset");
292 lttng_poll_clean(poll_set
);
297 void fini_thread_state(struct rotation_thread
*state
)
299 lttng_poll_clean(&state
->events
);
300 if (rotate_notification_channel
) {
301 lttng_notification_channel_destroy(rotate_notification_channel
);
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("[rotation-thread] 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("[rotation-thread] Could not create notification channel");
327 ret
= lttng_poll_add(&state
->events
, rotate_notification_channel
->socket
,
330 ERR("[rotation-thread] Failed to add notification fd to pollset");
339 int check_session_rotation_pending_local_on_consumer(
340 const struct ltt_session
*session
,
341 struct consumer_socket
*socket
, bool *rotation_completed
)
345 pthread_mutex_lock(socket
->lock
);
346 DBG("[rotation-thread] Checking for locally pending rotation on the %s consumer for session %s",
347 lttng_consumer_type_str(socket
->type
),
349 ret
= consumer_check_rotation_pending_local(socket
,
351 session
->current_archive_id
- 1);
352 pthread_mutex_unlock(socket
->lock
);
355 /* Rotation was completed on this consumer. */
356 DBG("[rotation-thread] Local rotation of trace archive %" PRIu64
" of session \"%s\" was completed on the %s consumer",
357 session
->current_archive_id
- 1,
359 lttng_consumer_type_str(socket
->type
));
360 *rotation_completed
= true;
361 } else if (ret
== 1) {
362 /* Rotation pending on this consumer. */
363 DBG("[rotation-thread] Local rotation of trace archive %" PRIu64
" of session \"%s\" is pending on the %s consumer",
364 session
->current_archive_id
- 1,
366 lttng_consumer_type_str(socket
->type
));
367 *rotation_completed
= false;
370 /* Not a fatal error. */
371 ERR("[rotation-thread] Encountered an error when checking if local rotation of trace archive %" PRIu64
" of session \"%s\" is pending on the %s consumer",
372 session
->current_archive_id
- 1,
374 lttng_consumer_type_str(socket
->type
));
375 *rotation_completed
= false;
381 int check_session_rotation_pending_local(struct ltt_session
*session
)
384 struct consumer_socket
*socket
;
385 struct cds_lfht_iter iter
;
386 bool rotation_completed
= true;
389 * Check for a local pending rotation on all consumers (32-bit
390 * user space, 64-bit user space, and kernel).
392 DBG("[rotation-thread] Checking for pending local rotation on session \"%s\", trace archive %" PRIu64
,
393 session
->name
, session
->current_archive_id
- 1);
396 if (!session
->ust_session
) {
399 cds_lfht_for_each_entry(session
->ust_session
->consumer
->socks
->ht
,
400 &iter
, socket
, node
.node
) {
401 ret
= check_session_rotation_pending_local_on_consumer(session
,
402 socket
, &rotation_completed
);
403 if (ret
|| !rotation_completed
) {
409 if (!session
->kernel_session
) {
412 cds_lfht_for_each_entry(session
->kernel_session
->consumer
->socks
->ht
,
413 &iter
, socket
, node
.node
) {
414 ret
= check_session_rotation_pending_local_on_consumer(session
,
415 socket
, &rotation_completed
);
416 if (ret
|| !rotation_completed
) {
424 if (rotation_completed
) {
425 DBG("[rotation-thread] Local rotation of trace archive %" PRIu64
" of session \"%s\" is complete on all consumers",
426 session
->current_archive_id
- 1,
428 session
->rotation_pending_local
= false;
431 ret
= session_reset_rotation_state(session
,
432 LTTNG_ROTATION_STATE_ERROR
);
434 ERR("Failed to reset rotation state of session \"%s\"",
442 int check_session_rotation_pending_relay(struct ltt_session
*session
)
445 struct consumer_socket
*socket
;
446 struct cds_lfht_iter iter
;
447 bool rotation_completed
= true;
448 const struct consumer_output
*output
;
451 * Check for a pending rotation on any consumer as we only use
452 * it as a "tunnel" to the relayd.
456 if (session
->ust_session
) {
457 cds_lfht_first(session
->ust_session
->consumer
->socks
->ht
,
459 output
= session
->ust_session
->consumer
;
461 cds_lfht_first(session
->kernel_session
->consumer
->socks
->ht
,
463 output
= session
->kernel_session
->consumer
;
465 assert(cds_lfht_iter_get_node(&iter
));
467 socket
= caa_container_of(cds_lfht_iter_get_node(&iter
),
468 typeof(*socket
), node
.node
);
470 pthread_mutex_lock(socket
->lock
);
471 DBG("[rotation-thread] Checking for pending relay rotation on session \"%s\", trace archive %" PRIu64
" through the %s consumer",
472 session
->name
, session
->current_archive_id
- 1,
473 lttng_consumer_type_str(socket
->type
));
474 ret
= consumer_check_rotation_pending_relay(socket
,
477 session
->current_archive_id
- 1);
478 pthread_mutex_unlock(socket
->lock
);
481 /* Rotation was completed on the relay. */
482 DBG("[rotation-thread] Relay rotation of trace archive %" PRIu64
" of session \"%s\" was completed",
483 session
->current_archive_id
- 1,
485 } else if (ret
== 1) {
486 /* Rotation pending on relay. */
487 DBG("[rotation-thread] Relay rotation of trace archive %" PRIu64
" of session \"%s\" is pending",
488 session
->current_archive_id
- 1,
490 rotation_completed
= false;
492 /* Not a fatal error. */
493 ERR("[rotation-thread] Encountered an error when checking if rotation of trace archive %" PRIu64
" of session \"%s\" is pending on the relay",
494 session
->current_archive_id
- 1,
496 ret
= session_reset_rotation_state(session
,
497 LTTNG_ROTATION_STATE_ERROR
);
499 ERR("Failed to reset rotation state of session \"%s\"",
502 rotation_completed
= false;
507 if (rotation_completed
) {
508 DBG("[rotation-thread] Rotation of trace archive %" PRIu64
" of session \"%s\" is complete on the relay",
509 session
->current_archive_id
- 1,
511 session
->rotation_pending_relay
= false;
517 * Check if the last rotation was completed, called with session lock held.
520 int check_session_rotation_pending(struct ltt_session
*session
,
521 struct notification_thread_handle
*notification_thread_handle
)
524 struct lttng_trace_archive_location
*location
;
527 DBG("[rotation-thread] Checking for pending rotation on session \"%s\", trace archive %" PRIu64
,
528 session
->name
, session
->current_archive_id
- 1);
531 * The rotation-pending check timer of a session is launched in
532 * one-shot mode. If the rotation is incomplete, the rotation
533 * thread will re-enable the pending-check timer.
535 * The timer thread can't stop the timer itself since it is involved
536 * in the check for the timer's quiescence.
538 ret
= timer_session_rotation_pending_check_stop(session
);
543 if (session
->rotation_pending_local
) {
544 /* Updates session->rotation_pending_local as needed. */
545 ret
= check_session_rotation_pending_local(session
);
551 * No need to check for a pending rotation on the relay
552 * since the rotation is not even completed locally yet.
554 if (session
->rotation_pending_local
) {
559 if (session
->rotation_pending_relay
) {
560 /* Updates session->rotation_pending_relay as needed. */
561 ret
= check_session_rotation_pending_relay(session
);
566 if (session
->rotation_pending_relay
) {
571 DBG("[rotation-thread] Rotation of trace archive %" PRIu64
" completed for "
572 "session %s", session
->current_archive_id
- 1,
575 /* Rename the completed trace archive's location. */
577 if (now
== (time_t) -1) {
578 ret
= session_reset_rotation_state(session
,
579 LTTNG_ROTATION_STATE_ERROR
);
581 ERR("Failed to reset rotation state of session \"%s\"",
588 ret
= rename_completed_chunk(session
, now
);
590 ERR("Failed to rename completed rotation chunk");
593 session
->last_chunk_start_ts
= session
->current_chunk_start_ts
;
596 * Now we can clear the "ONGOING" state in the session. New
597 * rotations can start now.
599 session
->rotation_state
= LTTNG_ROTATION_STATE_COMPLETED
;
601 /* Ownership of location is transferred. */
602 location
= session_get_trace_archive_location(session
);
603 ret
= notification_thread_command_session_rotation_completed(
604 notification_thread_handle
,
608 session
->current_archive_id
,
610 if (ret
!= LTTNG_OK
) {
611 ERR("[rotation-thread] Failed to notify notification thread of completed rotation for session %s",
615 if (!session
->active
) {
617 * A stop command was issued during the rotation, it is
618 * up to the rotation completion check to perform the
619 * renaming of the last chunk that was produced.
621 ret
= notification_thread_command_session_rotation_ongoing(
622 notification_thread_handle
,
626 session
->current_archive_id
);
627 if (ret
!= LTTNG_OK
) {
628 ERR("[rotation-thread] Failed to notify notification thread of completed rotation for session %s",
632 ret
= rename_active_chunk(session
);
634 ERR("[rotation-thread] Failed to rename active rotation chunk");
638 /* Ownership of location is transferred. */
639 location
= session_get_trace_archive_location(session
);
640 ret
= notification_thread_command_session_rotation_completed(
641 notification_thread_handle
,
645 session
->current_archive_id
,
647 if (ret
!= LTTNG_OK
) {
648 ERR("[rotation-thread] Failed to notify notification thread of completed rotation for session %s",
655 if (session
->rotation_state
== LTTNG_ROTATION_STATE_ONGOING
) {
656 DBG("[rotation-thread] Rotation of trace archive %" PRIu64
" is still pending for session %s",
657 session
->current_archive_id
- 1, session
->name
);
658 ret
= timer_session_rotation_pending_check_start(session
,
659 DEFAULT_ROTATE_PENDING_TIMER
);
661 ERR("Re-enabling rotate pending timer");
670 /* Call with the session and session_list locks held. */
672 int launch_session_rotation(struct ltt_session
*session
)
675 struct lttng_rotate_session_return rotation_return
;
677 DBG("[rotation-thread] Launching scheduled time-based rotation on session \"%s\"",
680 ret
= cmd_rotate_session(session
, &rotation_return
);
681 if (ret
== LTTNG_OK
) {
682 DBG("[rotation-thread] Scheduled time-based rotation successfully launched on session \"%s\"",
685 /* Don't consider errors as fatal. */
686 DBG("[rotation-thread] Scheduled time-based rotation aborted for session %s: %s",
687 session
->name
, lttng_strerror(ret
));
693 int run_job(struct rotation_thread_job
*job
, struct ltt_session
*session
,
694 struct notification_thread_handle
*notification_thread_handle
)
699 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
700 ret
= launch_session_rotation(session
);
702 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
703 ret
= check_session_rotation_pending(session
,
704 notification_thread_handle
);
713 int handle_job_queue(struct rotation_thread_handle
*handle
,
714 struct rotation_thread
*state
,
715 struct rotation_thread_timer_queue
*queue
)
718 int fd
= lttng_pipe_get_readfd(queue
->event_pipe
);
721 ret
= lttng_read(fd
, &buf
, 1);
723 ERR("[rotation-thread] Failed to read from wakeup pipe (fd = %i)", fd
);
729 struct ltt_session
*session
;
730 struct rotation_thread_job
*job
;
732 /* Take the queue lock only to pop an element from the list. */
733 pthread_mutex_lock(&queue
->lock
);
734 if (cds_list_empty(&queue
->list
)) {
735 pthread_mutex_unlock(&queue
->lock
);
738 job
= cds_list_first_entry(&queue
->list
,
740 cds_list_del(&job
->head
);
741 pthread_mutex_unlock(&queue
->lock
);
744 session
= job
->session
;
746 DBG("[rotation-thread] Session \"%s\" not found",
749 * This is a non-fatal error, and we cannot report it to
750 * the user (timer), so just print the error and
751 * continue the processing.
753 * While the timer thread will purge pending signals for
754 * a session on the session's destruction, it is
755 * possible for a job targeting that session to have
756 * already been queued before it was destroyed.
758 session_unlock_list();
760 session_put(session
);
764 session_lock(session
);
765 ret
= run_job(job
, session
, handle
->notification_thread_handle
);
766 session_unlock(session
);
767 /* Release reference held by the job. */
768 session_put(session
);
769 session_unlock_list();
783 int handle_condition(const struct lttng_condition
*condition
,
784 const struct lttng_evaluation
*evaluation
,
785 struct notification_thread_handle
*notification_thread_handle
)
788 const char *condition_session_name
= NULL
;
789 enum lttng_condition_type condition_type
;
790 enum lttng_condition_status condition_status
;
791 enum lttng_evaluation_status evaluation_status
;
793 struct ltt_session
*session
;
795 condition_type
= lttng_condition_get_type(condition
);
797 if (condition_type
!= LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
) {
799 ERR("[rotation-thread] Condition type and session usage type are not the same");
803 /* Fetch info to test */
804 condition_status
= lttng_condition_session_consumed_size_get_session_name(
805 condition
, &condition_session_name
);
806 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
807 ERR("[rotation-thread] Session name could not be fetched");
811 evaluation_status
= lttng_evaluation_session_consumed_size_get_consumed_size(evaluation
,
813 if (evaluation_status
!= LTTNG_EVALUATION_STATUS_OK
) {
814 ERR("[rotation-thread] Failed to get evaluation");
820 session
= session_find_by_name(condition_session_name
);
823 session_unlock_list();
824 ERR("[rotation-thread] Session \"%s\" not found",
825 condition_session_name
);
828 session_lock(session
);
829 session_unlock_list();
831 ret
= unsubscribe_session_consumed_size_rotation(session
,
832 notification_thread_handle
);
837 ret
= cmd_rotate_session(session
, NULL
);
838 if (ret
== -LTTNG_ERR_ROTATION_PENDING
) {
839 DBG("Rotate already pending, subscribe to the next threshold value");
840 } else if (ret
!= LTTNG_OK
) {
841 ERR("[rotation-thread] Failed to rotate on size notification with error: %s",
842 lttng_strerror(ret
));
846 ret
= subscribe_session_consumed_size_rotation(session
,
847 consumed
+ session
->rotate_size
,
848 notification_thread_handle
);
850 ERR("[rotation-thread] Failed to subscribe to session consumed size condition");
856 session_unlock(session
);
857 session_put(session
);
863 int handle_notification_channel(int fd
,
864 struct rotation_thread_handle
*handle
,
865 struct rotation_thread
*state
)
868 bool notification_pending
;
869 struct lttng_notification
*notification
= NULL
;
870 enum lttng_notification_channel_status status
;
871 const struct lttng_evaluation
*notification_evaluation
;
872 const struct lttng_condition
*notification_condition
;
874 status
= lttng_notification_channel_has_pending_notification(
875 rotate_notification_channel
, ¬ification_pending
);
876 if (status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
877 ERR("[rotation-thread ]Error occurred while checking for pending notification");
882 if (!notification_pending
) {
887 /* Receive the next notification. */
888 status
= lttng_notification_channel_get_next_notification(
889 rotate_notification_channel
,
893 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
:
895 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
:
896 /* Not an error, we will wait for the next one */
899 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED
:
900 ERR("Notification channel was closed");
904 /* Unhandled conditions / errors. */
905 ERR("Unknown notification channel status");
910 notification_condition
= lttng_notification_get_condition(notification
);
911 notification_evaluation
= lttng_notification_get_evaluation(notification
);
913 ret
= handle_condition(notification_condition
, notification_evaluation
,
914 handle
->notification_thread_handle
);
917 lttng_notification_destroy(notification
);
921 void *thread_rotation(void *data
)
924 struct rotation_thread_handle
*handle
= data
;
925 struct rotation_thread thread
;
927 DBG("[rotation-thread] Started rotation thread");
930 ERR("[rotation-thread] Invalid thread context provided");
934 rcu_register_thread();
937 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
938 health_code_update();
940 ret
= init_thread_state(handle
, &thread
);
945 /* Ready to handle client connections. */
946 sessiond_notify_ready();
952 DBG("[rotation-thread] Entering poll wait");
953 ret
= lttng_poll_wait(&thread
.events
, -1);
954 DBG("[rotation-thread] Poll wait returned (%i)", ret
);
958 * Restart interrupted system call.
960 if (errno
== EINTR
) {
963 ERR("[rotation-thread] Error encountered during lttng_poll_wait (%i)", ret
);
968 for (i
= 0; i
< fd_count
; i
++) {
969 int fd
= LTTNG_POLL_GETFD(&thread
.events
, i
);
970 uint32_t revents
= LTTNG_POLL_GETEV(&thread
.events
, i
);
972 DBG("[rotation-thread] Handling fd (%i) activity (%u)",
975 if (revents
& LPOLLERR
) {
976 ERR("[rotation-thread] Polling returned an error on fd %i", fd
);
980 if (sessiond_check_thread_quit_pipe(fd
, revents
)) {
981 DBG("[rotation-thread] Quit pipe activity");
982 /* TODO flush the queue. */
984 } else if (fd
== lttng_pipe_get_readfd(handle
->rotation_timer_queue
->event_pipe
)) {
985 ret
= handle_job_queue(handle
, &thread
,
986 handle
->rotation_timer_queue
);
988 ERR("[rotation-thread] Failed to handle rotation timer pipe event");
991 } else if (fd
== rotate_notification_channel
->socket
) {
992 ret
= handle_notification_channel(fd
, handle
,
995 ERR("[rotation-thread] Error occurred while handling activity on notification channel socket");
1003 DBG("[rotation-thread] Exit");
1004 fini_thread_state(&thread
);
1005 health_unregister(health_sessiond
);
1006 rcu_thread_offline();
1007 rcu_unregister_thread();