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,
 296                         LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
 297         if (ret < 0) {
 298                 ERR("Failed to add notification channel socket to pollset");
 299                 goto error;
 300         }
 301         ret = lttng_poll_add(poll_set, handle->cmd_queue.event_fd,
 302                         LPOLLIN | LPOLLERR);
 303         if (ret < 0) {
 304                 ERR("Failed to add notification command queue event fd to pollset");
 305                 goto error;
 306         }
 307         ret = lttng_poll_add(poll_set,
 308                         handle->channel_monitoring_pipes.ust32_consumer,
 309                         LPOLLIN | LPOLLERR);
 310         if (ret < 0) {
 311                 ERR("Failed to add ust-32 channel monitoring pipe fd to pollset");
 312                 goto error;
 313         }
 314         ret = lttng_poll_add(poll_set,
 315                         handle->channel_monitoring_pipes.ust64_consumer,
 316                         LPOLLIN | LPOLLERR);
 317         if (ret < 0) {
 318                 ERR("Failed to add ust-64 channel monitoring pipe fd to pollset");
 319                 goto error;
 320         }
 321         if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
 322                 goto end;
 323         }
 324         ret = lttng_poll_add(poll_set,
 325                         handle->channel_monitoring_pipes.kernel_consumer,
 326                         LPOLLIN | LPOLLERR);
 327         if (ret < 0) {
 328                 ERR("Failed to add kernel channel monitoring pipe fd to pollset");
 329                 goto error;
 330         }
 331 end:
 332         return ret;
 333 error:
 334         lttng_poll_clean(poll_set);
 335         return ret;
 336 }
 337
 338 static
 339 void fini_thread_state(struct notification_thread_state *state)
 340 {
 341         int ret;
 342
 343         if (state->client_socket_ht) {
 344                 ret = handle_notification_thread_client_disconnect_all(state);
 345                 LTTNG_ASSERT(!ret);
 346                 ret = cds_lfht_destroy(state->client_socket_ht, NULL);
 347                 LTTNG_ASSERT(!ret);
 348         }
 349         if (state->client_id_ht) {
 350                 ret = cds_lfht_destroy(state->client_id_ht, NULL);
 351                 LTTNG_ASSERT(!ret);
 352         }
 353         if (state->triggers_ht) {
 354                 ret = handle_notification_thread_trigger_unregister_all(state);
 355                 LTTNG_ASSERT(!ret);
 356                 ret = cds_lfht_destroy(state->triggers_ht, NULL);
 357                 LTTNG_ASSERT(!ret);
 358         }
 359         if (state->channel_triggers_ht) {
 360                 ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
 361                 LTTNG_ASSERT(!ret);
 362         }
 363         if (state->channel_state_ht) {
 364                 ret = cds_lfht_destroy(state->channel_state_ht, NULL);
 365                 LTTNG_ASSERT(!ret);
 366         }
 367         if (state->notification_trigger_clients_ht) {
 368                 ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
 369                                 NULL);
 370                 LTTNG_ASSERT(!ret);
 371         }
 372         if (state->channels_ht) {
 373                 ret = cds_lfht_destroy(state->channels_ht, NULL);
 374                 LTTNG_ASSERT(!ret);
 375         }
 376         if (state->sessions_ht) {
 377                 ret = cds_lfht_destroy(state->sessions_ht, NULL);
 378                 LTTNG_ASSERT(!ret);
 379         }
 380         if (state->triggers_by_name_uid_ht) {
 381                 ret = cds_lfht_destroy(state->triggers_by_name_uid_ht, NULL);
 382                 LTTNG_ASSERT(!ret);
 383         }
 384         if (state->trigger_tokens_ht) {
 385                 ret = cds_lfht_destroy(state->trigger_tokens_ht, NULL);
 386                 LTTNG_ASSERT(!ret);
 387         }
 388         /*
 389          * Must be destroyed after all channels have been destroyed.
 390          * See comment in struct lttng_session_trigger_list.
 391          */
 392         if (state->session_triggers_ht) {
 393                 ret = cds_lfht_destroy(state->session_triggers_ht, NULL);
 394                 LTTNG_ASSERT(!ret);
 395         }
 396         if (state->notification_channel_socket >= 0) {
 397                 notification_channel_socket_destroy(
 398                                 state->notification_channel_socket);
 399         }
 400
 401         LTTNG_ASSERT(cds_list_empty(&state->tracer_event_sources_list));
 402
 403         if (state->executor) {
 404                 action_executor_destroy(state->executor);
 405         }
 406         lttng_poll_clean(&state->events);
 407 }
 408
 409 static
 410 void mark_thread_as_ready(struct notification_thread_handle *handle)
 411 {
 412         DBG("Marking notification thread as ready");
 413         sem_post(&handle->ready);
 414 }
 415
 416 static
 417 void wait_until_thread_is_ready(struct notification_thread_handle *handle)
 418 {
 419         DBG("Waiting for notification thread to be ready");
 420         sem_wait(&handle->ready);
 421         DBG("Notification thread is ready");
 422 }
 423
 424 static
 425 int init_thread_state(struct notification_thread_handle *handle,
 426                 struct notification_thread_state *state)
 427 {
 428         int ret;
 429
 430         memset(state, 0, sizeof(*state));
 431         state->notification_channel_socket = -1;
 432         state->trigger_id.next_tracer_token = 1;
 433         lttng_poll_init(&state->events);
 434
 435         ret = notification_channel_socket_create();
 436         if (ret < 0) {
 437                 goto end;
 438         }
 439         state->notification_channel_socket = ret;
 440
 441         ret = init_poll_set(&state->events, handle,
 442                         state->notification_channel_socket);
 443         if (ret) {
 444                 goto end;
 445         }
 446
 447         DBG("Listening on notification channel socket");
 448         ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
 449         if (ret < 0) {
 450                 ERR("Listen failed on notification channel socket");
 451                 goto error;
 452         }
 453
 454         state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 455                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 456         if (!state->client_socket_ht) {
 457                 goto error;
 458         }
 459
 460         state->client_id_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 461                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 462         if (!state->client_id_ht) {
 463                 goto error;
 464         }
 465
 466         state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 467                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 468         if (!state->channel_triggers_ht) {
 469                 goto error;
 470         }
 471
 472         state->session_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 473                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 474         if (!state->session_triggers_ht) {
 475                 goto error;
 476         }
 477
 478         state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 479                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 480         if (!state->channel_state_ht) {
 481                 goto error;
 482         }
 483
 484         state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 485                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 486         if (!state->notification_trigger_clients_ht) {
 487                 goto error;
 488         }
 489
 490         state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 491                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 492         if (!state->channels_ht) {
 493                 goto error;
 494         }
 495         state->sessions_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 496                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 497         if (!state->sessions_ht) {
 498                 goto error;
 499         }
 500         state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 501                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 502         if (!state->triggers_ht) {
 503                 goto error;
 504         }
 505         state->triggers_by_name_uid_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 506                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 507         if (!state->triggers_by_name_uid_ht) {
 508                 goto error;
 509         }
 510
 511         state->trigger_tokens_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 512                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 513         if (!state->trigger_tokens_ht) {
 514                 goto error;
 515         }
 516
 517         CDS_INIT_LIST_HEAD(&state->tracer_event_sources_list);
 518
 519         state->executor = action_executor_create(handle);
 520         if (!state->executor) {
 521                 goto error;
 522         }
 523
 524         state->restart_poll = false;
 525
 526         mark_thread_as_ready(handle);
 527 end:
 528         return 0;
 529 error:
 530         fini_thread_state(state);
 531         return -1;
 532 }
 533
 534 static
 535 int handle_channel_monitoring_pipe(int fd, uint32_t revents,
 536                 struct notification_thread_handle *handle,
 537                 struct notification_thread_state *state)
 538 {
 539         int ret = 0;
 540         enum lttng_domain_type domain;
 541
 542         if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
 543                         fd == handle->channel_monitoring_pipes.ust64_consumer) {
 544                 domain = LTTNG_DOMAIN_UST;
 545         } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
 546                 domain = LTTNG_DOMAIN_KERNEL;
 547         } else {
 548                 abort();
 549         }
 550
 551         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 552                 ret = lttng_poll_del(&state->events, fd);
 553                 if (ret) {
 554                         ERR("Failed to remove consumer monitoring pipe from poll set");
 555                 }
 556                 goto end;
 557         }
 558
 559         ret = handle_notification_thread_channel_sample(
 560                         state, fd, domain);
 561         if (ret) {
 562                 ERR("Consumer sample handling error occurred");
 563                 ret = -1;
 564                 goto end;
 565         }
 566 end:
 567         return ret;
 568 }
 569
 570 static int handle_event_notification_pipe(int event_source_fd,
 571                 enum lttng_domain_type domain,
 572                 uint32_t revents,
 573                 struct notification_thread_state *state)
 574 {
 575         int ret = 0;
 576
 577         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 578                 ret = handle_notification_thread_tracer_event_source_died(
 579                                 state, event_source_fd);
 580                 if (ret) {
 581                         ERR("Failed to remove event notification pipe from poll set: fd = %d",
 582                                         event_source_fd);
 583                 }
 584                 goto end;
 585         }
 586
 587         if (testpoint(sessiond_handle_notifier_event_pipe)) {
 588                 ret = 0;
 589                 goto end;
 590         }
 591
 592         if (caa_unlikely(notifier_consumption_paused)) {
 593                 DBG("Event notifier notification consumption paused, sleeping...");
 594                 sleep(1);
 595                 goto end;
 596         }
 597
 598         ret = handle_notification_thread_event_notification(
 599                         state, event_source_fd, domain);
 600         if (ret) {
 601                 ERR("Event notification handling error occurred for fd: %d",
 602                                 event_source_fd);
 603                 ret = -1;
 604                 goto end;
 605         }
 606
 607 end:
 608         return ret;
 609 }
 610
 611 /*
 612  * Return the event source domain type via parameter.
 613  */
 614 static bool fd_is_event_notification_source(const struct notification_thread_state *state,
 615                 int fd,
 616                 enum lttng_domain_type *domain)
 617 {
 618         struct notification_event_tracer_event_source_element *source_element;
 619
 620         LTTNG_ASSERT(domain);
 621
 622         cds_list_for_each_entry(source_element,
 623                         &state->tracer_event_sources_list, node) {
 624                 if (source_element->fd != fd) {
 625                         continue;
 626                 }
 627
 628                 *domain = source_element->domain;
 629                 return true;
 630         }
 631
 632         return false;
 633 }
 634
 635 /*
 636  * This thread services notification channel clients and commands received
 637  * from various lttng-sessiond components over a command queue.
 638  */
 639 static
 640 void *thread_notification(void *data)
 641 {
 642         int ret;
 643         struct notification_thread_handle *handle = (notification_thread_handle *) data;
 644         struct notification_thread_state state;
 645         enum lttng_domain_type domain;
 646
 647         DBG("Started notification thread");
 648
 649         health_register(the_health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
 650         rcu_register_thread();
 651         rcu_thread_online();
 652
 653         if (!handle) {
 654                 ERR("Invalid thread context provided");
 655                 goto end;
 656         }
 657
 658         health_code_update();
 659
 660         ret = init_thread_state(handle, &state);
 661         if (ret) {
 662                 goto end;
 663         }
 664
 665         if (testpoint(sessiond_thread_notification)) {
 666                 goto end;
 667         }
 668
 669         while (true) {
 670                 int fd_count, i;
 671
 672                 health_poll_entry();
 673                 DBG("Entering poll wait");
 674                 ret = lttng_poll_wait(&state.events, -1);
 675                 DBG("Poll wait returned (%i)", ret);
 676                 health_poll_exit();
 677                 if (ret < 0) {
 678                         /*
 679                          * Restart interrupted system call.
 680                          */
 681                         if (errno == EINTR) {
 682                                 continue;
 683                         }
 684                         ERR("Error encountered during lttng_poll_wait (%i)", ret);
 685                         goto error;
 686                 }
 687
 688                 /*
 689                  * Reset restart_poll flag so that calls below might turn it
 690                  * on.
 691                  */
 692                 state.restart_poll = false;
 693
 694                 fd_count = ret;
 695                 for (i = 0; i < fd_count; i++) {
 696                         int fd = LTTNG_POLL_GETFD(&state.events, i);
 697                         uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
 698
 699                         DBG("Handling fd (%i) activity (%u)", fd, revents);
 700
 701                         if (fd == state.notification_channel_socket) {
 702                                 if (revents & LPOLLIN) {
 703                                         ret = handle_notification_thread_client_connect(
 704                                                         &state);
 705                                         if (ret < 0) {
 706                                                 goto error;
 707                                         }
 708                                 } else if (revents &
 709                                                 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 710                                         ERR("Notification socket poll error");
 711                                         goto error;
 712                                 } else {
 713                                         ERR("Unexpected poll events %u for notification socket %i", revents, fd);
 714                                         goto error;
 715                                 }
 716                         } else if (fd == handle->cmd_queue.event_fd) {
 717                                 ret = handle_notification_thread_command(handle,
 718                                                 &state);
 719                                 if (ret < 0) {
 720                                         DBG("Error encountered while servicing command queue");
 721                                         goto error;
 722                                 } else if (ret > 0) {
 723                                         goto exit;
 724                                 }
 725                         } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
 726                                         fd == handle->channel_monitoring_pipes.ust64_consumer ||
 727                                         fd == handle->channel_monitoring_pipes.kernel_consumer) {
 728                                 ret = handle_channel_monitoring_pipe(fd,
 729                                                 revents, handle, &state);
 730                                 if (ret) {
 731                                         goto error;
 732                                 }
 733                         } else if (fd_is_event_notification_source(&state, fd, &domain)) {
 734                                 ret = handle_event_notification_pipe(fd, domain, revents, &state);
 735                                 if (ret) {
 736                                         goto error;
 737                                 }
 738                         } else {
 739                                 /* Activity on a client's socket. */
 740                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 741                                         /*
 742                                          * It doesn't matter if a command was
 743                                          * pending on the client socket at this
 744                                          * point since it now has no way to
 745                                          * receive the notifications to which
 746                                          * it was subscribing or unsubscribing.
 747                                          */
 748                                         ret = handle_notification_thread_client_disconnect(
 749                                                         fd, &state);
 750                                         if (ret) {
 751                                                 goto error;
 752                                         }
 753                                 } else {
 754                                         if (revents & LPOLLIN) {
 755                                                 ret = handle_notification_thread_client_in(
 756                                                         &state, fd);
 757                                                 if (ret) {
 758                                                         goto error;
 759                                                 }
 760                                         }
 761
 762                                         if (revents & LPOLLOUT) {
 763                                                 ret = handle_notification_thread_client_out(
 764                                                         &state, fd);
 765                                                 if (ret) {
 766                                                         goto error;
 767                                                 }
 768                                         }
 769                                 }
 770                         }
 771
 772                         /*
 773                          * Calls above might have changed the state of the
 774                          * FDs in `state.events`. Call _poll_wait() again to
 775                          * ensure we have a consistent state.
 776                          */
 777                         if (state.restart_poll) {
 778                                 break;
 779                         }
 780                 }
 781         }
 782 exit:
 783 error:
 784         fini_thread_state(&state);
 785 end:
 786         rcu_thread_offline();
 787         rcu_unregister_thread();
 788         health_unregister(the_health_sessiond);
 789         return NULL;
 790 }
 791
 792 static
 793 bool shutdown_notification_thread(void *thread_data)
 794 {
 795         struct notification_thread_handle *handle = (notification_thread_handle *) thread_data;
 796
 797         notification_thread_command_quit(handle);
 798         return true;
 799 }
 800
 801 struct lttng_thread *launch_notification_thread(
 802                 struct notification_thread_handle *handle)
 803 {
 804         struct lttng_thread *thread;
 805
 806         thread = lttng_thread_create("Notification",
 807                         thread_notification,
 808                         shutdown_notification_thread,
 809                         NULL,
 810                         handle);
 811         if (!thread) {
 812                 goto error;
 813         }
 814
 815         /*
 816          * Wait for the thread to be marked as "ready" before returning
 817          * as other subsystems depend on the notification subsystem
 818          * (e.g. rotation thread).
 819          */
 820         wait_until_thread_is_ready(handle);
 821         return thread;
 822 error:
 823         return NULL;
 824 }