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 <lttng/trigger/trigger.h>
  11 #include <common/error.hpp>
  12 #include <common/config/session-config.hpp>
  13 #include <common/defaults.hpp>
  14 #include <common/utils.hpp>
  15 #include <common/futex.hpp>
  16 #include <common/align.hpp>
  17 #include <common/time.hpp>
  18 #include <common/hashtable/utils.hpp>
  19 #include <sys/stat.h>
  20 #include <time.h>
  21 #include <signal.h>
  22 #include <inttypes.h>
  23
  24 #include <common/kernel-ctl/kernel-ctl.hpp>
  25 #include <lttng/notification/channel-internal.hpp>
  26 #include <lttng/rotate-internal.hpp>
  27 #include <lttng/location-internal.hpp>
  28 #include <lttng/condition/condition-internal.hpp>
  29 #include <lttng/notification/notification-internal.hpp>
  30
  31 #include "rotation-thread.hpp"
  32 #include "lttng-sessiond.hpp"
  33 #include "health-sessiond.hpp"
  34 #include "rotate.hpp"
  35 #include "cmd.hpp"
  36 #include "session.hpp"
  37 #include "timer.hpp"
  38 #include "notification-thread-commands.hpp"
  39 #include "utils.hpp"
  40 #include "thread.hpp"
  41
  42 #include <urcu.h>
  43 #include <urcu/list.h>
  44
  45 struct lttng_notification_channel *rotate_notification_channel = NULL;
  46
  47 struct rotation_thread {
  48         struct lttng_poll_event events;
  49 };
  50
  51 /*
  52  * The timer thread enqueues jobs and wakes up the rotation thread.
  53  * When the rotation thread wakes up, it empties the queue.
  54  */
  55 struct rotation_thread_timer_queue {
  56         struct lttng_pipe *event_pipe;
  57         struct cds_list_head list;
  58         pthread_mutex_t lock;
  59 };
  60
  61 struct rotation_thread_handle {
  62         struct rotation_thread_timer_queue *rotation_timer_queue;
  63         /* Access to the notification thread cmd_queue */
  64         struct notification_thread_handle *notification_thread_handle;
  65         /* Thread-specific quit pipe. */
  66         struct lttng_pipe *quit_pipe;
  67 };
  68
  69 namespace {
  70 struct rotation_thread_job {
  71         enum rotation_thread_job_type type;
  72         struct ltt_session *session;
  73         /* List member in struct rotation_thread_timer_queue. */
  74         struct cds_list_head head;
  75 };
  76 } /* namespace */
  77
  78 static
  79 const char *get_job_type_str(enum rotation_thread_job_type job_type)
  80 {
  81         switch (job_type) {
  82         case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION:
  83                 return "CHECK_PENDING_ROTATION";
  84         case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION:
  85                 return "SCHEDULED_ROTATION";
  86         default:
  87                 abort();
  88         }
  89 }
  90
  91 struct rotation_thread_timer_queue *rotation_thread_timer_queue_create(void)
  92 {
  93         struct rotation_thread_timer_queue *queue = NULL;
  94
  95         queue = zmalloc<rotation_thread_timer_queue>();
  96         if (!queue) {
  97                 PERROR("Failed to allocate timer rotate queue");
  98                 goto end;
  99         }
 100
 101         queue->event_pipe = lttng_pipe_open(FD_CLOEXEC | O_NONBLOCK);
 102         CDS_INIT_LIST_HEAD(&queue->list);
 103         pthread_mutex_init(&queue->lock, NULL);
 104 end:
 105         return queue;
 106 }
 107
 108 void rotation_thread_timer_queue_destroy(
 109                 struct rotation_thread_timer_queue *queue)
 110 {
 111         if (!queue) {
 112                 return;
 113         }
 114
 115         lttng_pipe_destroy(queue->event_pipe);
 116
 117         pthread_mutex_lock(&queue->lock);
 118         LTTNG_ASSERT(cds_list_empty(&queue->list));
 119         pthread_mutex_unlock(&queue->lock);
 120         pthread_mutex_destroy(&queue->lock);
 121         free(queue);
 122 }
 123
 124 /*
 125  * Destroy the thread data previously created by the init function.
 126  */
 127 void rotation_thread_handle_destroy(
 128                 struct rotation_thread_handle *handle)
 129 {
 130         lttng_pipe_destroy(handle->quit_pipe);
 131         free(handle);
 132 }
 133
 134 struct rotation_thread_handle *rotation_thread_handle_create(
 135                 struct rotation_thread_timer_queue *rotation_timer_queue,
 136                 struct notification_thread_handle *notification_thread_handle)
 137 {
 138         struct rotation_thread_handle *handle;
 139
 140         handle = zmalloc<rotation_thread_handle>();
 141         if (!handle) {
 142                 goto end;
 143         }
 144
 145         handle->rotation_timer_queue = rotation_timer_queue;
 146         handle->notification_thread_handle = notification_thread_handle;
 147         handle->quit_pipe = lttng_pipe_open(FD_CLOEXEC);
 148         if (!handle->quit_pipe) {
 149                 goto error;
 150         }
 151
 152 end:
 153         return handle;
 154 error:
 155         rotation_thread_handle_destroy(handle);
 156         return NULL;
 157 }
 158
 159 /*
 160  * Called with the rotation_thread_timer_queue lock held.
 161  * Return true if the same timer job already exists in the queue, false if not.
 162  */
 163 static
 164 bool timer_job_exists(const struct rotation_thread_timer_queue *queue,
 165                 enum rotation_thread_job_type job_type,
 166                 struct ltt_session *session)
 167 {
 168         bool exists = false;
 169         struct rotation_thread_job *job;
 170
 171         cds_list_for_each_entry(job, &queue->list, head) {
 172                 if (job->session == session && job->type == job_type) {
 173                         exists = true;
 174                         goto end;
 175                 }
 176         }
 177 end:
 178         return exists;
 179 }
 180
 181 void rotation_thread_enqueue_job(struct rotation_thread_timer_queue *queue,
 182                 enum rotation_thread_job_type job_type,
 183                 struct ltt_session *session)
 184 {
 185         int ret;
 186         const char dummy = '!';
 187         struct rotation_thread_job *job = NULL;
 188         const char *job_type_str = get_job_type_str(job_type);
 189
 190         pthread_mutex_lock(&queue->lock);
 191         if (timer_job_exists(queue, job_type, session)) {
 192                 /*
 193                  * This timer job is already pending, we don't need to add
 194                  * it.
 195                  */
 196                 goto end;
 197         }
 198
 199         job = zmalloc<rotation_thread_job>();
 200         if (!job) {
 201                 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
 202                                 job_type_str, session->name);
 203                 goto end;
 204         }
 205         /* No reason for this to fail as the caller must hold a reference. */
 206         (void) session_get(session);
 207
 208         job->session = session;
 209         job->type = job_type;
 210         cds_list_add_tail(&job->head, &queue->list);
 211
 212         ret = lttng_write(lttng_pipe_get_writefd(queue->event_pipe), &dummy,
 213                         sizeof(dummy));
 214         if (ret < 0) {
 215                 /*
 216                  * We do not want to block in the timer handler, the job has
 217                  * been enqueued in the list, the wakeup pipe is probably full,
 218                  * the job will be processed when the rotation_thread catches
 219                  * up.
 220                  */
 221                 DIAGNOSTIC_PUSH
 222                 DIAGNOSTIC_IGNORE_LOGICAL_OP
 223                 if (errno == EAGAIN || errno == EWOULDBLOCK) {
 224                 DIAGNOSTIC_POP
 225                         /*
 226                          * Not an error, but would be surprising and indicate
 227                          * that the rotation thread can't keep up with the
 228                          * current load.
 229                          */
 230                         DBG("Wake-up pipe of rotation thread job queue is full");
 231                         goto end;
 232                 }
 233                 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
 234                                 job_type_str, session->name);
 235                 goto end;
 236         }
 237
 238 end:
 239         pthread_mutex_unlock(&queue->lock);
 240 }
 241
 242 static
 243 int init_poll_set(struct lttng_poll_event *poll_set,
 244                 struct rotation_thread_handle *handle)
 245 {
 246         int ret;
 247
 248         /*
 249          * Create pollset with size 3:
 250          *      - rotation thread quit pipe,
 251          *      - rotation thread timer queue pipe,
 252          *      - notification channel sock,
 253          */
 254         ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
 255         if (ret < 0) {
 256                 goto error;
 257         }
 258
 259         ret = lttng_poll_add(poll_set,
 260                         lttng_pipe_get_readfd(handle->quit_pipe), LPOLLIN);
 261         if (ret < 0) {
 262                 ERR("Failed to add quit pipe read fd to poll set");
 263                 goto error;
 264         }
 265
 266         ret = lttng_poll_add(poll_set,
 267                         lttng_pipe_get_readfd(handle->rotation_timer_queue->event_pipe),
 268                         LPOLLIN);
 269         if (ret < 0) {
 270                 ERR("Failed to add rotate_pending fd to poll set");
 271                 goto error;
 272         }
 273
 274         return ret;
 275 error:
 276         lttng_poll_clean(poll_set);
 277         return ret;
 278 }
 279
 280 static
 281 void fini_thread_state(struct rotation_thread *state)
 282 {
 283         lttng_poll_clean(&state->events);
 284         if (rotate_notification_channel) {
 285                 lttng_notification_channel_destroy(rotate_notification_channel);
 286         }
 287 }
 288
 289 static
 290 int init_thread_state(struct rotation_thread_handle *handle,
 291                 struct rotation_thread *state)
 292 {
 293         int ret;
 294
 295         memset(state, 0, sizeof(*state));
 296         lttng_poll_init(&state->events);
 297
 298         ret = init_poll_set(&state->events, handle);
 299         if (ret) {
 300                 ERR("Failed to initialize rotation thread poll set");
 301                 goto end;
 302         }
 303
 304         rotate_notification_channel = lttng_notification_channel_create(
 305                         lttng_session_daemon_notification_endpoint);
 306         if (!rotate_notification_channel) {
 307                 ERR("Could not create notification channel");
 308                 ret = -1;
 309                 goto end;
 310         }
 311         ret = lttng_poll_add(&state->events, rotate_notification_channel->socket,
 312                         LPOLLIN);
 313         if (ret < 0) {
 314                 ERR("Failed to add notification fd to pollset");
 315                 goto end;
 316         }
 317
 318 end:
 319         return ret;
 320 }
 321
 322 static
 323 void check_session_rotation_pending_on_consumers(struct ltt_session *session,
 324                 bool *_rotation_completed)
 325 {
 326         int ret = 0;
 327         struct consumer_socket *socket;
 328         struct cds_lfht_iter iter;
 329         enum consumer_trace_chunk_exists_status exists_status;
 330         uint64_t relayd_id;
 331         bool chunk_exists_on_peer = false;
 332         enum lttng_trace_chunk_status chunk_status;
 333
 334         LTTNG_ASSERT(session->chunk_being_archived);
 335
 336         /*
 337          * Check for a local pending rotation on all consumers (32-bit
 338          * user space, 64-bit user space, and kernel).
 339          */
 340         rcu_read_lock();
 341         if (!session->ust_session) {
 342                 goto skip_ust;
 343         }
 344         cds_lfht_for_each_entry(session->ust_session->consumer->socks->ht,
 345                         &iter, socket, node.node) {
 346                 relayd_id = session->ust_session->consumer->type == CONSUMER_DST_LOCAL ?
 347                                 -1ULL :
 348                                 session->ust_session->consumer->net_seq_index;
 349
 350                 pthread_mutex_lock(socket->lock);
 351                 ret = consumer_trace_chunk_exists(socket,
 352                                 relayd_id,
 353                                 session->id, session->chunk_being_archived,
 354                                 &exists_status);
 355                 if (ret) {
 356                         pthread_mutex_unlock(socket->lock);
 357                         ERR("Error occurred while checking rotation status on consumer daemon");
 358                         goto end;
 359                 }
 360
 361                 if (exists_status != CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK) {
 362                         pthread_mutex_unlock(socket->lock);
 363                         chunk_exists_on_peer = true;
 364                         goto end;
 365                 }
 366                 pthread_mutex_unlock(socket->lock);
 367         }
 368
 369 skip_ust:
 370         if (!session->kernel_session) {
 371                 goto skip_kernel;
 372         }
 373         cds_lfht_for_each_entry(session->kernel_session->consumer->socks->ht,
 374                                 &iter, socket, node.node) {
 375                 pthread_mutex_lock(socket->lock);
 376                 relayd_id = session->kernel_session->consumer->type == CONSUMER_DST_LOCAL ?
 377                                 -1ULL :
 378                                 session->kernel_session->consumer->net_seq_index;
 379
 380                 ret = consumer_trace_chunk_exists(socket,
 381                                 relayd_id,
 382                                 session->id, session->chunk_being_archived,
 383                                 &exists_status);
 384                 if (ret) {
 385                         pthread_mutex_unlock(socket->lock);
 386                         ERR("Error occurred while checking rotation status on consumer daemon");
 387                         goto end;
 388                 }
 389
 390                 if (exists_status != CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK) {
 391                         pthread_mutex_unlock(socket->lock);
 392                         chunk_exists_on_peer = true;
 393                         goto end;
 394                 }
 395                 pthread_mutex_unlock(socket->lock);
 396         }
 397 skip_kernel:
 398 end:
 399         rcu_read_unlock();
 400
 401         if (!chunk_exists_on_peer) {
 402                 uint64_t chunk_being_archived_id;
 403
 404                 chunk_status = lttng_trace_chunk_get_id(
 405                                 session->chunk_being_archived,
 406                                 &chunk_being_archived_id);
 407                 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 408                 DBG("Rotation of trace archive %" PRIu64 " of session \"%s\" is complete on all consumers",
 409                                 chunk_being_archived_id,
 410                                 session->name);
 411         }
 412         *_rotation_completed = !chunk_exists_on_peer;
 413         if (ret) {
 414                 ret = session_reset_rotation_state(session,
 415                                 LTTNG_ROTATION_STATE_ERROR);
 416                 if (ret) {
 417                         ERR("Failed to reset rotation state of session \"%s\"",
 418                                         session->name);
 419                 }
 420         }
 421 }
 422
 423 /*
 424  * Check if the last rotation was completed, called with session lock held.
 425  * Should only return non-zero in the event of a fatal error. Doing so will
 426  * shutdown the thread.
 427  */
 428 static
 429 int check_session_rotation_pending(struct ltt_session *session,
 430                 struct notification_thread_handle *notification_thread_handle)
 431 {
 432         int ret;
 433         struct lttng_trace_archive_location *location;
 434         enum lttng_trace_chunk_status chunk_status;
 435         bool rotation_completed = false;
 436         const char *archived_chunk_name;
 437         uint64_t chunk_being_archived_id;
 438
 439         if (!session->chunk_being_archived) {
 440                 ret = 0;
 441                 goto end;
 442         }
 443
 444         chunk_status = lttng_trace_chunk_get_id(session->chunk_being_archived,
 445                         &chunk_being_archived_id);
 446         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 447
 448         DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64,
 449                         session->name, chunk_being_archived_id);
 450
 451         /*
 452          * The rotation-pending check timer of a session is launched in
 453          * one-shot mode. If the rotation is incomplete, the rotation
 454          * thread will re-enable the pending-check timer.
 455          *
 456          * The timer thread can't stop the timer itself since it is involved
 457          * in the check for the timer's quiescence.
 458          */
 459         ret = timer_session_rotation_pending_check_stop(session);
 460         if (ret) {
 461                 goto check_ongoing_rotation;
 462         }
 463
 464         check_session_rotation_pending_on_consumers(session,
 465                         &rotation_completed);
 466         if (!rotation_completed ||
 467                         session->rotation_state == LTTNG_ROTATION_STATE_ERROR) {
 468                 goto check_ongoing_rotation;
 469         }
 470
 471         /*
 472          * Now we can clear the "ONGOING" state in the session. New
 473          * rotations can start now.
 474          */
 475         chunk_status = lttng_trace_chunk_get_name(session->chunk_being_archived,
 476                         &archived_chunk_name, NULL);
 477         LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 478         free(session->last_archived_chunk_name);
 479         session->last_archived_chunk_name = strdup(archived_chunk_name);
 480         if (!session->last_archived_chunk_name) {
 481                 PERROR("Failed to duplicate archived chunk name");
 482         }
 483         session_reset_rotation_state(session, LTTNG_ROTATION_STATE_COMPLETED);
 484
 485         if (!session->quiet_rotation) {
 486                 location = session_get_trace_archive_location(session);
 487                 ret = notification_thread_command_session_rotation_completed(
 488                                 notification_thread_handle,
 489                                 session->id,
 490                                 session->last_archived_chunk_id.value,
 491                                 location);
 492                 lttng_trace_archive_location_put(location);
 493                 if (ret != LTTNG_OK) {
 494                         ERR("Failed to notify notification thread of completed rotation for session %s",
 495                                         session->name);
 496                 }
 497         }
 498
 499         ret = 0;
 500 check_ongoing_rotation:
 501         if (session->rotation_state == LTTNG_ROTATION_STATE_ONGOING) {
 502                 chunk_status = lttng_trace_chunk_get_id(
 503                                 session->chunk_being_archived,
 504                                 &chunk_being_archived_id);
 505                 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 506
 507                 DBG("Rotation of trace archive %" PRIu64 " is still pending for session %s",
 508                                 chunk_being_archived_id, session->name);
 509                 ret = timer_session_rotation_pending_check_start(session,
 510                                 DEFAULT_ROTATE_PENDING_TIMER);
 511                 if (ret) {
 512                         ERR("Failed to re-enable rotation pending timer");
 513                         ret = -1;
 514                         goto end;
 515                 }
 516         }
 517
 518 end:
 519         return ret;
 520 }
 521
 522 /* Call with the session and session_list locks held. */
 523 static
 524 int launch_session_rotation(struct ltt_session *session)
 525 {
 526         int ret;
 527         struct lttng_rotate_session_return rotation_return;
 528
 529         DBG("Launching scheduled time-based rotation on session \"%s\"",
 530                         session->name);
 531
 532         ret = cmd_rotate_session(session, &rotation_return, false,
 533                 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
 534         if (ret == LTTNG_OK) {
 535                 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
 536                                 session->name);
 537         } else {
 538                 /* Don't consider errors as fatal. */
 539                 DBG("Scheduled time-based rotation aborted for session %s: %s",
 540                                 session->name, lttng_strerror(ret));
 541         }
 542         return 0;
 543 }
 544
 545 static
 546 int run_job(struct rotation_thread_job *job, struct ltt_session *session,
 547                 struct notification_thread_handle *notification_thread_handle)
 548 {
 549         int ret;
 550
 551         switch (job->type) {
 552         case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION:
 553                 ret = launch_session_rotation(session);
 554                 break;
 555         case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION:
 556                 ret = check_session_rotation_pending(session,
 557                                 notification_thread_handle);
 558                 break;
 559         default:
 560                 abort();
 561         }
 562         return ret;
 563 }
 564
 565 static
 566 int handle_job_queue(struct rotation_thread_handle *handle,
 567                 struct rotation_thread *state __attribute__((unused)),
 568                 struct rotation_thread_timer_queue *queue)
 569 {
 570         int ret = 0;
 571
 572         for (;;) {
 573                 struct ltt_session *session;
 574                 struct rotation_thread_job *job;
 575
 576                 /* Take the queue lock only to pop an element from the list. */
 577                 pthread_mutex_lock(&queue->lock);
 578                 if (cds_list_empty(&queue->list)) {
 579                         pthread_mutex_unlock(&queue->lock);
 580                         break;
 581                 }
 582                 job = cds_list_first_entry(&queue->list,
 583                                 typeof(*job), head);
 584                 cds_list_del(&job->head);
 585                 pthread_mutex_unlock(&queue->lock);
 586
 587                 session_lock_list();
 588                 session = job->session;
 589                 if (!session) {
 590                         DBG("Session \"%s\" not found",
 591                                         session->name != NULL ? session->name : "");
 592                         /*
 593                          * This is a non-fatal error, and we cannot report it to
 594                          * the user (timer), so just print the error and
 595                          * continue the processing.
 596                          *
 597                          * While the timer thread will purge pending signals for
 598                          * a session on the session's destruction, it is
 599                          * possible for a job targeting that session to have
 600                          * already been queued before it was destroyed.
 601                          */
 602                         free(job);
 603                         session_put(session);
 604                         session_unlock_list();
 605                         continue;
 606                 }
 607
 608                 session_lock(session);
 609                 ret = run_job(job, session, handle->notification_thread_handle);
 610                 session_unlock(session);
 611                 /* Release reference held by the job. */
 612                 session_put(session);
 613                 session_unlock_list();
 614                 free(job);
 615                 if (ret) {
 616                         goto end;
 617                 }
 618         }
 619
 620         ret = 0;
 621
 622 end:
 623         return ret;
 624 }
 625
 626 static
 627 int handle_condition(const struct lttng_notification *notification,
 628                 struct notification_thread_handle *notification_thread_handle)
 629 {
 630         int ret = 0;
 631         const char *condition_session_name = NULL;
 632         enum lttng_condition_type condition_type;
 633         enum lttng_condition_status condition_status;
 634         enum lttng_evaluation_status evaluation_status;
 635         uint64_t consumed;
 636         struct ltt_session *session;
 637         const struct lttng_condition *condition =
 638                         lttng_notification_get_const_condition(notification);
 639         const struct lttng_evaluation *evaluation =
 640                         lttng_notification_get_const_evaluation(notification);
 641
 642         condition_type = lttng_condition_get_type(condition);
 643
 644         if (condition_type != LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE) {
 645                 ret = -1;
 646                 ERR("Condition type and session usage type are not the same");
 647                 goto end;
 648         }
 649
 650         /* Fetch info to test */
 651         condition_status = lttng_condition_session_consumed_size_get_session_name(
 652                         condition, &condition_session_name);
 653         if (condition_status != LTTNG_CONDITION_STATUS_OK) {
 654                 ERR("Session name could not be fetched");
 655                 ret = -1;
 656                 goto end;
 657         }
 658         evaluation_status = lttng_evaluation_session_consumed_size_get_consumed_size(evaluation,
 659                         &consumed);
 660         if (evaluation_status != LTTNG_EVALUATION_STATUS_OK) {
 661                 ERR("Failed to get evaluation");
 662                 ret = -1;
 663                 goto end;
 664         }
 665
 666         session_lock_list();
 667         session = session_find_by_name(condition_session_name);
 668         if (!session) {
 669                 DBG("Failed to find session while handling notification: notification type = %s, session name = `%s`",
 670                                 lttng_condition_type_str(condition_type),
 671                                 condition_session_name);
 672                 /*
 673                  * Not a fatal error: a session can be destroyed before we get
 674                  * the chance to handle the notification.
 675                  */
 676                 ret = 0;
 677                 session_unlock_list();
 678                 goto end;
 679         }
 680         session_lock(session);
 681
 682         if (!lttng_trigger_is_equal(session->rotate_trigger,
 683                         lttng_notification_get_const_trigger(notification))) {
 684                 /* Notification does not originate from our rotation trigger. */
 685                 ret = 0;
 686                 goto end_unlock;
 687         }
 688
 689         ret = unsubscribe_session_consumed_size_rotation(session,
 690                         notification_thread_handle);
 691         if (ret) {
 692                 goto end_unlock;
 693         }
 694
 695         ret = cmd_rotate_session(
 696                         session, NULL, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
 697         switch (ret) {
 698         case LTTNG_OK:
 699                 break;
 700         case -LTTNG_ERR_ROTATION_PENDING:
 701                 DBG("Rotate already pending, subscribe to the next threshold value");
 702                 break;
 703         case -LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
 704                 DBG("Rotation already happened since last stop, subscribe to the next threshold value");
 705                 break;
 706         case -LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
 707                 DBG("Rotation already happened since last stop and clear, subscribe to the next threshold value");
 708                 break;
 709         default:
 710                 ERR("Failed to rotate on size notification with error: %s", lttng_strerror(ret));
 711                 ret = -1;
 712                 goto end_unlock;
 713         }
 714
 715         ret = subscribe_session_consumed_size_rotation(
 716                         session, consumed + session->rotate_size, notification_thread_handle);
 717         if (ret) {
 718                 ERR("Failed to subscribe to session consumed size condition");
 719                 goto end_unlock;
 720         }
 721         ret = 0;
 722
 723 end_unlock:
 724         session_unlock(session);
 725         session_put(session);
 726         session_unlock_list();
 727 end:
 728         return ret;
 729 }
 730
 731 static
 732 int handle_notification_channel(int fd __attribute__((unused)),
 733                 struct rotation_thread_handle *handle,
 734                 struct rotation_thread *state __attribute__((unused)))
 735 {
 736         int ret;
 737         bool notification_pending;
 738         struct lttng_notification *notification = NULL;
 739         enum lttng_notification_channel_status status;
 740
 741         status = lttng_notification_channel_has_pending_notification(
 742                         rotate_notification_channel, &notification_pending);
 743         if (status != LTTNG_NOTIFICATION_CHANNEL_STATUS_OK) {
 744                 ERR("Error occurred while checking for pending notification");
 745                 ret = -1;
 746                 goto end;
 747         }
 748
 749         if (!notification_pending) {
 750                 ret = 0;
 751                 goto end;
 752         }
 753
 754         /* Receive the next notification. */
 755         status = lttng_notification_channel_get_next_notification(
 756                         rotate_notification_channel,
 757                         &notification);
 758
 759         switch (status) {
 760         case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK:
 761                 break;
 762         case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED:
 763                 /* Not an error, we will wait for the next one */
 764                 ret = 0;
 765                 goto end;;
 766         case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED:
 767                 ERR("Notification channel was closed");
 768                 ret = -1;
 769                 goto end;
 770         default:
 771                 /* Unhandled conditions / errors. */
 772                 ERR("Unknown notification channel status");
 773                 ret = -1;
 774                 goto end;
 775         }
 776
 777         ret = handle_condition(notification,
 778                         handle->notification_thread_handle);
 779
 780 end:
 781         lttng_notification_destroy(notification);
 782         return ret;
 783 }
 784
 785 static
 786 void *thread_rotation(void *data)
 787 {
 788         int ret;
 789         struct rotation_thread_handle *handle = (rotation_thread_handle *) data;
 790         struct rotation_thread thread;
 791         int queue_pipe_fd;
 792
 793         DBG("Started rotation thread");
 794         rcu_register_thread();
 795         rcu_thread_online();
 796         health_register(the_health_sessiond, HEALTH_SESSIOND_TYPE_ROTATION);
 797         health_code_update();
 798
 799         if (!handle) {
 800                 ERR("Invalid thread context provided");
 801                 goto end;
 802         }
 803
 804         queue_pipe_fd = lttng_pipe_get_readfd(
 805                         handle->rotation_timer_queue->event_pipe);
 806
 807
 808         ret = init_thread_state(handle, &thread);
 809         if (ret) {
 810                 goto error;
 811         }
 812
 813         while (true) {
 814                 int fd_count, i;
 815
 816                 health_poll_entry();
 817                 DBG("Entering poll wait");
 818                 ret = lttng_poll_wait(&thread.events, -1);
 819                 DBG("Poll wait returned (%i)", ret);
 820                 health_poll_exit();
 821                 if (ret < 0) {
 822                         /*
 823                          * Restart interrupted system call.
 824                          */
 825                         if (errno == EINTR) {
 826                                 continue;
 827                         }
 828                         ERR("Error encountered during lttng_poll_wait (%i)", ret);
 829                         goto error;
 830                 }
 831
 832                 fd_count = ret;
 833                 for (i = 0; i < fd_count; i++) {
 834                         int fd = LTTNG_POLL_GETFD(&thread.events, i);
 835                         uint32_t revents = LTTNG_POLL_GETEV(&thread.events, i);
 836
 837                         DBG("Handling fd (%i) activity (%u)",
 838                                         fd, revents);
 839
 840                         if (revents & LPOLLERR) {
 841                                 ERR("Polling returned an error on fd %i", fd);
 842                                 goto error;
 843                         }
 844
 845                         if (fd == rotate_notification_channel->socket) {
 846                                 ret = handle_notification_channel(fd, handle,
 847                                                 &thread);
 848                                 if (ret) {
 849                                         ERR("Error occurred while handling activity on notification channel socket");
 850                                         goto error;
 851                                 }
 852                         } else {
 853                                 /* Job queue or quit pipe activity. */
 854
 855                                 /*
 856                                  * The job queue is serviced if there is
 857                                  * activity on the quit pipe to ensure it is
 858                                  * flushed and all references held in the queue
 859                                  * are released.
 860                                  */
 861                                 ret = handle_job_queue(handle, &thread,
 862                                                 handle->rotation_timer_queue);
 863                                 if (ret) {
 864                                         ERR("Failed to handle rotation timer pipe event");
 865                                         goto error;
 866                                 }
 867
 868                                 if (fd == queue_pipe_fd) {
 869                                         char buf;
 870
 871                                         ret = lttng_read(fd, &buf, 1);
 872                                         if (ret != 1) {
 873                                                 ERR("Failed to read from wakeup pipe (fd = %i)", fd);
 874                                                 goto error;
 875                                         }
 876                                 } else {
 877                                         DBG("Quit pipe activity");
 878                                         goto exit;
 879                                 }
 880                         }
 881                 }
 882         }
 883 exit:
 884 error:
 885         DBG("Thread exit");
 886         fini_thread_state(&thread);
 887 end:
 888         health_unregister(the_health_sessiond);
 889         rcu_thread_offline();
 890         rcu_unregister_thread();
 891         return NULL;
 892 }
 893
 894 static
 895 bool shutdown_rotation_thread(void *thread_data)
 896 {
 897         struct rotation_thread_handle *handle = (rotation_thread_handle *) thread_data;
 898         const int write_fd = lttng_pipe_get_writefd(handle->quit_pipe);
 899
 900         return notify_thread_pipe(write_fd) == 1;
 901 }
 902
 903 bool launch_rotation_thread(struct rotation_thread_handle *handle)
 904 {
 905         struct lttng_thread *thread;
 906
 907         thread = lttng_thread_create("Rotation",
 908                         thread_rotation,
 909                         shutdown_rotation_thread,
 910                         NULL,
 911                         handle);
 912         if (!thread) {
 913                 goto error;
 914         }
 915         lttng_thread_put(thread);
 916         return true;
 917 error:
 918         return false;
 919 }