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
  30 #include "rotation-thread.hpp"
  31 #include "lttng-sessiond.hpp"
  32 #include "health-sessiond.hpp"
  33 #include "rotate.hpp"
  34 #include "cmd.hpp"
  35 #include "session.hpp"
  36 #include "timer.hpp"
  37 #include "notification-thread-commands.hpp"
  38 #include "utils.hpp"
  39 #include "thread.hpp"
  40
  41 #include <urcu.h>
  42 #include <urcu/list.h>
  43
  44 struct lttng_notification_channel *rotate_notification_channel = NULL;
  45
  46 struct rotation_thread {
  47         struct lttng_poll_event events;
  48 };
  49
  50 /*
  51  * The timer thread enqueues jobs and wakes up the rotation thread.
  52  * When the rotation thread wakes up, it empties the queue.
  53  */
  54 struct rotation_thread_timer_queue {
  55         struct lttng_pipe *event_pipe;
  56         struct cds_list_head list;
  57         pthread_mutex_t lock;
  58 };
  59
  60 struct rotation_thread_handle {
  61         struct rotation_thread_timer_queue *rotation_timer_queue;
  62         /* Access to the notification thread cmd_queue */
  63         struct notification_thread_handle *notification_thread_handle;
  64         /* Thread-specific quit pipe. */
  65         struct lttng_pipe *quit_pipe;
  66 };
  67
  68 namespace {
  69 struct rotation_thread_job {
  70         enum rotation_thread_job_type type;
  71         struct ltt_session *session;
  72         /* List member in struct rotation_thread_timer_queue. */
  73         struct cds_list_head head;
  74 };
  75 } /* namespace */
  76
  77 static
  78 const char *get_job_type_str(enum rotation_thread_job_type job_type)
  79 {
  80         switch (job_type) {
  81         case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION:
  82                 return "CHECK_PENDING_ROTATION";
  83         case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION:
  84                 return "SCHEDULED_ROTATION";
  85         default:
  86                 abort();
  87         }
  88 }
  89
  90 struct rotation_thread_timer_queue *rotation_thread_timer_queue_create(void)
  91 {
  92         struct rotation_thread_timer_queue *queue = NULL;
  93
  94         queue = zmalloc<rotation_thread_timer_queue>();
  95         if (!queue) {
  96                 PERROR("Failed to allocate timer rotate queue");
  97                 goto end;
  98         }
  99
 100         queue->event_pipe = lttng_pipe_open(FD_CLOEXEC | O_NONBLOCK);
 101         CDS_INIT_LIST_HEAD(&queue->list);
 102         pthread_mutex_init(&queue->lock, NULL);
 103 end:
 104         return queue;
 105 }
 106
 107 void rotation_thread_timer_queue_destroy(
 108                 struct rotation_thread_timer_queue *queue)
 109 {
 110         if (!queue) {
 111                 return;
 112         }
 113
 114         lttng_pipe_destroy(queue->event_pipe);
 115
 116         pthread_mutex_lock(&queue->lock);
 117         LTTNG_ASSERT(cds_list_empty(&queue->list));
 118         pthread_mutex_unlock(&queue->lock);
 119         pthread_mutex_destroy(&queue->lock);
 120         free(queue);
 121 }
 122
 123 /*
 124  * Destroy the thread data previously created by the init function.
 125  */
 126 void rotation_thread_handle_destroy(
 127                 struct rotation_thread_handle *handle)
 128 {
 129         lttng_pipe_destroy(handle->quit_pipe);
 130         free(handle);
 131 }
 132
 133 struct rotation_thread_handle *rotation_thread_handle_create(
 134                 struct rotation_thread_timer_queue *rotation_timer_queue,
 135                 struct notification_thread_handle *notification_thread_handle)
 136 {
 137         struct rotation_thread_handle *handle;
 138
 139         handle = zmalloc<rotation_thread_handle>();
 140         if (!handle) {
 141                 goto end;
 142         }
 143
 144         handle->rotation_timer_queue = rotation_timer_queue;
 145         handle->notification_thread_handle = notification_thread_handle;
 146         handle->quit_pipe = lttng_pipe_open(FD_CLOEXEC);
 147         if (!handle->quit_pipe) {
 148                 goto error;
 149         }
 150
 151 end:
 152         return handle;
 153 error:
 154         rotation_thread_handle_destroy(handle);
 155         return NULL;
 156 }
 157
 158 /*
 159  * Called with the rotation_thread_timer_queue lock held.
 160  * Return true if the same timer job already exists in the queue, false if not.
 161  */
 162 static
 163 bool timer_job_exists(const struct rotation_thread_timer_queue *queue,
 164                 enum rotation_thread_job_type job_type,
 165                 struct ltt_session *session)
 166 {
 167         bool exists = false;
 168         struct rotation_thread_job *job;
 169
 170         cds_list_for_each_entry(job, &queue->list, head) {
 171                 if (job->session == session && job->type == job_type) {
 172                         exists = true;
 173                         goto end;
 174                 }
 175         }
 176 end:
 177         return exists;
 178 }
 179
 180 void rotation_thread_enqueue_job(struct rotation_thread_timer_queue *queue,
 181                 enum rotation_thread_job_type job_type,
 182                 struct ltt_session *session)
 183 {
 184         int ret;
 185         const char dummy = '!';
 186         struct rotation_thread_job *job = NULL;
 187         const char *job_type_str = get_job_type_str(job_type);
 188
 189         pthread_mutex_lock(&queue->lock);
 190         if (timer_job_exists(queue, job_type, session)) {
 191                 /*
 192                  * This timer job is already pending, we don't need to add
 193                  * it.
 194                  */
 195                 goto end;
 196         }
 197
 198         job = zmalloc<rotation_thread_job>();
 199         if (!job) {
 200                 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
 201                                 job_type_str, session->name);
 202                 goto end;
 203         }
 204         /* No reason for this to fail as the caller must hold a reference. */
 205         (void) session_get(session);
 206
 207         job->session = session;
 208         job->type = job_type;
 209         cds_list_add_tail(&job->head, &queue->list);
 210
 211         ret = lttng_write(lttng_pipe_get_writefd(queue->event_pipe), &dummy,
 212                         sizeof(dummy));
 213         if (ret < 0) {
 214                 /*
 215                  * We do not want to block in the timer handler, the job has
 216                  * been enqueued in the list, the wakeup pipe is probably full,
 217                  * the job will be processed when the rotation_thread catches
 218                  * up.
 219                  */
 220                 DIAGNOSTIC_PUSH
 221                 DIAGNOSTIC_IGNORE_LOGICAL_OP
 222                 if (errno == EAGAIN || errno == EWOULDBLOCK) {
 223                 DIAGNOSTIC_POP
 224                         /*
 225                          * Not an error, but would be surprising and indicate
 226                          * that the rotation thread can't keep up with the
 227                          * current load.
 228                          */
 229                         DBG("Wake-up pipe of rotation thread job queue is full");
 230                         goto end;
 231                 }
 232                 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
 233                                 job_type_str, session->name);
 234                 goto end;
 235         }
 236
 237 end:
 238         pthread_mutex_unlock(&queue->lock);
 239 }
 240
 241 static
 242 int init_poll_set(struct lttng_poll_event *poll_set,
 243                 struct rotation_thread_handle *handle)
 244 {
 245         int ret;
 246
 247         /*
 248          * Create pollset with size 3:
 249          *      - rotation thread quit pipe,
 250          *      - rotation thread timer queue pipe,
 251          *      - notification channel sock,
 252          */
 253         ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
 254         if (ret < 0) {
 255                 goto error;
 256         }
 257
 258         ret = lttng_poll_add(poll_set,
 259                         lttng_pipe_get_readfd(handle->quit_pipe),
 260                         LPOLLIN | LPOLLERR);
 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 | LPOLLERR);
 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 | LPOLLERR);
 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->name,
 490                                 session->uid,
 491                                 session->gid,
 492                                 session->last_archived_chunk_id.value,
 493                                 location);
 494                 lttng_trace_archive_location_put(location);
 495                 if (ret != LTTNG_OK) {
 496                         ERR("Failed to notify notification thread of completed rotation for session %s",
 497                                         session->name);
 498                 }
 499         }
 500
 501         ret = 0;
 502 check_ongoing_rotation:
 503         if (session->rotation_state == LTTNG_ROTATION_STATE_ONGOING) {
 504                 chunk_status = lttng_trace_chunk_get_id(
 505                                 session->chunk_being_archived,
 506                                 &chunk_being_archived_id);
 507                 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
 508
 509                 DBG("Rotation of trace archive %" PRIu64 " is still pending for session %s",
 510                                 chunk_being_archived_id, session->name);
 511                 ret = timer_session_rotation_pending_check_start(session,
 512                                 DEFAULT_ROTATE_PENDING_TIMER);
 513                 if (ret) {
 514                         ERR("Failed to re-enable rotation pending timer");
 515                         ret = -1;
 516                         goto end;
 517                 }
 518         }
 519
 520 end:
 521         return ret;
 522 }
 523
 524 /* Call with the session and session_list locks held. */
 525 static
 526 int launch_session_rotation(struct ltt_session *session)
 527 {
 528         int ret;
 529         struct lttng_rotate_session_return rotation_return;
 530
 531         DBG("Launching scheduled time-based rotation on session \"%s\"",
 532                         session->name);
 533
 534         ret = cmd_rotate_session(session, &rotation_return, false,
 535                 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
 536         if (ret == LTTNG_OK) {
 537                 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
 538                                 session->name);
 539         } else {
 540                 /* Don't consider errors as fatal. */
 541                 DBG("Scheduled time-based rotation aborted for session %s: %s",
 542                                 session->name, lttng_strerror(ret));
 543         }
 544         return 0;
 545 }
 546
 547 static
 548 int run_job(struct rotation_thread_job *job, struct ltt_session *session,
 549                 struct notification_thread_handle *notification_thread_handle)
 550 {
 551         int ret;
 552
 553         switch (job->type) {
 554         case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION:
 555                 ret = launch_session_rotation(session);
 556                 break;
 557         case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION:
 558                 ret = check_session_rotation_pending(session,
 559                                 notification_thread_handle);
 560                 break;
 561         default:
 562                 abort();
 563         }
 564         return ret;
 565 }
 566
 567 static
 568 int handle_job_queue(struct rotation_thread_handle *handle,
 569                 struct rotation_thread *state __attribute__((unused)),
 570                 struct rotation_thread_timer_queue *queue)
 571 {
 572         int ret = 0;
 573
 574         for (;;) {
 575                 struct ltt_session *session;
 576                 struct rotation_thread_job *job;
 577
 578                 /* Take the queue lock only to pop an element from the list. */
 579                 pthread_mutex_lock(&queue->lock);
 580                 if (cds_list_empty(&queue->list)) {
 581                         pthread_mutex_unlock(&queue->lock);
 582                         break;
 583                 }
 584                 job = cds_list_first_entry(&queue->list,
 585                                 typeof(*job), head);
 586                 cds_list_del(&job->head);
 587                 pthread_mutex_unlock(&queue->lock);
 588
 589                 session_lock_list();
 590                 session = job->session;
 591                 if (!session) {
 592                         DBG("Session \"%s\" not found",
 593                                         session->name != NULL ? session->name : "");
 594                         /*
 595                          * This is a non-fatal error, and we cannot report it to
 596                          * the user (timer), so just print the error and
 597                          * continue the processing.
 598                          *
 599                          * While the timer thread will purge pending signals for
 600                          * a session on the session's destruction, it is
 601                          * possible for a job targeting that session to have
 602                          * already been queued before it was destroyed.
 603                          */
 604                         free(job);
 605                         session_put(session);
 606                         session_unlock_list();
 607                         continue;
 608                 }
 609
 610                 session_lock(session);
 611                 ret = run_job(job, session, handle->notification_thread_handle);
 612                 session_unlock(session);
 613                 /* Release reference held by the job. */
 614                 session_put(session);
 615                 session_unlock_list();
 616                 free(job);
 617                 if (ret) {
 618                         goto end;
 619                 }
 620         }
 621
 622         ret = 0;
 623
 624 end:
 625         return ret;
 626 }
 627
 628 static
 629 int handle_condition(const struct lttng_condition *condition,
 630                 const struct lttng_evaluation *evaluation,
 631                 struct notification_thread_handle *notification_thread_handle)
 632 {
 633         int ret = 0;
 634         const char *condition_session_name = NULL;
 635         enum lttng_condition_type condition_type;
 636         enum lttng_condition_status condition_status;
 637         enum lttng_evaluation_status evaluation_status;
 638         uint64_t consumed;
 639         struct ltt_session *session;
 640
 641         condition_type = lttng_condition_get_type(condition);
 642
 643         if (condition_type != LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE) {
 644                 ret = -1;
 645                 ERR("Condition type and session usage type are not the same");
 646                 goto end;
 647         }
 648
 649         /* Fetch info to test */
 650         condition_status = lttng_condition_session_consumed_size_get_session_name(
 651                         condition, &condition_session_name);
 652         if (condition_status != LTTNG_CONDITION_STATUS_OK) {
 653                 ERR("Session name could not be fetched");
 654                 ret = -1;
 655                 goto end;
 656         }
 657         evaluation_status = lttng_evaluation_session_consumed_size_get_consumed_size(evaluation,
 658                         &consumed);
 659         if (evaluation_status != LTTNG_EVALUATION_STATUS_OK) {
 660                 ERR("Failed to get evaluation");
 661                 ret = -1;
 662                 goto end;
 663         }
 664
 665         session_lock_list();
 666         session = session_find_by_name(condition_session_name);
 667         if (!session) {
 668                 DBG("Failed to find session while handling notification: notification type = %s, session name = `%s`",
 669                                 lttng_condition_type_str(condition_type),
 670                                 condition_session_name);
 671                 /*
 672                  * Not a fatal error: a session can be destroyed before we get
 673                  * the chance to handle the notification.
 674                  */
 675                 ret = 0;
 676                 session_unlock_list();
 677                 goto end;
 678         }
 679         session_lock(session);
 680
 681         ret = unsubscribe_session_consumed_size_rotation(session,
 682                         notification_thread_handle);
 683         if (ret) {
 684                 goto end_unlock;
 685         }
 686
 687         ret = cmd_rotate_session(session, NULL, false,
 688                 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
 689         if (ret == -LTTNG_ERR_ROTATION_PENDING) {
 690                 DBG("Rotate already pending, subscribe to the next threshold value");
 691         } else if (ret != LTTNG_OK) {
 692                 ERR("Failed to rotate on size notification with error: %s",
 693                                 lttng_strerror(ret));
 694                 ret = -1;
 695                 goto end_unlock;
 696         }
 697         ret = subscribe_session_consumed_size_rotation(session,
 698                         consumed + session->rotate_size,
 699                         notification_thread_handle);
 700         if (ret) {
 701                 ERR("Failed to subscribe to session consumed size condition");
 702                 goto end_unlock;
 703         }
 704         ret = 0;
 705
 706 end_unlock:
 707         session_unlock(session);
 708         session_put(session);
 709         session_unlock_list();
 710 end:
 711         return ret;
 712 }
 713
 714 static
 715 int handle_notification_channel(int fd __attribute__((unused)),
 716                 struct rotation_thread_handle *handle,
 717                 struct rotation_thread *state __attribute__((unused)))
 718 {
 719         int ret;
 720         bool notification_pending;
 721         struct lttng_notification *notification = NULL;
 722         enum lttng_notification_channel_status status;
 723         const struct lttng_evaluation *notification_evaluation;
 724         const struct lttng_condition *notification_condition;
 725
 726         status = lttng_notification_channel_has_pending_notification(
 727                         rotate_notification_channel, &notification_pending);
 728         if (status != LTTNG_NOTIFICATION_CHANNEL_STATUS_OK) {
 729                 ERR("Error occurred while checking for pending notification");
 730                 ret = -1;
 731                 goto end;
 732         }
 733
 734         if (!notification_pending) {
 735                 ret = 0;
 736                 goto end;
 737         }
 738
 739         /* Receive the next notification. */
 740         status = lttng_notification_channel_get_next_notification(
 741                         rotate_notification_channel,
 742                         &notification);
 743
 744         switch (status) {
 745         case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK:
 746                 break;
 747         case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED:
 748                 /* Not an error, we will wait for the next one */
 749                 ret = 0;
 750                 goto end;;
 751         case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED:
 752                 ERR("Notification channel was closed");
 753                 ret = -1;
 754                 goto end;
 755         default:
 756                 /* Unhandled conditions / errors. */
 757                 ERR("Unknown notification channel status");
 758                 ret = -1;
 759                 goto end;
 760         }
 761
 762         notification_condition = lttng_notification_get_condition(notification);
 763         notification_evaluation = lttng_notification_get_evaluation(notification);
 764
 765         ret = handle_condition(notification_condition, notification_evaluation,
 766                         handle->notification_thread_handle);
 767
 768 end:
 769         lttng_notification_destroy(notification);
 770         return ret;
 771 }
 772
 773 static
 774 void *thread_rotation(void *data)
 775 {
 776         int ret;
 777         struct rotation_thread_handle *handle = (rotation_thread_handle *) data;
 778         struct rotation_thread thread;
 779         int queue_pipe_fd;
 780
 781         DBG("Started rotation thread");
 782         rcu_register_thread();
 783         rcu_thread_online();
 784         health_register(the_health_sessiond, HEALTH_SESSIOND_TYPE_ROTATION);
 785         health_code_update();
 786
 787         if (!handle) {
 788                 ERR("Invalid thread context provided");
 789                 goto end;
 790         }
 791
 792         queue_pipe_fd = lttng_pipe_get_readfd(
 793                         handle->rotation_timer_queue->event_pipe);
 794
 795
 796         ret = init_thread_state(handle, &thread);
 797         if (ret) {
 798                 goto error;
 799         }
 800
 801         while (true) {
 802                 int fd_count, i;
 803
 804                 health_poll_entry();
 805                 DBG("Entering poll wait");
 806                 ret = lttng_poll_wait(&thread.events, -1);
 807                 DBG("Poll wait returned (%i)", ret);
 808                 health_poll_exit();
 809                 if (ret < 0) {
 810                         /*
 811                          * Restart interrupted system call.
 812                          */
 813                         if (errno == EINTR) {
 814                                 continue;
 815                         }
 816                         ERR("Error encountered during lttng_poll_wait (%i)", ret);
 817                         goto error;
 818                 }
 819
 820                 fd_count = ret;
 821                 for (i = 0; i < fd_count; i++) {
 822                         int fd = LTTNG_POLL_GETFD(&thread.events, i);
 823                         uint32_t revents = LTTNG_POLL_GETEV(&thread.events, i);
 824
 825                         DBG("Handling fd (%i) activity (%u)",
 826                                         fd, revents);
 827
 828                         if (revents & LPOLLERR) {
 829                                 ERR("Polling returned an error on fd %i", fd);
 830                                 goto error;
 831                         }
 832
 833                         if (fd == rotate_notification_channel->socket) {
 834                                 ret = handle_notification_channel(fd, handle,
 835                                                 &thread);
 836                                 if (ret) {
 837                                         ERR("Error occurred while handling activity on notification channel socket");
 838                                         goto error;
 839                                 }
 840                         } else {
 841                                 /* Job queue or quit pipe activity. */
 842
 843                                 /*
 844                                  * The job queue is serviced if there is
 845                                  * activity on the quit pipe to ensure it is
 846                                  * flushed and all references held in the queue
 847                                  * are released.
 848                                  */
 849                                 ret = handle_job_queue(handle, &thread,
 850                                                 handle->rotation_timer_queue);
 851                                 if (ret) {
 852                                         ERR("Failed to handle rotation timer pipe event");
 853                                         goto error;
 854                                 }
 855
 856                                 if (fd == queue_pipe_fd) {
 857                                         char buf;
 858
 859                                         ret = lttng_read(fd, &buf, 1);
 860                                         if (ret != 1) {
 861                                                 ERR("Failed to read from wakeup pipe (fd = %i)", fd);
 862                                                 goto error;
 863                                         }
 864                                 } else {
 865                                         DBG("Quit pipe activity");
 866                                         goto exit;
 867                                 }
 868                         }
 869                 }
 870         }
 871 exit:
 872 error:
 873         DBG("Thread exit");
 874         fini_thread_state(&thread);
 875 end:
 876         health_unregister(the_health_sessiond);
 877         rcu_thread_offline();
 878         rcu_unregister_thread();
 879         return NULL;
 880 }
 881
 882 static
 883 bool shutdown_rotation_thread(void *thread_data)
 884 {
 885         struct rotation_thread_handle *handle = (rotation_thread_handle *) thread_data;
 886         const int write_fd = lttng_pipe_get_writefd(handle->quit_pipe);
 887
 888         return notify_thread_pipe(write_fd) == 1;
 889 }
 890
 891 bool launch_rotation_thread(struct rotation_thread_handle *handle)
 892 {
 893         struct lttng_thread *thread;
 894
 895         thread = lttng_thread_create("Rotation",
 896                         thread_rotation,
 897                         shutdown_rotation_thread,
 898                         NULL,
 899                         handle);
 900         if (!thread) {
 901                 goto error;
 902         }
 903         lttng_thread_put(thread);
 904         return true;
 905 error:
 906         return false;
 907 }