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         if (!session->active) {
 602                 /*
 603                  * A stop command was issued during the rotation, it is
 604                  * up to the rotation completion check to perform the
 605                  * renaming of the last chunk that was produced.
 606                  */
 607                 ret = notification_thread_command_session_rotation_ongoing(
 608                                 notification_thread_handle,
 609                                 session->name,
 610                                 session->uid,
 611                                 session->gid,
 612                                 session->current_archive_id);
 613                 if (ret != LTTNG_OK) {
 614                         ERR("[rotation-thread] Failed to notify notification thread of completed rotation for session %s",
 615                                         session->name);
 616                 }
 617
 618                 ret = rename_active_chunk(session);
 619                 if (ret < 0) {
 620                         ERR("[rotation-thread] Failed to rename active rotation chunk");
 621                         goto end;
 622                 }
 623
 624                 /* Ownership of location is transferred. */
 625                 location = session_get_trace_archive_location(session);
 626                 ret = notification_thread_command_session_rotation_completed(
 627                                 notification_thread_handle,
 628                                 session->name,
 629                                 session->uid,
 630                                 session->gid,
 631                                 session->current_archive_id,
 632                                 location);
 633                 if (ret != LTTNG_OK) {
 634                         ERR("[rotation-thread] Failed to notify notification thread of completed rotation for session %s",
 635                                         session->name);
 636                 }
 637         }
 638
 639         ret = 0;
 640 end:
 641         if (session->rotation_state == LTTNG_ROTATION_STATE_ONGOING) {
 642                 DBG("[rotation-thread] Rotation of trace archive %" PRIu64 " is still pending for session %s",
 643                                 session->current_archive_id - 1, session->name);
 644                 ret = timer_session_rotation_pending_check_start(session,
 645                                 DEFAULT_ROTATE_PENDING_TIMER);
 646                 if (ret) {
 647                         ERR("Re-enabling rotate pending timer");
 648                         ret = -1;
 649                         goto end;
 650                 }
 651         }
 652
 653         return ret;
 654 }
 655
 656 /* Call with the session lock held. */
 657 static
 658 int launch_session_rotation(struct ltt_session *session)
 659 {
 660         int ret;
 661         struct lttng_rotate_session_return rotation_return;
 662
 663         DBG("[rotation-thread] Launching scheduled time-based rotation on session \"%s\"",
 664                         session->name);
 665
 666         ret = cmd_rotate_session(session, &rotation_return);
 667         if (ret == LTTNG_OK) {
 668                 DBG("[rotation-thread] Scheduled time-based rotation successfully launched on session \"%s\"",
 669                                 session->name);
 670         } else {
 671                 /* Don't consider errors as fatal. */
 672                 DBG("[rotation-thread] Scheduled time-based rotation aborted for session %s: %s",
 673                                 session->name, lttng_strerror(ret));
 674         }
 675         return 0;
 676 }
 677
 678 static
 679 int run_job(struct rotation_thread_job *job, struct ltt_session *session,
 680                 struct notification_thread_handle *notification_thread_handle)
 681 {
 682         int ret;
 683
 684         switch (job->type) {
 685         case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION:
 686                 ret = launch_session_rotation(session);
 687                 break;
 688         case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION:
 689                 ret = check_session_rotation_pending(session,
 690                                 notification_thread_handle);
 691                 break;
 692         default:
 693                 abort();
 694         }
 695         return ret;
 696 }
 697
 698 static
 699 int handle_job_queue(struct rotation_thread_handle *handle,
 700                 struct rotation_thread *state,
 701                 struct rotation_thread_timer_queue *queue)
 702 {
 703         int ret = 0;
 704         int fd = lttng_pipe_get_readfd(queue->event_pipe);
 705         struct ltt_session *session;
 706         char buf;
 707
 708         ret = lttng_read(fd, &buf, 1);
 709         if (ret != 1) {
 710                 ERR("[rotation-thread] Failed to read from wakeup pipe (fd = %i)", fd);
 711                 ret = -1;
 712                 goto end;
 713         }
 714
 715         for (;;) {
 716                 struct rotation_thread_job *job;
 717
 718                 /* Take the queue lock only to pop an element from the list. */
 719                 pthread_mutex_lock(&queue->lock);
 720                 if (cds_list_empty(&queue->list)) {
 721                         pthread_mutex_unlock(&queue->lock);
 722                         break;
 723                 }
 724                 job = cds_list_first_entry(&queue->list,
 725                                 typeof(*job), head);
 726                 cds_list_del(&job->head);
 727                 pthread_mutex_unlock(&queue->lock);
 728
 729                 session_lock_list();
 730                 session = session_find_by_id(job->session_id);
 731                 if (!session) {
 732                         DBG("[rotation-thread] Session %" PRIu64 " not found",
 733                                         job->session_id);
 734                         /*
 735                          * This is a non-fatal error, and we cannot report it to
 736                          * the user (timer), so just print the error and
 737                          * continue the processing.
 738                          *
 739                          * While the timer thread will purge pending signals for
 740                          * a session on the session's destruction, it is
 741                          * possible for a job targeting that session to have
 742                          * already been queued before it was destroyed.
 743                          */
 744                         session_unlock_list();
 745                         free(job);
 746                         continue;
 747                 }
 748
 749                 session_lock(session);
 750                 session_unlock_list();
 751
 752                 ret = run_job(job, session, handle->notification_thread_handle);
 753                 session_unlock(session);
 754                 free(job);
 755                 if (ret) {
 756                         goto end;
 757                 }
 758         }
 759
 760         ret = 0;
 761
 762 end:
 763         return ret;
 764 }
 765
 766 static
 767 int handle_condition(const struct lttng_condition *condition,
 768                 const struct lttng_evaluation *evaluation,
 769                 struct notification_thread_handle *notification_thread_handle)
 770 {
 771         int ret = 0;
 772         const char *condition_session_name = NULL;
 773         enum lttng_condition_type condition_type;
 774         enum lttng_condition_status condition_status;
 775         enum lttng_evaluation_status evaluation_status;
 776         uint64_t consumed;
 777         struct ltt_session *session;
 778
 779         condition_type = lttng_condition_get_type(condition);
 780
 781         if (condition_type != LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE) {
 782                 ret = -1;
 783                 ERR("[rotation-thread] Condition type and session usage type are not the same");
 784                 goto end;
 785         }
 786
 787         /* Fetch info to test */
 788         condition_status = lttng_condition_session_consumed_size_get_session_name(
 789                         condition, &condition_session_name);
 790         if (condition_status != LTTNG_CONDITION_STATUS_OK) {
 791                 ERR("[rotation-thread] Session name could not be fetched");
 792                 ret = -1;
 793                 goto end;
 794         }
 795         evaluation_status = lttng_evaluation_session_consumed_size_get_consumed_size(evaluation,
 796                         &consumed);
 797         if (evaluation_status != LTTNG_EVALUATION_STATUS_OK) {
 798                 ERR("[rotation-thread] Failed to get evaluation");
 799                 ret = -1;
 800                 goto end;
 801         }
 802
 803         session_lock_list();
 804         session = session_find_by_name(condition_session_name);
 805         if (!session) {
 806                 ret = -1;
 807                 session_unlock_list();
 808                 ERR("[rotation-thread] Session \"%s\" not found",
 809                                 condition_session_name);
 810                 goto end;
 811         }
 812         session_lock(session);
 813         session_unlock_list();
 814
 815         ret = unsubscribe_session_consumed_size_rotation(session,
 816                         notification_thread_handle);
 817         if (ret) {
 818                 goto end;
 819         }
 820
 821         ret = cmd_rotate_session(session, NULL);
 822         if (ret == -LTTNG_ERR_ROTATION_PENDING) {
 823                 DBG("Rotate already pending, subscribe to the next threshold value");
 824         } else if (ret != LTTNG_OK) {
 825                 ERR("[rotation-thread] Failed to rotate on size notification with error: %s",
 826                                 lttng_strerror(ret));
 827                 ret = -1;
 828                 goto end_unlock;
 829         }
 830         ret = subscribe_session_consumed_size_rotation(session,
 831                         consumed + session->rotate_size,
 832                         notification_thread_handle);
 833         if (ret) {
 834                 ERR("[rotation-thread] Failed to subscribe to session consumed size condition");
 835                 goto end_unlock;
 836         }
 837         ret = 0;
 838
 839 end_unlock:
 840         session_unlock(session);
 841 end:
 842         return ret;
 843 }
 844
 845 static
 846 int handle_notification_channel(int fd,
 847                 struct rotation_thread_handle *handle,
 848                 struct rotation_thread *state)
 849 {
 850         int ret;
 851         bool notification_pending;
 852         struct lttng_notification *notification = NULL;
 853         enum lttng_notification_channel_status status;
 854         const struct lttng_evaluation *notification_evaluation;
 855         const struct lttng_condition *notification_condition;
 856
 857         status = lttng_notification_channel_has_pending_notification(
 858                         rotate_notification_channel, &notification_pending);
 859         if (status != LTTNG_NOTIFICATION_CHANNEL_STATUS_OK) {
 860                 ERR("[rotation-thread ]Error occured while checking for pending notification");
 861                 ret = -1;
 862                 goto end;
 863         }
 864
 865         if (!notification_pending) {
 866                 ret = 0;
 867                 goto end;
 868         }
 869
 870         /* Receive the next notification. */
 871         status = lttng_notification_channel_get_next_notification(
 872                         rotate_notification_channel,
 873                         &notification);
 874
 875         switch (status) {
 876         case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK:
 877                 break;
 878         case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED:
 879                 /* Not an error, we will wait for the next one */
 880                 ret = 0;
 881                 goto end;;
 882         case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED:
 883                 ERR("Notification channel was closed");
 884                 ret = -1;
 885                 goto end;
 886         default:
 887                 /* Unhandled conditions / errors. */
 888                 ERR("Unknown notification channel status");
 889                 ret = -1;
 890                 goto end;
 891         }
 892
 893         notification_condition = lttng_notification_get_condition(notification);
 894         notification_evaluation = lttng_notification_get_evaluation(notification);
 895
 896         ret = handle_condition(notification_condition, notification_evaluation,
 897                         handle->notification_thread_handle);
 898
 899 end:
 900         lttng_notification_destroy(notification);
 901         return ret;
 902 }
 903
 904 void *thread_rotation(void *data)
 905 {
 906         int ret;
 907         struct rotation_thread_handle *handle = data;
 908         struct rotation_thread thread;
 909
 910         DBG("[rotation-thread] Started rotation thread");
 911
 912         if (!handle) {
 913                 ERR("[rotation-thread] Invalid thread context provided");
 914                 goto end;
 915         }
 916
 917         rcu_register_thread();
 918         rcu_thread_online();
 919
 920         health_register(health_sessiond, HEALTH_SESSIOND_TYPE_ROTATION);
 921         health_code_update();
 922
 923         ret = init_thread_state(handle, &thread);
 924         if (ret) {
 925                 goto end;
 926         }
 927
 928         /* Ready to handle client connections. */
 929         sessiond_notify_ready();
 930
 931         while (true) {
 932                 int fd_count, i;
 933
 934                 health_poll_entry();
 935                 DBG("[rotation-thread] Entering poll wait");
 936                 ret = lttng_poll_wait(&thread.events, -1);
 937                 DBG("[rotation-thread] Poll wait returned (%i)", ret);
 938                 health_poll_exit();
 939                 if (ret < 0) {
 940                         /*
 941                          * Restart interrupted system call.
 942                          */
 943                         if (errno == EINTR) {
 944                                 continue;
 945                         }
 946                         ERR("[rotation-thread] Error encountered during lttng_poll_wait (%i)", ret);
 947                         goto error;
 948                 }
 949
 950                 fd_count = ret;
 951                 for (i = 0; i < fd_count; i++) {
 952                         int fd = LTTNG_POLL_GETFD(&thread.events, i);
 953                         uint32_t revents = LTTNG_POLL_GETEV(&thread.events, i);
 954
 955                         DBG("[rotation-thread] Handling fd (%i) activity (%u)",
 956                                         fd, revents);
 957
 958                         if (revents & LPOLLERR) {
 959                                 ERR("[rotation-thread] Polling returned an error on fd %i", fd);
 960                                 goto error;
 961                         }
 962
 963                         if (fd == handle->quit_pipe) {
 964                                 DBG("[rotation-thread] Quit pipe activity");
 965                                 /* TODO flush the queue. */
 966                                 goto exit;
 967                         } else if (fd == lttng_pipe_get_readfd(handle->rotation_timer_queue->event_pipe)) {
 968                                 ret = handle_job_queue(handle, &thread,
 969                                                 handle->rotation_timer_queue);
 970                                 if (ret) {
 971                                         ERR("[rotation-thread] Failed to handle rotation timer pipe event");
 972                                         goto error;
 973                                 }
 974                         } else if (fd == rotate_notification_channel->socket) {
 975                                 ret = handle_notification_channel(fd, handle,
 976                                                 &thread);
 977                                 if (ret) {
 978                                         ERR("[rotation-thread] Error occured while handling activity on notification channel socket");
 979                                         goto error;
 980                                 }
 981                         }
 982                 }
 983         }
 984 exit:
 985 error:
 986         DBG("[rotation-thread] Exit");
 987         fini_thread_state(&thread);
 988         health_unregister(health_sessiond);
 989         rcu_thread_offline();
 990         rcu_unregister_thread();
 991 end:
 992         return NULL;
 993 }