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(ltt_session
& session
, bool& _rotation_completed
)
118 struct consumer_socket
*socket
;
119 struct cds_lfht_iter iter
;
120 enum consumer_trace_chunk_exists_status exists_status
;
122 bool chunk_exists_on_peer
= false;
123 enum lttng_trace_chunk_status chunk_status
;
124 lttng::urcu::read_lock_guard read_lock
;
126 LTTNG_ASSERT(session
.chunk_being_archived
);
129 * Check for a local pending rotation on all consumers (32-bit
130 * user space, 64-bit user space, and kernel).
132 if (!session
.ust_session
) {
136 cds_lfht_for_each_entry (
137 session
.ust_session
->consumer
->socks
->ht
, &iter
, socket
, node
.node
) {
138 relayd_id
= session
.ust_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
140 session
.ust_session
->consumer
->net_seq_index
;
142 lttng::pthread::lock_guard
socket_lock(*socket
->lock
);
143 ret
= consumer_trace_chunk_exists(socket
,
146 session
.chunk_being_archived
,
149 ERR("Error occurred while checking rotation status on consumer daemon");
153 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
154 chunk_exists_on_peer
= true;
160 if (!session
.kernel_session
) {
164 cds_lfht_for_each_entry (
165 session
.kernel_session
->consumer
->socks
->ht
, &iter
, socket
, node
.node
) {
166 lttng::pthread::lock_guard
socket_lock(*socket
->lock
);
168 relayd_id
= session
.kernel_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
170 session
.kernel_session
->consumer
->net_seq_index
;
172 ret
= consumer_trace_chunk_exists(socket
,
175 session
.chunk_being_archived
,
178 ERR("Error occurred while checking rotation status on consumer daemon");
182 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
183 chunk_exists_on_peer
= true;
190 if (!chunk_exists_on_peer
) {
191 uint64_t chunk_being_archived_id
;
193 chunk_status
= lttng_trace_chunk_get_id(session
.chunk_being_archived
,
194 &chunk_being_archived_id
);
195 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
196 DBG("Rotation of trace archive %" PRIu64
197 " of session \"%s\" is complete on all consumers",
198 chunk_being_archived_id
,
202 _rotation_completed
= !chunk_exists_on_peer
;
204 ret
= session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_ERROR
);
206 ERR("Failed to reset rotation state of session \"%s\"", session
.name
);
212 * Check if the last rotation was completed, called with session lock held.
213 * Should only return non-zero in the event of a fatal error. Doing so will
214 * shutdown the thread.
216 int check_session_rotation_pending(ltt_session
& session
,
217 notification_thread_handle
& notification_thread_handle
)
220 struct lttng_trace_archive_location
*location
;
221 enum lttng_trace_chunk_status chunk_status
;
222 bool rotation_completed
= false;
223 const char *archived_chunk_name
;
224 uint64_t chunk_being_archived_id
;
226 if (!session
.chunk_being_archived
) {
232 lttng_trace_chunk_get_id(session
.chunk_being_archived
, &chunk_being_archived_id
);
233 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
235 DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64
,
237 chunk_being_archived_id
);
240 * The rotation-pending check timer of a session is launched in
241 * one-shot mode. If the rotation is incomplete, the rotation
242 * thread will re-enable the pending-check timer.
244 * The timer thread can't stop the timer itself since it is involved
245 * in the check for the timer's quiescence.
247 ret
= timer_session_rotation_pending_check_stop(session
);
249 goto check_ongoing_rotation
;
252 check_session_rotation_pending_on_consumers(session
, rotation_completed
);
253 if (!rotation_completed
|| session
.rotation_state
== LTTNG_ROTATION_STATE_ERROR
) {
254 goto check_ongoing_rotation
;
258 * Now we can clear the "ONGOING" state in the session. New
259 * rotations can start now.
261 chunk_status
= lttng_trace_chunk_get_name(
262 session
.chunk_being_archived
, &archived_chunk_name
, nullptr);
263 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
264 free(session
.last_archived_chunk_name
);
265 session
.last_archived_chunk_name
= strdup(archived_chunk_name
);
266 if (!session
.last_archived_chunk_name
) {
267 PERROR("Failed to duplicate archived chunk name");
270 session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_COMPLETED
);
272 if (!session
.quiet_rotation
) {
273 location
= session_get_trace_archive_location(&session
);
274 ret
= notification_thread_command_session_rotation_completed(
275 ¬ification_thread_handle
,
277 session
.last_archived_chunk_id
.value
,
279 lttng_trace_archive_location_put(location
);
280 if (ret
!= LTTNG_OK
) {
281 ERR("Failed to notify notification thread of completed rotation for session %s",
287 check_ongoing_rotation
:
288 if (session
.rotation_state
== LTTNG_ROTATION_STATE_ONGOING
) {
289 chunk_status
= lttng_trace_chunk_get_id(session
.chunk_being_archived
,
290 &chunk_being_archived_id
);
291 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
293 DBG("Rotation of trace archive %" PRIu64
" is still pending for session %s",
294 chunk_being_archived_id
,
296 ret
= timer_session_rotation_pending_check_start(&session
,
297 DEFAULT_ROTATE_PENDING_TIMER
);
299 ERR("Failed to re-enable rotation pending timer");
309 /* Call with the session and session_list locks held. */
310 int launch_session_rotation(ltt_session
& session
)
313 struct lttng_rotate_session_return rotation_return
;
315 DBG("Launching scheduled time-based rotation on session \"%s\"", session
.name
);
317 ASSERT_SESSION_LIST_LOCKED();
318 ASSERT_LOCKED(session
.lock
);
320 ret
= cmd_rotate_session(&session
,
323 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
324 if (ret
== LTTNG_OK
) {
325 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
328 /* Don't consider errors as fatal. */
329 DBG("Scheduled time-based rotation aborted for session %s: %s",
331 lttng_strerror(ret
));
337 int run_job(const rotation_thread_job
& job
,
338 ltt_session
& session
,
339 notification_thread_handle
& notification_thread_handle
)
344 case ls::rotation_thread_job_type::SCHEDULED_ROTATION
:
345 ret
= launch_session_rotation(session
);
347 case ls::rotation_thread_job_type::CHECK_PENDING_ROTATION
:
348 ret
= check_session_rotation_pending(session
, notification_thread_handle
);
357 bool shutdown_rotation_thread(void *thread_data
)
359 auto *handle
= reinterpret_cast<const ls::rotation_thread
*>(thread_data
);
361 return handle
->shutdown();
365 ls::rotation_thread_timer_queue
*ls::rotation_thread_timer_queue_create()
367 auto queue
= zmalloc
<ls::rotation_thread_timer_queue
>();
369 PERROR("Failed to allocate timer rotate queue");
373 queue
->event_pipe
= lttng_pipe_open(FD_CLOEXEC
| O_NONBLOCK
);
374 CDS_INIT_LIST_HEAD(&queue
->list
);
375 pthread_mutex_init(&queue
->lock
, nullptr);
380 void ls::rotation_thread_timer_queue_destroy(struct rotation_thread_timer_queue
*queue
)
386 lttng_pipe_destroy(queue
->event_pipe
);
389 lttng::pthread::lock_guard
queue_lock(queue
->lock
);
391 LTTNG_ASSERT(cds_list_empty(&queue
->list
));
394 pthread_mutex_destroy(&queue
->lock
);
398 ls::rotation_thread::rotation_thread(rotation_thread_timer_queue
& rotation_timer_queue
,
399 notification_thread_handle
& notification_thread_handle
) :
400 _rotation_timer_queue(rotation_timer_queue
),
401 _notification_thread_handle(notification_thread_handle
)
403 _quit_pipe
.reset([]() {
404 auto raw_pipe
= lttng_pipe_open(FD_CLOEXEC
);
406 LTTNG_THROW_POSIX("Failed to rotation thread's quit pipe", errno
);
412 _notification_channel
.reset([]() {
413 auto channel
= lttng_notification_channel_create(
414 lttng_session_daemon_notification_endpoint
);
417 "Failed to create notification channel of rotation thread");
423 lttng_poll_init(&_events
);
426 * Create pollset with size 4:
427 * - rotation thread quit pipe,
428 * - rotation thread timer queue pipe,
429 * - notification channel sock,
430 * - subscribtion change event fd
432 if (lttng_poll_create(&_events
, 4, LTTNG_CLOEXEC
) < 0) {
433 LTTNG_THROW_ERROR("Failed to create poll object for rotation thread");
436 if (lttng_poll_add(&_events
, lttng_pipe_get_readfd(_quit_pipe
.get()), LPOLLIN
) < 0) {
437 LTTNG_THROW_ERROR("Failed to add quit pipe read fd to poll set");
440 if (lttng_poll_add(&_events
,
441 lttng_pipe_get_readfd(_rotation_timer_queue
.event_pipe
),
443 LTTNG_THROW_ERROR("Failed to add rotation timer queue event pipe fd to poll set");
446 if (lttng_poll_add(&_events
,
447 _notification_channel_subscribtion_change_eventfd
.fd(),
450 "Failed to add rotation thread notification channel subscription change eventfd to poll set");
453 if (lttng_poll_add(&_events
, _notification_channel
->socket
, LPOLLIN
) < 0) {
454 LTTNG_THROW_ERROR("Failed to add notification channel socket fd to pollset");
458 ls::rotation_thread::~rotation_thread()
460 lttng_poll_clean(&_events
);
463 void ls::rotation_thread_enqueue_job(ls::rotation_thread_timer_queue
*queue
,
464 ls::rotation_thread_job_type job_type
,
465 ltt_session
*session
)
467 const char dummy
= '!';
468 struct rotation_thread_job
*job
= nullptr;
469 const char *job_type_str
= get_job_type_str(job_type
);
470 lttng::pthread::lock_guard
queue_lock(queue
->lock
);
472 if (timer_job_exists(queue
, job_type
, session
)) {
474 * This timer job is already pending, we don't need to add
480 job
= zmalloc
<rotation_thread_job
>();
482 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
488 /* No reason for this to fail as the caller must hold a reference. */
489 (void) session_get(session
);
491 job
->session
= session
;
492 job
->type
= job_type
;
493 cds_list_add_tail(&job
->head
, &queue
->list
);
495 const int write_ret
=
496 lttng_write(lttng_pipe_get_writefd(queue
->event_pipe
), &dummy
, sizeof(dummy
));
499 * We do not want to block in the timer handler, the job has
500 * been enqueued in the list, the wakeup pipe is probably full,
501 * the job will be processed when the rotation_thread catches
505 DIAGNOSTIC_IGNORE_LOGICAL_OP
506 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
509 * Not an error, but would be surprising and indicate
510 * that the rotation thread can't keep up with the
513 DBG("Wake-up pipe of rotation thread job queue is full");
517 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
524 void ls::rotation_thread::_handle_job_queue()
527 rotation_thread_job::uptr job
;
530 /* Take the queue lock only to pop an element from the list. */
531 lttng::pthread::lock_guard
rotation_timer_queue_lock(
532 _rotation_timer_queue
.lock
);
533 if (cds_list_empty(&_rotation_timer_queue
.list
)) {
537 job
.reset(cds_list_first_entry(
538 &_rotation_timer_queue
.list
, typeof(rotation_thread_job
), head
));
539 cds_list_del(&job
->head
);
543 const auto unlock_list
=
544 lttng::make_scope_exit([]() noexcept
{ session_unlock_list(); });
546 /* locked_ptr will unlock the session and release the ref held by the job. */
547 session_lock(job
->session
);
548 auto session
= ltt_session::locked_ptr(job
->session
);
550 if (run_job(*job
, *session
, _notification_thread_handle
)) {
556 void ls::rotation_thread::_handle_notification(const lttng_notification
& notification
)
559 const char *condition_session_name
= nullptr;
560 enum lttng_condition_status condition_status
;
561 enum lttng_evaluation_status evaluation_status
;
563 auto *condition
= lttng_notification_get_const_condition(¬ification
);
564 auto *evaluation
= lttng_notification_get_const_evaluation(¬ification
);
565 const auto condition_type
= lttng_condition_get_type(condition
);
567 if (condition_type
!= LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
) {
568 LTTNG_THROW_ERROR("Unexpected condition type");
571 /* Fetch info to test. */
572 condition_status
= lttng_condition_session_consumed_size_get_session_name(
573 condition
, &condition_session_name
);
574 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
575 LTTNG_THROW_ERROR("Session name could not be fetched from notification");
579 lttng_evaluation_session_consumed_size_get_consumed_size(evaluation
, &consumed
);
580 if (evaluation_status
!= LTTNG_EVALUATION_STATUS_OK
) {
581 LTTNG_THROW_ERROR("Failed to get consumed size from evaluation");
584 DBG_FMT("Handling session consumed size condition: session_name=`{}`, consumed_size={}",
585 condition_session_name
,
589 const auto unlock_list
= lttng::make_scope_exit([]() noexcept
{ session_unlock_list(); });
591 ltt_session::locked_ptr session
{ [&condition_session_name
]() {
592 auto raw_session_ptr
= session_find_by_name(condition_session_name
);
594 if (raw_session_ptr
) {
595 session_lock(raw_session_ptr
);
598 return raw_session_ptr
;
601 DBG_FMT("Failed to find session while handling notification: notification_type={}, session name=`{}`",
602 lttng_condition_type_str(condition_type
),
603 condition_session_name
);
605 * Not a fatal error: a session can be destroyed before we get
606 * the chance to handle the notification.
611 if (!lttng_trigger_is_equal(session
->rotate_trigger
,
612 lttng_notification_get_const_trigger(¬ification
))) {
613 DBG("Notification does not originate from the internal size-based scheduled rotation trigger, skipping");
617 unsubscribe_session_consumed_size_rotation(*session
);
619 ret
= cmd_rotate_session(
620 session
.get(), nullptr, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
624 case -LTTNG_ERR_ROTATION_PENDING
:
625 DBG("Rotate already pending, subscribe to the next threshold value");
627 case -LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
628 DBG("Rotation already happened since last stop, subscribe to the next threshold value");
630 case -LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
631 DBG("Rotation already happened since last stop and clear, subscribe to the next threshold value");
634 LTTNG_THROW_CTL("Failed to rotate on consumed size notification",
635 static_cast<lttng_error_code
>(-ret
));
638 subscribe_session_consumed_size_rotation(*session
, consumed
+ session
->rotate_size
);
641 void ls::rotation_thread::_handle_notification_channel_activity()
643 bool notification_pending
= true;
646 * A notification channel may have multiple notifications queued-up internally in
647 * its buffers. This is because a notification channel multiplexes command replies
648 * and notifications. The current protocol specifies that multiple notifications can be
649 * received before the reply to a command.
651 * In such cases, the notification channel client implementation internally queues them and
652 * provides them on the next calls to lttng_notification_channel_get_next_notification().
653 * This is correct with respect to the public API, which is intended to be used in "blocking
656 * However, this internal user relies on poll/epoll to wake-up when data is available
657 * on the notification channel's socket. As such, it can't assume that a wake-up means only
658 * one notification is available for consumption since many of them may have been queued in
659 * the channel's internal buffers.
661 while (notification_pending
) {
662 const auto pending_status
= lttng_notification_channel_has_pending_notification(
663 _notification_channel
.get(), ¬ification_pending
);
664 if (pending_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
665 LTTNG_THROW_ERROR("Error occurred while checking for pending notification");
668 if (!notification_pending
) {
672 /* Receive the next notification. */
673 lttng_notification::uptr notification
;
674 enum lttng_notification_channel_status next_notification_status
;
677 struct lttng_notification
*raw_notification_ptr
;
679 next_notification_status
= lttng_notification_channel_get_next_notification(
680 _notification_channel
.get(), &raw_notification_ptr
);
681 notification
.reset(raw_notification_ptr
);
684 switch (next_notification_status
) {
685 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
:
687 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
:
688 WARN("Dropped notification detected on notification channel used by the rotation management thread.");
690 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED
:
691 LTTNG_THROW_ERROR("Notification channel was closed");
693 /* Unhandled conditions / errors. */
694 LTTNG_THROW_ERROR("Unknown notification channel status");
697 _handle_notification(*notification
);
701 void ls::rotation_thread::_thread_function() noexcept
703 DBG("Started rotation thread");
707 } catch (const std::exception
& e
) {
708 ERR_FMT("Fatal rotation thread error: {}", e
.what());
714 void ls::rotation_thread::_run()
716 rcu_register_thread();
717 const auto unregister_rcu_thread
=
718 lttng::make_scope_exit([]() noexcept
{ rcu_unregister_thread(); });
721 const auto offline_rcu_thread
=
722 lttng::make_scope_exit([]() noexcept
{ rcu_thread_offline(); });
724 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
725 health_code_update();
726 const auto unregister_health
=
727 lttng::make_scope_exit([]() noexcept
{ health_unregister(the_health_sessiond
); });
729 const auto queue_pipe_fd
= lttng_pipe_get_readfd(_rotation_timer_queue
.event_pipe
);
733 DBG("Entering poll wait");
734 auto poll_wait_ret
= lttng_poll_wait(&_events
, -1);
735 DBG_FMT("Poll wait returned: ret={}", poll_wait_ret
);
737 if (poll_wait_ret
< 0) {
739 * Restart interrupted system call.
741 if (errno
== EINTR
) {
745 LTTNG_THROW_POSIX("Error encountered during lttng_poll_wait", errno
);
748 const auto fd_count
= poll_wait_ret
;
749 for (int i
= 0; i
< fd_count
; i
++) {
750 const auto fd
= LTTNG_POLL_GETFD(&_events
, i
);
751 const auto revents
= LTTNG_POLL_GETEV(&_events
, i
);
753 DBG_FMT("Handling descriptor activity: fd={}, events={:b}", fd
, revents
);
755 if (revents
& LPOLLERR
) {
756 LTTNG_THROW_ERROR(lttng::format(
757 "Polling returned an error on fd: fd={}", fd
));
760 if (fd
== _notification_channel
->socket
||
761 fd
== _notification_channel_subscribtion_change_eventfd
.fd()) {
763 _handle_notification_channel_activity();
764 } catch (const lttng::ctl::error
& e
) {
766 * The only non-fatal error (rotation failed), others
767 * are caught at the top-level.
769 DBG_FMT("Control error occurred while handling activity on notification channel socket: {}",
774 if (fd
== _notification_channel_subscribtion_change_eventfd
.fd()) {
775 _notification_channel_subscribtion_change_eventfd
779 /* Job queue or quit pipe activity. */
782 * The job queue is serviced if there is
783 * activity on the quit pipe to ensure it is
784 * flushed and all references held in the queue
788 if (fd
== queue_pipe_fd
) {
791 if (lttng_read(fd
, &buf
, 1) != 1) {
794 "Failed to read from wakeup pipe: fd={}",
799 DBG("Quit pipe activity");
807 bool ls::rotation_thread::shutdown() const noexcept
809 const int write_fd
= lttng_pipe_get_writefd(_quit_pipe
.get());
811 return notify_thread_pipe(write_fd
) == 1;
814 void ls::rotation_thread::launch_thread()
816 auto thread
= lttng_thread_create(
819 auto handle
= reinterpret_cast<rotation_thread
*>(ptr
);
821 handle
->_thread_function();
822 return static_cast<void *>(nullptr);
824 shutdown_rotation_thread
,
828 LTTNG_THROW_ERROR("Failed to launch rotation thread");
831 lttng_thread_put(thread
);
834 void ls::rotation_thread::subscribe_session_consumed_size_rotation(ltt_session
& session
,
837 const struct lttng_credentials session_creds
= {
838 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session
.uid
),
839 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session
.gid
),
842 ASSERT_LOCKED(session
.lock
);
844 auto rotate_condition
= lttng::make_unique_wrapper
<lttng_condition
, lttng_condition_put
>(
845 lttng_condition_session_consumed_size_create());
846 if (!rotate_condition
) {
847 LTTNG_THROW_POSIX("Failed to create session consumed size condition object", errno
);
850 auto condition_status
=
851 lttng_condition_session_consumed_size_set_threshold(rotate_condition
.get(), size
);
852 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
853 LTTNG_THROW_ERROR(lttng::format(
854 "Could not set session consumed size condition threshold: size={}", size
));
857 condition_status
= lttng_condition_session_consumed_size_set_session_name(
858 rotate_condition
.get(), session
.name
);
859 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
860 LTTNG_THROW_ERROR(lttng::format(
861 "Could not set session consumed size condition session name: name=`{}`",
865 auto notify_action
= lttng::make_unique_wrapper
<lttng_action
, lttng_action_put
>(
866 lttng_action_notify_create());
867 if (!notify_action
) {
868 LTTNG_THROW_POSIX("Could not create notify action", errno
);
871 LTTNG_ASSERT(!session
.rotate_trigger
);
872 /* trigger acquires its own reference to condition and action on success. */
873 auto trigger
= lttng::make_unique_wrapper
<lttng_trigger
, lttng_trigger_put
>(
874 lttng_trigger_create(rotate_condition
.get(), notify_action
.get()));
876 LTTNG_THROW_POSIX("Could not create size-based rotation trigger", errno
);
879 /* Ensure this trigger is not visible to external users. */
880 lttng_trigger_set_hidden(trigger
.get());
881 lttng_trigger_set_credentials(trigger
.get(), &session_creds
);
883 auto nc_status
= lttng_notification_channel_subscribe(_notification_channel
.get(),
884 rotate_condition
.get());
885 if (nc_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
886 LTTNG_THROW_ERROR("Could not subscribe to session consumed size notification");
890 * Ensure any notification queued during the subscription are consumed by queueing an
893 _notification_channel_subscribtion_change_eventfd
.increment();
895 const auto register_ret
= notification_thread_command_register_trigger(
896 &_notification_thread_handle
, trigger
.get(), true);
897 if (register_ret
!= LTTNG_OK
) {
900 "Failed to register trigger for automatic size-based rotation: session_name{}, size={}",
906 /* Ownership transferred to the session. */
907 session
.rotate_trigger
= trigger
.release();
910 void ls::rotation_thread::unsubscribe_session_consumed_size_rotation(ltt_session
& session
)
912 LTTNG_ASSERT(session
.rotate_trigger
);
914 const auto remove_session_trigger
= lttng::make_scope_exit([&session
]() noexcept
{
915 lttng_trigger_put(session
.rotate_trigger
);
916 session
.rotate_trigger
= nullptr;
919 const auto unsubscribe_status
= lttng_notification_channel_unsubscribe(
920 _notification_channel
.get(),
921 lttng_trigger_get_const_condition(session
.rotate_trigger
));
922 if (unsubscribe_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
923 LTTNG_THROW_ERROR(lttng::format(
924 "Failed to unsubscribe from consumed size condition used to control automatic size-based rotations: session_name=`{}` return_code={}",
926 static_cast<int>(unsubscribe_status
)));
930 * Ensure any notification queued during the un-subscription are consumed by queueing an
933 _notification_channel_subscribtion_change_eventfd
.increment();
935 const auto unregister_status
= notification_thread_command_unregister_trigger(
936 &_notification_thread_handle
, session
.rotate_trigger
);
937 if (unregister_status
!= LTTNG_OK
) {
940 "Failed to unregister trigger for automatic size-based rotation: session_name{}",