src/bin/lttng-sessiond/rotation-thread.cpp

   1 /*
   2  * Copyright (C) 2017 Julien Desfossez <jdesfossez@efficios.com>
   3  * Copyright (C) 2018 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  *
   5  * SPDX-License-Identifier: GPL-2.0-only
   6  *
   7  */
   8
   9 #define _LGPL_SOURCE
  10 #include "cmd.hpp"
  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"
  16 #include "thread.hpp"
  17 #include "timer.hpp"
  18 #include "utils.hpp"
  19
  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>
  38
  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>
  46
  47 #include <fcntl.h>
  48 #include <inttypes.h>
  49 #include <memory>
  50 #include <signal.h>
  51 #include <sys/eventfd.h>
  52 #include <sys/stat.h>
  53 #include <time.h>
  54 #include <urcu.h>
  55 #include <urcu/list.h>
  56
  57 namespace ls = lttng::sessiond;
  58
  59 /*
  60  * The timer thread enqueues jobs and wakes up the rotation thread.
  61  * When the rotation thread wakes up, it empties the queue.
  62  */
  63 struct ls::rotation_thread_timer_queue {
  64         struct lttng_pipe *event_pipe;
  65         struct cds_list_head list;
  66         pthread_mutex_t lock;
  67 };
  68
  69 namespace {
  70 struct rotation_thread_job {
  71         using uptr =
  72                 std::unique_ptr<rotation_thread_job,
  73                                 lttng::memory::create_deleter_class<rotation_thread_job,
  74                                                                     lttng::memory::free>::deleter>;
  75
  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;
  80 };
  81
  82 const char *get_job_type_str(enum ls::rotation_thread_job_type job_type)
  83 {
  84         switch (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";
  89         default:
  90                 abort();
  91         }
  92 }
  93
  94 /*
  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.
  97  */
  98 bool timer_job_exists(const ls::rotation_thread_timer_queue *queue,
  99                       ls::rotation_thread_job_type job_type,
 100                       ltt_session *session)
 101 {
 102         bool exists = false;
 103         struct rotation_thread_job *job;
 104
 105         cds_list_for_each_entry (job, &queue->list, head) {
 106                 if (job->session == session && job->type == job_type) {
 107                         exists = true;
 108                         goto end;
 109                 }
 110         }
 111 end:
 112         return exists;
 113 }
 114
 115 void check_session_rotation_pending_on_consumers(ltt_session& session, bool& _rotation_completed)
 116 {
 117         int ret = 0;
 118         struct consumer_socket *socket;
 119         struct cds_lfht_iter iter;
 120         enum consumer_trace_chunk_exists_status exists_status;
 121         uint64_t relayd_id;
 122         bool chunk_exists_on_peer = false;
 123         enum lttng_trace_chunk_status chunk_status;
 124         lttng::urcu::read_lock_guard read_lock;
 125
 126         LTTNG_ASSERT(session.chunk_being_archived);
 127
 128         /*
 129          * Check for a local pending rotation on all consumers (32-bit
 130          * user space, 64-bit user space, and kernel).
 131          */
 132         if (!session.ust_session) {
 133                 goto skip_ust;
 134         }
 135
 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 ?
 139                         -1ULL :
 140                         session.ust_session->consumer->net_seq_index;
 141
 142                 lttng::pthread::lock_guard socket_lock(*socket->lock);
 143                 ret = consumer_trace_chunk_exists(socket,
 144                                                   relayd_id,
 145                                                   session.id,
 146                                                   session.chunk_being_archived,
 147                                                   &exists_status);
 148                 if (ret) {
 149                         ERR("Error occurred while checking rotation status on consumer daemon");
 150                         goto end;
 151                 }
 152
 153                 if (exists_status != CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK) {
 154                         chunk_exists_on_peer = true;
 155                         goto end;
 156                 }
 157         }
 158
 159 skip_ust:
 160         if (!session.kernel_session) {
 161                 goto skip_kernel;
 162         }
 163
 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);
 167
 168                 relayd_id = session.kernel_session->consumer->type == CONSUMER_DST_LOCAL ?
 169                         -1ULL :
 170                         session.kernel_session->consumer->net_seq_index;
 171
 172                 ret = consumer_trace_chunk_exists(socket,
 173                                                   relayd_id,
 174                                                   session.id,
 175                                                   session.chunk_being_archived,
 176                                                   &exists_status);
 177                 if (ret) {
 178                         ERR("Error occurred while checking rotation status on consumer daemon");
 179                         goto end;
 180                 }
 181
 182                 if (exists_status != CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK) {
 183                         chunk_exists_on_peer = true;
 184                         goto end;
 185                 }
 186         }
 187 skip_kernel:
 188 end:
 189
 190         if (!chunk_exists_on_peer) {
 191                 uint64_t chunk_being_archived_id;
 192
 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,
 199                     session.name);
 200         }
 201
 202         _rotation_completed = !chunk_exists_on_peer;
 203         if (ret) {
 204                 ret = session_reset_rotation_state(session, LTTNG_ROTATION_STATE_ERROR);
 205                 if (ret) {
 206                         ERR("Failed to reset rotation state of session \"%s\"", session.name);
 207                 }
 208         }
 209 }
 210
 211 /*
 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.
 215  */
 216 int check_session_rotation_pending(ltt_session& session,
 217                                    notification_thread_handle& notification_thread_handle)
 218 {
 219         int ret;
 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;
 225
 226         if (!session.chunk_being_archived) {
 227                 ret = 0;
 228                 goto end;
 229         }
 230
 231         chunk_status =
 232                 lttng_trace_chunk_get_id(session.chunk_being_archived, &chunk_being_archived_id);
 233         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 234
 235         DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64,
 236             session.name,
 237             chunk_being_archived_id);
 238
 239         /*
 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.
 243          *
 244          * The timer thread can't stop the timer itself since it is involved
 245          * in the check for the timer's quiescence.
 246          */
 247         ret = timer_session_rotation_pending_check_stop(session);
 248         if (ret) {
 249                 goto check_ongoing_rotation;
 250         }
 251
 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;
 255         }
 256
 257         /*
 258          * Now we can clear the "ONGOING" state in the session. New
 259          * rotations can start now.
 260          */
 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");
 268         }
 269
 270         session_reset_rotation_state(session, LTTNG_ROTATION_STATE_COMPLETED);
 271
 272         if (!session.quiet_rotation) {
 273                 location = session_get_trace_archive_location(&session);
 274                 ret = notification_thread_command_session_rotation_completed(
 275                         &notification_thread_handle,
 276                         session.id,
 277                         session.last_archived_chunk_id.value,
 278                         location);
 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",
 282                             session.name);
 283                 }
 284         }
 285
 286         ret = 0;
 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);
 292
 293                 DBG("Rotation of trace archive %" PRIu64 " is still pending for session %s",
 294                     chunk_being_archived_id,
 295                     session.name);
 296                 ret = timer_session_rotation_pending_check_start(&session,
 297                                                                  DEFAULT_ROTATE_PENDING_TIMER);
 298                 if (ret) {
 299                         ERR("Failed to re-enable rotation pending timer");
 300                         ret = -1;
 301                         goto end;
 302                 }
 303         }
 304
 305 end:
 306         return ret;
 307 }
 308
 309 /* Call with the session and session_list locks held. */
 310 int launch_session_rotation(ltt_session& session)
 311 {
 312         int ret;
 313         struct lttng_rotate_session_return rotation_return;
 314
 315         DBG("Launching scheduled time-based rotation on session \"%s\"", session.name);
 316
 317         ASSERT_SESSION_LIST_LOCKED();
 318         ASSERT_LOCKED(session.lock);
 319
 320         ret = cmd_rotate_session(&session,
 321                                  &rotation_return,
 322                                  false,
 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\"",
 326                     session.name);
 327         } else {
 328                 /* Don't consider errors as fatal. */
 329                 DBG("Scheduled time-based rotation aborted for session %s: %s",
 330                     session.name,
 331                     lttng_strerror(ret));
 332         }
 333
 334         return 0;
 335 }
 336
 337 int run_job(const rotation_thread_job& job,
 338             ltt_session& session,
 339             notification_thread_handle& notification_thread_handle)
 340 {
 341         int ret;
 342
 343         switch (job.type) {
 344         case ls::rotation_thread_job_type::SCHEDULED_ROTATION:
 345                 ret = launch_session_rotation(session);
 346                 break;
 347         case ls::rotation_thread_job_type::CHECK_PENDING_ROTATION:
 348                 ret = check_session_rotation_pending(session, notification_thread_handle);
 349                 break;
 350         default:
 351                 abort();
 352         }
 353
 354         return ret;
 355 }
 356
 357 bool shutdown_rotation_thread(void *thread_data)
 358 {
 359         auto *handle = reinterpret_cast<const ls::rotation_thread *>(thread_data);
 360
 361         return handle->shutdown();
 362 }
 363 } /* namespace */
 364
 365 ls::rotation_thread_timer_queue *ls::rotation_thread_timer_queue_create()
 366 {
 367         auto queue = zmalloc<ls::rotation_thread_timer_queue>();
 368         if (!queue) {
 369                 PERROR("Failed to allocate timer rotate queue");
 370                 goto end;
 371         }
 372
 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);
 376 end:
 377         return queue;
 378 }
 379
 380 void ls::rotation_thread_timer_queue_destroy(struct rotation_thread_timer_queue *queue)
 381 {
 382         if (!queue) {
 383                 return;
 384         }
 385
 386         lttng_pipe_destroy(queue->event_pipe);
 387
 388         {
 389                 lttng::pthread::lock_guard queue_lock(queue->lock);
 390
 391                 LTTNG_ASSERT(cds_list_empty(&queue->list));
 392         }
 393
 394         pthread_mutex_destroy(&queue->lock);
 395         free(queue);
 396 }
 397
 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)
 402 {
 403         _quit_pipe.reset([]() {
 404                 auto raw_pipe = lttng_pipe_open(FD_CLOEXEC);
 405                 if (!raw_pipe) {
 406                         LTTNG_THROW_POSIX("Failed to rotation thread's quit pipe", errno);
 407                 }
 408
 409                 return raw_pipe;
 410         }());
 411
 412         _notification_channel.reset([]() {
 413                 auto channel = lttng_notification_channel_create(
 414                         lttng_session_daemon_notification_endpoint);
 415                 if (!channel) {
 416                         LTTNG_THROW_ERROR(
 417                                 "Failed to create notification channel of rotation thread");
 418                 }
 419
 420                 return channel;
 421         }());
 422
 423         lttng_poll_init(&_events);
 424
 425         /*
 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
 431          */
 432         if (lttng_poll_create(&_events, 4, LTTNG_CLOEXEC) < 0) {
 433                 LTTNG_THROW_ERROR("Failed to create poll object for rotation thread");
 434         }
 435
 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");
 438         }
 439
 440         if (lttng_poll_add(&_events,
 441                            lttng_pipe_get_readfd(_rotation_timer_queue.event_pipe),
 442                            LPOLLIN) < 0) {
 443                 LTTNG_THROW_ERROR("Failed to add rotation timer queue event pipe fd to poll set");
 444         }
 445
 446         if (lttng_poll_add(&_events,
 447                            _notification_channel_subscribtion_change_eventfd.fd(),
 448                            LPOLLIN) < 0) {
 449                 LTTNG_THROW_ERROR(
 450                         "Failed to add rotation thread notification channel subscription change eventfd to poll set");
 451         }
 452
 453         if (lttng_poll_add(&_events, _notification_channel->socket, LPOLLIN) < 0) {
 454                 LTTNG_THROW_ERROR("Failed to add notification channel socket fd to pollset");
 455         }
 456 }
 457
 458 ls::rotation_thread::~rotation_thread()
 459 {
 460         lttng_poll_clean(&_events);
 461 }
 462
 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)
 466 {
 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);
 471
 472         if (timer_job_exists(queue, job_type, session)) {
 473                 /*
 474                  * This timer job is already pending, we don't need to add
 475                  * it.
 476                  */
 477                 return;
 478         }
 479
 480         job = zmalloc<rotation_thread_job>();
 481         if (!job) {
 482                 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
 483                        job_type_str,
 484                        session->name);
 485                 return;
 486         }
 487
 488         /* No reason for this to fail as the caller must hold a reference. */
 489         (void) session_get(session);
 490
 491         job->session = session;
 492         job->type = job_type;
 493         cds_list_add_tail(&job->head, &queue->list);
 494
 495         const int write_ret =
 496                 lttng_write(lttng_pipe_get_writefd(queue->event_pipe), &dummy, sizeof(dummy));
 497         if (write_ret < 0) {
 498                 /*
 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
 502                  * up.
 503                  */
 504                 DIAGNOSTIC_PUSH
 505                 DIAGNOSTIC_IGNORE_LOGICAL_OP
 506                 if (errno == EAGAIN || errno == EWOULDBLOCK) {
 507                         DIAGNOSTIC_POP
 508                         /*
 509                          * Not an error, but would be surprising and indicate
 510                          * that the rotation thread can't keep up with the
 511                          * current load.
 512                          */
 513                         DBG("Wake-up pipe of rotation thread job queue is full");
 514                         return;
 515                 }
 516
 517                 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
 518                        job_type_str,
 519                        session->name);
 520                 return;
 521         }
 522 }
 523
 524 void ls::rotation_thread::_handle_job_queue()
 525 {
 526         for (;;) {
 527                 rotation_thread_job::uptr job;
 528
 529                 {
 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)) {
 534                                 break;
 535                         }
 536
 537                         job.reset(cds_list_first_entry(
 538                                 &_rotation_timer_queue.list, typeof(rotation_thread_job), head));
 539                         cds_list_del(&job->head);
 540                 }
 541
 542                 session_lock_list();
 543                 const auto unlock_list =
 544                         lttng::make_scope_exit([]() noexcept { session_unlock_list(); });
 545
 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);
 549
 550                 if (run_job(*job, *session, _notification_thread_handle)) {
 551                         return;
 552                 }
 553         }
 554 }
 555
 556 void ls::rotation_thread::_handle_notification(const lttng_notification& notification)
 557 {
 558         int ret = 0;
 559         const char *condition_session_name = nullptr;
 560         enum lttng_condition_status condition_status;
 561         enum lttng_evaluation_status evaluation_status;
 562         uint64_t consumed;
 563         auto *condition = lttng_notification_get_const_condition(&notification);
 564         auto *evaluation = lttng_notification_get_const_evaluation(&notification);
 565         const auto condition_type = lttng_condition_get_type(condition);
 566
 567         if (condition_type != LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE) {
 568                 LTTNG_THROW_ERROR("Unexpected condition type");
 569         }
 570
 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");
 576         }
 577
 578         evaluation_status =
 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");
 582         }
 583
 584         DBG_FMT("Handling session consumed size condition: session_name=`{}`, consumed_size={}",
 585                 condition_session_name,
 586                 consumed);
 587
 588         session_lock_list();
 589         const auto unlock_list = lttng::make_scope_exit([]() noexcept { session_unlock_list(); });
 590
 591         ltt_session::locked_ptr session{ [&condition_session_name]() {
 592                 auto raw_session_ptr = session_find_by_name(condition_session_name);
 593
 594                 if (raw_session_ptr) {
 595                         session_lock(raw_session_ptr);
 596                 }
 597
 598                 return raw_session_ptr;
 599         }() };
 600         if (!session) {
 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);
 604                 /*
 605                  * Not a fatal error: a session can be destroyed before we get
 606                  * the chance to handle the notification.
 607                  */
 608                 return;
 609         }
 610
 611         if (!lttng_trigger_is_equal(session->rotate_trigger,
 612                                     lttng_notification_get_const_trigger(&notification))) {
 613                 DBG("Notification does not originate from the internal size-based scheduled rotation trigger, skipping");
 614                 return;
 615         }
 616
 617         unsubscribe_session_consumed_size_rotation(*session);
 618
 619         ret = cmd_rotate_session(
 620                 session.get(), nullptr, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
 621         switch (ret) {
 622         case LTTNG_OK:
 623                 break;
 624         case -LTTNG_ERR_ROTATION_PENDING:
 625                 DBG("Rotate already pending, subscribe to the next threshold value");
 626                 break;
 627         case -LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
 628                 DBG("Rotation already happened since last stop, subscribe to the next threshold value");
 629                 break;
 630         case -LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
 631                 DBG("Rotation already happened since last stop and clear, subscribe to the next threshold value");
 632                 break;
 633         default:
 634                 LTTNG_THROW_CTL("Failed to rotate on consumed size notification",
 635                                 static_cast<lttng_error_code>(-ret));
 636         }
 637
 638         subscribe_session_consumed_size_rotation(*session, consumed + session->rotate_size);
 639 }
 640
 641 void ls::rotation_thread::_handle_notification_channel_activity()
 642 {
 643         bool notification_pending = true;
 644
 645         /*
 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.
 650          *
 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
 654          * mode".
 655          *
 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.
 660          */
 661         while (notification_pending) {
 662                 const auto pending_status = lttng_notification_channel_has_pending_notification(
 663                         _notification_channel.get(), &notification_pending);
 664                 if (pending_status != LTTNG_NOTIFICATION_CHANNEL_STATUS_OK) {
 665                         LTTNG_THROW_ERROR("Error occurred while checking for pending notification");
 666                 }
 667
 668                 if (!notification_pending) {
 669                         return;
 670                 }
 671
 672                 /* Receive the next notification. */
 673                 lttng_notification::uptr notification;
 674                 enum lttng_notification_channel_status next_notification_status;
 675
 676                 {
 677                         struct lttng_notification *raw_notification_ptr;
 678
 679                         next_notification_status = lttng_notification_channel_get_next_notification(
 680                                 _notification_channel.get(), &raw_notification_ptr);
 681                         notification.reset(raw_notification_ptr);
 682                 }
 683
 684                 switch (next_notification_status) {
 685                 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK:
 686                         break;
 687                 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED:
 688                         WARN("Dropped notification detected on notification channel used by the rotation management thread.");
 689                         return;
 690                 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED:
 691                         LTTNG_THROW_ERROR("Notification channel was closed");
 692                 default:
 693                         /* Unhandled conditions / errors. */
 694                         LTTNG_THROW_ERROR("Unknown notification channel status");
 695                 }
 696
 697                 _handle_notification(*notification);
 698         }
 699 }
 700
 701 void ls::rotation_thread::_thread_function() noexcept
 702 {
 703         DBG("Started rotation thread");
 704
 705         try {
 706                 _run();
 707         } catch (const std::exception& e) {
 708                 ERR_FMT("Fatal rotation thread error: {}", e.what());
 709         }
 710
 711         DBG("Thread exit");
 712 }
 713
 714 void ls::rotation_thread::_run()
 715 {
 716         rcu_register_thread();
 717         const auto unregister_rcu_thread =
 718                 lttng::make_scope_exit([]() noexcept { rcu_unregister_thread(); });
 719
 720         rcu_thread_online();
 721         const auto offline_rcu_thread =
 722                 lttng::make_scope_exit([]() noexcept { rcu_thread_offline(); });
 723
 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); });
 728
 729         const auto queue_pipe_fd = lttng_pipe_get_readfd(_rotation_timer_queue.event_pipe);
 730
 731         while (true) {
 732                 health_poll_entry();
 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);
 736                 health_poll_exit();
 737                 if (poll_wait_ret < 0) {
 738                         /*
 739                          * Restart interrupted system call.
 740                          */
 741                         if (errno == EINTR) {
 742                                 continue;
 743                         }
 744
 745                         LTTNG_THROW_POSIX("Error encountered during lttng_poll_wait", errno);
 746                 }
 747
 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);
 752
 753                         DBG_FMT("Handling descriptor activity: fd={}, events={:b}", fd, revents);
 754
 755                         if (revents & LPOLLERR) {
 756                                 LTTNG_THROW_ERROR(lttng::format(
 757                                         "Polling returned an error on fd: fd={}", fd));
 758                         }
 759
 760                         if (fd == _notification_channel->socket ||
 761                             fd == _notification_channel_subscribtion_change_eventfd.fd()) {
 762                                 try {
 763                                         _handle_notification_channel_activity();
 764                                 } catch (const lttng::ctl::error& e) {
 765                                         /*
 766                                          * The only non-fatal error (rotation failed), others
 767                                          * are caught at the top-level.
 768                                          */
 769                                         DBG_FMT("Control error occurred while handling activity on notification channel socket: {}",
 770                                                 e.what());
 771                                         continue;
 772                                 }
 773
 774                                 if (fd == _notification_channel_subscribtion_change_eventfd.fd()) {
 775                                         _notification_channel_subscribtion_change_eventfd
 776                                                 .decrement();
 777                                 }
 778                         } else {
 779                                 /* Job queue or quit pipe activity. */
 780
 781                                 /*
 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
 785                                  * are released.
 786                                  */
 787                                 _handle_job_queue();
 788                                 if (fd == queue_pipe_fd) {
 789                                         char buf;
 790
 791                                         if (lttng_read(fd, &buf, 1) != 1) {
 792                                                 LTTNG_THROW_POSIX(
 793                                                         lttng::format(
 794                                                                 "Failed to read from wakeup pipe: fd={}",
 795                                                                 fd),
 796                                                         errno);
 797                                         }
 798                                 } else {
 799                                         DBG("Quit pipe activity");
 800                                         return;
 801                                 }
 802                         }
 803                 }
 804         }
 805 }
 806
 807 bool ls::rotation_thread::shutdown() const noexcept
 808 {
 809         const int write_fd = lttng_pipe_get_writefd(_quit_pipe.get());
 810
 811         return notify_thread_pipe(write_fd) == 1;
 812 }
 813
 814 void ls::rotation_thread::launch_thread()
 815 {
 816         auto thread = lttng_thread_create(
 817                 "Rotation",
 818                 [](void *ptr) {
 819                         auto handle = reinterpret_cast<rotation_thread *>(ptr);
 820
 821                         handle->_thread_function();
 822                         return static_cast<void *>(nullptr);
 823                 },
 824                 shutdown_rotation_thread,
 825                 nullptr,
 826                 this);
 827         if (!thread) {
 828                 LTTNG_THROW_ERROR("Failed to launch rotation thread");
 829         }
 830
 831         lttng_thread_put(thread);
 832 }
 833
 834 void ls::rotation_thread::subscribe_session_consumed_size_rotation(ltt_session& session,
 835                                                                    std::uint64_t size)
 836 {
 837         const struct lttng_credentials session_creds = {
 838                 .uid = LTTNG_OPTIONAL_INIT_VALUE(session.uid),
 839                 .gid = LTTNG_OPTIONAL_INIT_VALUE(session.gid),
 840         };
 841
 842         ASSERT_LOCKED(session.lock);
 843
 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);
 848         }
 849
 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));
 855         }
 856
 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=`{}`",
 862                         session.name));
 863         }
 864
 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);
 869         }
 870
 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()));
 875         if (!trigger) {
 876                 LTTNG_THROW_POSIX("Could not create size-based rotation trigger", errno);
 877         }
 878
 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);
 882
 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");
 887         }
 888
 889         /*
 890          * Ensure any notification queued during the subscription are consumed by queueing an
 891          * event.
 892          */
 893         _notification_channel_subscribtion_change_eventfd.increment();
 894
 895         const auto register_ret = notification_thread_command_register_trigger(
 896                 &_notification_thread_handle, trigger.get(), true);
 897         if (register_ret != LTTNG_OK) {
 898                 LTTNG_THROW_CTL(
 899                         lttng::format(
 900                                 "Failed to register trigger for automatic size-based rotation: session_name{}, size={}",
 901                                 session.name,
 902                                 size),
 903                         register_ret);
 904         }
 905
 906         /* Ownership transferred to the session. */
 907         session.rotate_trigger = trigger.release();
 908 }
 909
 910 void ls::rotation_thread::unsubscribe_session_consumed_size_rotation(ltt_session& session)
 911 {
 912         LTTNG_ASSERT(session.rotate_trigger);
 913
 914         const auto remove_session_trigger = lttng::make_scope_exit([&session]() noexcept {
 915                 lttng_trigger_put(session.rotate_trigger);
 916                 session.rotate_trigger = nullptr;
 917         });
 918
 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={}",
 925                         session.name,
 926                         static_cast<int>(unsubscribe_status)));
 927         }
 928
 929         /*
 930          * Ensure any notification queued during the un-subscription are consumed by queueing an
 931          * event.
 932          */
 933         _notification_channel_subscribtion_change_eventfd.increment();
 934
 935         const auto unregister_status = notification_thread_command_unregister_trigger(
 936                 &_notification_thread_handle, session.rotate_trigger);
 937         if (unregister_status != LTTNG_OK) {
 938                 LTTNG_THROW_CTL(
 939                         lttng::format(
 940                                 "Failed to unregister trigger for automatic size-based rotation: session_name{}",
 941                                 session.name),
 942                         unregister_status);
 943         }
 944 }