src/bin/lttng-sessiond/dispatch.c

   1 /*
   2  * Copyright (C) 2011 David Goulet <david.goulet@polymtl.ca>
   3  * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   4  * Copyright (C) 2013 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   5  *
   6  * SPDX-License-Identifier: GPL-2.0-only
   7  *
   8  */
   9
  10 #include <stddef.h>
  11 #include <stdlib.h>
  12 #include <urcu.h>
  13 #include <common/futex.h>
  14 #include <common/macros.h>
  15
  16 #include "dispatch.h"
  17 #include "ust-app.h"
  18 #include "testpoint.h"
  19 #include "fd-limit.h"
  20 #include "health-sessiond.h"
  21 #include "lttng-sessiond.h"
  22 #include "thread.h"
  23
  24 struct thread_notifiers {
  25         struct ust_cmd_queue *ust_cmd_queue;
  26         int apps_cmd_pipe_write_fd;
  27         int apps_cmd_notify_pipe_write_fd;
  28         int dispatch_thread_exit;
  29 };
  30
  31 /*
  32  * For each tracing session, update newly registered apps. The session list
  33  * lock MUST be acquired before calling this.
  34  */
  35 static void update_ust_app(int app_sock)
  36 {
  37         struct ltt_session *sess, *stmp;
  38         const struct ltt_session_list *session_list = session_get_list();
  39         struct ust_app *app;
  40
  41         /* Consumer is in an ERROR state. Stop any application update. */
  42         if (uatomic_read(&the_ust_consumerd_state) == CONSUMER_ERROR) {
  43                 /* Stop the update process since the consumer is dead. */
  44                 return;
  45         }
  46
  47         rcu_read_lock();
  48         assert(app_sock >= 0);
  49         app = ust_app_find_by_sock(app_sock);
  50         if (app == NULL) {
  51                 /*
  52                  * Application can be unregistered before so
  53                  * this is possible hence simply stopping the
  54                  * update.
  55                  */
  56                 DBG3("UST app update failed to find app sock %d",
  57                         app_sock);
  58                 goto unlock_rcu;
  59         }
  60
  61         /* Update all event notifiers for the app. */
  62         ust_app_global_update_event_notifier_rules(app);
  63
  64         /* For all tracing session(s) */
  65         cds_list_for_each_entry_safe(sess, stmp, &session_list->head, list) {
  66                 if (!session_get(sess)) {
  67                         continue;
  68                 }
  69                 session_lock(sess);
  70                 if (!sess->active || !sess->ust_session) {
  71                         goto unlock_session;
  72                 }
  73
  74                 ust_app_global_update(sess->ust_session, app);
  75         unlock_session:
  76                 session_unlock(sess);
  77                 session_put(sess);
  78         }
  79
  80 unlock_rcu:
  81         rcu_read_unlock();
  82 }
  83
  84 /*
  85  * Sanitize the wait queue of the dispatch registration thread meaning removing
  86  * invalid nodes from it. This is to avoid memory leaks for the case the UST
  87  * notify socket is never received.
  88  */
  89 static void sanitize_wait_queue(struct ust_reg_wait_queue *wait_queue)
  90 {
  91         int ret, nb_fd = 0, i;
  92         unsigned int fd_added = 0;
  93         struct lttng_poll_event events;
  94         struct ust_reg_wait_node *wait_node = NULL, *tmp_wait_node;
  95
  96         assert(wait_queue);
  97
  98         lttng_poll_init(&events);
  99
 100         /* Just skip everything for an empty queue. */
 101         if (!wait_queue->count) {
 102                 goto end;
 103         }
 104
 105         ret = lttng_poll_create(&events, wait_queue->count, LTTNG_CLOEXEC);
 106         if (ret < 0) {
 107                 goto error_create;
 108         }
 109
 110         cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
 111                         &wait_queue->head, head) {
 112                 assert(wait_node->app);
 113                 ret = lttng_poll_add(&events, wait_node->app->sock,
 114                                 LPOLLHUP | LPOLLERR);
 115                 if (ret < 0) {
 116                         goto error;
 117                 }
 118
 119                 fd_added = 1;
 120         }
 121
 122         if (!fd_added) {
 123                 goto end;
 124         }
 125
 126         /*
 127          * Poll but don't block so we can quickly identify the faulty events and
 128          * clean them afterwards from the wait queue.
 129          */
 130         ret = lttng_poll_wait(&events, 0);
 131         if (ret < 0) {
 132                 goto error;
 133         }
 134         nb_fd = ret;
 135
 136         for (i = 0; i < nb_fd; i++) {
 137                 /* Get faulty FD. */
 138                 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
 139                 int pollfd = LTTNG_POLL_GETFD(&events, i);
 140
 141                 cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
 142                                 &wait_queue->head, head) {
 143                         if (pollfd == wait_node->app->sock &&
 144                                         (revents & (LPOLLHUP | LPOLLERR))) {
 145                                 cds_list_del(&wait_node->head);
 146                                 wait_queue->count--;
 147                                 ust_app_destroy(wait_node->app);
 148                                 free(wait_node);
 149                                 /*
 150                                  * Silence warning of use-after-free in
 151                                  * cds_list_for_each_entry_safe which uses
 152                                  * __typeof__(*wait_node).
 153                                  */
 154                                 wait_node = NULL;
 155                                 break;
 156                         } else {
 157                                 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
 158                                 goto error;
 159                         }
 160                 }
 161         }
 162
 163         if (nb_fd > 0) {
 164                 DBG("Wait queue sanitized, %d node were cleaned up", nb_fd);
 165         }
 166
 167 end:
 168         lttng_poll_clean(&events);
 169         return;
 170
 171 error:
 172         lttng_poll_clean(&events);
 173 error_create:
 174         ERR("Unable to sanitize wait queue");
 175         return;
 176 }
 177
 178 /*
 179  * Send a socket to a thread This is called from the dispatch UST registration
 180  * thread once all sockets are set for the application.
 181  *
 182  * The sock value can be invalid, we don't really care, the thread will handle
 183  * it and make the necessary cleanup if so.
 184  *
 185  * On success, return 0 else a negative value being the errno message of the
 186  * write().
 187  */
 188 static int send_socket_to_thread(int fd, int sock)
 189 {
 190         ssize_t ret;
 191
 192         /*
 193          * It's possible that the FD is set as invalid with -1 concurrently just
 194          * before calling this function being a shutdown state of the thread.
 195          */
 196         if (fd < 0) {
 197                 ret = -EBADF;
 198                 goto error;
 199         }
 200
 201         ret = lttng_write(fd, &sock, sizeof(sock));
 202         if (ret < sizeof(sock)) {
 203                 PERROR("write apps pipe %d", fd);
 204                 if (ret < 0) {
 205                         ret = -errno;
 206                 }
 207                 goto error;
 208         }
 209
 210         /* All good. Don't send back the write positive ret value. */
 211         ret = 0;
 212 error:
 213         return (int) ret;
 214 }
 215
 216 static void cleanup_ust_dispatch_thread(void *data)
 217 {
 218         free(data);
 219 }
 220
 221 /*
 222  * Dispatch request from the registration threads to the application
 223  * communication thread.
 224  */
 225 static void *thread_dispatch_ust_registration(void *data)
 226 {
 227         int ret, err = -1;
 228         struct cds_wfcq_node *node;
 229         struct ust_command *ust_cmd = NULL;
 230         struct ust_reg_wait_node *wait_node = NULL, *tmp_wait_node;
 231         struct ust_reg_wait_queue wait_queue = {
 232                 .count = 0,
 233         };
 234         struct thread_notifiers *notifiers = data;
 235
 236         rcu_register_thread();
 237
 238         health_register(the_health_sessiond,
 239                         HEALTH_SESSIOND_TYPE_APP_REG_DISPATCH);
 240
 241         if (testpoint(sessiond_thread_app_reg_dispatch)) {
 242                 goto error_testpoint;
 243         }
 244
 245         health_code_update();
 246
 247         CDS_INIT_LIST_HEAD(&wait_queue.head);
 248
 249         DBG("[thread] Dispatch UST command started");
 250
 251         for (;;) {
 252                 health_code_update();
 253
 254                 /* Atomically prepare the queue futex */
 255                 futex_nto1_prepare(&notifiers->ust_cmd_queue->futex);
 256
 257                 if (CMM_LOAD_SHARED(notifiers->dispatch_thread_exit)) {
 258                         break;
 259                 }
 260
 261                 do {
 262                         struct ust_app *app = NULL;
 263                         ust_cmd = NULL;
 264
 265                         /*
 266                          * Make sure we don't have node(s) that have hung up before receiving
 267                          * the notify socket. This is to clean the list in order to avoid
 268                          * memory leaks from notify socket that are never seen.
 269                          */
 270                         sanitize_wait_queue(&wait_queue);
 271
 272                         health_code_update();
 273                         /* Dequeue command for registration */
 274                         node = cds_wfcq_dequeue_blocking(
 275                                         &notifiers->ust_cmd_queue->head,
 276                                         &notifiers->ust_cmd_queue->tail);
 277                         if (node == NULL) {
 278                                 DBG("Woken up but nothing in the UST command queue");
 279                                 /* Continue thread execution */
 280                                 break;
 281                         }
 282
 283                         ust_cmd = caa_container_of(node, struct ust_command, node);
 284
 285                         DBG("Dispatching UST registration pid:%d ppid:%d uid:%d"
 286                                         " gid:%d sock:%d name:%s (version %d.%d)",
 287                                         ust_cmd->reg_msg.pid, ust_cmd->reg_msg.ppid,
 288                                         ust_cmd->reg_msg.uid, ust_cmd->reg_msg.gid,
 289                                         ust_cmd->sock, ust_cmd->reg_msg.name,
 290                                         ust_cmd->reg_msg.major, ust_cmd->reg_msg.minor);
 291
 292                         if (ust_cmd->reg_msg.type == USTCTL_SOCKET_CMD) {
 293                                 wait_node = zmalloc(sizeof(*wait_node));
 294                                 if (!wait_node) {
 295                                         PERROR("zmalloc wait_node dispatch");
 296                                         ret = close(ust_cmd->sock);
 297                                         if (ret < 0) {
 298                                                 PERROR("close ust sock dispatch %d", ust_cmd->sock);
 299                                         }
 300                                         lttng_fd_put(LTTNG_FD_APPS, 1);
 301                                         free(ust_cmd);
 302                                         ust_cmd = NULL;
 303                                         goto error;
 304                                 }
 305                                 CDS_INIT_LIST_HEAD(&wait_node->head);
 306
 307                                 /* Create application object if socket is CMD. */
 308                                 wait_node->app = ust_app_create(&ust_cmd->reg_msg,
 309                                                 ust_cmd->sock);
 310                                 if (!wait_node->app) {
 311                                         ret = close(ust_cmd->sock);
 312                                         if (ret < 0) {
 313                                                 PERROR("close ust sock dispatch %d", ust_cmd->sock);
 314                                         }
 315                                         lttng_fd_put(LTTNG_FD_APPS, 1);
 316                                         free(wait_node);
 317                                         wait_node = NULL;
 318                                         free(ust_cmd);
 319                                         ust_cmd = NULL;
 320                                         continue;
 321                                 }
 322                                 /*
 323                                  * Add application to the wait queue so we can set the notify
 324                                  * socket before putting this object in the global ht.
 325                                  */
 326                                 cds_list_add(&wait_node->head, &wait_queue.head);
 327                                 wait_queue.count++;
 328
 329                                 free(ust_cmd);
 330                                 ust_cmd = NULL;
 331                                 /*
 332                                  * We have to continue here since we don't have the notify
 333                                  * socket and the application MUST be added to the hash table
 334                                  * only at that moment.
 335                                  */
 336                                 continue;
 337                         } else {
 338                                 /*
 339                                  * Look for the application in the local wait queue and set the
 340                                  * notify socket if found.
 341                                  */
 342                                 cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
 343                                                 &wait_queue.head, head) {
 344                                         health_code_update();
 345                                         if (wait_node->app->pid == ust_cmd->reg_msg.pid) {
 346                                                 wait_node->app->notify_sock = ust_cmd->sock;
 347                                                 cds_list_del(&wait_node->head);
 348                                                 wait_queue.count--;
 349                                                 app = wait_node->app;
 350                                                 free(wait_node);
 351                                                 wait_node = NULL;
 352                                                 DBG3("UST app notify socket %d is set", ust_cmd->sock);
 353                                                 break;
 354                                         }
 355                                 }
 356
 357                                 /*
 358                                  * With no application at this stage the received socket is
 359                                  * basically useless so close it before we free the cmd data
 360                                  * structure for good.
 361                                  */
 362                                 if (!app) {
 363                                         ret = close(ust_cmd->sock);
 364                                         if (ret < 0) {
 365                                                 PERROR("close ust sock dispatch %d", ust_cmd->sock);
 366                                         }
 367                                         lttng_fd_put(LTTNG_FD_APPS, 1);
 368                                 }
 369                                 free(ust_cmd);
 370                                 ust_cmd = NULL;
 371                         }
 372
 373                         if (app) {
 374                                 /*
 375                                  * @session_lock_list
 376                                  *
 377                                  * Lock the global session list so from the register up to the
 378                                  * registration done message, no thread can see the application
 379                                  * and change its state.
 380                                  */
 381                                 session_lock_list();
 382                                 rcu_read_lock();
 383
 384                                 /*
 385                                  * Add application to the global hash table. This needs to be
 386                                  * done before the update to the UST registry can locate the
 387                                  * application.
 388                                  */
 389                                 ust_app_add(app);
 390
 391                                 /* Set app version. This call will print an error if needed. */
 392                                 (void) ust_app_version(app);
 393
 394                                 (void) ust_app_setup_event_notifier_group(app);
 395
 396                                 /* Send notify socket through the notify pipe. */
 397                                 ret = send_socket_to_thread(
 398                                                 notifiers->apps_cmd_notify_pipe_write_fd,
 399                                                 app->notify_sock);
 400                                 if (ret < 0) {
 401                                         rcu_read_unlock();
 402                                         session_unlock_list();
 403                                         /*
 404                                          * No notify thread, stop the UST tracing. However, this is
 405                                          * not an internal error of the this thread thus setting
 406                                          * the health error code to a normal exit.
 407                                          */
 408                                         err = 0;
 409                                         goto error;
 410                                 }
 411
 412                                 /*
 413                                  * Update newly registered application with the tracing
 414                                  * registry info already enabled information.
 415                                  */
 416                                 update_ust_app(app->sock);
 417
 418                                 /*
 419                                  * Don't care about return value. Let the manage apps threads
 420                                  * handle app unregistration upon socket close.
 421                                  */
 422                                 (void) ust_app_register_done(app);
 423
 424                                 /*
 425                                  * Even if the application socket has been closed, send the app
 426                                  * to the thread and unregistration will take place at that
 427                                  * place.
 428                                  */
 429                                 ret = send_socket_to_thread(
 430                                                 notifiers->apps_cmd_pipe_write_fd,
 431                                                 app->sock);
 432                                 if (ret < 0) {
 433                                         rcu_read_unlock();
 434                                         session_unlock_list();
 435                                         /*
 436                                          * No apps. thread, stop the UST tracing. However, this is
 437                                          * not an internal error of the this thread thus setting
 438                                          * the health error code to a normal exit.
 439                                          */
 440                                         err = 0;
 441                                         goto error;
 442                                 }
 443
 444                                 rcu_read_unlock();
 445                                 session_unlock_list();
 446                         }
 447                 } while (node != NULL);
 448
 449                 health_poll_entry();
 450                 /* Futex wait on queue. Blocking call on futex() */
 451                 futex_nto1_wait(&notifiers->ust_cmd_queue->futex);
 452                 health_poll_exit();
 453         }
 454         /* Normal exit, no error */
 455         err = 0;
 456
 457 error:
 458         /* Clean up wait queue. */
 459         cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
 460                         &wait_queue.head, head) {
 461                 cds_list_del(&wait_node->head);
 462                 wait_queue.count--;
 463                 free(wait_node);
 464         }
 465
 466         /* Empty command queue. */
 467         for (;;) {
 468                 /* Dequeue command for registration */
 469                 node = cds_wfcq_dequeue_blocking(
 470                                 &notifiers->ust_cmd_queue->head,
 471                                 &notifiers->ust_cmd_queue->tail);
 472                 if (node == NULL) {
 473                         break;
 474                 }
 475                 ust_cmd = caa_container_of(node, struct ust_command, node);
 476                 ret = close(ust_cmd->sock);
 477                 if (ret < 0) {
 478                         PERROR("close ust sock exit dispatch %d", ust_cmd->sock);
 479                 }
 480                 lttng_fd_put(LTTNG_FD_APPS, 1);
 481                 free(ust_cmd);
 482         }
 483
 484 error_testpoint:
 485         DBG("Dispatch thread dying");
 486         if (err) {
 487                 health_error();
 488                 ERR("Health error occurred in %s", __func__);
 489         }
 490         health_unregister(the_health_sessiond);
 491         rcu_unregister_thread();
 492         return NULL;
 493 }
 494
 495 static bool shutdown_ust_dispatch_thread(void *data)
 496 {
 497         struct thread_notifiers *notifiers = data;
 498
 499         CMM_STORE_SHARED(notifiers->dispatch_thread_exit, 1);
 500         futex_nto1_wake(&notifiers->ust_cmd_queue->futex);
 501         return true;
 502 }
 503
 504 bool launch_ust_dispatch_thread(struct ust_cmd_queue *cmd_queue,
 505                 int apps_cmd_pipe_write_fd,
 506                 int apps_cmd_notify_pipe_write_fd)
 507 {
 508         struct lttng_thread *thread;
 509         struct thread_notifiers *notifiers;
 510
 511         notifiers = zmalloc(sizeof(*notifiers));
 512         if (!notifiers) {
 513                 goto error;
 514         }
 515         notifiers->ust_cmd_queue = cmd_queue;
 516         notifiers->apps_cmd_pipe_write_fd = apps_cmd_pipe_write_fd;
 517         notifiers->apps_cmd_notify_pipe_write_fd = apps_cmd_notify_pipe_write_fd;
 518
 519         thread = lttng_thread_create("UST registration dispatch",
 520                         thread_dispatch_ust_registration,
 521                         shutdown_ust_dispatch_thread,
 522                         cleanup_ust_dispatch_thread,
 523                         notifiers);
 524         if (!thread) {
 525                 goto error;
 526         }
 527         lttng_thread_put(thread);
 528         return true;
 529 error:
 530         free(notifiers);
 531         return false;
 532 }