src/bin/lttng-sessiond/dispatch.c

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