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 <sys/eventfd.h>
  22 #include <time.h>
  23 #include <signal.h>
  24
  25 #include "notification-thread.h"
  26 #include "notification-thread-events.h"
  27 #include "notification-thread-commands.h"
  28 #include "lttng-sessiond.h"
  29 #include "health-sessiond.h"
  30 #include "thread.h"
  31 #include "testpoint.h"
  32
  33 #include "kernel.h"
  34 #include <common/kernel-ctl/kernel-ctl.h>
  35
  36 #include <urcu.h>
  37 #include <urcu/list.h>
  38 #include <urcu/rculfhash.h>
  39
  40
  41 int notifier_consumption_paused;
  42 /*
  43  * Destroy the thread data previously created by the init function.
  44  */
  45 void notification_thread_handle_destroy(
  46                 struct notification_thread_handle *handle)
  47 {
  48         int ret;
  49
  50         if (!handle) {
  51                 goto end;
  52         }
  53
  54         assert(cds_list_empty(&handle->cmd_queue.list));
  55         pthread_mutex_destroy(&handle->cmd_queue.lock);
  56         sem_destroy(&handle->ready);
  57
  58         if (handle->cmd_queue.event_fd >= 0) {
  59                 ret = close(handle->cmd_queue.event_fd);
  60                 if (ret < 0) {
  61                         PERROR("Failed to close notification command queue event fd");
  62                 }
  63         }
  64         if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
  65                 ret = close(handle->channel_monitoring_pipes.ust32_consumer);
  66                 if (ret) {
  67                         PERROR("close 32-bit consumer channel monitoring pipe");
  68                 }
  69         }
  70         if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
  71                 ret = close(handle->channel_monitoring_pipes.ust64_consumer);
  72                 if (ret) {
  73                         PERROR("close 64-bit consumer channel monitoring pipe");
  74                 }
  75         }
  76         if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
  77                 ret = close(handle->channel_monitoring_pipes.kernel_consumer);
  78                 if (ret) {
  79                         PERROR("close kernel consumer channel monitoring pipe");
  80                 }
  81         }
  82
  83 end:
  84         free(handle);
  85 }
  86
  87 struct notification_thread_handle *notification_thread_handle_create(
  88                 struct lttng_pipe *ust32_channel_monitor_pipe,
  89                 struct lttng_pipe *ust64_channel_monitor_pipe,
  90                 struct lttng_pipe *kernel_channel_monitor_pipe)
  91 {
  92         int ret;
  93         struct notification_thread_handle *handle;
  94         int event_fd = -1;
  95
  96         handle = zmalloc(sizeof(*handle));
  97         if (!handle) {
  98                 goto end;
  99         }
 100
 101         sem_init(&handle->ready, 0, 0);
 102
 103         event_fd =  eventfd(0, EFD_CLOEXEC);
 104         if (event_fd < 0) {
 105                 PERROR("event_fd creation");
 106                 goto error;
 107         }
 108
 109         handle->cmd_queue.event_fd = event_fd;
 110
 111         CDS_INIT_LIST_HEAD(&handle->cmd_queue.list);
 112         ret = pthread_mutex_init(&handle->cmd_queue.lock, NULL);
 113         if (ret) {
 114                 goto error;
 115         }
 116
 117         if (ust32_channel_monitor_pipe) {
 118                 handle->channel_monitoring_pipes.ust32_consumer =
 119                                 lttng_pipe_release_readfd(
 120                                         ust32_channel_monitor_pipe);
 121                 if (handle->channel_monitoring_pipes.ust32_consumer < 0) {
 122                         goto error;
 123                 }
 124         } else {
 125                 handle->channel_monitoring_pipes.ust32_consumer = -1;
 126         }
 127         if (ust64_channel_monitor_pipe) {
 128                 handle->channel_monitoring_pipes.ust64_consumer =
 129                                 lttng_pipe_release_readfd(
 130                                         ust64_channel_monitor_pipe);
 131                 if (handle->channel_monitoring_pipes.ust64_consumer < 0) {
 132                         goto error;
 133                 }
 134         } else {
 135                 handle->channel_monitoring_pipes.ust64_consumer = -1;
 136         }
 137         if (kernel_channel_monitor_pipe) {
 138                 handle->channel_monitoring_pipes.kernel_consumer =
 139                                 lttng_pipe_release_readfd(
 140                                         kernel_channel_monitor_pipe);
 141                 if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
 142                         goto error;
 143                 }
 144         } else {
 145                 handle->channel_monitoring_pipes.kernel_consumer = -1;
 146         }
 147
 148 end:
 149         return handle;
 150 error:
 151         notification_thread_handle_destroy(handle);
 152         return NULL;
 153 }
 154
 155 static
 156 char *get_notification_channel_sock_path(void)
 157 {
 158         int ret;
 159         bool is_root = !getuid();
 160         char *sock_path;
 161
 162         sock_path = zmalloc(LTTNG_PATH_MAX);
 163         if (!sock_path) {
 164                 goto error;
 165         }
 166
 167         if (is_root) {
 168                 ret = snprintf(sock_path, LTTNG_PATH_MAX,
 169                                 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
 170                 if (ret < 0) {
 171                         goto error;
 172                 }
 173         } else {
 174                 const char *home_path = utils_get_home_dir();
 175
 176                 if (!home_path) {
 177                         ERR("Can't get HOME directory for socket creation");
 178                         goto error;
 179                 }
 180
 181                 ret = snprintf(sock_path, LTTNG_PATH_MAX,
 182                                 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
 183                                 home_path);
 184                 if (ret < 0) {
 185                         goto error;
 186                 }
 187         }
 188
 189         return sock_path;
 190 error:
 191         free(sock_path);
 192         return NULL;
 193 }
 194
 195 static
 196 void notification_channel_socket_destroy(int fd)
 197 {
 198         int ret;
 199         char *sock_path = get_notification_channel_sock_path();
 200
 201         DBG("Destroying notification channel socket");
 202
 203         if (sock_path) {
 204                 ret = unlink(sock_path);
 205                 free(sock_path);
 206                 if (ret < 0) {
 207                         PERROR("unlink notification channel socket");
 208                 }
 209         }
 210
 211         ret = close(fd);
 212         if (ret) {
 213                 PERROR("close notification channel socket");
 214         }
 215 }
 216
 217 static
 218 int notification_channel_socket_create(void)
 219 {
 220         int fd = -1, ret;
 221         char *sock_path = get_notification_channel_sock_path();
 222
 223         DBG("Creating notification channel UNIX socket at %s",
 224                         sock_path);
 225
 226         ret = lttcomm_create_unix_sock(sock_path);
 227         if (ret < 0) {
 228                 ERR("Failed to create notification socket");
 229                 goto error;
 230         }
 231         fd = ret;
 232
 233         ret = chmod(sock_path, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
 234         if (ret < 0) {
 235                 ERR("Set file permissions failed: %s", sock_path);
 236                 PERROR("chmod notification channel socket");
 237                 goto error;
 238         }
 239
 240         if (getuid() == 0) {
 241                 gid_t gid;
 242
 243                 ret = utils_get_group_id(the_config.tracing_group_name.value,
 244                                 true, &gid);
 245                 if (ret) {
 246                         /* Default to root group. */
 247                         gid = 0;
 248                 }
 249
 250                 ret = chown(sock_path, 0, gid);
 251                 if (ret) {
 252                         ERR("Failed to set the notification channel socket's group");
 253                         ret = -1;
 254                         goto error;
 255                 }
 256         }
 257
 258         DBG("Notification channel UNIX socket created (fd = %i)",
 259                         fd);
 260         free(sock_path);
 261         return fd;
 262 error:
 263         if (fd >= 0 && close(fd) < 0) {
 264                 PERROR("close notification channel socket");
 265         }
 266         free(sock_path);
 267         return ret;
 268 }
 269
 270 static
 271 int init_poll_set(struct lttng_poll_event *poll_set,
 272                 struct notification_thread_handle *handle,
 273                 int notification_channel_socket)
 274 {
 275         int ret;
 276
 277         /*
 278          * Create pollset with size 5:
 279          *      - notification channel socket (listen for new connections),
 280          *      - command queue event fd (internal sessiond commands),
 281          *      - consumerd (32-bit user space) channel monitor pipe,
 282          *      - consumerd (64-bit user space) channel monitor pipe,
 283          *      - consumerd (kernel) channel monitor pipe.
 284          */
 285         ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
 286         if (ret < 0) {
 287                 goto end;
 288         }
 289
 290         ret = lttng_poll_add(poll_set, notification_channel_socket,
 291                         LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
 292         if (ret < 0) {
 293                 ERR("Failed to add notification channel socket to pollset");
 294                 goto error;
 295         }
 296         ret = lttng_poll_add(poll_set, handle->cmd_queue.event_fd,
 297                         LPOLLIN | LPOLLERR);
 298         if (ret < 0) {
 299                 ERR("Failed to add notification command queue event fd to pollset");
 300                 goto error;
 301         }
 302         ret = lttng_poll_add(poll_set,
 303                         handle->channel_monitoring_pipes.ust32_consumer,
 304                         LPOLLIN | LPOLLERR);
 305         if (ret < 0) {
 306                 ERR("Failed to add ust-32 channel monitoring pipe fd to pollset");
 307                 goto error;
 308         }
 309         ret = lttng_poll_add(poll_set,
 310                         handle->channel_monitoring_pipes.ust64_consumer,
 311                         LPOLLIN | LPOLLERR);
 312         if (ret < 0) {
 313                 ERR("Failed to add ust-64 channel monitoring pipe fd to pollset");
 314                 goto error;
 315         }
 316         if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
 317                 goto end;
 318         }
 319         ret = lttng_poll_add(poll_set,
 320                         handle->channel_monitoring_pipes.kernel_consumer,
 321                         LPOLLIN | LPOLLERR);
 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                 assert(!ret);
 341                 ret = cds_lfht_destroy(state->client_socket_ht, NULL);
 342                 assert(!ret);
 343         }
 344         if (state->client_id_ht) {
 345                 ret = cds_lfht_destroy(state->client_id_ht, NULL);
 346                 assert(!ret);
 347         }
 348         if (state->triggers_ht) {
 349                 ret = handle_notification_thread_trigger_unregister_all(state);
 350                 assert(!ret);
 351                 ret = cds_lfht_destroy(state->triggers_ht, NULL);
 352                 assert(!ret);
 353         }
 354         if (state->channel_triggers_ht) {
 355                 ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
 356                 assert(!ret);
 357         }
 358         if (state->channel_state_ht) {
 359                 ret = cds_lfht_destroy(state->channel_state_ht, NULL);
 360                 assert(!ret);
 361         }
 362         if (state->notification_trigger_clients_ht) {
 363                 ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
 364                                 NULL);
 365                 assert(!ret);
 366         }
 367         if (state->channels_ht) {
 368                 ret = cds_lfht_destroy(state->channels_ht, NULL);
 369                 assert(!ret);
 370         }
 371         if (state->sessions_ht) {
 372                 ret = cds_lfht_destroy(state->sessions_ht, NULL);
 373                 assert(!ret);
 374         }
 375         if (state->triggers_by_name_uid_ht) {
 376                 ret = cds_lfht_destroy(state->triggers_by_name_uid_ht, NULL);
 377                 assert(!ret);
 378         }
 379         if (state->trigger_tokens_ht) {
 380                 ret = cds_lfht_destroy(state->trigger_tokens_ht, NULL);
 381                 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                 assert(!ret);
 390         }
 391         if (state->notification_channel_socket >= 0) {
 392                 notification_channel_socket_destroy(
 393                                 state->notification_channel_socket);
 394         }
 395
 396         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         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 = 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 = 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 }