src/bin/lttng-sessiond/notification-thread.cpp

   1 /*
   2  * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  *
   4  * SPDX-License-Identifier: GPL-2.0-only
   5  *
   6  */
   7
   8 #define _LGPL_SOURCE
   9 #include <lttng/trigger/trigger.h>
  10 #include <lttng/notification/channel-internal.hpp>
  11 #include <lttng/notification/notification-internal.hpp>
  12 #include <lttng/condition/condition-internal.hpp>
  13 #include <lttng/condition/buffer-usage-internal.hpp>
  14 #include <common/error.hpp>
  15 #include <common/config/session-config.hpp>
  16 #include <common/defaults.hpp>
  17 #include <common/utils.hpp>
  18 #include <common/align.hpp>
  19 #include <common/time.hpp>
  20 #include <sys/stat.h>
  21 #include <sys/eventfd.h>
  22 #include <time.h>
  23 #include <signal.h>
  24
  25 #include "notification-thread.hpp"
  26 #include "notification-thread-events.hpp"
  27 #include "notification-thread-commands.hpp"
  28 #include "lttng-sessiond.hpp"
  29 #include "health-sessiond.hpp"
  30 #include "thread.hpp"
  31 #include "testpoint.hpp"
  32
  33 #include "kernel.hpp"
  34 #include <common/kernel-ctl/kernel-ctl.hpp>
  35
  36 #include <urcu.h>
  37 #include <urcu/list.h>
  38 #include <urcu/rculfhash.h>
  39
  40 /*
  41  * Flag used to temporarily pause data consumption from testpoints.
  42  *
  43  * This variable is dlsym-ed from a test, so needs to be exported.
  44  */
  45 LTTNG_EXPORT int notifier_consumption_paused;
  46
  47 /*
  48  * Destroy the thread data previously created by the init function.
  49  */
  50 void notification_thread_handle_destroy(
  51                 struct notification_thread_handle *handle)
  52 {
  53         int ret;
  54
  55         if (!handle) {
  56                 goto end;
  57         }
  58
  59         LTTNG_ASSERT(cds_list_empty(&handle->cmd_queue.list));
  60         pthread_mutex_destroy(&handle->cmd_queue.lock);
  61         sem_destroy(&handle->ready);
  62
  63         if (handle->cmd_queue.event_fd >= 0) {
  64                 ret = close(handle->cmd_queue.event_fd);
  65                 if (ret < 0) {
  66                         PERROR("Failed to close notification command queue event fd");
  67                 }
  68         }
  69         if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
  70                 ret = close(handle->channel_monitoring_pipes.ust32_consumer);
  71                 if (ret) {
  72                         PERROR("close 32-bit consumer channel monitoring pipe");
  73                 }
  74         }
  75         if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
  76                 ret = close(handle->channel_monitoring_pipes.ust64_consumer);
  77                 if (ret) {
  78                         PERROR("close 64-bit consumer channel monitoring pipe");
  79                 }
  80         }
  81         if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
  82                 ret = close(handle->channel_monitoring_pipes.kernel_consumer);
  83                 if (ret) {
  84                         PERROR("close kernel consumer channel monitoring pipe");
  85                 }
  86         }
  87
  88 end:
  89         free(handle);
  90 }
  91
  92 struct notification_thread_handle *notification_thread_handle_create(
  93                 struct lttng_pipe *ust32_channel_monitor_pipe,
  94                 struct lttng_pipe *ust64_channel_monitor_pipe,
  95                 struct lttng_pipe *kernel_channel_monitor_pipe)
  96 {
  97         int ret;
  98         struct notification_thread_handle *handle;
  99         int event_fd = -1;
 100
 101         handle = zmalloc<notification_thread_handle>();
 102         if (!handle) {
 103                 goto end;
 104         }
 105
 106         sem_init(&handle->ready, 0, 0);
 107
 108         event_fd =  eventfd(0, EFD_CLOEXEC | EFD_SEMAPHORE);
 109         if (event_fd < 0) {
 110                 PERROR("event_fd creation");
 111                 goto error;
 112         }
 113
 114         handle->cmd_queue.event_fd = event_fd;
 115
 116         CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
 117         ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
 118         if (ret) {
 119                 goto error;
 120         }
 121
 122         if (ust32_channel_monitor_pipe) {
 123                 handle->channel_monitoring_pipes.ust32_consumer =
 124                                 lttng_pipe_release_readfd(
 125                                         ust32_channel_monitor_pipe);
 126                 if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
 127                         goto error;
 128                 }
 129         } else {
 130                 handle->channel_monitoring_pipes.ust32_consumer = -1;
 131         }
 132         if (ust64_channel_monitor_pipe) {
 133                 handle->channel_monitoring_pipes.ust64_consumer =
 134                                 lttng_pipe_release_readfd(
 135                                         ust64_channel_monitor_pipe);
 136                 if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
 137                         goto error;
 138                 }
 139         } else {
 140                 handle->channel_monitoring_pipes.ust64_consumer = -1;
 141         }
 142         if (kernel_channel_monitor_pipe) {
 143                 handle->channel_monitoring_pipes.kernel_consumer =
 144                                 lttng_pipe_release_readfd(
 145                                         kernel_channel_monitor_pipe);
 146                 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
 147                         goto error;
 148                 }
 149         } else {
 150                 handle->channel_monitoring_pipes.kernel_consumer = -1;
 151         }
 152
 153 end:
 154         return handle;
 155 error:
 156         notification_thread_handle_destroy(handle);
 157         return NULL;
 158 }
 159
 160 static
 161 char *get_notification_channel_sock_path(void)
 162 {
 163         int ret;
 164         bool is_root = !getuid();
 165         char *sock_path;
 166
 167         sock_path = calloc<char>(LTTNG_PATH_MAX);
 168         if (!sock_path) {
 169                 goto error;
 170         }
 171
 172         if (is_root) {
 173                 ret = snprintf(sock_path, LTTNG_PATH_MAX,
 174                                 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
 175                 if (ret < 0) {
 176                         goto error;
 177                 }
 178         } else {
 179                 const char *home_path = utils_get_home_dir();
 180
 181                 if (!home_path) {
 182                         ERR("Can't get HOME directory for socket creation");
 183                         goto error;
 184                 }
 185
 186                 ret = snprintf(sock_path, LTTNG_PATH_MAX,
 187                                 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
 188                                 home_path);
 189                 if (ret < 0) {
 190                         goto error;
 191                 }
 192         }
 193
 194         return sock_path;
 195 error:
 196         free(sock_path);
 197         return NULL;
 198 }
 199
 200 static
 201 void notification_channel_socket_destroy(int fd)
 202 {
 203         int ret;
 204         char *sock_path = get_notification_channel_sock_path();
 205
 206         DBG("Destroying notification channel socket");
 207
 208         if (sock_path) {
 209                 ret = unlink(sock_path);
 210                 free(sock_path);
 211                 if (ret < 0) {
 212                         PERROR("unlink notification channel socket");
 213                 }
 214         }
 215
 216         ret = close(fd);
 217         if (ret) {
 218                 PERROR("close notification channel socket");
 219         }
 220 }
 221
 222 static
 223 int notification_channel_socket_create(void)
 224 {
 225         int fd = -1, ret;
 226         char *sock_path = get_notification_channel_sock_path();
 227
 228         DBG("Creating notification channel UNIX socket at %s",
 229                         sock_path);
 230
 231         ret = lttcomm_create_unix_sock(sock_path);
 232         if (ret < 0) {
 233                 ERR("Failed to create notification socket");
 234                 goto error;
 235         }
 236         fd = ret;
 237
 238         ret = chmod(sock_path, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
 239         if (ret < 0) {
 240                 ERR("Set file permissions failed: %s", sock_path);
 241                 PERROR("chmod notification channel socket");
 242                 goto error;
 243         }
 244
 245         if (getuid() == 0) {
 246                 gid_t gid;
 247
 248                 ret = utils_get_group_id(the_config.tracing_group_name.value,
 249                                 true, &gid);
 250                 if (ret) {
 251                         /* Default to root group. */
 252                         gid = 0;
 253                 }
 254
 255                 ret = chown(sock_path, 0, gid);
 256                 if (ret) {
 257                         ERR("Failed to set the notification channel socket's group");
 258                         ret = -1;
 259                         goto error;
 260                 }
 261         }
 262
 263         DBG("Notification channel UNIX socket created (fd = %i)",
 264                         fd);
 265         free(sock_path);
 266         return fd;
 267 error:
 268         if (fd >= 0 && close(fd) < 0) {
 269                 PERROR("close notification channel socket");
 270         }
 271         free(sock_path);
 272         return ret;
 273 }
 274
 275 static
 276 int init_poll_set(struct lttng_poll_event *poll_set,
 277                 struct notification_thread_handle *handle,
 278                 int notification_channel_socket)
 279 {
 280         int ret;
 281
 282         /*
 283          * Create pollset with size 5:
 284          *      - notification channel socket (listen for new connections),
 285          *      - command queue event fd (internal sessiond commands),
 286          *      - consumerd (32-bit user space) channel monitor pipe,
 287          *      - consumerd (64-bit user space) channel monitor pipe,
 288          *      - consumerd (kernel) channel monitor pipe.
 289          */
 290         ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
 291         if (ret < 0) {
 292                 goto end;
 293         }
 294
 295         ret = lttng_poll_add(poll_set, notification_channel_socket, LPOLLIN | LPOLLRDHUP);
 296         if (ret < 0) {
 297                 ERR("Failed to add notification channel socket to pollset");
 298                 goto error;
 299         }
 300         ret = lttng_poll_add(poll_set, handle->cmd_queue.event_fd, LPOLLIN);
 301         if (ret < 0) {
 302                 ERR("Failed to add notification command queue event fd to pollset");
 303                 goto error;
 304         }
 305         ret = lttng_poll_add(poll_set,
 306                         handle->channel_monitoring_pipes.ust32_consumer, LPOLLIN);
 307         if (ret < 0) {
 308                 ERR("Failed to add ust-32 channel monitoring pipe fd to pollset");
 309                 goto error;
 310         }
 311         ret = lttng_poll_add(poll_set,
 312                         handle->channel_monitoring_pipes.ust64_consumer, LPOLLIN);
 313         if (ret < 0) {
 314                 ERR("Failed to add ust-64 channel monitoring pipe fd to pollset");
 315                 goto error;
 316         }
 317         if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
 318                 goto end;
 319         }
 320         ret = lttng_poll_add(poll_set,
 321                         handle->channel_monitoring_pipes.kernel_consumer, LPOLLIN);
 322         if (ret < 0) {
 323                 ERR("Failed to add kernel channel monitoring pipe fd to pollset");
 324                 goto error;
 325         }
 326 end:
 327         return ret;
 328 error:
 329         lttng_poll_clean(poll_set);
 330         return ret;
 331 }
 332
 333 static
 334 void fini_thread_state(struct notification_thread_state *state)
 335 {
 336         int ret;
 337
 338         if (state->client_socket_ht) {
 339                 ret = handle_notification_thread_client_disconnect_all(state);
 340                 LTTNG_ASSERT(!ret);
 341                 ret = cds_lfht_destroy(state->client_socket_ht, NULL);
 342                 LTTNG_ASSERT(!ret);
 343         }
 344         if (state->client_id_ht) {
 345                 ret = cds_lfht_destroy(state->client_id_ht, NULL);
 346                 LTTNG_ASSERT(!ret);
 347         }
 348         if (state->triggers_ht) {
 349                 ret = handle_notification_thread_trigger_unregister_all(state);
 350                 LTTNG_ASSERT(!ret);
 351                 ret = cds_lfht_destroy(state->triggers_ht, NULL);
 352                 LTTNG_ASSERT(!ret);
 353         }
 354         if (state->channel_triggers_ht) {
 355                 ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
 356                 LTTNG_ASSERT(!ret);
 357         }
 358         if (state->channel_state_ht) {
 359                 ret = cds_lfht_destroy(state->channel_state_ht, NULL);
 360                 LTTNG_ASSERT(!ret);
 361         }
 362         if (state->notification_trigger_clients_ht) {
 363                 ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
 364                                 NULL);
 365                 LTTNG_ASSERT(!ret);
 366         }
 367         if (state->channels_ht) {
 368                 ret = cds_lfht_destroy(state->channels_ht, NULL);
 369                 LTTNG_ASSERT(!ret);
 370         }
 371         if (state->sessions_ht) {
 372                 ret = cds_lfht_destroy(state->sessions_ht, NULL);
 373                 LTTNG_ASSERT(!ret);
 374         }
 375         if (state->triggers_by_name_uid_ht) {
 376                 ret = cds_lfht_destroy(state->triggers_by_name_uid_ht, NULL);
 377                 LTTNG_ASSERT(!ret);
 378         }
 379         if (state->trigger_tokens_ht) {
 380                 ret = cds_lfht_destroy(state->trigger_tokens_ht, NULL);
 381                 LTTNG_ASSERT(!ret);
 382         }
 383         /*
 384          * Must be destroyed after all channels have been destroyed.
 385          * See comment in struct lttng_session_trigger_list.
 386          */
 387         if (state->session_triggers_ht) {
 388                 ret = cds_lfht_destroy(state->session_triggers_ht, NULL);
 389                 LTTNG_ASSERT(!ret);
 390         }
 391         if (state->notification_channel_socket >= 0) {
 392                 notification_channel_socket_destroy(
 393                                 state->notification_channel_socket);
 394         }
 395
 396         LTTNG_ASSERT(cds_list_empty(&state->tracer_event_sources_list));
 397
 398         if (state->executor) {
 399                 action_executor_destroy(state->executor);
 400         }
 401         lttng_poll_clean(&state->events);
 402 }
 403
 404 static
 405 void mark_thread_as_ready(struct notification_thread_handle *handle)
 406 {
 407         DBG("Marking notification thread as ready");
 408         sem_post(&handle->ready);
 409 }
 410
 411 static
 412 void wait_until_thread_is_ready(struct notification_thread_handle *handle)
 413 {
 414         DBG("Waiting for notification thread to be ready");
 415         sem_wait(&handle->ready);
 416         DBG("Notification thread is ready");
 417 }
 418
 419 static
 420 int init_thread_state(struct notification_thread_handle *handle,
 421                 struct notification_thread_state *state)
 422 {
 423         int ret;
 424
 425         memset(state, 0, sizeof(*state));
 426         state->notification_channel_socket = -1;
 427         state->trigger_id.next_tracer_token = 1;
 428         lttng_poll_init(&state->events);
 429
 430         ret = notification_channel_socket_create();
 431         if (ret < 0) {
 432                 goto end;
 433         }
 434         state->notification_channel_socket = ret;
 435
 436         ret = init_poll_set(&state->events, handle,
 437                         state->notification_channel_socket);
 438         if (ret) {
 439                 goto end;
 440         }
 441
 442         DBG("Listening on notification channel socket");
 443         ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
 444         if (ret < 0) {
 445                 ERR("Listen failed on notification channel socket");
 446                 goto error;
 447         }
 448
 449         state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 450                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 451         if (!state->client_socket_ht) {
 452                 goto error;
 453         }
 454
 455         state->client_id_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 456                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 457         if (!state->client_id_ht) {
 458                 goto error;
 459         }
 460
 461         state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 462                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 463         if (!state->channel_triggers_ht) {
 464                 goto error;
 465         }
 466
 467         state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 468                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 469         if (!state->session_triggers_ht) {
 470                 goto error;
 471         }
 472
 473         state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 474                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 475         if (!state->channel_state_ht) {
 476                 goto error;
 477         }
 478
 479         state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 480                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 481         if (!state->notification_trigger_clients_ht) {
 482                 goto error;
 483         }
 484
 485         state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 486                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 487         if (!state->channels_ht) {
 488                 goto error;
 489         }
 490         state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 491                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 492         if (!state->sessions_ht) {
 493                 goto error;
 494         }
 495         state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 496                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 497         if (!state->triggers_ht) {
 498                 goto error;
 499         }
 500         state->triggers_by_name_uid_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 501                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 502         if (!state->triggers_by_name_uid_ht) {
 503                 goto error;
 504         }
 505
 506         state->trigger_tokens_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 507                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 508         if (!state->trigger_tokens_ht) {
 509                 goto error;
 510         }
 511
 512         CDS_INIT_LIST_HEAD(&state->tracer_event_sources_list);
 513
 514         state->executor = action_executor_create(handle);
 515         if (!state->executor) {
 516                 goto error;
 517         }
 518
 519         state->restart_poll = false;
 520
 521         mark_thread_as_ready(handle);
 522 end:
 523         return 0;
 524 error:
 525         fini_thread_state(state);
 526         return -1;
 527 }
 528
 529 static
 530 int handle_channel_monitoring_pipe(int fd, uint32_t revents,
 531                 struct notification_thread_handle *handle,
 532                 struct notification_thread_state *state)
 533 {
 534         int ret = 0;
 535         enum lttng_domain_type domain;
 536
 537         if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
 538                         fd == handle->channel_monitoring_pipes.ust64_consumer) {
 539                 domain = LTTNG_DOMAIN_UST;
 540         } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
 541                 domain = LTTNG_DOMAIN_KERNEL;
 542         } else {
 543                 abort();
 544         }
 545
 546         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 547                 ret = lttng_poll_del(&state->events, fd);
 548                 if (ret) {
 549                         ERR("Failed to remove consumer monitoring pipe from poll set");
 550                 }
 551                 goto end;
 552         }
 553
 554         ret = handle_notification_thread_channel_sample(
 555                         state, fd, domain);
 556         if (ret) {
 557                 ERR("Consumer sample handling error occurred");
 558                 ret = -1;
 559                 goto end;
 560         }
 561 end:
 562         return ret;
 563 }
 564
 565 static int handle_event_notification_pipe(int event_source_fd,
 566                 enum lttng_domain_type domain,
 567                 uint32_t revents,
 568                 struct notification_thread_state *state)
 569 {
 570         int ret = 0;
 571
 572         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 573                 ret = handle_notification_thread_tracer_event_source_died(
 574                                 state, event_source_fd);
 575                 if (ret) {
 576                         ERR("Failed to remove event notification pipe from poll set: fd = %d",
 577                                         event_source_fd);
 578                 }
 579                 goto end;
 580         }
 581
 582         if (testpoint(sessiond_handle_notifier_event_pipe)) {
 583                 ret = 0;
 584                 goto end;
 585         }
 586
 587         if (caa_unlikely(notifier_consumption_paused)) {
 588                 DBG("Event notifier notification consumption paused, sleeping...");
 589                 sleep(1);
 590                 goto end;
 591         }
 592
 593         ret = handle_notification_thread_event_notification(
 594                         state, event_source_fd, domain);
 595         if (ret) {
 596                 ERR("Event notification handling error occurred for fd: %d",
 597                                 event_source_fd);
 598                 ret = -1;
 599                 goto end;
 600         }
 601
 602 end:
 603         return ret;
 604 }
 605
 606 /*
 607  * Return the event source domain type via parameter.
 608  */
 609 static bool fd_is_event_notification_source(const struct notification_thread_state *state,
 610                 int fd,
 611                 enum lttng_domain_type *domain)
 612 {
 613         struct notification_event_tracer_event_source_element *source_element;
 614
 615         LTTNG_ASSERT(domain);
 616
 617         cds_list_for_each_entry(source_element,
 618                         &state->tracer_event_sources_list, node) {
 619                 if (source_element->fd != fd) {
 620                         continue;
 621                 }
 622
 623                 *domain = source_element->domain;
 624                 return true;
 625         }
 626
 627         return false;
 628 }
 629
 630 /*
 631  * This thread services notification channel clients and commands received
 632  * from various lttng-sessiond components over a command queue.
 633  */
 634 static
 635 void *thread_notification(void *data)
 636 {
 637         int ret;
 638         struct notification_thread_handle *handle = (notification_thread_handle *) data;
 639         struct notification_thread_state state;
 640         enum lttng_domain_type domain;
 641
 642         DBG("Started notification thread");
 643
 644         health_register(the_health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
 645         rcu_register_thread();
 646         rcu_thread_online();
 647
 648         if (!handle) {
 649                 ERR("Invalid thread context provided");
 650                 goto end;
 651         }
 652
 653         health_code_update();
 654
 655         ret = init_thread_state(handle, &state);
 656         if (ret) {
 657                 goto end;
 658         }
 659
 660         if (testpoint(sessiond_thread_notification)) {
 661                 goto end;
 662         }
 663
 664         while (true) {
 665                 int fd_count, i;
 666
 667                 health_poll_entry();
 668                 DBG("Entering poll wait");
 669                 ret = lttng_poll_wait(&state.events, -1);
 670                 DBG("Poll wait returned (%i)", ret);
 671                 health_poll_exit();
 672                 if (ret < 0) {
 673                         /*
 674                          * Restart interrupted system call.
 675                          */
 676                         if (errno == EINTR) {
 677                                 continue;
 678                         }
 679                         ERR("Error encountered during lttng_poll_wait (%i)", ret);
 680                         goto error;
 681                 }
 682
 683                 /*
 684                  * Reset restart_poll flag so that calls below might turn it
 685                  * on.
 686                  */
 687                 state.restart_poll = false;
 688
 689                 fd_count = ret;
 690                 for (i = 0; i < fd_count; i++) {
 691                         int fd = LTTNG_POLL_GETFD(&state.events, i);
 692                         uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
 693
 694                         DBG("Handling fd (%i) activity (%u)", fd, revents);
 695
 696                         if (fd == state.notification_channel_socket) {
 697                                 if (revents & LPOLLIN) {
 698                                         ret = handle_notification_thread_client_connect(
 699                                                         &state);
 700                                         if (ret < 0) {
 701                                                 goto error;
 702                                         }
 703                                 } else if (revents &
 704                                                 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 705                                         ERR("Notification socket poll error");
 706                                         goto error;
 707                                 } else {
 708                                         ERR("Unexpected poll events %u for notification socket %i", revents, fd);
 709                                         goto error;
 710                                 }
 711                         } else if (fd == handle->cmd_queue.event_fd) {
 712                                 ret = handle_notification_thread_command(handle,
 713                                                 &state);
 714                                 if (ret < 0) {
 715                                         DBG("Error encountered while servicing command queue");
 716                                         goto error;
 717                                 } else if (ret > 0) {
 718                                         goto exit;
 719                                 }
 720                         } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
 721                                         fd == handle->channel_monitoring_pipes.ust64_consumer ||
 722                                         fd == handle->channel_monitoring_pipes.kernel_consumer) {
 723                                 ret = handle_channel_monitoring_pipe(fd,
 724                                                 revents, handle, &state);
 725                                 if (ret) {
 726                                         goto error;
 727                                 }
 728                         } else if (fd_is_event_notification_source(&state, fd, &domain)) {
 729                                 ret = handle_event_notification_pipe(fd, domain, revents, &state);
 730                                 if (ret) {
 731                                         goto error;
 732                                 }
 733                         } else {
 734                                 /* Activity on a client's socket. */
 735                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 736                                         /*
 737                                          * It doesn't matter if a command was
 738                                          * pending on the client socket at this
 739                                          * point since it now has no way to
 740                                          * receive the notifications to which
 741                                          * it was subscribing or unsubscribing.
 742                                          */
 743                                         ret = handle_notification_thread_client_disconnect(
 744                                                         fd, &state);
 745                                         if (ret) {
 746                                                 goto error;
 747                                         }
 748                                 } else {
 749                                         if (revents & LPOLLIN) {
 750                                                 ret = handle_notification_thread_client_in(
 751                                                         &state, fd);
 752                                                 if (ret) {
 753                                                         goto error;
 754                                                 }
 755                                         }
 756
 757                                         if (revents & LPOLLOUT) {
 758                                                 ret = handle_notification_thread_client_out(
 759                                                         &state, fd);
 760                                                 if (ret) {
 761                                                         goto error;
 762                                                 }
 763                                         }
 764                                 }
 765                         }
 766
 767                         /*
 768                          * Calls above might have changed the state of the
 769                          * FDs in `state.events`. Call _poll_wait() again to
 770                          * ensure we have a consistent state.
 771                          */
 772                         if (state.restart_poll) {
 773                                 break;
 774                         }
 775                 }
 776         }
 777 exit:
 778 error:
 779         fini_thread_state(&state);
 780 end:
 781         rcu_thread_offline();
 782         rcu_unregister_thread();
 783         health_unregister(the_health_sessiond);
 784         return NULL;
 785 }
 786
 787 static
 788 bool shutdown_notification_thread(void *thread_data)
 789 {
 790         struct notification_thread_handle *handle = (notification_thread_handle *) thread_data;
 791
 792         notification_thread_command_quit(handle);
 793         return true;
 794 }
 795
 796 struct lttng_thread *launch_notification_thread(
 797                 struct notification_thread_handle *handle)
 798 {
 799         struct lttng_thread *thread;
 800
 801         thread = lttng_thread_create("Notification",
 802                         thread_notification,
 803                         shutdown_notification_thread,
 804                         NULL,
 805                         handle);
 806         if (!thread) {
 807                 goto error;
 808         }
 809
 810         /*
 811          * Wait for the thread to be marked as "ready" before returning
 812          * as other subsystems depend on the notification subsystem
 813          * (e.g. rotation thread).
 814          */
 815         wait_until_thread_is_ready(handle);
 816         return thread;
 817 error:
 818         return NULL;
 819 }