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

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