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>
24 #include <common/kernel-ctl/kernel-ctl.h>
25 #include <lttng/notification/channel-internal.h>
26 #include <lttng/rotate-internal.h>
28 #include "rotation-thread.h"
29 #include "lttng-sessiond.h"
30 #include "health-sessiond.h"
35 #include "notification-thread-commands.h"
40 #include <urcu/list.h>
42 struct lttng_notification_channel
*rotate_notification_channel
= NULL
;
44 struct rotation_thread
{
45 struct lttng_poll_event events
;
48 struct rotation_thread_job
{
49 enum rotation_thread_job_type type
;
50 struct ltt_session
*session
;
51 /* List member in struct rotation_thread_timer_queue. */
52 struct cds_list_head head
;
56 * The timer thread enqueues jobs and wakes up the rotation thread.
57 * When the rotation thread wakes up, it empties the queue.
59 struct rotation_thread_timer_queue
{
60 struct lttng_pipe
*event_pipe
;
61 struct cds_list_head list
;
65 struct rotation_thread_handle
{
66 struct rotation_thread_timer_queue
*rotation_timer_queue
;
67 /* Access to the notification thread cmd_queue */
68 struct notification_thread_handle
*notification_thread_handle
;
69 /* Thread-specific quit pipe. */
70 struct lttng_pipe
*quit_pipe
;
74 const char *get_job_type_str(enum rotation_thread_job_type job_type
)
77 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
78 return "CHECK_PENDING_ROTATION";
79 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
80 return "SCHEDULED_ROTATION";
86 struct rotation_thread_timer_queue
*rotation_thread_timer_queue_create(void)
88 struct rotation_thread_timer_queue
*queue
= NULL
;
90 queue
= zmalloc(sizeof(*queue
));
92 PERROR("Failed to allocate timer rotate queue");
96 queue
->event_pipe
= lttng_pipe_open(FD_CLOEXEC
| O_NONBLOCK
);
97 CDS_INIT_LIST_HEAD(&queue
->list
);
98 pthread_mutex_init(&queue
->lock
, NULL
);
103 void rotation_thread_timer_queue_destroy(
104 struct rotation_thread_timer_queue
*queue
)
110 lttng_pipe_destroy(queue
->event_pipe
);
112 pthread_mutex_lock(&queue
->lock
);
113 assert(cds_list_empty(&queue
->list
));
114 pthread_mutex_unlock(&queue
->lock
);
115 pthread_mutex_destroy(&queue
->lock
);
120 * Destroy the thread data previously created by the init function.
122 void rotation_thread_handle_destroy(
123 struct rotation_thread_handle
*handle
)
125 lttng_pipe_destroy(handle
->quit_pipe
);
129 struct rotation_thread_handle
*rotation_thread_handle_create(
130 struct rotation_thread_timer_queue
*rotation_timer_queue
,
131 struct notification_thread_handle
*notification_thread_handle
)
133 struct rotation_thread_handle
*handle
;
135 handle
= zmalloc(sizeof(*handle
));
140 handle
->rotation_timer_queue
= rotation_timer_queue
;
141 handle
->notification_thread_handle
= notification_thread_handle
;
142 handle
->quit_pipe
= lttng_pipe_open(FD_CLOEXEC
);
143 if (!handle
->quit_pipe
) {
150 rotation_thread_handle_destroy(handle
);
155 * Called with the rotation_thread_timer_queue lock held.
156 * Return true if the same timer job already exists in the queue, false if not.
159 bool timer_job_exists(const struct rotation_thread_timer_queue
*queue
,
160 enum rotation_thread_job_type job_type
,
161 struct ltt_session
*session
)
164 struct rotation_thread_job
*job
;
166 cds_list_for_each_entry(job
, &queue
->list
, head
) {
167 if (job
->session
== session
&& job
->type
== job_type
) {
176 void rotation_thread_enqueue_job(struct rotation_thread_timer_queue
*queue
,
177 enum rotation_thread_job_type job_type
,
178 struct ltt_session
*session
)
181 const char dummy
= '!';
182 struct rotation_thread_job
*job
= NULL
;
183 const char *job_type_str
= get_job_type_str(job_type
);
185 pthread_mutex_lock(&queue
->lock
);
186 if (timer_job_exists(queue
, job_type
, session
)) {
188 * This timer job is already pending, we don't need to add
194 job
= zmalloc(sizeof(struct rotation_thread_job
));
196 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
197 job_type_str
, session
->name
);
200 /* No reason for this to fail as the caller must hold a reference. */
201 (void) session_get(session
);
203 job
->session
= session
;
204 job
->type
= job_type
;
205 cds_list_add_tail(&job
->head
, &queue
->list
);
207 ret
= lttng_write(lttng_pipe_get_writefd(queue
->event_pipe
), &dummy
,
211 * We do not want to block in the timer handler, the job has
212 * been enqueued in the list, the wakeup pipe is probably full,
213 * the job will be processed when the rotation_thread catches
216 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
218 * Not an error, but would be surprising and indicate
219 * that the rotation thread can't keep up with the
222 DBG("Wake-up pipe of rotation thread job queue is full");
225 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
226 job_type_str
, session
->name
);
231 pthread_mutex_unlock(&queue
->lock
);
235 int init_poll_set(struct lttng_poll_event
*poll_set
,
236 struct rotation_thread_handle
*handle
)
241 * Create pollset with size 3:
242 * - rotation thread quit pipe,
243 * - rotation thread timer queue pipe,
244 * - notification channel sock,
246 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
251 ret
= lttng_poll_add(poll_set
,
252 lttng_pipe_get_readfd(handle
->quit_pipe
),
255 ERR("Failed to add quit pipe read fd to poll set");
259 ret
= lttng_poll_add(poll_set
,
260 lttng_pipe_get_readfd(handle
->rotation_timer_queue
->event_pipe
),
263 ERR("Failed to add rotate_pending fd to poll set");
269 lttng_poll_clean(poll_set
);
274 void fini_thread_state(struct rotation_thread
*state
)
276 lttng_poll_clean(&state
->events
);
277 if (rotate_notification_channel
) {
278 lttng_notification_channel_destroy(rotate_notification_channel
);
283 int init_thread_state(struct rotation_thread_handle
*handle
,
284 struct rotation_thread
*state
)
288 memset(state
, 0, sizeof(*state
));
289 lttng_poll_init(&state
->events
);
291 ret
= init_poll_set(&state
->events
, handle
);
293 ERR("Failed to initialize rotation thread poll set");
297 rotate_notification_channel
= lttng_notification_channel_create(
298 lttng_session_daemon_notification_endpoint
);
299 if (!rotate_notification_channel
) {
300 ERR("Could not create notification channel");
304 ret
= lttng_poll_add(&state
->events
, rotate_notification_channel
->socket
,
307 ERR("Failed to add notification fd to pollset");
316 void check_session_rotation_pending_on_consumers(struct ltt_session
*session
,
317 bool *_rotation_completed
)
320 struct consumer_socket
*socket
;
321 struct cds_lfht_iter iter
;
322 enum consumer_trace_chunk_exists_status exists_status
;
324 bool chunk_exists_on_peer
= false;
325 enum lttng_trace_chunk_status chunk_status
;
327 assert(session
->chunk_being_archived
);
330 * Check for a local pending rotation on all consumers (32-bit
331 * user space, 64-bit user space, and kernel).
334 if (!session
->ust_session
) {
337 cds_lfht_for_each_entry(session
->ust_session
->consumer
->socks
->ht
,
338 &iter
, socket
, node
.node
) {
339 relayd_id
= session
->ust_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
341 session
->ust_session
->consumer
->net_seq_index
;
343 pthread_mutex_lock(socket
->lock
);
344 ret
= consumer_trace_chunk_exists(socket
,
346 session
->id
, session
->chunk_being_archived
,
349 pthread_mutex_unlock(socket
->lock
);
350 ERR("Error occurred while checking rotation status on consumer daemon");
354 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
355 pthread_mutex_unlock(socket
->lock
);
356 chunk_exists_on_peer
= true;
359 pthread_mutex_unlock(socket
->lock
);
363 if (!session
->kernel_session
) {
366 cds_lfht_for_each_entry(session
->kernel_session
->consumer
->socks
->ht
,
367 &iter
, socket
, node
.node
) {
368 pthread_mutex_lock(socket
->lock
);
369 relayd_id
= session
->kernel_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
371 session
->kernel_session
->consumer
->net_seq_index
;
373 ret
= consumer_trace_chunk_exists(socket
,
375 session
->id
, session
->chunk_being_archived
,
378 pthread_mutex_unlock(socket
->lock
);
379 ERR("Error occurred while checking rotation status on consumer daemon");
383 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
384 pthread_mutex_unlock(socket
->lock
);
385 chunk_exists_on_peer
= true;
388 pthread_mutex_unlock(socket
->lock
);
394 if (!chunk_exists_on_peer
) {
395 uint64_t chunk_being_archived_id
;
397 chunk_status
= lttng_trace_chunk_get_id(
398 session
->chunk_being_archived
,
399 &chunk_being_archived_id
);
400 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
401 DBG("Rotation of trace archive %" PRIu64
" of session \"%s\" is complete on all consumers",
402 chunk_being_archived_id
,
405 *_rotation_completed
= !chunk_exists_on_peer
;
407 ret
= session_reset_rotation_state(session
,
408 LTTNG_ROTATION_STATE_ERROR
);
410 ERR("Failed to reset rotation state of session \"%s\"",
417 * Check if the last rotation was completed, called with session lock held.
418 * Should only return non-zero in the event of a fatal error. Doing so will
419 * shutdown the thread.
422 int check_session_rotation_pending(struct ltt_session
*session
,
423 struct notification_thread_handle
*notification_thread_handle
)
426 struct lttng_trace_archive_location
*location
;
427 enum lttng_trace_chunk_status chunk_status
;
428 bool rotation_completed
= false;
429 const char *archived_chunk_name
;
430 uint64_t chunk_being_archived_id
;
432 if (!session
->chunk_being_archived
) {
437 chunk_status
= lttng_trace_chunk_get_id(session
->chunk_being_archived
,
438 &chunk_being_archived_id
);
439 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
441 DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64
,
442 session
->name
, chunk_being_archived_id
);
445 * The rotation-pending check timer of a session is launched in
446 * one-shot mode. If the rotation is incomplete, the rotation
447 * thread will re-enable the pending-check timer.
449 * The timer thread can't stop the timer itself since it is involved
450 * in the check for the timer's quiescence.
452 ret
= timer_session_rotation_pending_check_stop(session
);
454 goto check_ongoing_rotation
;
457 check_session_rotation_pending_on_consumers(session
,
458 &rotation_completed
);
459 if (!rotation_completed
||
460 session
->rotation_state
== LTTNG_ROTATION_STATE_ERROR
) {
461 goto check_ongoing_rotation
;
465 * Now we can clear the "ONGOING" state in the session. New
466 * rotations can start now.
468 chunk_status
= lttng_trace_chunk_get_name(session
->chunk_being_archived
,
469 &archived_chunk_name
, NULL
);
470 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
471 free(session
->last_archived_chunk_name
);
472 session
->last_archived_chunk_name
= strdup(archived_chunk_name
);
473 if (!session
->last_archived_chunk_name
) {
474 PERROR("Failed to duplicate archived chunk name");
476 session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_COMPLETED
);
478 if (!session
->quiet_rotation
) {
479 location
= session_get_trace_archive_location(session
);
480 /* Ownership of location is transferred. */
481 ret
= notification_thread_command_session_rotation_completed(
482 notification_thread_handle
,
486 session
->last_archived_chunk_id
.value
,
488 if (ret
!= LTTNG_OK
) {
489 ERR("Failed to notify notification thread of completed rotation for session %s",
495 check_ongoing_rotation
:
496 if (session
->rotation_state
== LTTNG_ROTATION_STATE_ONGOING
) {
497 chunk_status
= lttng_trace_chunk_get_id(
498 session
->chunk_being_archived
,
499 &chunk_being_archived_id
);
500 assert(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
502 DBG("Rotation of trace archive %" PRIu64
" is still pending for session %s",
503 chunk_being_archived_id
, session
->name
);
504 ret
= timer_session_rotation_pending_check_start(session
,
505 DEFAULT_ROTATE_PENDING_TIMER
);
507 ERR("Failed to re-enable rotation pending timer");
517 /* Call with the session and session_list locks held. */
519 int launch_session_rotation(struct ltt_session
*session
)
522 struct lttng_rotate_session_return rotation_return
;
524 DBG("Launching scheduled time-based rotation on session \"%s\"",
527 ret
= cmd_rotate_session(session
, &rotation_return
, false,
528 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
529 if (ret
== LTTNG_OK
) {
530 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
533 /* Don't consider errors as fatal. */
534 DBG("Scheduled time-based rotation aborted for session %s: %s",
535 session
->name
, lttng_strerror(ret
));
541 int run_job(struct rotation_thread_job
*job
, struct ltt_session
*session
,
542 struct notification_thread_handle
*notification_thread_handle
)
547 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
548 ret
= launch_session_rotation(session
);
550 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
551 ret
= check_session_rotation_pending(session
,
552 notification_thread_handle
);
561 int handle_job_queue(struct rotation_thread_handle
*handle
,
562 struct rotation_thread
*state
,
563 struct rotation_thread_timer_queue
*queue
)
568 struct ltt_session
*session
;
569 struct rotation_thread_job
*job
;
571 /* Take the queue lock only to pop an element from the list. */
572 pthread_mutex_lock(&queue
->lock
);
573 if (cds_list_empty(&queue
->list
)) {
574 pthread_mutex_unlock(&queue
->lock
);
577 job
= cds_list_first_entry(&queue
->list
,
579 cds_list_del(&job
->head
);
580 pthread_mutex_unlock(&queue
->lock
);
583 session
= job
->session
;
585 DBG("Session \"%s\" not found",
588 * This is a non-fatal error, and we cannot report it to
589 * the user (timer), so just print the error and
590 * continue the processing.
592 * While the timer thread will purge pending signals for
593 * a session on the session's destruction, it is
594 * possible for a job targeting that session to have
595 * already been queued before it was destroyed.
598 session_put(session
);
599 session_unlock_list();
603 session_lock(session
);
604 ret
= run_job(job
, session
, handle
->notification_thread_handle
);
605 session_unlock(session
);
606 /* Release reference held by the job. */
607 session_put(session
);
608 session_unlock_list();
622 int handle_condition(const struct lttng_condition
*condition
,
623 const struct lttng_evaluation
*evaluation
,
624 struct notification_thread_handle
*notification_thread_handle
)
627 const char *condition_session_name
= NULL
;
628 enum lttng_condition_type condition_type
;
629 enum lttng_condition_status condition_status
;
630 enum lttng_evaluation_status evaluation_status
;
632 struct ltt_session
*session
;
634 condition_type
= lttng_condition_get_type(condition
);
636 if (condition_type
!= LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
) {
638 ERR("Condition type and session usage type are not the same");
642 /* Fetch info to test */
643 condition_status
= lttng_condition_session_consumed_size_get_session_name(
644 condition
, &condition_session_name
);
645 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
646 ERR("Session name could not be fetched");
650 evaluation_status
= lttng_evaluation_session_consumed_size_get_consumed_size(evaluation
,
652 if (evaluation_status
!= LTTNG_EVALUATION_STATUS_OK
) {
653 ERR("Failed to get evaluation");
659 session
= session_find_by_name(condition_session_name
);
662 session_unlock_list();
663 ERR("Session \"%s\" not found",
664 condition_session_name
);
667 session_lock(session
);
669 ret
= unsubscribe_session_consumed_size_rotation(session
,
670 notification_thread_handle
);
675 ret
= cmd_rotate_session(session
, NULL
, false,
676 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
677 if (ret
== -LTTNG_ERR_ROTATION_PENDING
) {
678 DBG("Rotate already pending, subscribe to the next threshold value");
679 } else if (ret
!= LTTNG_OK
) {
680 ERR("Failed to rotate on size notification with error: %s",
681 lttng_strerror(ret
));
685 ret
= subscribe_session_consumed_size_rotation(session
,
686 consumed
+ session
->rotate_size
,
687 notification_thread_handle
);
689 ERR("Failed to subscribe to session consumed size condition");
695 session_unlock(session
);
696 session_put(session
);
697 session_unlock_list();
703 int handle_notification_channel(int fd
,
704 struct rotation_thread_handle
*handle
,
705 struct rotation_thread
*state
)
708 bool notification_pending
;
709 struct lttng_notification
*notification
= NULL
;
710 enum lttng_notification_channel_status status
;
711 const struct lttng_evaluation
*notification_evaluation
;
712 const struct lttng_condition
*notification_condition
;
714 status
= lttng_notification_channel_has_pending_notification(
715 rotate_notification_channel
, ¬ification_pending
);
716 if (status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
717 ERR("Error occurred while checking for pending notification");
722 if (!notification_pending
) {
727 /* Receive the next notification. */
728 status
= lttng_notification_channel_get_next_notification(
729 rotate_notification_channel
,
733 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
:
735 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
:
736 /* Not an error, we will wait for the next one */
739 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED
:
740 ERR("Notification channel was closed");
744 /* Unhandled conditions / errors. */
745 ERR("Unknown notification channel status");
750 notification_condition
= lttng_notification_get_condition(notification
);
751 notification_evaluation
= lttng_notification_get_evaluation(notification
);
753 ret
= handle_condition(notification_condition
, notification_evaluation
,
754 handle
->notification_thread_handle
);
757 lttng_notification_destroy(notification
);
762 void *thread_rotation(void *data
)
765 struct rotation_thread_handle
*handle
= data
;
766 struct rotation_thread thread
;
769 DBG("Started rotation thread");
770 rcu_register_thread();
772 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
773 health_code_update();
776 ERR("Invalid thread context provided");
780 queue_pipe_fd
= lttng_pipe_get_readfd(
781 handle
->rotation_timer_queue
->event_pipe
);
784 ret
= init_thread_state(handle
, &thread
);
793 DBG("Entering poll wait");
794 ret
= lttng_poll_wait(&thread
.events
, -1);
795 DBG("Poll wait returned (%i)", ret
);
799 * Restart interrupted system call.
801 if (errno
== EINTR
) {
804 ERR("Error encountered during lttng_poll_wait (%i)", ret
);
809 for (i
= 0; i
< fd_count
; i
++) {
810 int fd
= LTTNG_POLL_GETFD(&thread
.events
, i
);
811 uint32_t revents
= LTTNG_POLL_GETEV(&thread
.events
, i
);
813 DBG("Handling fd (%i) activity (%u)",
816 if (revents
& LPOLLERR
) {
817 ERR("Polling returned an error on fd %i", fd
);
821 if (fd
== rotate_notification_channel
->socket
) {
822 ret
= handle_notification_channel(fd
, handle
,
825 ERR("Error occurred while handling activity on notification channel socket");
829 /* Job queue or quit pipe activity. */
832 * The job queue is serviced if there is
833 * activity on the quit pipe to ensure it is
834 * flushed and all references held in the queue
837 ret
= handle_job_queue(handle
, &thread
,
838 handle
->rotation_timer_queue
);
840 ERR("Failed to handle rotation timer pipe event");
844 if (fd
== queue_pipe_fd
) {
847 ret
= lttng_read(fd
, &buf
, 1);
849 ERR("Failed to read from wakeup pipe (fd = %i)", fd
);
853 DBG("Quit pipe activity");
862 fini_thread_state(&thread
);
864 health_unregister(the_health_sessiond
);
865 rcu_thread_offline();
866 rcu_unregister_thread();
871 bool shutdown_rotation_thread(void *thread_data
)
873 struct rotation_thread_handle
*handle
= thread_data
;
874 const int write_fd
= lttng_pipe_get_writefd(handle
->quit_pipe
);
876 return notify_thread_pipe(write_fd
) == 1;
879 bool launch_rotation_thread(struct rotation_thread_handle
*handle
)
881 struct lttng_thread
*thread
;
883 thread
= lttng_thread_create("Rotation",
885 shutdown_rotation_thread
,
891 lttng_thread_put(thread
);