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

   1 /*
   2  * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  *
   4  * This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License, version 2 only, as
   6  * published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  11  * more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along with
  14  * this program; if not, write to the Free Software Foundation, Inc., 51
  15  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _LGPL_SOURCE
  19 #include <lttng/trigger/trigger.h>
  20 #include <lttng/notification/channel-internal.h>
  21 #include <lttng/notification/notification-internal.h>
  22 #include <lttng/condition/condition-internal.h>
  23 #include <lttng/condition/buffer-usage-internal.h>
  24 #include <common/error.h>
  25 #include <common/config/session-config.h>
  26 #include <common/defaults.h>
  27 #include <common/utils.h>
  28 #include <common/align.h>
  29 #include <common/time.h>
  30 #include <sys/eventfd.h>
  31 #include <sys/stat.h>
  32 #include <time.h>
  33 #include <signal.h>
  34
  35 #include "notification-thread.h"
  36 #include "notification-thread-events.h"
  37 #include "notification-thread-commands.h"
  38 #include "lttng-sessiond.h"
  39 #include "health-sessiond.h"
  40
  41 #include <urcu.h>
  42 #include <urcu/list.h>
  43 #include <urcu/rculfhash.h>
  44
  45 /*
  46  * Destroy the thread data previously created by the init function.
  47  */
  48 void notification_thread_handle_destroy(
  49                 struct notification_thread_handle *handle)
  50 {
  51         int ret;
  52
  53         if (!handle) {
  54                 goto end;
  55         }
  56
  57         assert(cds_list_empty(&handle->cmd_queue.list));
  58         pthread_mutex_destroy(&handle->cmd_queue.lock);
  59
  60         if (handle->cmd_queue.event_pipe) {
  61                 lttng_pipe_destroy(handle->cmd_queue.event_pipe);
  62         }
  63         if (handle->channel_monitoring_pipes.ust32_consumer >= 0) {
  64                 ret = close(handle->channel_monitoring_pipes.ust32_consumer);
  65                 if (ret) {
  66                         PERROR("close 32-bit consumer channel monitoring pipe");
  67                 }
  68         }
  69         if (handle->channel_monitoring_pipes.ust64_consumer >= 0) {
  70                 ret = close(handle->channel_monitoring_pipes.ust64_consumer);
  71                 if (ret) {
  72                         PERROR("close 64-bit consumer channel monitoring pipe");
  73                 }
  74         }
  75         if (handle->channel_monitoring_pipes.kernel_consumer >= 0) {
  76                 ret = close(handle->channel_monitoring_pipes.kernel_consumer);
  77                 if (ret) {
  78                         PERROR("close kernel consumer channel monitoring pipe");
  79                 }
  80         }
  81 end:
  82         free(handle);
  83 }
  84
  85 struct notification_thread_handle *notification_thread_handle_create(
  86                 struct lttng_pipe *ust32_channel_monitor_pipe,
  87                 struct lttng_pipe *ust64_channel_monitor_pipe,
  88                 struct lttng_pipe *kernel_channel_monitor_pipe)
  89 {
  90         int ret;
  91         struct notification_thread_handle *handle;
  92         struct lttng_pipe *event_pipe = NULL;
  93
  94         handle = zmalloc(sizeof(*handle));
  95         if (!handle) {
  96                 goto end;
  97         }
  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 end:
 145         return handle;
 146 error:
 147         lttng_pipe_destroy(event_pipe);
 148         notification_thread_handle_destroy(handle);
 149         return NULL;
 150 }
 151
 152 static
 153 char *get_notification_channel_sock_path(void)
 154 {
 155         int ret;
 156         bool is_root = !getuid();
 157         char *sock_path;
 158
 159         sock_path = zmalloc(LTTNG_PATH_MAX);
 160         if (!sock_path) {
 161                 goto error;
 162         }
 163
 164         if (is_root) {
 165                 ret = snprintf(sock_path, LTTNG_PATH_MAX,
 166                                 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK);
 167                 if (ret < 0) {
 168                         goto error;
 169                 }
 170         } else {
 171                 char *home_path = utils_get_home_dir();
 172
 173                 if (!home_path) {
 174                         ERR("Can't get HOME directory for socket creation");
 175                         goto error;
 176                 }
 177
 178                 ret = snprintf(sock_path, LTTNG_PATH_MAX,
 179                                 DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
 180                                 home_path);
 181                 if (ret < 0) {
 182                         goto error;
 183                 }
 184         }
 185
 186         return sock_path;
 187 error:
 188         free(sock_path);
 189         return NULL;
 190 }
 191
 192 static
 193 void notification_channel_socket_destroy(int fd)
 194 {
 195         int ret;
 196         char *sock_path = get_notification_channel_sock_path();
 197
 198         DBG("[notification-thread] Destroying notification channel socket");
 199
 200         if (sock_path) {
 201                 ret = unlink(sock_path);
 202                 free(sock_path);
 203                 if (ret < 0) {
 204                         PERROR("unlink notification channel socket");
 205                 }
 206         }
 207
 208         ret = close(fd);
 209         if (ret) {
 210                 PERROR("close notification channel socket");
 211         }
 212 }
 213
 214 static
 215 int notification_channel_socket_create(void)
 216 {
 217         int fd = -1, ret;
 218         char *sock_path = get_notification_channel_sock_path();
 219
 220         DBG("[notification-thread] Creating notification channel UNIX socket at %s",
 221                         sock_path);
 222
 223         ret = lttcomm_create_unix_sock(sock_path);
 224         if (ret < 0) {
 225                 ERR("[notification-thread] Failed to create notification socket");
 226                 goto error;
 227         }
 228         fd = ret;
 229
 230         ret = chmod(sock_path, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
 231         if (ret < 0) {
 232                 ERR("Set file permissions failed: %s", sock_path);
 233                 PERROR("chmod notification channel socket");
 234                 goto error;
 235         }
 236
 237         if (getuid() == 0) {
 238                 gid_t gid;
 239
 240                 ret =  utils_get_group_id(config.tracing_group_name.value, true,
 241                                 &gid);
 242                 if (ret) {
 243                         /* Default to root group. */
 244                         gid = 0;
 245                 }
 246
 247                 ret = chown(sock_path, 0, gid);
 248                 if (ret) {
 249                         ERR("Failed to set the notification channel socket's group");
 250                         ret = -1;
 251                         goto error;
 252                 }
 253         }
 254
 255         DBG("[notification-thread] Notification channel UNIX socket created (fd = %i)",
 256                         fd);
 257         free(sock_path);
 258         return fd;
 259 error:
 260         if (fd >= 0 && close(fd) < 0) {
 261                 PERROR("close notification channel socket");
 262         }
 263         free(sock_path);
 264         return ret;
 265 }
 266
 267 static
 268 int init_poll_set(struct lttng_poll_event *poll_set,
 269                 struct notification_thread_handle *handle,
 270                 int notification_channel_socket)
 271 {
 272         int ret;
 273
 274         /*
 275          * Create pollset with size 5:
 276          *      - notification channel socket (listen for new connections),
 277          *      - command queue event fd (internal sessiond commands),
 278          *      - consumerd (32-bit user space) channel monitor pipe,
 279          *      - consumerd (64-bit user space) channel monitor pipe,
 280          *      - consumerd (kernel) channel monitor pipe.
 281          */
 282         ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
 283         if (ret < 0) {
 284                 goto end;
 285         }
 286
 287         ret = lttng_poll_add(poll_set, notification_channel_socket,
 288                         LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
 289         if (ret < 0) {
 290                 ERR("[notification-thread] Failed to add notification channel socket to pollset");
 291                 goto error;
 292         }
 293         ret = lttng_poll_add(poll_set, lttng_pipe_get_readfd(handle->cmd_queue.event_pipe),
 294                         LPOLLIN | LPOLLERR);
 295         if (ret < 0) {
 296                 ERR("[notification-thread] Failed to add notification command queue event fd to pollset");
 297                 goto error;
 298         }
 299         ret = lttng_poll_add(poll_set,
 300                         handle->channel_monitoring_pipes.ust32_consumer,
 301                         LPOLLIN | LPOLLERR);
 302         if (ret < 0) {
 303                 ERR("[notification-thread] Failed to add ust-32 channel monitoring pipe fd to pollset");
 304                 goto error;
 305         }
 306         ret = lttng_poll_add(poll_set,
 307                         handle->channel_monitoring_pipes.ust64_consumer,
 308                         LPOLLIN | LPOLLERR);
 309         if (ret < 0) {
 310                 ERR("[notification-thread] Failed to add ust-64 channel monitoring pipe fd to pollset");
 311                 goto error;
 312         }
 313         if (handle->channel_monitoring_pipes.kernel_consumer < 0) {
 314                 goto end;
 315         }
 316         ret = lttng_poll_add(poll_set,
 317                         handle->channel_monitoring_pipes.kernel_consumer,
 318                         LPOLLIN | LPOLLERR);
 319         if (ret < 0) {
 320                 ERR("[notification-thread] Failed to add kernel channel monitoring pipe fd to pollset");
 321                 goto error;
 322         }
 323 end:
 324         return ret;
 325 error:
 326         lttng_poll_clean(poll_set);
 327         return ret;
 328 }
 329
 330 static
 331 void fini_thread_state(struct notification_thread_state *state)
 332 {
 333         int ret;
 334
 335         if (state->client_socket_ht) {
 336                 ret = handle_notification_thread_client_disconnect_all(state);
 337                 assert(!ret);
 338                 ret = cds_lfht_destroy(state->client_socket_ht, NULL);
 339                 assert(!ret);
 340         }
 341         if (state->triggers_ht) {
 342                 ret = handle_notification_thread_trigger_unregister_all(state);
 343                 assert(!ret);
 344                 ret = cds_lfht_destroy(state->triggers_ht, NULL);
 345                 assert(!ret);
 346         }
 347         if (state->channel_triggers_ht) {
 348                 ret = cds_lfht_destroy(state->channel_triggers_ht, NULL);
 349                 assert(!ret);
 350         }
 351         if (state->channel_state_ht) {
 352                 ret = cds_lfht_destroy(state->channel_state_ht, NULL);
 353                 assert(!ret);
 354         }
 355         if (state->notification_trigger_clients_ht) {
 356                 ret = cds_lfht_destroy(state->notification_trigger_clients_ht,
 357                                 NULL);
 358                 assert(!ret);
 359         }
 360         if (state->channels_ht) {
 361                 ret = cds_lfht_destroy(state->channels_ht,
 362                                 NULL);
 363                 assert(!ret);
 364         }
 365
 366         if (state->notification_channel_socket >= 0) {
 367                 notification_channel_socket_destroy(
 368                                 state->notification_channel_socket);
 369         }
 370         lttng_poll_clean(&state->events);
 371 }
 372
 373 static
 374 int init_thread_state(struct notification_thread_handle *handle,
 375                 struct notification_thread_state *state)
 376 {
 377         int ret;
 378
 379         memset(state, 0, sizeof(*state));
 380         state->notification_channel_socket = -1;
 381         lttng_poll_init(&state->events);
 382
 383         ret = notification_channel_socket_create();
 384         if (ret < 0) {
 385                 goto end;
 386         }
 387         state->notification_channel_socket = ret;
 388
 389         ret = init_poll_set(&state->events, handle,
 390                         state->notification_channel_socket);
 391         if (ret) {
 392                 goto end;
 393         }
 394
 395         DBG("[notification-thread] Listening on notification channel socket");
 396         ret = lttcomm_listen_unix_sock(state->notification_channel_socket);
 397         if (ret < 0) {
 398                 ERR("[notification-thread] Listen failed on notification channel socket");
 399                 goto error;
 400         }
 401
 402         state->client_socket_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 403                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 404         if (!state->client_socket_ht) {
 405                 goto error;
 406         }
 407
 408         state->channel_triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 409                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 410         if (!state->channel_triggers_ht) {
 411                 goto error;
 412         }
 413
 414         state->channel_state_ht = cds_lfht_new(DEFAULT_HT_SIZE, 1, 0,
 415                         CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 416         if (!state->channel_state_ht) {
 417                 goto error;
 418         }
 419
 420         state->notification_trigger_clients_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 421                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 422         if (!state->notification_trigger_clients_ht) {
 423                 goto error;
 424         }
 425
 426         state->channels_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 427                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 428         if (!state->channels_ht) {
 429                 goto error;
 430         }
 431
 432         state->triggers_ht = cds_lfht_new(DEFAULT_HT_SIZE,
 433                         1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
 434         if (!state->triggers_ht) {
 435                 goto error;
 436         }
 437 end:
 438         return 0;
 439 error:
 440         fini_thread_state(state);
 441         return -1;
 442 }
 443
 444 static
 445 int handle_channel_monitoring_pipe(int fd, uint32_t revents,
 446                 struct notification_thread_handle *handle,
 447                 struct notification_thread_state *state)
 448 {
 449         int ret = 0;
 450         enum lttng_domain_type domain;
 451
 452         if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
 453                         fd == handle->channel_monitoring_pipes.ust64_consumer) {
 454                 domain = LTTNG_DOMAIN_UST;
 455         } else if (fd == handle->channel_monitoring_pipes.kernel_consumer) {
 456                 domain = LTTNG_DOMAIN_KERNEL;
 457         } else {
 458                 abort();
 459         }
 460
 461         if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 462                 ret = lttng_poll_del(&state->events, fd);
 463                 if (ret) {
 464                         ERR("[notification-thread] Failed to remove consumer monitoring pipe from poll set");
 465                 }
 466                 goto end;
 467         }
 468
 469         ret = handle_notification_thread_channel_sample(
 470                         state, fd, domain);
 471         if (ret) {
 472                 ERR("[notification-thread] Consumer sample handling error occurred");
 473                 ret = -1;
 474                 goto end;
 475         }
 476 end:
 477         return ret;
 478 }
 479
 480 /*
 481  * This thread services notification channel clients and commands received
 482  * from various lttng-sessiond components over a command queue.
 483  */
 484 void *thread_notification(void *data)
 485 {
 486         int ret;
 487         struct notification_thread_handle *handle = data;
 488         struct notification_thread_state state;
 489
 490         DBG("[notification-thread] Started notification thread");
 491
 492         if (!handle) {
 493                 ERR("[notification-thread] Invalid thread context provided");
 494                 goto end;
 495         }
 496
 497         rcu_register_thread();
 498         rcu_thread_online();
 499
 500         health_register(health_sessiond, HEALTH_SESSIOND_TYPE_NOTIFICATION);
 501         health_code_update();
 502
 503         ret = init_thread_state(handle, &state);
 504         if (ret) {
 505                 goto end;
 506         }
 507
 508         /* Ready to handle client connections. */
 509         sessiond_notify_ready();
 510
 511         while (true) {
 512                 int fd_count, i;
 513
 514                 health_poll_entry();
 515                 DBG("[notification-thread] Entering poll wait");
 516                 ret = lttng_poll_wait(&state.events, -1);
 517                 DBG("[notification-thread] Poll wait returned (%i)", ret);
 518                 health_poll_exit();
 519                 if (ret < 0) {
 520                         /*
 521                          * Restart interrupted system call.
 522                          */
 523                         if (errno == EINTR) {
 524                                 continue;
 525                         }
 526                         ERR("[notification-thread] Error encountered during lttng_poll_wait (%i)", ret);
 527                         goto error;
 528                 }
 529
 530                 fd_count = ret;
 531                 for (i = 0; i < fd_count; i++) {
 532                         int fd = LTTNG_POLL_GETFD(&state.events, i);
 533                         uint32_t revents = LTTNG_POLL_GETEV(&state.events, i);
 534
 535                         if (!revents) {
 536                                 continue;
 537                         }
 538                         DBG("[notification-thread] Handling fd (%i) activity (%u)", fd, revents);
 539
 540                         if (fd == state.notification_channel_socket) {
 541                                 if (revents & LPOLLIN) {
 542                                         ret = handle_notification_thread_client_connect(
 543                                                         &state);
 544                                         if (ret < 0) {
 545                                                 goto error;
 546                                         }
 547                                 } else if (revents &
 548                                                 (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 549                                         ERR("[notification-thread] Notification socket poll error");
 550                                         goto error;
 551                                 } else {
 552                                         ERR("[notification-thread] Unexpected poll events %u for notification socket %i", revents, fd);
 553                                         goto error;
 554                                 }
 555                         } else if (fd == lttng_pipe_get_readfd(handle->cmd_queue.event_pipe)) {
 556                                 ret = handle_notification_thread_command(handle,
 557                                                 &state);
 558                                 if (ret < 0) {
 559                                         DBG("[notification-thread] Error encountered while servicing command queue");
 560                                         goto error;
 561                                 } else if (ret > 0) {
 562                                         goto exit;
 563                                 }
 564                         } else if (fd == handle->channel_monitoring_pipes.ust32_consumer ||
 565                                         fd == handle->channel_monitoring_pipes.ust64_consumer ||
 566                                         fd == handle->channel_monitoring_pipes.kernel_consumer) {
 567                                 ret = handle_channel_monitoring_pipe(fd,
 568                                                 revents, handle, &state);
 569                                 if (ret) {
 570                                         goto error;
 571                                 }
 572                         } else {
 573                                 /* Activity on a client's socket. */
 574                                 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 575                                         /*
 576                                          * It doesn't matter if a command was
 577                                          * pending on the client socket at this
 578                                          * point since it now has no way to
 579                                          * receive the notifications to which
 580                                          * it was subscribing or unsubscribing.
 581                                          */
 582                                         ret = handle_notification_thread_client_disconnect(
 583                                                         fd, &state);
 584                                         if (ret) {
 585                                                 goto error;
 586                                         }
 587                                 } else {
 588                                         if (revents & LPOLLIN) {
 589                                                 ret = handle_notification_thread_client_in(
 590                                                         &state, fd);
 591                                                 if (ret) {
 592                                                         goto error;
 593                                                 }
 594                                         }
 595
 596                                         if (revents & LPOLLOUT) {
 597                                                 ret = handle_notification_thread_client_out(
 598                                                         &state, fd);
 599                                                 if (ret) {
 600                                                         goto error;
 601                                                 }
 602                                         }
 603                                 }
 604                         }
 605                 }
 606         }
 607 exit:
 608 error:
 609         fini_thread_state(&state);
 610         health_unregister(health_sessiond);
 611         rcu_thread_offline();
 612         rcu_unregister_thread();
 613 end:
 614         return NULL;
 615 }