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"
14 #include "rotation-thread.hpp"
15 #include "session.hpp"
20 #include <common/align.hpp>
21 #include <common/config/session-config.hpp>
22 #include <common/defaults.hpp>
23 #include <common/error.hpp>
24 #include <common/eventfd.hpp>
25 #include <common/exception.hpp>
26 #include <common/file-descriptor.hpp>
27 #include <common/format.hpp>
28 #include <common/futex.hpp>
29 #include <common/hashtable/utils.hpp>
30 #include <common/kernel-ctl/kernel-ctl.hpp>
31 #include <common/locked-reference.hpp>
32 #include <common/make-unique-wrapper.hpp>
33 #include <common/pthread-lock.hpp>
34 #include <common/scope-exit.hpp>
35 #include <common/time.hpp>
36 #include <common/urcu.hpp>
37 #include <common/utils.hpp>
39 #include <lttng/action/action-internal.hpp>
40 #include <lttng/condition/condition-internal.hpp>
41 #include <lttng/location-internal.hpp>
42 #include <lttng/notification/channel-internal.hpp>
43 #include <lttng/notification/notification-internal.hpp>
44 #include <lttng/rotate-internal.hpp>
45 #include <lttng/trigger/trigger.h>
51 #include <sys/eventfd.h>
55 #include <urcu/list.h>
57 namespace ls
= lttng::sessiond
;
60 * The timer thread enqueues jobs and wakes up the rotation thread.
61 * When the rotation thread wakes up, it empties the queue.
63 struct ls::rotation_thread_timer_queue
{
64 struct lttng_pipe
*event_pipe
;
65 struct cds_list_head list
;
70 struct rotation_thread_job
{
72 std::unique_ptr
<rotation_thread_job
,
73 lttng::memory::create_deleter_class
<rotation_thread_job
,
74 lttng::memory::free
>::deleter
>;
76 enum ls::rotation_thread_job_type type
;
77 struct ltt_session
*session
;
78 /* List member in struct rotation_thread_timer_queue. */
79 struct cds_list_head head
;
82 const char *get_job_type_str(enum ls::rotation_thread_job_type job_type
)
85 case ls::rotation_thread_job_type::CHECK_PENDING_ROTATION
:
86 return "CHECK_PENDING_ROTATION";
87 case ls::rotation_thread_job_type::SCHEDULED_ROTATION
:
88 return "SCHEDULED_ROTATION";
95 * Called with the rotation_thread_timer_queue lock held.
96 * Return true if the same timer job already exists in the queue, false if not.
98 bool timer_job_exists(const ls::rotation_thread_timer_queue
*queue
,
99 ls::rotation_thread_job_type job_type
,
100 ltt_session
*session
)
103 struct rotation_thread_job
*job
;
105 cds_list_for_each_entry (job
, &queue
->list
, head
) {
106 if (job
->session
== session
&& job
->type
== job_type
) {
115 void check_session_rotation_pending_on_consumers(const ltt_session::locked_ref
& session
,
116 bool& _rotation_completed
)
119 struct consumer_socket
*socket
;
120 struct cds_lfht_iter iter
;
121 enum consumer_trace_chunk_exists_status exists_status
;
123 bool chunk_exists_on_peer
= false;
124 enum lttng_trace_chunk_status chunk_status
;
125 lttng::urcu::read_lock_guard read_lock
;
127 LTTNG_ASSERT(session
->chunk_being_archived
);
130 * Check for a local pending rotation on all consumers (32-bit
131 * user space, 64-bit user space, and kernel).
133 if (!session
->ust_session
) {
137 cds_lfht_for_each_entry (
138 session
->ust_session
->consumer
->socks
->ht
, &iter
, socket
, node
.node
) {
139 relayd_id
= session
->ust_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
141 session
->ust_session
->consumer
->net_seq_index
;
143 lttng::pthread::lock_guard
socket_lock(*socket
->lock
);
144 ret
= consumer_trace_chunk_exists(socket
,
147 session
->chunk_being_archived
,
150 ERR("Error occurred while checking rotation status on consumer daemon");
154 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
155 chunk_exists_on_peer
= true;
161 if (!session
->kernel_session
) {
165 cds_lfht_for_each_entry (
166 session
->kernel_session
->consumer
->socks
->ht
, &iter
, socket
, node
.node
) {
167 lttng::pthread::lock_guard
socket_lock(*socket
->lock
);
169 relayd_id
= session
->kernel_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
171 session
->kernel_session
->consumer
->net_seq_index
;
173 ret
= consumer_trace_chunk_exists(socket
,
176 session
->chunk_being_archived
,
179 ERR("Error occurred while checking rotation status on consumer daemon");
183 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
184 chunk_exists_on_peer
= true;
191 if (!chunk_exists_on_peer
) {
192 uint64_t chunk_being_archived_id
;
194 chunk_status
= lttng_trace_chunk_get_id(session
->chunk_being_archived
,
195 &chunk_being_archived_id
);
196 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
197 DBG("Rotation of trace archive %" PRIu64
198 " of session \"%s\" is complete on all consumers",
199 chunk_being_archived_id
,
203 _rotation_completed
= !chunk_exists_on_peer
;
205 ret
= session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_ERROR
);
207 ERR("Failed to reset rotation state of session \"%s\"", session
->name
);
213 * Check if the last rotation was completed, called with session lock held.
214 * Should only return non-zero in the event of a fatal error. Doing so will
215 * shutdown the thread.
217 int check_session_rotation_pending(const ltt_session::locked_ref
& session
,
218 notification_thread_handle
& notification_thread_handle
)
221 struct lttng_trace_archive_location
*location
;
222 enum lttng_trace_chunk_status chunk_status
;
223 bool rotation_completed
= false;
224 const char *archived_chunk_name
;
225 uint64_t chunk_being_archived_id
;
227 if (!session
->chunk_being_archived
) {
233 lttng_trace_chunk_get_id(session
->chunk_being_archived
, &chunk_being_archived_id
);
234 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
236 DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64
,
238 chunk_being_archived_id
);
241 * The rotation-pending check timer of a session is launched in
242 * one-shot mode. If the rotation is incomplete, the rotation
243 * thread will re-enable the pending-check timer.
245 * The timer thread can't stop the timer itself since it is involved
246 * in the check for the timer's quiescence.
248 ret
= timer_session_rotation_pending_check_stop(session
);
250 goto check_ongoing_rotation
;
253 check_session_rotation_pending_on_consumers(session
, rotation_completed
);
254 if (!rotation_completed
|| session
->rotation_state
== LTTNG_ROTATION_STATE_ERROR
) {
255 goto check_ongoing_rotation
;
259 * Now we can clear the "ONGOING" state in the session. New
260 * rotations can start now.
262 chunk_status
= lttng_trace_chunk_get_name(
263 session
->chunk_being_archived
, &archived_chunk_name
, nullptr);
264 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
265 free(session
->last_archived_chunk_name
);
266 session
->last_archived_chunk_name
= strdup(archived_chunk_name
);
267 if (!session
->last_archived_chunk_name
) {
268 PERROR("Failed to duplicate archived chunk name");
271 session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_COMPLETED
);
273 if (!session
->quiet_rotation
) {
274 location
= session_get_trace_archive_location(session
);
275 ret
= notification_thread_command_session_rotation_completed(
276 ¬ification_thread_handle
,
278 session
->last_archived_chunk_id
.value
,
280 lttng_trace_archive_location_put(location
);
281 if (ret
!= LTTNG_OK
) {
282 ERR("Failed to notify notification thread of completed rotation for session %s",
288 check_ongoing_rotation
:
289 if (session
->rotation_state
== LTTNG_ROTATION_STATE_ONGOING
) {
290 chunk_status
= lttng_trace_chunk_get_id(session
->chunk_being_archived
,
291 &chunk_being_archived_id
);
292 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
294 DBG("Rotation of trace archive %" PRIu64
" is still pending for session %s",
295 chunk_being_archived_id
,
297 ret
= timer_session_rotation_pending_check_start(session
,
298 DEFAULT_ROTATE_PENDING_TIMER
);
300 ERR("Failed to re-enable rotation pending timer");
310 /* Call with the session and session_list locks held. */
311 void launch_session_rotation(const ltt_session::locked_ref
& session
)
315 DBG_FMT("Launching scheduled time-based rotation: session_name='{}'", session
->name
);
317 ASSERT_SESSION_LIST_LOCKED();
319 ret
= cmd_rotate_session(
320 session
, nullptr, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
321 if (ret
!= LTTNG_OK
) {
322 LTTNG_THROW_CTL(fmt::format("Failed to launch session rotation: session_name={}",
324 static_cast<lttng_error_code
>(ret
));
326 /* Don't consider errors as fatal. */
327 DBG_FMT("Scheduled time-based rotation aborted session_name=`{}`, error='{}'",
329 lttng_strerror(ret
));
333 int run_job(const rotation_thread_job
& job
,
334 const ltt_session::locked_ref
& session
,
335 notification_thread_handle
& notification_thread_handle
)
340 case ls::rotation_thread_job_type::SCHEDULED_ROTATION
:
342 launch_session_rotation(session
);
343 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
345 } catch (const lttng::ctl::error
& ctl_ex
) {
346 /* Don't consider errors as fatal. */
347 DBG("Scheduled time-based rotation aborted for session %s: %s",
349 lttng_strerror(ctl_ex
.code()));
352 case ls::rotation_thread_job_type::CHECK_PENDING_ROTATION
:
353 ret
= check_session_rotation_pending(session
, notification_thread_handle
);
362 bool shutdown_rotation_thread(void *thread_data
)
364 auto *handle
= reinterpret_cast<const ls::rotation_thread
*>(thread_data
);
366 return handle
->shutdown();
370 ls::rotation_thread_timer_queue
*ls::rotation_thread_timer_queue_create()
372 auto queue
= zmalloc
<ls::rotation_thread_timer_queue
>();
374 PERROR("Failed to allocate timer rotate queue");
378 queue
->event_pipe
= lttng_pipe_open(FD_CLOEXEC
| O_NONBLOCK
);
379 CDS_INIT_LIST_HEAD(&queue
->list
);
380 pthread_mutex_init(&queue
->lock
, nullptr);
385 void ls::rotation_thread_timer_queue_destroy(struct rotation_thread_timer_queue
*queue
)
391 lttng_pipe_destroy(queue
->event_pipe
);
394 lttng::pthread::lock_guard
queue_lock(queue
->lock
);
396 LTTNG_ASSERT(cds_list_empty(&queue
->list
));
399 pthread_mutex_destroy(&queue
->lock
);
403 ls::rotation_thread::rotation_thread(rotation_thread_timer_queue
& rotation_timer_queue
,
404 notification_thread_handle
& notification_thread_handle
) :
405 _rotation_timer_queue(rotation_timer_queue
),
406 _notification_thread_handle(notification_thread_handle
)
408 _quit_pipe
.reset([]() {
409 auto raw_pipe
= lttng_pipe_open(FD_CLOEXEC
);
411 LTTNG_THROW_POSIX("Failed to rotation thread's quit pipe", errno
);
417 _notification_channel
.reset([]() {
418 auto channel
= lttng_notification_channel_create(
419 lttng_session_daemon_notification_endpoint
);
422 "Failed to create notification channel of rotation thread");
428 lttng_poll_init(&_events
);
431 * Create pollset with size 4:
432 * - rotation thread quit pipe,
433 * - rotation thread timer queue pipe,
434 * - notification channel sock,
435 * - subscribtion change event fd
437 if (lttng_poll_create(&_events
, 4, LTTNG_CLOEXEC
) < 0) {
438 LTTNG_THROW_ERROR("Failed to create poll object for rotation thread");
441 if (lttng_poll_add(&_events
, lttng_pipe_get_readfd(_quit_pipe
.get()), LPOLLIN
) < 0) {
442 LTTNG_THROW_ERROR("Failed to add quit pipe read fd to poll set");
445 if (lttng_poll_add(&_events
,
446 lttng_pipe_get_readfd(_rotation_timer_queue
.event_pipe
),
448 LTTNG_THROW_ERROR("Failed to add rotation timer queue event pipe fd to poll set");
451 if (lttng_poll_add(&_events
,
452 _notification_channel_subscribtion_change_eventfd
.fd(),
455 "Failed to add rotation thread notification channel subscription change eventfd to poll set");
458 if (lttng_poll_add(&_events
, _notification_channel
->socket
, LPOLLIN
) < 0) {
459 LTTNG_THROW_ERROR("Failed to add notification channel socket fd to pollset");
463 ls::rotation_thread::~rotation_thread()
465 lttng_poll_clean(&_events
);
468 void ls::rotation_thread_enqueue_job(ls::rotation_thread_timer_queue
*queue
,
469 ls::rotation_thread_job_type job_type
,
470 ltt_session
*session
)
472 const char dummy
= '!';
473 struct rotation_thread_job
*job
= nullptr;
474 const char *job_type_str
= get_job_type_str(job_type
);
475 lttng::pthread::lock_guard
queue_lock(queue
->lock
);
477 if (timer_job_exists(queue
, job_type
, session
)) {
479 * This timer job is already pending, we don't need to add
485 job
= zmalloc
<rotation_thread_job
>();
487 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
493 /* No reason for this to fail as the caller must hold a reference. */
494 (void) session_get(session
);
496 job
->session
= session
;
497 job
->type
= job_type
;
498 cds_list_add_tail(&job
->head
, &queue
->list
);
500 const int write_ret
=
501 lttng_write(lttng_pipe_get_writefd(queue
->event_pipe
), &dummy
, sizeof(dummy
));
504 * We do not want to block in the timer handler, the job has
505 * been enqueued in the list, the wakeup pipe is probably full,
506 * the job will be processed when the rotation_thread catches
510 DIAGNOSTIC_IGNORE_LOGICAL_OP
511 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
514 * Not an error, but would be surprising and indicate
515 * that the rotation thread can't keep up with the
518 DBG("Wake-up pipe of rotation thread job queue is full");
522 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
529 void ls::rotation_thread::_handle_job_queue()
532 rotation_thread_job::uptr job
;
535 /* Take the queue lock only to pop an element from the list. */
536 lttng::pthread::lock_guard
rotation_timer_queue_lock(
537 _rotation_timer_queue
.lock
);
538 if (cds_list_empty(&_rotation_timer_queue
.list
)) {
542 job
.reset(cds_list_first_entry(
543 &_rotation_timer_queue
.list
, typeof(rotation_thread_job
), head
));
544 cds_list_del(&job
->head
);
547 const auto list_lock
= lttng::sessiond::lock_session_list();
549 /* locked_ref will unlock the session and release the ref held by the job. */
550 session_lock(job
->session
);
551 auto session
= ltt_session::locked_ref(*job
->session
);
553 if (run_job(*job
, session
, _notification_thread_handle
)) {
559 void ls::rotation_thread::_handle_notification(const lttng_notification
& notification
)
562 const char *condition_session_name
= nullptr;
563 enum lttng_condition_status condition_status
;
564 enum lttng_evaluation_status evaluation_status
;
566 auto *condition
= lttng_notification_get_const_condition(¬ification
);
567 auto *evaluation
= lttng_notification_get_const_evaluation(¬ification
);
568 const auto condition_type
= lttng_condition_get_type(condition
);
570 if (condition_type
!= LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
) {
571 LTTNG_THROW_ERROR("Unexpected condition type");
574 /* Fetch info to test. */
575 condition_status
= lttng_condition_session_consumed_size_get_session_name(
576 condition
, &condition_session_name
);
577 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
578 LTTNG_THROW_ERROR("Session name could not be fetched from notification");
582 lttng_evaluation_session_consumed_size_get_consumed_size(evaluation
, &consumed
);
583 if (evaluation_status
!= LTTNG_EVALUATION_STATUS_OK
) {
584 LTTNG_THROW_ERROR("Failed to get consumed size from evaluation");
587 DBG_FMT("Handling session consumed size condition: session_name=`{}`, consumed_size={}",
588 condition_session_name
,
592 * Mind the order of the declaration of list_lock vs session:
593 * the session list lock must always be released _after_ the release of
594 * a session's reference (the destruction of a ref/locked_ref) to ensure
595 * since the reference's release may unpublish the session from the list of
598 const auto list_lock
= lttng::sessiond::lock_session_list();
600 const auto session
= ltt_session::find_locked_session(condition_session_name
);
602 if (!lttng_trigger_is_equal(session
->rotate_trigger
,
603 lttng_notification_get_const_trigger(¬ification
))) {
604 DBG("Notification does not originate from the internal size-based scheduled rotation trigger, skipping");
608 unsubscribe_session_consumed_size_rotation(*session
);
610 ret
= cmd_rotate_session(
611 session
, nullptr, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
612 if (ret
!= LTTNG_OK
) {
616 case -LTTNG_ERR_ROTATION_PENDING
:
617 DBG("Rotate already pending, subscribe to the next threshold value");
619 case -LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
620 DBG("Rotation already happened since last stop, subscribe to the next threshold value");
622 case -LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
623 DBG("Rotation already happened since last stop and clear, subscribe to the next threshold value");
626 LTTNG_THROW_CTL("Failed to rotate on consumed size notification",
627 static_cast<lttng_error_code
>(-ret
));
631 subscribe_session_consumed_size_rotation(*session
, consumed
+ session
->rotate_size
);
632 } catch (const lttng::sessiond::exceptions::session_not_found_error
& ex
) {
633 DBG_FMT("Failed to find session while handling notification: notification_type={}, session name=`{}`",
634 lttng_condition_type_str(condition_type
),
635 condition_session_name
);
637 * Not a fatal error: a session can be destroyed before we get
638 * the chance to handle the notification.
644 void ls::rotation_thread::_handle_notification_channel_activity()
646 bool notification_pending
= true;
649 * A notification channel may have multiple notifications queued-up internally in
650 * its buffers. This is because a notification channel multiplexes command replies
651 * and notifications. The current protocol specifies that multiple notifications can be
652 * received before the reply to a command.
654 * In such cases, the notification channel client implementation internally queues them and
655 * provides them on the next calls to lttng_notification_channel_get_next_notification().
656 * This is correct with respect to the public API, which is intended to be used in "blocking
659 * However, this internal user relies on poll/epoll to wake-up when data is available
660 * on the notification channel's socket. As such, it can't assume that a wake-up means only
661 * one notification is available for consumption since many of them may have been queued in
662 * the channel's internal buffers.
664 while (notification_pending
) {
665 const auto pending_status
= lttng_notification_channel_has_pending_notification(
666 _notification_channel
.get(), ¬ification_pending
);
667 if (pending_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
668 LTTNG_THROW_ERROR("Error occurred while checking for pending notification");
671 if (!notification_pending
) {
675 /* Receive the next notification. */
676 lttng_notification::uptr notification
;
677 enum lttng_notification_channel_status next_notification_status
;
680 struct lttng_notification
*raw_notification_ptr
;
682 next_notification_status
= lttng_notification_channel_get_next_notification(
683 _notification_channel
.get(), &raw_notification_ptr
);
684 notification
.reset(raw_notification_ptr
);
687 switch (next_notification_status
) {
688 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
:
690 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
:
691 WARN("Dropped notification detected on notification channel used by the rotation management thread.");
693 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED
:
694 LTTNG_THROW_ERROR("Notification channel was closed");
696 /* Unhandled conditions / errors. */
697 LTTNG_THROW_ERROR("Unknown notification channel status");
700 _handle_notification(*notification
);
704 void ls::rotation_thread::_thread_function() noexcept
706 DBG("Started rotation thread");
710 } catch (const std::exception
& e
) {
711 ERR_FMT("Fatal rotation thread error: {}", e
.what());
717 void ls::rotation_thread::_run()
719 rcu_register_thread();
720 const auto unregister_rcu_thread
=
721 lttng::make_scope_exit([]() noexcept
{ rcu_unregister_thread(); });
724 const auto offline_rcu_thread
=
725 lttng::make_scope_exit([]() noexcept
{ rcu_thread_offline(); });
727 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
728 health_code_update();
729 const auto unregister_health
=
730 lttng::make_scope_exit([]() noexcept
{ health_unregister(the_health_sessiond
); });
732 const auto queue_pipe_fd
= lttng_pipe_get_readfd(_rotation_timer_queue
.event_pipe
);
736 DBG("Entering poll wait");
737 auto poll_wait_ret
= lttng_poll_wait(&_events
, -1);
738 DBG_FMT("Poll wait returned: ret={}", poll_wait_ret
);
740 if (poll_wait_ret
< 0) {
742 * Restart interrupted system call.
744 if (errno
== EINTR
) {
748 LTTNG_THROW_POSIX("Error encountered during lttng_poll_wait", errno
);
751 const auto fd_count
= poll_wait_ret
;
752 for (int i
= 0; i
< fd_count
; i
++) {
753 const auto fd
= LTTNG_POLL_GETFD(&_events
, i
);
754 const auto revents
= LTTNG_POLL_GETEV(&_events
, i
);
756 DBG_FMT("Handling descriptor activity: fd={}, events={:b}", fd
, revents
);
758 if (revents
& LPOLLERR
) {
759 LTTNG_THROW_ERROR(lttng::format(
760 "Polling returned an error on fd: fd={}", fd
));
763 if (fd
== _notification_channel
->socket
||
764 fd
== _notification_channel_subscribtion_change_eventfd
.fd()) {
766 _handle_notification_channel_activity();
767 } catch (const lttng::ctl::error
& e
) {
769 * The only non-fatal error (rotation failed), others
770 * are caught at the top-level.
772 DBG_FMT("Control error occurred while handling activity on notification channel socket: {}",
777 if (fd
== _notification_channel_subscribtion_change_eventfd
.fd()) {
778 _notification_channel_subscribtion_change_eventfd
782 /* Job queue or quit pipe activity. */
785 * The job queue is serviced if there is
786 * activity on the quit pipe to ensure it is
787 * flushed and all references held in the queue
791 if (fd
== queue_pipe_fd
) {
794 if (lttng_read(fd
, &buf
, 1) != 1) {
797 "Failed to read from wakeup pipe: fd={}",
802 DBG("Quit pipe activity");
810 bool ls::rotation_thread::shutdown() const noexcept
812 const int write_fd
= lttng_pipe_get_writefd(_quit_pipe
.get());
814 return notify_thread_pipe(write_fd
) == 1;
817 void ls::rotation_thread::launch_thread()
819 auto thread
= lttng_thread_create(
822 auto handle
= reinterpret_cast<rotation_thread
*>(ptr
);
824 handle
->_thread_function();
825 return static_cast<void *>(nullptr);
827 shutdown_rotation_thread
,
831 LTTNG_THROW_ERROR("Failed to launch rotation thread");
834 lttng_thread_put(thread
);
837 void ls::rotation_thread::subscribe_session_consumed_size_rotation(ltt_session
& session
,
840 const struct lttng_credentials session_creds
= {
841 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session
.uid
),
842 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session
.gid
),
845 ASSERT_LOCKED(session
._lock
);
847 auto rotate_condition
= lttng::make_unique_wrapper
<lttng_condition
, lttng_condition_put
>(
848 lttng_condition_session_consumed_size_create());
849 if (!rotate_condition
) {
850 LTTNG_THROW_POSIX("Failed to create session consumed size condition object", errno
);
853 auto condition_status
=
854 lttng_condition_session_consumed_size_set_threshold(rotate_condition
.get(), size
);
855 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
856 LTTNG_THROW_ERROR(lttng::format(
857 "Could not set session consumed size condition threshold: size={}", size
));
860 condition_status
= lttng_condition_session_consumed_size_set_session_name(
861 rotate_condition
.get(), session
.name
);
862 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
863 LTTNG_THROW_ERROR(lttng::format(
864 "Could not set session consumed size condition session name: name=`{}`",
868 auto notify_action
= lttng::make_unique_wrapper
<lttng_action
, lttng_action_put
>(
869 lttng_action_notify_create());
870 if (!notify_action
) {
871 LTTNG_THROW_POSIX("Could not create notify action", errno
);
874 LTTNG_ASSERT(!session
.rotate_trigger
);
875 /* trigger acquires its own reference to condition and action on success. */
876 auto trigger
= lttng::make_unique_wrapper
<lttng_trigger
, lttng_trigger_put
>(
877 lttng_trigger_create(rotate_condition
.get(), notify_action
.get()));
879 LTTNG_THROW_POSIX("Could not create size-based rotation trigger", errno
);
882 /* Ensure this trigger is not visible to external users. */
883 lttng_trigger_set_hidden(trigger
.get());
884 lttng_trigger_set_credentials(trigger
.get(), &session_creds
);
886 auto nc_status
= lttng_notification_channel_subscribe(_notification_channel
.get(),
887 rotate_condition
.get());
888 if (nc_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
889 LTTNG_THROW_ERROR("Could not subscribe to session consumed size notification");
893 * Ensure any notification queued during the subscription are consumed by queueing an
896 _notification_channel_subscribtion_change_eventfd
.increment();
898 const auto register_ret
= notification_thread_command_register_trigger(
899 &_notification_thread_handle
, trigger
.get(), true);
900 if (register_ret
!= LTTNG_OK
) {
903 "Failed to register trigger for automatic size-based rotation: session_name{}, size={}",
909 /* Ownership transferred to the session. */
910 session
.rotate_trigger
= trigger
.release();
913 void ls::rotation_thread::unsubscribe_session_consumed_size_rotation(ltt_session
& session
)
915 LTTNG_ASSERT(session
.rotate_trigger
);
917 const auto remove_session_trigger
= lttng::make_scope_exit([&session
]() noexcept
{
918 lttng_trigger_put(session
.rotate_trigger
);
919 session
.rotate_trigger
= nullptr;
922 const auto unsubscribe_status
= lttng_notification_channel_unsubscribe(
923 _notification_channel
.get(),
924 lttng_trigger_get_const_condition(session
.rotate_trigger
));
925 if (unsubscribe_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
926 LTTNG_THROW_ERROR(lttng::format(
927 "Failed to unsubscribe from consumed size condition used to control automatic size-based rotations: session_name=`{}` return_code={}",
929 static_cast<int>(unsubscribe_status
)));
933 * Ensure any notification queued during the un-subscription are consumed by queueing an
936 _notification_channel_subscribtion_change_eventfd
.increment();
938 const auto unregister_status
= notification_thread_command_unregister_trigger(
939 &_notification_thread_handle
, session
.rotate_trigger
);
940 if (unregister_status
!= LTTNG_OK
) {
943 "Failed to unregister trigger for automatic size-based rotation: session_name{}",