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.hpp>
12 #include <common/config/session-config.hpp>
13 #include <common/defaults.hpp>
14 #include <common/utils.hpp>
15 #include <common/futex.hpp>
16 #include <common/align.hpp>
17 #include <common/time.hpp>
18 #include <common/hashtable/utils.hpp>
24 #include <common/kernel-ctl/kernel-ctl.hpp>
25 #include <lttng/notification/channel-internal.hpp>
26 #include <lttng/rotate-internal.hpp>
27 #include <lttng/location-internal.hpp>
28 #include <lttng/condition/condition-internal.hpp>
29 #include <lttng/notification/notification-internal.hpp>
31 #include "rotation-thread.hpp"
32 #include "lttng-sessiond.hpp"
33 #include "health-sessiond.hpp"
36 #include "session.hpp"
38 #include "notification-thread-commands.hpp"
43 #include <urcu/list.h>
45 struct lttng_notification_channel
*rotate_notification_channel
= NULL
;
47 struct rotation_thread
{
48 struct lttng_poll_event events
;
52 * The timer thread enqueues jobs and wakes up the rotation thread.
53 * When the rotation thread wakes up, it empties the queue.
55 struct rotation_thread_timer_queue
{
56 struct lttng_pipe
*event_pipe
;
57 struct cds_list_head list
;
61 struct rotation_thread_handle
{
62 struct rotation_thread_timer_queue
*rotation_timer_queue
;
63 /* Access to the notification thread cmd_queue */
64 struct notification_thread_handle
*notification_thread_handle
;
65 /* Thread-specific quit pipe. */
66 struct lttng_pipe
*quit_pipe
;
70 struct rotation_thread_job
{
71 enum rotation_thread_job_type type
;
72 struct ltt_session
*session
;
73 /* List member in struct rotation_thread_timer_queue. */
74 struct cds_list_head head
;
79 const char *get_job_type_str(enum rotation_thread_job_type job_type
)
82 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
83 return "CHECK_PENDING_ROTATION";
84 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
85 return "SCHEDULED_ROTATION";
91 struct rotation_thread_timer_queue
*rotation_thread_timer_queue_create(void)
93 struct rotation_thread_timer_queue
*queue
= NULL
;
95 queue
= zmalloc
<rotation_thread_timer_queue
>();
97 PERROR("Failed to allocate timer rotate queue");
101 queue
->event_pipe
= lttng_pipe_open(FD_CLOEXEC
| O_NONBLOCK
);
102 CDS_INIT_LIST_HEAD(&queue
->list
);
103 pthread_mutex_init(&queue
->lock
, NULL
);
108 void rotation_thread_timer_queue_destroy(
109 struct rotation_thread_timer_queue
*queue
)
115 lttng_pipe_destroy(queue
->event_pipe
);
117 pthread_mutex_lock(&queue
->lock
);
118 LTTNG_ASSERT(cds_list_empty(&queue
->list
));
119 pthread_mutex_unlock(&queue
->lock
);
120 pthread_mutex_destroy(&queue
->lock
);
125 * Destroy the thread data previously created by the init function.
127 void rotation_thread_handle_destroy(
128 struct rotation_thread_handle
*handle
)
130 lttng_pipe_destroy(handle
->quit_pipe
);
134 struct rotation_thread_handle
*rotation_thread_handle_create(
135 struct rotation_thread_timer_queue
*rotation_timer_queue
,
136 struct notification_thread_handle
*notification_thread_handle
)
138 struct rotation_thread_handle
*handle
;
140 handle
= zmalloc
<rotation_thread_handle
>();
145 handle
->rotation_timer_queue
= rotation_timer_queue
;
146 handle
->notification_thread_handle
= notification_thread_handle
;
147 handle
->quit_pipe
= lttng_pipe_open(FD_CLOEXEC
);
148 if (!handle
->quit_pipe
) {
155 rotation_thread_handle_destroy(handle
);
160 * Called with the rotation_thread_timer_queue lock held.
161 * Return true if the same timer job already exists in the queue, false if not.
164 bool timer_job_exists(const struct rotation_thread_timer_queue
*queue
,
165 enum rotation_thread_job_type job_type
,
166 struct ltt_session
*session
)
169 struct rotation_thread_job
*job
;
171 cds_list_for_each_entry(job
, &queue
->list
, head
) {
172 if (job
->session
== session
&& job
->type
== job_type
) {
181 void rotation_thread_enqueue_job(struct rotation_thread_timer_queue
*queue
,
182 enum rotation_thread_job_type job_type
,
183 struct ltt_session
*session
)
186 const char dummy
= '!';
187 struct rotation_thread_job
*job
= NULL
;
188 const char *job_type_str
= get_job_type_str(job_type
);
190 pthread_mutex_lock(&queue
->lock
);
191 if (timer_job_exists(queue
, job_type
, session
)) {
193 * This timer job is already pending, we don't need to add
199 job
= zmalloc
<rotation_thread_job
>();
201 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
202 job_type_str
, session
->name
);
205 /* No reason for this to fail as the caller must hold a reference. */
206 (void) session_get(session
);
208 job
->session
= session
;
209 job
->type
= job_type
;
210 cds_list_add_tail(&job
->head
, &queue
->list
);
212 ret
= lttng_write(lttng_pipe_get_writefd(queue
->event_pipe
), &dummy
,
216 * We do not want to block in the timer handler, the job has
217 * been enqueued in the list, the wakeup pipe is probably full,
218 * the job will be processed when the rotation_thread catches
222 DIAGNOSTIC_IGNORE_LOGICAL_OP
223 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
226 * Not an error, but would be surprising and indicate
227 * that the rotation thread can't keep up with the
230 DBG("Wake-up pipe of rotation thread job queue is full");
233 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
234 job_type_str
, session
->name
);
239 pthread_mutex_unlock(&queue
->lock
);
243 int init_poll_set(struct lttng_poll_event
*poll_set
,
244 struct rotation_thread_handle
*handle
)
249 * Create pollset with size 3:
250 * - rotation thread quit pipe,
251 * - rotation thread timer queue pipe,
252 * - notification channel sock,
254 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
259 ret
= lttng_poll_add(poll_set
,
260 lttng_pipe_get_readfd(handle
->quit_pipe
),
263 ERR("Failed to add quit pipe read fd to poll set");
267 ret
= lttng_poll_add(poll_set
,
268 lttng_pipe_get_readfd(handle
->rotation_timer_queue
->event_pipe
),
271 ERR("Failed to add rotate_pending fd to poll set");
277 lttng_poll_clean(poll_set
);
282 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
);
291 int init_thread_state(struct rotation_thread_handle
*handle
,
292 struct rotation_thread
*state
)
296 memset(state
, 0, sizeof(*state
));
297 lttng_poll_init(&state
->events
);
299 ret
= init_poll_set(&state
->events
, handle
);
301 ERR("Failed to initialize rotation thread poll set");
305 rotate_notification_channel
= lttng_notification_channel_create(
306 lttng_session_daemon_notification_endpoint
);
307 if (!rotate_notification_channel
) {
308 ERR("Could not create notification channel");
312 ret
= lttng_poll_add(&state
->events
, rotate_notification_channel
->socket
,
315 ERR("Failed to add notification fd to pollset");
324 void check_session_rotation_pending_on_consumers(struct ltt_session
*session
,
325 bool *_rotation_completed
)
328 struct consumer_socket
*socket
;
329 struct cds_lfht_iter iter
;
330 enum consumer_trace_chunk_exists_status exists_status
;
332 bool chunk_exists_on_peer
= false;
333 enum lttng_trace_chunk_status chunk_status
;
335 LTTNG_ASSERT(session
->chunk_being_archived
);
338 * Check for a local pending rotation on all consumers (32-bit
339 * user space, 64-bit user space, and kernel).
342 if (!session
->ust_session
) {
345 cds_lfht_for_each_entry(session
->ust_session
->consumer
->socks
->ht
,
346 &iter
, socket
, node
.node
) {
347 relayd_id
= session
->ust_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
349 session
->ust_session
->consumer
->net_seq_index
;
351 pthread_mutex_lock(socket
->lock
);
352 ret
= consumer_trace_chunk_exists(socket
,
354 session
->id
, session
->chunk_being_archived
,
357 pthread_mutex_unlock(socket
->lock
);
358 ERR("Error occurred while checking rotation status on consumer daemon");
362 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
363 pthread_mutex_unlock(socket
->lock
);
364 chunk_exists_on_peer
= true;
367 pthread_mutex_unlock(socket
->lock
);
371 if (!session
->kernel_session
) {
374 cds_lfht_for_each_entry(session
->kernel_session
->consumer
->socks
->ht
,
375 &iter
, socket
, node
.node
) {
376 pthread_mutex_lock(socket
->lock
);
377 relayd_id
= session
->kernel_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
379 session
->kernel_session
->consumer
->net_seq_index
;
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
);
402 if (!chunk_exists_on_peer
) {
403 uint64_t chunk_being_archived_id
;
405 chunk_status
= lttng_trace_chunk_get_id(
406 session
->chunk_being_archived
,
407 &chunk_being_archived_id
);
408 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
409 DBG("Rotation of trace archive %" PRIu64
" of session \"%s\" is complete on all consumers",
410 chunk_being_archived_id
,
413 *_rotation_completed
= !chunk_exists_on_peer
;
415 ret
= session_reset_rotation_state(session
,
416 LTTNG_ROTATION_STATE_ERROR
);
418 ERR("Failed to reset rotation state of session \"%s\"",
425 * Check if the last rotation was completed, called with session lock held.
426 * Should only return non-zero in the event of a fatal error. Doing so will
427 * shutdown the thread.
430 int check_session_rotation_pending(struct ltt_session
*session
,
431 struct notification_thread_handle
*notification_thread_handle
)
434 struct lttng_trace_archive_location
*location
;
435 enum lttng_trace_chunk_status chunk_status
;
436 bool rotation_completed
= false;
437 const char *archived_chunk_name
;
438 uint64_t chunk_being_archived_id
;
440 if (!session
->chunk_being_archived
) {
445 chunk_status
= lttng_trace_chunk_get_id(session
->chunk_being_archived
,
446 &chunk_being_archived_id
);
447 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
449 DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64
,
450 session
->name
, chunk_being_archived_id
);
453 * The rotation-pending check timer of a session is launched in
454 * one-shot mode. If the rotation is incomplete, the rotation
455 * thread will re-enable the pending-check timer.
457 * The timer thread can't stop the timer itself since it is involved
458 * in the check for the timer's quiescence.
460 ret
= timer_session_rotation_pending_check_stop(session
);
462 goto check_ongoing_rotation
;
465 check_session_rotation_pending_on_consumers(session
,
466 &rotation_completed
);
467 if (!rotation_completed
||
468 session
->rotation_state
== LTTNG_ROTATION_STATE_ERROR
) {
469 goto check_ongoing_rotation
;
473 * Now we can clear the "ONGOING" state in the session. New
474 * rotations can start now.
476 chunk_status
= lttng_trace_chunk_get_name(session
->chunk_being_archived
,
477 &archived_chunk_name
, NULL
);
478 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
479 free(session
->last_archived_chunk_name
);
480 session
->last_archived_chunk_name
= strdup(archived_chunk_name
);
481 if (!session
->last_archived_chunk_name
) {
482 PERROR("Failed to duplicate archived chunk name");
484 session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_COMPLETED
);
486 if (!session
->quiet_rotation
) {
487 location
= session_get_trace_archive_location(session
);
488 ret
= notification_thread_command_session_rotation_completed(
489 notification_thread_handle
,
493 session
->last_archived_chunk_id
.value
,
495 lttng_trace_archive_location_put(location
);
496 if (ret
!= LTTNG_OK
) {
497 ERR("Failed to notify notification thread of completed rotation for session %s",
503 check_ongoing_rotation
:
504 if (session
->rotation_state
== LTTNG_ROTATION_STATE_ONGOING
) {
505 chunk_status
= lttng_trace_chunk_get_id(
506 session
->chunk_being_archived
,
507 &chunk_being_archived_id
);
508 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
510 DBG("Rotation of trace archive %" PRIu64
" is still pending for session %s",
511 chunk_being_archived_id
, session
->name
);
512 ret
= timer_session_rotation_pending_check_start(session
,
513 DEFAULT_ROTATE_PENDING_TIMER
);
515 ERR("Failed to re-enable rotation pending timer");
525 /* Call with the session and session_list locks held. */
527 int launch_session_rotation(struct ltt_session
*session
)
530 struct lttng_rotate_session_return rotation_return
;
532 DBG("Launching scheduled time-based rotation on session \"%s\"",
535 ret
= cmd_rotate_session(session
, &rotation_return
, false,
536 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
537 if (ret
== LTTNG_OK
) {
538 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
541 /* Don't consider errors as fatal. */
542 DBG("Scheduled time-based rotation aborted for session %s: %s",
543 session
->name
, lttng_strerror(ret
));
549 int run_job(struct rotation_thread_job
*job
, struct ltt_session
*session
,
550 struct notification_thread_handle
*notification_thread_handle
)
555 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION
:
556 ret
= launch_session_rotation(session
);
558 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION
:
559 ret
= check_session_rotation_pending(session
,
560 notification_thread_handle
);
569 int handle_job_queue(struct rotation_thread_handle
*handle
,
570 struct rotation_thread
*state
__attribute__((unused
)),
571 struct rotation_thread_timer_queue
*queue
)
576 struct ltt_session
*session
;
577 struct rotation_thread_job
*job
;
579 /* Take the queue lock only to pop an element from the list. */
580 pthread_mutex_lock(&queue
->lock
);
581 if (cds_list_empty(&queue
->list
)) {
582 pthread_mutex_unlock(&queue
->lock
);
585 job
= cds_list_first_entry(&queue
->list
,
587 cds_list_del(&job
->head
);
588 pthread_mutex_unlock(&queue
->lock
);
591 session
= job
->session
;
593 DBG("Session \"%s\" not found",
594 session
->name
!= NULL
? session
->name
: "");
596 * This is a non-fatal error, and we cannot report it to
597 * the user (timer), so just print the error and
598 * continue the processing.
600 * While the timer thread will purge pending signals for
601 * a session on the session's destruction, it is
602 * possible for a job targeting that session to have
603 * already been queued before it was destroyed.
606 session_put(session
);
607 session_unlock_list();
611 session_lock(session
);
612 ret
= run_job(job
, session
, handle
->notification_thread_handle
);
613 session_unlock(session
);
614 /* Release reference held by the job. */
615 session_put(session
);
616 session_unlock_list();
630 int handle_condition(const struct lttng_notification
*notification
,
631 struct notification_thread_handle
*notification_thread_handle
)
634 const char *condition_session_name
= NULL
;
635 enum lttng_condition_type condition_type
;
636 enum lttng_condition_status condition_status
;
637 enum lttng_evaluation_status evaluation_status
;
639 struct ltt_session
*session
;
640 const struct lttng_condition
*condition
=
641 lttng_notification_get_const_condition(notification
);
642 const struct lttng_evaluation
*evaluation
=
643 lttng_notification_get_const_evaluation(notification
);
645 condition_type
= lttng_condition_get_type(condition
);
647 if (condition_type
!= LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
) {
649 ERR("Condition type and session usage type are not the same");
653 /* Fetch info to test */
654 condition_status
= lttng_condition_session_consumed_size_get_session_name(
655 condition
, &condition_session_name
);
656 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
657 ERR("Session name could not be fetched");
661 evaluation_status
= lttng_evaluation_session_consumed_size_get_consumed_size(evaluation
,
663 if (evaluation_status
!= LTTNG_EVALUATION_STATUS_OK
) {
664 ERR("Failed to get evaluation");
670 session
= session_find_by_name(condition_session_name
);
672 DBG("Failed to find session while handling notification: notification type = %s, session name = `%s`",
673 lttng_condition_type_str(condition_type
),
674 condition_session_name
);
676 * Not a fatal error: a session can be destroyed before we get
677 * the chance to handle the notification.
680 session_unlock_list();
683 session_lock(session
);
685 if (!lttng_trigger_is_equal(session
->rotate_trigger
,
686 lttng_notification_get_const_trigger(notification
))) {
687 /* Notification does not originate from our rotation trigger. */
692 ret
= unsubscribe_session_consumed_size_rotation(session
,
693 notification_thread_handle
);
698 ret
= cmd_rotate_session(session
, NULL
, false,
699 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
700 if (ret
== -LTTNG_ERR_ROTATION_PENDING
) {
701 DBG("Rotate already pending, subscribe to the next threshold value");
702 } else if (ret
!= LTTNG_OK
) {
703 ERR("Failed to rotate on size notification with error: %s",
704 lttng_strerror(ret
));
708 ret
= subscribe_session_consumed_size_rotation(session
,
709 consumed
+ session
->rotate_size
,
710 notification_thread_handle
);
712 ERR("Failed to subscribe to session consumed size condition");
718 session_unlock(session
);
719 session_put(session
);
720 session_unlock_list();
726 int handle_notification_channel(int fd
__attribute__((unused
)),
727 struct rotation_thread_handle
*handle
,
728 struct rotation_thread
*state
__attribute__((unused
)))
731 bool notification_pending
;
732 struct lttng_notification
*notification
= NULL
;
733 enum lttng_notification_channel_status status
;
735 status
= lttng_notification_channel_has_pending_notification(
736 rotate_notification_channel
, ¬ification_pending
);
737 if (status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
738 ERR("Error occurred while checking for pending notification");
743 if (!notification_pending
) {
748 /* Receive the next notification. */
749 status
= lttng_notification_channel_get_next_notification(
750 rotate_notification_channel
,
754 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
:
756 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
:
757 /* Not an error, we will wait for the next one */
760 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED
:
761 ERR("Notification channel was closed");
765 /* Unhandled conditions / errors. */
766 ERR("Unknown notification channel status");
771 ret
= handle_condition(notification
,
772 handle
->notification_thread_handle
);
775 lttng_notification_destroy(notification
);
780 void *thread_rotation(void *data
)
783 struct rotation_thread_handle
*handle
= (rotation_thread_handle
*) data
;
784 struct rotation_thread thread
;
787 DBG("Started rotation thread");
788 rcu_register_thread();
790 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
791 health_code_update();
794 ERR("Invalid thread context provided");
798 queue_pipe_fd
= lttng_pipe_get_readfd(
799 handle
->rotation_timer_queue
->event_pipe
);
802 ret
= init_thread_state(handle
, &thread
);
811 DBG("Entering poll wait");
812 ret
= lttng_poll_wait(&thread
.events
, -1);
813 DBG("Poll wait returned (%i)", ret
);
817 * Restart interrupted system call.
819 if (errno
== EINTR
) {
822 ERR("Error encountered during lttng_poll_wait (%i)", ret
);
827 for (i
= 0; i
< fd_count
; i
++) {
828 int fd
= LTTNG_POLL_GETFD(&thread
.events
, i
);
829 uint32_t revents
= LTTNG_POLL_GETEV(&thread
.events
, i
);
831 DBG("Handling fd (%i) activity (%u)",
834 if (revents
& LPOLLERR
) {
835 ERR("Polling returned an error on fd %i", fd
);
839 if (fd
== rotate_notification_channel
->socket
) {
840 ret
= handle_notification_channel(fd
, handle
,
843 ERR("Error occurred while handling activity on notification channel socket");
847 /* Job queue or quit pipe activity. */
850 * The job queue is serviced if there is
851 * activity on the quit pipe to ensure it is
852 * flushed and all references held in the queue
855 ret
= handle_job_queue(handle
, &thread
,
856 handle
->rotation_timer_queue
);
858 ERR("Failed to handle rotation timer pipe event");
862 if (fd
== queue_pipe_fd
) {
865 ret
= lttng_read(fd
, &buf
, 1);
867 ERR("Failed to read from wakeup pipe (fd = %i)", fd
);
871 DBG("Quit pipe activity");
880 fini_thread_state(&thread
);
882 health_unregister(the_health_sessiond
);
883 rcu_thread_offline();
884 rcu_unregister_thread();
889 bool shutdown_rotation_thread(void *thread_data
)
891 struct rotation_thread_handle
*handle
= (rotation_thread_handle
*) thread_data
;
892 const int write_fd
= lttng_pipe_get_writefd(handle
->quit_pipe
);
894 return notify_thread_pipe(write_fd
) == 1;
897 bool launch_rotation_thread(struct rotation_thread_handle
*handle
)
899 struct lttng_thread
*thread
;
901 thread
= lttng_thread_create("Rotation",
903 shutdown_rotation_thread
,
909 lttng_thread_put(thread
);