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

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