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
{
71 using uptr
= std::unique_ptr
<
73 lttng::memory::create_deleter_class
<rotation_thread_job
, lttng::free
>::deleter
>;
75 enum ls::rotation_thread_job_type type
;
76 struct ltt_session
*session
;
77 /* List member in struct rotation_thread_timer_queue. */
78 struct cds_list_head head
;
81 const char *get_job_type_str(enum ls::rotation_thread_job_type job_type
)
84 case ls::rotation_thread_job_type::CHECK_PENDING_ROTATION
:
85 return "CHECK_PENDING_ROTATION";
86 case ls::rotation_thread_job_type::SCHEDULED_ROTATION
:
87 return "SCHEDULED_ROTATION";
94 * Called with the rotation_thread_timer_queue lock held.
95 * Return true if the same timer job already exists in the queue, false if not.
97 bool timer_job_exists(const ls::rotation_thread_timer_queue
*queue
,
98 ls::rotation_thread_job_type job_type
,
102 struct rotation_thread_job
*job
;
104 cds_list_for_each_entry (job
, &queue
->list
, head
) {
105 if (job
->session
== session
&& job
->type
== job_type
) {
114 void check_session_rotation_pending_on_consumers(ltt_session
& session
, bool& _rotation_completed
)
117 struct consumer_socket
*socket
;
118 struct cds_lfht_iter iter
;
119 enum consumer_trace_chunk_exists_status exists_status
;
121 bool chunk_exists_on_peer
= false;
122 enum lttng_trace_chunk_status chunk_status
;
123 lttng::urcu::read_lock_guard read_lock
;
125 LTTNG_ASSERT(session
.chunk_being_archived
);
128 * Check for a local pending rotation on all consumers (32-bit
129 * user space, 64-bit user space, and kernel).
131 if (!session
.ust_session
) {
135 cds_lfht_for_each_entry (
136 session
.ust_session
->consumer
->socks
->ht
, &iter
, socket
, node
.node
) {
137 relayd_id
= session
.ust_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
139 session
.ust_session
->consumer
->net_seq_index
;
141 lttng::pthread::lock_guard
socket_lock(*socket
->lock
);
142 ret
= consumer_trace_chunk_exists(socket
,
145 session
.chunk_being_archived
,
148 ERR("Error occurred while checking rotation status on consumer daemon");
152 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
153 chunk_exists_on_peer
= true;
159 if (!session
.kernel_session
) {
163 cds_lfht_for_each_entry (
164 session
.kernel_session
->consumer
->socks
->ht
, &iter
, socket
, node
.node
) {
165 lttng::pthread::lock_guard
socket_lock(*socket
->lock
);
167 relayd_id
= session
.kernel_session
->consumer
->type
== CONSUMER_DST_LOCAL
?
169 session
.kernel_session
->consumer
->net_seq_index
;
171 ret
= consumer_trace_chunk_exists(socket
,
174 session
.chunk_being_archived
,
177 ERR("Error occurred while checking rotation status on consumer daemon");
181 if (exists_status
!= CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK
) {
182 chunk_exists_on_peer
= true;
189 if (!chunk_exists_on_peer
) {
190 uint64_t chunk_being_archived_id
;
192 chunk_status
= lttng_trace_chunk_get_id(session
.chunk_being_archived
,
193 &chunk_being_archived_id
);
194 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
195 DBG("Rotation of trace archive %" PRIu64
196 " of session \"%s\" is complete on all consumers",
197 chunk_being_archived_id
,
201 _rotation_completed
= !chunk_exists_on_peer
;
203 ret
= session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_ERROR
);
205 ERR("Failed to reset rotation state of session \"%s\"", session
.name
);
211 * Check if the last rotation was completed, called with session lock held.
212 * Should only return non-zero in the event of a fatal error. Doing so will
213 * shutdown the thread.
215 int check_session_rotation_pending(ltt_session
& session
,
216 notification_thread_handle
& notification_thread_handle
)
219 struct lttng_trace_archive_location
*location
;
220 enum lttng_trace_chunk_status chunk_status
;
221 bool rotation_completed
= false;
222 const char *archived_chunk_name
;
223 uint64_t chunk_being_archived_id
;
225 if (!session
.chunk_being_archived
) {
231 lttng_trace_chunk_get_id(session
.chunk_being_archived
, &chunk_being_archived_id
);
232 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
234 DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64
,
236 chunk_being_archived_id
);
239 * The rotation-pending check timer of a session is launched in
240 * one-shot mode. If the rotation is incomplete, the rotation
241 * thread will re-enable the pending-check timer.
243 * The timer thread can't stop the timer itself since it is involved
244 * in the check for the timer's quiescence.
246 ret
= timer_session_rotation_pending_check_stop(session
);
248 goto check_ongoing_rotation
;
251 check_session_rotation_pending_on_consumers(session
, rotation_completed
);
252 if (!rotation_completed
|| session
.rotation_state
== LTTNG_ROTATION_STATE_ERROR
) {
253 goto check_ongoing_rotation
;
257 * Now we can clear the "ONGOING" state in the session. New
258 * rotations can start now.
260 chunk_status
= lttng_trace_chunk_get_name(
261 session
.chunk_being_archived
, &archived_chunk_name
, nullptr);
262 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
263 free(session
.last_archived_chunk_name
);
264 session
.last_archived_chunk_name
= strdup(archived_chunk_name
);
265 if (!session
.last_archived_chunk_name
) {
266 PERROR("Failed to duplicate archived chunk name");
269 session_reset_rotation_state(session
, LTTNG_ROTATION_STATE_COMPLETED
);
271 if (!session
.quiet_rotation
) {
272 location
= session_get_trace_archive_location(&session
);
273 ret
= notification_thread_command_session_rotation_completed(
274 ¬ification_thread_handle
,
276 session
.last_archived_chunk_id
.value
,
278 lttng_trace_archive_location_put(location
);
279 if (ret
!= LTTNG_OK
) {
280 ERR("Failed to notify notification thread of completed rotation for session %s",
286 check_ongoing_rotation
:
287 if (session
.rotation_state
== LTTNG_ROTATION_STATE_ONGOING
) {
288 chunk_status
= lttng_trace_chunk_get_id(session
.chunk_being_archived
,
289 &chunk_being_archived_id
);
290 LTTNG_ASSERT(chunk_status
== LTTNG_TRACE_CHUNK_STATUS_OK
);
292 DBG("Rotation of trace archive %" PRIu64
" is still pending for session %s",
293 chunk_being_archived_id
,
295 ret
= timer_session_rotation_pending_check_start(&session
,
296 DEFAULT_ROTATE_PENDING_TIMER
);
298 ERR("Failed to re-enable rotation pending timer");
308 /* Call with the session and session_list locks held. */
309 int launch_session_rotation(ltt_session
& session
)
312 struct lttng_rotate_session_return rotation_return
;
314 DBG("Launching scheduled time-based rotation on session \"%s\"", session
.name
);
316 ASSERT_SESSION_LIST_LOCKED();
317 ASSERT_LOCKED(session
.lock
);
319 ret
= cmd_rotate_session(&session
,
322 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
323 if (ret
== LTTNG_OK
) {
324 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
327 /* Don't consider errors as fatal. */
328 DBG("Scheduled time-based rotation aborted for session %s: %s",
330 lttng_strerror(ret
));
336 int run_job(const rotation_thread_job
& job
,
337 ltt_session
& session
,
338 notification_thread_handle
& notification_thread_handle
)
343 case ls::rotation_thread_job_type::SCHEDULED_ROTATION
:
344 ret
= launch_session_rotation(session
);
346 case ls::rotation_thread_job_type::CHECK_PENDING_ROTATION
:
347 ret
= check_session_rotation_pending(session
, notification_thread_handle
);
356 bool shutdown_rotation_thread(void *thread_data
)
358 auto *handle
= reinterpret_cast<const ls::rotation_thread
*>(thread_data
);
360 return handle
->shutdown();
364 ls::rotation_thread_timer_queue
*ls::rotation_thread_timer_queue_create()
366 auto queue
= zmalloc
<ls::rotation_thread_timer_queue
>();
368 PERROR("Failed to allocate timer rotate queue");
372 queue
->event_pipe
= lttng_pipe_open(FD_CLOEXEC
| O_NONBLOCK
);
373 CDS_INIT_LIST_HEAD(&queue
->list
);
374 pthread_mutex_init(&queue
->lock
, nullptr);
379 void ls::rotation_thread_timer_queue_destroy(struct rotation_thread_timer_queue
*queue
)
385 lttng_pipe_destroy(queue
->event_pipe
);
388 lttng::pthread::lock_guard
queue_lock(queue
->lock
);
390 LTTNG_ASSERT(cds_list_empty(&queue
->list
));
393 pthread_mutex_destroy(&queue
->lock
);
397 ls::rotation_thread::rotation_thread(rotation_thread_timer_queue
& rotation_timer_queue
,
398 notification_thread_handle
& notification_thread_handle
) :
399 _rotation_timer_queue(rotation_timer_queue
),
400 _notification_thread_handle(notification_thread_handle
)
402 _quit_pipe
.reset([]() {
403 auto raw_pipe
= lttng_pipe_open(FD_CLOEXEC
);
405 LTTNG_THROW_POSIX("Failed to rotation thread's quit pipe", errno
);
411 _notification_channel
.reset([]() {
412 auto channel
= lttng_notification_channel_create(
413 lttng_session_daemon_notification_endpoint
);
416 "Failed to create notification channel of rotation thread");
422 lttng_poll_init(&_events
);
425 * Create pollset with size 4:
426 * - rotation thread quit pipe,
427 * - rotation thread timer queue pipe,
428 * - notification channel sock,
429 * - subscribtion change event fd
431 if (lttng_poll_create(&_events
, 4, LTTNG_CLOEXEC
) < 0) {
432 LTTNG_THROW_ERROR("Failed to create poll object for rotation thread");
435 if (lttng_poll_add(&_events
, lttng_pipe_get_readfd(_quit_pipe
.get()), LPOLLIN
) < 0) {
436 LTTNG_THROW_ERROR("Failed to add quit pipe read fd to poll set");
439 if (lttng_poll_add(&_events
,
440 lttng_pipe_get_readfd(_rotation_timer_queue
.event_pipe
),
442 LTTNG_THROW_ERROR("Failed to add rotation timer queue event pipe fd to poll set");
445 if (lttng_poll_add(&_events
,
446 _notification_channel_subscribtion_change_eventfd
.fd(),
449 "Failed to add rotation thread notification channel subscription change eventfd to poll set");
452 if (lttng_poll_add(&_events
, _notification_channel
->socket
, LPOLLIN
) < 0) {
453 LTTNG_THROW_ERROR("Failed to add notification channel socket fd to pollset");
457 ls::rotation_thread::~rotation_thread()
459 lttng_poll_clean(&_events
);
462 void ls::rotation_thread_enqueue_job(ls::rotation_thread_timer_queue
*queue
,
463 ls::rotation_thread_job_type job_type
,
464 ltt_session
*session
)
466 const char dummy
= '!';
467 struct rotation_thread_job
*job
= nullptr;
468 const char *job_type_str
= get_job_type_str(job_type
);
469 lttng::pthread::lock_guard
queue_lock(queue
->lock
);
471 if (timer_job_exists(queue
, job_type
, session
)) {
473 * This timer job is already pending, we don't need to add
479 job
= zmalloc
<rotation_thread_job
>();
481 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
487 /* No reason for this to fail as the caller must hold a reference. */
488 (void) session_get(session
);
490 job
->session
= session
;
491 job
->type
= job_type
;
492 cds_list_add_tail(&job
->head
, &queue
->list
);
494 const int write_ret
=
495 lttng_write(lttng_pipe_get_writefd(queue
->event_pipe
), &dummy
, sizeof(dummy
));
498 * We do not want to block in the timer handler, the job has
499 * been enqueued in the list, the wakeup pipe is probably full,
500 * the job will be processed when the rotation_thread catches
504 DIAGNOSTIC_IGNORE_LOGICAL_OP
505 if (errno
== EAGAIN
|| errno
== EWOULDBLOCK
) {
508 * Not an error, but would be surprising and indicate
509 * that the rotation thread can't keep up with the
512 DBG("Wake-up pipe of rotation thread job queue is full");
516 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
523 void ls::rotation_thread::_handle_job_queue()
526 rotation_thread_job::uptr job
;
529 /* Take the queue lock only to pop an element from the list. */
530 lttng::pthread::lock_guard
rotation_timer_queue_lock(
531 _rotation_timer_queue
.lock
);
532 if (cds_list_empty(&_rotation_timer_queue
.list
)) {
536 job
.reset(cds_list_first_entry(
537 &_rotation_timer_queue
.list
, typeof(rotation_thread_job
), head
));
538 cds_list_del(&job
->head
);
542 const auto unlock_list
=
543 lttng::make_scope_exit([]() noexcept
{ session_unlock_list(); });
545 /* locked_ptr will unlock the session and release the ref held by the job. */
546 session_lock(job
->session
);
547 auto session
= ltt_session::locked_ptr(job
->session
);
549 if (run_job(*job
, *session
, _notification_thread_handle
)) {
555 void ls::rotation_thread::_handle_notification(const lttng_notification
& notification
)
558 const char *condition_session_name
= nullptr;
559 enum lttng_condition_status condition_status
;
560 enum lttng_evaluation_status evaluation_status
;
562 auto *condition
= lttng_notification_get_const_condition(¬ification
);
563 auto *evaluation
= lttng_notification_get_const_evaluation(¬ification
);
564 const auto condition_type
= lttng_condition_get_type(condition
);
566 if (condition_type
!= LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE
) {
567 LTTNG_THROW_ERROR("Unexpected condition type");
570 /* Fetch info to test. */
571 condition_status
= lttng_condition_session_consumed_size_get_session_name(
572 condition
, &condition_session_name
);
573 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
574 LTTNG_THROW_ERROR("Session name could not be fetched from notification");
578 lttng_evaluation_session_consumed_size_get_consumed_size(evaluation
, &consumed
);
579 if (evaluation_status
!= LTTNG_EVALUATION_STATUS_OK
) {
580 LTTNG_THROW_ERROR("Failed to get consumed size from evaluation");
583 DBG_FMT("Handling session consumed size condition: session_name=`{}`, consumed_size={}",
584 condition_session_name
,
588 const auto unlock_list
= lttng::make_scope_exit([]() noexcept
{ session_unlock_list(); });
590 ltt_session::locked_ptr session
{ [&condition_session_name
]() {
591 auto raw_session_ptr
= session_find_by_name(condition_session_name
);
593 if (raw_session_ptr
) {
594 session_lock(raw_session_ptr
);
597 return raw_session_ptr
;
600 DBG_FMT("Failed to find session while handling notification: notification_type={}, session name=`{}`",
601 lttng_condition_type_str(condition_type
),
602 condition_session_name
);
604 * Not a fatal error: a session can be destroyed before we get
605 * the chance to handle the notification.
610 if (!lttng_trigger_is_equal(session
->rotate_trigger
,
611 lttng_notification_get_const_trigger(¬ification
))) {
612 DBG("Notification does not originate from the internal size-based scheduled rotation trigger, skipping");
616 unsubscribe_session_consumed_size_rotation(*session
);
618 ret
= cmd_rotate_session(
619 session
.get(), nullptr, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED
);
623 case -LTTNG_ERR_ROTATION_PENDING
:
624 DBG("Rotate already pending, subscribe to the next threshold value");
626 case -LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP
:
627 DBG("Rotation already happened since last stop, subscribe to the next threshold value");
629 case -LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR
:
630 DBG("Rotation already happened since last stop and clear, subscribe to the next threshold value");
633 LTTNG_THROW_CTL("Failed to rotate on consumed size notification",
634 static_cast<lttng_error_code
>(-ret
));
637 subscribe_session_consumed_size_rotation(*session
, consumed
+ session
->rotate_size
);
640 void ls::rotation_thread::_handle_notification_channel_activity()
642 bool notification_pending
= true;
645 * A notification channel may have multiple notifications queued-up internally in
646 * its buffers. This is because a notification channel multiplexes command replies
647 * and notifications. The current protocol specifies that multiple notifications can be
648 * received before the reply to a command.
650 * In such cases, the notification channel client implementation internally queues them and
651 * provides them on the next calls to lttng_notification_channel_get_next_notification().
652 * This is correct with respect to the public API, which is intended to be used in "blocking
655 * However, this internal user relies on poll/epoll to wake-up when data is available
656 * on the notification channel's socket. As such, it can't assume that a wake-up means only
657 * one notification is available for consumption since many of them may have been queued in
658 * the channel's internal buffers.
660 while (notification_pending
) {
661 const auto pending_status
= lttng_notification_channel_has_pending_notification(
662 _notification_channel
.get(), ¬ification_pending
);
663 if (pending_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
664 LTTNG_THROW_ERROR("Error occurred while checking for pending notification");
667 if (!notification_pending
) {
671 /* Receive the next notification. */
672 lttng_notification::uptr notification
;
673 enum lttng_notification_channel_status next_notification_status
;
676 struct lttng_notification
*raw_notification_ptr
;
678 next_notification_status
= lttng_notification_channel_get_next_notification(
679 _notification_channel
.get(), &raw_notification_ptr
);
680 notification
.reset(raw_notification_ptr
);
683 switch (next_notification_status
) {
684 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
:
686 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED
:
687 WARN("Dropped notification detected on notification channel used by the rotation management thread.");
689 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED
:
690 LTTNG_THROW_ERROR("Notification channel was closed");
692 /* Unhandled conditions / errors. */
693 LTTNG_THROW_ERROR("Unknown notification channel status");
696 _handle_notification(*notification
);
700 void ls::rotation_thread::_thread_function() noexcept
702 DBG("Started rotation thread");
706 } catch (const std::exception
& e
) {
707 ERR_FMT("Fatal rotation thread error: {}", e
.what());
713 void ls::rotation_thread::_run()
715 rcu_register_thread();
716 const auto unregister_rcu_thread
=
717 lttng::make_scope_exit([]() noexcept
{ rcu_unregister_thread(); });
720 const auto offline_rcu_thread
=
721 lttng::make_scope_exit([]() noexcept
{ rcu_thread_offline(); });
723 health_register(the_health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
724 health_code_update();
725 const auto unregister_health
=
726 lttng::make_scope_exit([]() noexcept
{ health_unregister(the_health_sessiond
); });
728 const auto queue_pipe_fd
= lttng_pipe_get_readfd(_rotation_timer_queue
.event_pipe
);
732 DBG("Entering poll wait");
733 auto poll_wait_ret
= lttng_poll_wait(&_events
, -1);
734 DBG_FMT("Poll wait returned: ret={}", poll_wait_ret
);
736 if (poll_wait_ret
< 0) {
738 * Restart interrupted system call.
740 if (errno
== EINTR
) {
744 LTTNG_THROW_POSIX("Error encountered during lttng_poll_wait", errno
);
747 const auto fd_count
= poll_wait_ret
;
748 for (int i
= 0; i
< fd_count
; i
++) {
749 const auto fd
= LTTNG_POLL_GETFD(&_events
, i
);
750 const auto revents
= LTTNG_POLL_GETEV(&_events
, i
);
752 DBG_FMT("Handling descriptor activity: fd={}, events={:b}", fd
, revents
);
754 if (revents
& LPOLLERR
) {
755 LTTNG_THROW_ERROR(lttng::format(
756 "Polling returned an error on fd: fd={}", fd
));
759 if (fd
== _notification_channel
->socket
||
760 fd
== _notification_channel_subscribtion_change_eventfd
.fd()) {
762 _handle_notification_channel_activity();
763 } catch (const lttng::ctl::error
& e
) {
765 * The only non-fatal error (rotation failed), others
766 * are caught at the top-level.
768 DBG_FMT("Control error occurred while handling activity on notification channel socket: {}",
773 if (fd
== _notification_channel_subscribtion_change_eventfd
.fd()) {
774 _notification_channel_subscribtion_change_eventfd
778 /* Job queue or quit pipe activity. */
781 * The job queue is serviced if there is
782 * activity on the quit pipe to ensure it is
783 * flushed and all references held in the queue
787 if (fd
== queue_pipe_fd
) {
790 if (lttng_read(fd
, &buf
, 1) != 1) {
793 "Failed to read from wakeup pipe: fd={}",
798 DBG("Quit pipe activity");
806 bool ls::rotation_thread::shutdown() const noexcept
808 const int write_fd
= lttng_pipe_get_writefd(_quit_pipe
.get());
810 return notify_thread_pipe(write_fd
) == 1;
813 void ls::rotation_thread::launch_thread()
815 auto thread
= lttng_thread_create(
818 auto handle
= reinterpret_cast<rotation_thread
*>(ptr
);
820 handle
->_thread_function();
821 return static_cast<void *>(nullptr);
823 shutdown_rotation_thread
,
827 LTTNG_THROW_ERROR("Failed to launch rotation thread");
830 lttng_thread_put(thread
);
833 void ls::rotation_thread::subscribe_session_consumed_size_rotation(ltt_session
& session
,
836 const struct lttng_credentials session_creds
= {
837 .uid
= LTTNG_OPTIONAL_INIT_VALUE(session
.uid
),
838 .gid
= LTTNG_OPTIONAL_INIT_VALUE(session
.gid
),
841 ASSERT_LOCKED(session
.lock
);
843 auto rotate_condition
= lttng::make_unique_wrapper
<lttng_condition
, lttng_condition_put
>(
844 lttng_condition_session_consumed_size_create());
845 if (!rotate_condition
) {
846 LTTNG_THROW_POSIX("Failed to create session consumed size condition object", errno
);
849 auto condition_status
=
850 lttng_condition_session_consumed_size_set_threshold(rotate_condition
.get(), size
);
851 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
852 LTTNG_THROW_ERROR(lttng::format(
853 "Could not set session consumed size condition threshold: size={}", size
));
856 condition_status
= lttng_condition_session_consumed_size_set_session_name(
857 rotate_condition
.get(), session
.name
);
858 if (condition_status
!= LTTNG_CONDITION_STATUS_OK
) {
859 LTTNG_THROW_ERROR(lttng::format(
860 "Could not set session consumed size condition session name: name=`{}`",
864 auto notify_action
= lttng::make_unique_wrapper
<lttng_action
, lttng_action_put
>(
865 lttng_action_notify_create());
866 if (!notify_action
) {
867 LTTNG_THROW_POSIX("Could not create notify action", errno
);
870 LTTNG_ASSERT(!session
.rotate_trigger
);
871 /* trigger acquires its own reference to condition and action on success. */
872 auto trigger
= lttng::make_unique_wrapper
<lttng_trigger
, lttng_trigger_put
>(
873 lttng_trigger_create(rotate_condition
.get(), notify_action
.get()));
875 LTTNG_THROW_POSIX("Could not create size-based rotation trigger", errno
);
878 /* Ensure this trigger is not visible to external users. */
879 lttng_trigger_set_hidden(trigger
.get());
880 lttng_trigger_set_credentials(trigger
.get(), &session_creds
);
882 auto nc_status
= lttng_notification_channel_subscribe(_notification_channel
.get(),
883 rotate_condition
.get());
884 if (nc_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
885 LTTNG_THROW_ERROR("Could not subscribe to session consumed size notification");
889 * Ensure any notification queued during the subscription are consumed by queueing an
892 _notification_channel_subscribtion_change_eventfd
.increment();
894 const auto register_ret
= notification_thread_command_register_trigger(
895 &_notification_thread_handle
, trigger
.get(), true);
896 if (register_ret
!= LTTNG_OK
) {
899 "Failed to register trigger for automatic size-based rotation: session_name{}, size={}",
905 /* Ownership transferred to the session. */
906 session
.rotate_trigger
= trigger
.release();
909 void ls::rotation_thread::unsubscribe_session_consumed_size_rotation(ltt_session
& session
)
911 LTTNG_ASSERT(session
.rotate_trigger
);
913 const auto remove_session_trigger
= lttng::make_scope_exit([&session
]() noexcept
{
914 lttng_trigger_put(session
.rotate_trigger
);
915 session
.rotate_trigger
= nullptr;
918 const auto unsubscribe_status
= lttng_notification_channel_unsubscribe(
919 _notification_channel
.get(),
920 lttng_trigger_get_const_condition(session
.rotate_trigger
));
921 if (unsubscribe_status
!= LTTNG_NOTIFICATION_CHANNEL_STATUS_OK
) {
922 LTTNG_THROW_ERROR(lttng::format(
923 "Failed to unsubscribe from consumed size condition used to control automatic size-based rotations: session_name=`{}` return_code={}",
925 static_cast<int>(unsubscribe_status
)));
929 * Ensure any notification queued during the un-subscription are consumed by queueing an
932 _notification_channel_subscribtion_change_eventfd
.increment();
934 const auto unregister_status
= notification_thread_command_unregister_trigger(
935 &_notification_thread_handle
, session
.rotate_trigger
);
936 if (unregister_status
!= LTTNG_OK
) {
939 "Failed to unregister trigger for automatic size-based rotation: session_name{}",