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

   1 /*
   2  * Copyright (C) 2017 - Julien Desfossez <jdesfossez@efficios.com>
   3  *
   4  * This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License, version 2 only, as
   6  * published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  11  * more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along with
  14  * this program; if not, write to the Free Software Foundation, Inc., 51
  15  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _LGPL_SOURCE
  19 #include <lttng/trigger/trigger.h>
  20 #include <common/error.h>
  21 #include <common/config/session-config.h>
  22 #include <common/defaults.h>
  23 #include <common/utils.h>
  24 #include <common/futex.h>
  25 #include <common/align.h>
  26 #include <common/time.h>
  27 #include <common/hashtable/utils.h>
  28 #include <sys/eventfd.h>
  29 #include <sys/stat.h>
  30 #include <time.h>
  31 #include <signal.h>
  32 #include <inttypes.h>
  33
  34 #include <common/kernel-ctl/kernel-ctl.h>
  35 #include "rotation-thread.h"
  36 #include "lttng-sessiond.h"
  37 #include "health-sessiond.h"
  38 #include "rotate.h"
  39 #include "cmd.h"
  40 #include "sessiond-timer.h"
  41
  42 #include <urcu.h>
  43 #include <urcu/list.h>
  44 #include <urcu/rculfhash.h>
  45
  46 struct cds_lfht *channel_pending_rotate_ht;
  47
  48 static
  49 void channel_rotation_info_destroy(struct rotation_channel_info *channel_info)
  50 {
  51         assert(channel_info);
  52         free(channel_info);
  53 }
  54
  55 static
  56 int match_channel_info(struct cds_lfht_node *node, const void *key)
  57 {
  58         struct rotation_channel_key *channel_key = (struct rotation_channel_key *) key;
  59         struct rotation_channel_info *channel_info;
  60
  61         channel_info = caa_container_of(node, struct rotation_channel_info,
  62                         rotate_channels_ht_node);
  63
  64         return !!((channel_key->key == channel_info->channel_key.key) &&
  65                         (channel_key->domain == channel_info->channel_key.domain));
  66 }
  67
  68 static
  69 struct rotation_channel_info *lookup_channel_pending(uint64_t key,
  70                 enum lttng_domain_type domain)
  71 {
  72         struct cds_lfht_iter iter;
  73         struct cds_lfht_node *node;
  74         struct rotation_channel_info *channel_info = NULL;
  75         struct rotation_channel_key channel_key = { .key = key,
  76                 .domain = domain };
  77
  78         cds_lfht_lookup(channel_pending_rotate_ht,
  79                         hash_channel_key(&channel_key),
  80                         match_channel_info,
  81                         &channel_key, &iter);
  82         node = cds_lfht_iter_get_node(&iter);
  83         if (!node) {
  84                 goto end;
  85         }
  86
  87         channel_info = caa_container_of(node, struct rotation_channel_info,
  88                         rotate_channels_ht_node);
  89         cds_lfht_del(channel_pending_rotate_ht, node);
  90 end:
  91         return channel_info;
  92 }
  93
  94 /*
  95  * Destroy the thread data previously created by the init function.
  96  */
  97 void rotation_thread_handle_destroy(
  98                 struct rotation_thread_handle *handle)
  99 {
 100         int ret;
 101
 102         if (!handle) {
 103                 goto end;
 104         }
 105
 106         if (handle->ust32_consumer >= 0) {
 107                 ret = close(handle->ust32_consumer);
 108                 if (ret) {
 109                         PERROR("close 32-bit consumer channel rotation pipe");
 110                 }
 111         }
 112         if (handle->ust64_consumer >= 0) {
 113                 ret = close(handle->ust64_consumer);
 114                 if (ret) {
 115                         PERROR("close 64-bit consumer channel rotation pipe");
 116                 }
 117         }
 118         if (handle->kernel_consumer >= 0) {
 119                 ret = close(handle->kernel_consumer);
 120                 if (ret) {
 121                         PERROR("close kernel consumer channel rotation pipe");
 122                 }
 123         }
 124
 125 end:
 126         free(handle);
 127 }
 128
 129 struct rotation_thread_handle *rotation_thread_handle_create(
 130                 struct lttng_pipe *ust32_channel_rotate_pipe,
 131                 struct lttng_pipe *ust64_channel_rotate_pipe,
 132                 struct lttng_pipe *kernel_channel_rotate_pipe,
 133                 int thread_quit_pipe, int rotate_timer_pipe)
 134 {
 135         struct rotation_thread_handle *handle;
 136
 137         handle = zmalloc(sizeof(*handle));
 138         if (!handle) {
 139                 goto end;
 140         }
 141
 142         if (ust32_channel_rotate_pipe) {
 143                 handle->ust32_consumer =
 144                                 lttng_pipe_release_readfd(
 145                                         ust32_channel_rotate_pipe);
 146                 if (handle->ust32_consumer < 0) {
 147                         goto error;
 148                 }
 149         } else {
 150                 handle->ust32_consumer = -1;
 151         }
 152         if (ust64_channel_rotate_pipe) {
 153                 handle->ust64_consumer =
 154                                 lttng_pipe_release_readfd(
 155                                         ust64_channel_rotate_pipe);
 156                 if (handle->ust64_consumer < 0) {
 157                         goto error;
 158                 }
 159         } else {
 160                 handle->ust64_consumer = -1;
 161         }
 162         if (kernel_channel_rotate_pipe) {
 163                 handle->kernel_consumer =
 164                                 lttng_pipe_release_readfd(
 165                                         kernel_channel_rotate_pipe);
 166                 if (handle->kernel_consumer < 0) {
 167                         goto error;
 168                 }
 169         } else {
 170                 handle->kernel_consumer = -1;
 171         }
 172         handle->thread_quit_pipe = thread_quit_pipe;
 173         handle->rotate_timer_pipe = rotate_timer_pipe;
 174
 175 end:
 176         return handle;
 177 error:
 178         rotation_thread_handle_destroy(handle);
 179         return NULL;
 180 }
 181
 182 static
 183 int init_poll_set(struct lttng_poll_event *poll_set,
 184                 struct rotation_thread_handle *handle)
 185 {
 186         int ret;
 187
 188         /*
 189          * Create pollset with size 5:
 190          *      - sessiond quit pipe
 191          *      - consumerd (32-bit user space) channel rotate pipe,
 192          *      - consumerd (64-bit user space) channel rotate pipe,
 193          *      - consumerd (kernel) channel rotate pipe.
 194          *      - sessiond rotate pending pipe
 195          */
 196         ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
 197         if (ret < 0) {
 198                 goto end;
 199         }
 200
 201         ret = lttng_poll_add(poll_set, handle->thread_quit_pipe,
 202                         LPOLLIN | LPOLLERR);
 203         if (ret < 0) {
 204                 ERR("[rotation-thread] Failed to add thread_quit_pipe fd to pollset");
 205                 goto error;
 206         }
 207         ret = lttng_poll_add(poll_set, handle->rotate_timer_pipe,
 208                         LPOLLIN | LPOLLERR);
 209         if (ret < 0) {
 210                 ERR("[rotation-thread] Failed to add rotate_pending fd to pollset");
 211                 goto error;
 212         }
 213         ret = lttng_poll_add(poll_set, handle->ust32_consumer,
 214                         LPOLLIN | LPOLLERR);
 215         if (ret < 0) {
 216                 ERR("[rotation-thread] Failed to add ust-32 channel rotation pipe fd to pollset");
 217                 goto error;
 218         }
 219         ret = lttng_poll_add(poll_set, handle->ust64_consumer,
 220                         LPOLLIN | LPOLLERR);
 221         if (ret < 0) {
 222                 ERR("[rotation-thread] Failed to add ust-64 channel rotation pipe fd to pollset");
 223                 goto error;
 224         }
 225         if (handle->kernel_consumer < 0) {
 226                 goto end;
 227         }
 228         ret = lttng_poll_add(poll_set, handle->kernel_consumer,
 229                         LPOLLIN | LPOLLERR);
 230         if (ret < 0) {
 231                 ERR("[rotation-thread] Failed to add kernel channel rotation pipe fd to pollset");
 232                 goto error;
 233         }
 234
 235 end:
 236         return ret;
 237 error:
 238         lttng_poll_clean(poll_set);
 239         return ret;
 240 }
 241
 242 static
 243 void fini_thread_state(struct rotation_thread_state *state)
 244 {
 245         lttng_poll_clean(&state->events);
 246 }
 247
 248 static
 249 int init_thread_state(struct rotation_thread_handle *handle,
 250                 struct rotation_thread_state *state)
 251 {
 252         int ret;
 253
 254         memset(state, 0, sizeof(*state));
 255         lttng_poll_init(&state->events);
 256
 257         ret = init_poll_set(&state->events, handle);
 258         if (ret) {
 259                 goto end;
 260         }
 261
 262         channel_pending_rotate_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 263                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 264         if (!channel_pending_rotate_ht) {
 265                 ret = -1;
 266         }
 267
 268 end:
 269         return 0;
 270 }
 271
 272 static
 273 int handle_channel_rotation_pipe(int fd, uint32_t revents,
 274                 struct rotation_thread_handle *handle,
 275                 struct rotation_thread_state *state)
 276 {
 277         int ret = 0;
 278         enum lttng_domain_type domain;
 279         struct rotation_channel_info *channel_info;
 280         uint64_t key;
 281
 282         if (fd == handle->ust32_consumer ||
 283                         fd == handle->ust64_consumer) {
 284                 domain = LTTNG_DOMAIN_UST;
 285         } else if (fd == handle->kernel_consumer) {
 286                 domain = LTTNG_DOMAIN_KERNEL;
 287         } else {
 288                 abort();
 289         }
 290
 291         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 292                 ret = lttng_poll_del(&state->events, fd);
 293                 if (ret) {
 294                         ERR("[rotation-thread] Failed to remove consumer "
 295                                         "rotation pipe from poll set");
 296                 }
 297                 goto end;
 298         }
 299
 300         do {
 301                 ret = read(fd, &key, sizeof(key));
 302         } while (ret == -1 && errno == EINTR);
 303         if (ret != sizeof(key)) {
 304                 ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
 305                                 fd);
 306                 ret = -1;
 307                 goto end;
 308         }
 309
 310         DBG("[rotation-thread] Received notification for chan %" PRIu64
 311                         ", domain %d\n", key, domain);
 312
 313         channel_info = lookup_channel_pending(key, domain);
 314         if (!channel_info) {
 315                 ERR("[rotation-thread] Failed to find channel_info (key = %"
 316                                 PRIu64 ")", key);
 317                 ret = -1;
 318                 goto end;
 319         }
 320
 321         if (--channel_info->session->nr_chan_rotate_pending == 0) {
 322                 time_t now = time(NULL);
 323
 324                 if (now == (time_t) -1) {
 325                         channel_info->session->rotate_status = LTTNG_ROTATE_ERROR;
 326                         ret = LTTNG_ERR_ROTATE_NOT_AVAILABLE;
 327                         goto end;
 328                 }
 329
 330                 ret = rename_complete_chunk(channel_info->session, now);
 331                 if (ret < 0) {
 332                         ERR("Failed to rename completed rotation chunk");
 333                         goto end;
 334                 }
 335                 channel_info->session->rotate_pending = false;
 336                 if (channel_info->session->rotate_pending_relay) {
 337                         ret = sessiond_timer_rotate_pending_start(
 338                                         channel_info->session,
 339                                         DEFAULT_ROTATE_PENDING_RELAY_TIMER);
 340                         if (ret) {
 341                                 ERR("Enabling rotate pending timer");
 342                                 ret = -1;
 343                                 goto end;
 344                         }
 345                 }
 346         }
 347
 348         channel_rotation_info_destroy(channel_info);
 349
 350         ret = 0;
 351
 352 end:
 353         return ret;
 354 }
 355
 356 static
 357 int rotate_pending_relay_timer(struct ltt_session *session)
 358 {
 359         int ret;
 360
 361         DBG("[rotation-thread] Check rotate pending on session %" PRIu64,
 362                         session->id);
 363         ret = relay_rotate_pending(session, session->rotate_count - 1);
 364         if (ret < 0) {
 365                 ERR("[rotation-thread] Check relay rotate pending");
 366                 goto end;
 367         }
 368         if (ret == 0) {
 369                 DBG("[rotation-thread] Rotation completed on the relay for "
 370                                 "session %" PRIu64, session->id);
 371                 /*
 372                  * Stop the timer and clear the queue, the timers are currently
 373                  * ignored because of the rotate_pending_relay_check_in_progress
 374                  * flag.
 375                  */
 376                 sessiond_timer_rotate_pending_stop(session);
 377                 /*
 378                  * Now we can clear the pending flag in the session. New
 379                  * rotations can start now.
 380                  */
 381                 session->rotate_pending_relay = false;
 382         } else if (ret == 1) {
 383                 DBG("[rotation-thread] Rotation still pending on the relay for "
 384                                 "session %" PRIu64, session->id);
 385         }
 386         /*
 387          * Allow the timer thread to send other notifications when needed.
 388          */
 389         session->rotate_pending_relay_check_in_progress = false;
 390         fprintf(stderr, "RET PENDING: %d\n", ret);
 391
 392         ret = 0;
 393
 394 end:
 395         return ret;
 396 }
 397
 398 static
 399 int rotate_timer(struct ltt_session *session)
 400 {
 401         int ret;
 402
 403         DBG("[rotation-thread] Rotate timer on session %" PRIu64, session->id);
 404
 405         /*
 406          * If the session is stopped, we need to cancel this timer.
 407          */
 408         session_lock(session);
 409         if (!session->active && session->rotate_timer_enabled) {
 410                 sessiond_rotate_timer_stop(session);
 411         }
 412
 413         ret = cmd_rotate_session(session, NULL);
 414         session_unlock(session);
 415         fprintf(stderr, "RET ROTATE TIMER: %d\n", ret);
 416         if (ret == -LTTNG_ERR_ROTATE_PENDING) {
 417                 ret = 0;
 418                 goto end;
 419         } else if (ret != LTTNG_OK) {
 420                 ret = -1;
 421                 goto end;
 422         }
 423
 424
 425         ret = 0;
 426
 427 end:
 428         return ret;
 429 }
 430
 431 static
 432 int handle_rotate_timer_pipe(int fd, uint32_t revents,
 433                 struct rotation_thread_handle *handle,
 434                 struct rotation_thread_state *state)
 435 {
 436         int ret = 0;
 437         struct ltt_session *session;
 438         struct sessiond_rotation_timer timer_data;
 439
 440         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 441                 ret = lttng_poll_del(&state->events, fd);
 442                 if (ret) {
 443                         ERR("[rotation-thread] Failed to remove consumer "
 444                                         "rotate pending pipe from poll set");
 445                 }
 446                 goto end;
 447         }
 448
 449         memset(&timer_data, 0, sizeof(struct sessiond_rotation_timer));
 450
 451         do {
 452                 ret = read(fd, &timer_data, sizeof(timer_data));
 453         } while (ret == -1 && errno == EINTR);
 454         if (ret != sizeof(timer_data)) {
 455                 ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
 456                                 fd);
 457                 ret = -1;
 458                 goto end;
 459         }
 460
 461         rcu_read_lock();
 462         session_lock_list();
 463         session = session_find_by_id(timer_data.session_id);
 464         if (!session) {
 465                 ERR("[rotation-thread] Session %" PRIu64 " not found",
 466                                 timer_data.session_id);
 467                 ret = -1;
 468                 goto end_unlock;
 469         }
 470
 471         if (timer_data.signal == LTTNG_SESSIOND_SIG_ROTATE_PENDING) {
 472                 ret = rotate_pending_relay_timer(session);
 473         } else if (timer_data.signal == LTTNG_SESSIOND_SIG_ROTATE_TIMER) {
 474                 ret = rotate_timer(session);
 475         } else {
 476                 ERR("Unknown signal in rotate timer");
 477                 ret = -1;
 478         }
 479
 480 end_unlock:
 481         session_unlock_list();
 482         rcu_read_unlock();
 483 end:
 484         return ret;
 485 }
 486
 487 void *thread_rotation(void *data)
 488 {
 489         int ret;
 490         struct rotation_thread_handle *handle = data;
 491         struct rotation_thread_state state;
 492
 493         DBG("[rotation-thread] Started rotation thread");
 494
 495         if (!handle) {
 496                 ERR("[rotation-thread] Invalid thread context provided");
 497                 goto end;
 498         }
 499
 500         rcu_register_thread();
 501         rcu_thread_online();
 502
 503         health_register(health_sessiond, HEALTH_SESSIOND_TYPE_ROTATION);
 504         health_code_update();
 505
 506         ret = init_thread_state(handle, &state);
 507         if (ret) {
 508                 goto end;
 509         }
 510
 511         /* Ready to handle client connections. */
 512         sessiond_notify_ready();
 513
 514         while (true) {
 515                 int fd_count, i;
 516
 517                 health_poll_entry();
 518                 DBG("[rotation-thread] Entering poll wait");
 519                 ret = lttng_poll_wait(&state.events, -1);
 520                 DBG("[rotation-thread] Poll wait returned (%i)", ret);
 521                 health_poll_exit();
 522                 if (ret < 0) {
 523                         /*
 524                          * Restart interrupted system call.
 525                          */
 526                         if (errno == EINTR) {
 527                                 continue;
 528                         }
 529                         ERR("[rotation-thread] Error encountered during lttng_poll_wait (%i)", ret);
 530                         goto error;
 531                 }
 532
 533                 fd_count = ret;
 534                 for (i = 0; i < fd_count; i++) {
 535                         int fd = LTTNG_POLL_GETFD(&state.events, i);
 536                         uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
 537
 538                         DBG("[rotation-thread] Handling fd (%i) activity (%u)",
 539                                         fd, revents);
 540
 541                         if (fd == handle->thread_quit_pipe) {
 542                                 DBG("[rotation-thread] Quit pipe activity");
 543                                 goto exit;
 544                         } else if (fd == handle->rotate_timer_pipe) {
 545                                 ret = handle_rotate_timer_pipe(fd, revents,
 546                                                 handle, &state);
 547                                 if (ret) {
 548                                         ERR("[rotation-thread] Rotate timer");
 549                                         goto error;
 550                                 }
 551                         } else if (fd == handle->ust32_consumer ||
 552                                         fd == handle->ust64_consumer ||
 553                                         fd == handle->kernel_consumer) {
 554                                 ret = handle_channel_rotation_pipe(fd,
 555                                                 revents, handle, &state);
 556                                 if (ret) {
 557                                         ERR("[rotation-thread] Exit main loop");
 558                                         goto error;
 559                                 }
 560                         }
 561                 }
 562         }
 563 exit:
 564 error:
 565         fini_thread_state(&state);
 566         health_unregister(health_sessiond);
 567         rcu_thread_offline();
 568         rcu_unregister_thread();
 569 end:
 570         return NULL;
 571 }