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

   1 /*
   2  * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
   3  *
   4  * SPDX-License-Identifier: GPL-2.0-only
   5  *
   6  */
   7
   8 #define _LGPL_SOURCE
   9 #include <assert.h>
  10
  11 #include <common/common.h>
  12 #include <common/sessiond-comm/sessiond-comm.h>
  13 #include <common/uri.h>
  14 #include <common/utils.h>
  15
  16 #include <common/compat/endian.h>
  17
  18 #include "fd-limit.h"
  19 #include "agent-thread.h"
  20 #include "agent.h"
  21 #include "lttng-sessiond.h"
  22 #include "session.h"
  23 #include "utils.h"
  24 #include "thread.h"
  25
  26 struct thread_notifiers {
  27         struct lttng_pipe *quit_pipe;
  28         sem_t ready;
  29 };
  30
  31 static int agent_tracing_enabled = -1;
  32
  33 /*
  34  * Note that there is not port here. It's set after this URI is parsed so we
  35  * can let the user define a custom one. However, localhost is ALWAYS the
  36  * default listening address.
  37  */
  38 static const char *default_reg_uri =
  39         "tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS;
  40
  41 /*
  42  * Update agent application using the given socket. This is done just after
  43  * registration was successful.
  44  *
  45  * This is a quite heavy call in terms of locking since the session list lock
  46  * AND session lock are acquired.
  47  */
  48 static void update_agent_app(struct agent_app *app)
  49 {
  50         struct ltt_session *session, *stmp;
  51         struct ltt_session_list *list;
  52
  53         list = session_get_list();
  54         assert(list);
  55
  56         session_lock_list();
  57         cds_list_for_each_entry_safe(session, stmp, &list->head, list) {
  58                 if (!session_get(session)) {
  59                         continue;
  60                 }
  61
  62                 session_lock(session);
  63                 if (session->ust_session) {
  64                         struct agent *agt;
  65
  66                         rcu_read_lock();
  67                         agt = trace_ust_find_agent(session->ust_session, app->domain);
  68                         if (agt) {
  69                                 agent_update(agt, app->sock->fd);
  70                         }
  71                         rcu_read_unlock();
  72                 }
  73                 session_unlock(session);
  74                 session_put(session);
  75         }
  76         session_unlock_list();
  77 }
  78
  79 /*
  80  * Create and init socket from uri.
  81  */
  82 static struct lttcomm_sock *init_tcp_socket(void)
  83 {
  84         int ret;
  85         struct lttng_uri *uri = NULL;
  86         struct lttcomm_sock *sock = NULL;
  87         unsigned int port;
  88         bool bind_succeeded = false;
  89
  90         /*
  91          * This should never fail since the URI is hardcoded and the port is set
  92          * before this thread is launched.
  93          */
  94         ret = uri_parse(default_reg_uri, &uri);
  95         assert(ret);
  96         assert(config.agent_tcp_port.begin > 0);
  97         uri->port = config.agent_tcp_port.begin;
  98
  99         sock = lttcomm_alloc_sock_from_uri(uri);
 100         uri_free(uri);
 101         if (sock == NULL) {
 102                 ERR("[agent-thread] agent allocating TCP socket");
 103                 goto error;
 104         }
 105
 106         ret = lttcomm_create_sock(sock);
 107         if (ret < 0) {
 108                 goto error;
 109         }
 110
 111         for (port = config.agent_tcp_port.begin;
 112                         port <= config.agent_tcp_port.end; port++) {
 113                 ret = lttcomm_sock_set_port(sock, (uint16_t) port);
 114                 if (ret) {
 115                         ERR("[agent-thread] Failed to set port %u on socket",
 116                                         port);
 117                         goto error;
 118                 }
 119                 DBG3("[agent-thread] Trying to bind on port %u", port);
 120                 ret = sock->ops->bind(sock);
 121                 if (!ret) {
 122                         bind_succeeded = true;
 123                         break;
 124                 }
 125
 126                 if (errno == EADDRINUSE) {
 127                         DBG("Failed to bind to port %u since it is already in use",
 128                                         port);
 129                 } else {
 130                         PERROR("Failed to bind to port %u", port);
 131                         goto error;
 132                 }
 133         }
 134
 135         if (!bind_succeeded) {
 136                 if (config.agent_tcp_port.begin == config.agent_tcp_port.end) {
 137                         WARN("Another process is already using the agent port %i. "
 138                                         "Agent support will be deactivated.",
 139                                         config.agent_tcp_port.begin);
 140                         goto error;
 141                 } else {
 142                         WARN("All ports in the range [%i, %i] are already in use. "
 143                                         "Agent support will be deactivated.",
 144                                         config.agent_tcp_port.begin,
 145                                         config.agent_tcp_port.end);
 146                         goto error;
 147                 }
 148         }
 149
 150         ret = sock->ops->listen(sock, -1);
 151         if (ret < 0) {
 152                 goto error;
 153         }
 154
 155         DBG("[agent-thread] Listening on TCP port %u and socket %d",
 156                         port, sock->fd);
 157
 158         return sock;
 159
 160 error:
 161         if (sock) {
 162                 lttcomm_destroy_sock(sock);
 163         }
 164         return NULL;
 165 }
 166
 167 /*
 168  * Close and destroy the given TCP socket.
 169  */
 170 static void destroy_tcp_socket(struct lttcomm_sock *sock)
 171 {
 172         int ret;
 173         uint16_t port;
 174
 175         assert(sock);
 176
 177         ret = lttcomm_sock_get_port(sock, &port);
 178         if (ret) {
 179                 ERR("[agent-thread] Failed to get port of agent TCP socket");
 180                 port = 0;
 181         }
 182
 183         DBG3("[agent-thread] Destroy TCP socket on port %" PRIu16,
 184                         port);
 185
 186         /* This will return gracefully if fd is invalid. */
 187         sock->ops->close(sock);
 188         lttcomm_destroy_sock(sock);
 189 }
 190
 191 /*
 192  * Handle a new agent registration using the reg socket. After that, a new
 193  * agent application is added to the global hash table and attach to an UST app
 194  * object. If r_app is not NULL, the created app is set to the pointer.
 195  *
 196  * Return the new FD created upon accept() on success or else a negative errno
 197  * value.
 198  */
 199 static int handle_registration(struct lttcomm_sock *reg_sock,
 200                 struct agent_app **r_app)
 201 {
 202         int ret;
 203         pid_t pid;
 204         uint32_t major_version, minor_version;
 205         ssize_t size;
 206         enum lttng_domain_type domain;
 207         struct agent_app *app;
 208         struct agent_register_msg msg;
 209         struct lttcomm_sock *new_sock;
 210
 211         assert(reg_sock);
 212
 213         new_sock = reg_sock->ops->accept(reg_sock);
 214         if (!new_sock) {
 215                 ret = -ENOTCONN;
 216                 goto error;
 217         }
 218
 219         size = new_sock->ops->recvmsg(new_sock, &msg, sizeof(msg), 0);
 220         if (size < sizeof(msg)) {
 221                 ret = -EINVAL;
 222                 goto error_socket;
 223         }
 224         domain = be32toh(msg.domain);
 225         pid = be32toh(msg.pid);
 226         major_version = be32toh(msg.major_version);
 227         minor_version = be32toh(msg.minor_version);
 228
 229         /* Test communication protocol version of the registring agent. */
 230         if (major_version != AGENT_MAJOR_VERSION) {
 231                 ret = -EINVAL;
 232                 goto error_socket;
 233         }
 234         if (minor_version != AGENT_MINOR_VERSION) {
 235                 ret = -EINVAL;
 236                 goto error_socket;
 237         }
 238
 239         DBG2("[agent-thread] New registration for pid %d domain %d on socket %d",
 240                         pid, domain, new_sock->fd);
 241
 242         app = agent_create_app(pid, domain, new_sock);
 243         if (!app) {
 244                 ret = -ENOMEM;
 245                 goto error_socket;
 246         }
 247
 248         /*
 249          * Add before assigning the socket value to the UST app so it can be found
 250          * concurrently.
 251          */
 252         agent_add_app(app);
 253
 254         /*
 255          * We don't need to attach the agent app to the app. If we ever do so, we
 256          * should consider both registration order of agent before app and app
 257          * before agent.
 258          */
 259
 260         if (r_app) {
 261                 *r_app = app;
 262         }
 263
 264         return new_sock->fd;
 265
 266 error_socket:
 267         new_sock->ops->close(new_sock);
 268         lttcomm_destroy_sock(new_sock);
 269 error:
 270         return ret;
 271 }
 272
 273 bool agent_tracing_is_enabled(void)
 274 {
 275         int enabled;
 276
 277         enabled = uatomic_read(&agent_tracing_enabled);
 278         assert(enabled != -1);
 279         return enabled == 1;
 280 }
 281
 282 /*
 283  * Write agent TCP port using the rundir.
 284  */
 285 static int write_agent_port(uint16_t port)
 286 {
 287         return utils_create_pid_file((pid_t) port,
 288                         config.agent_port_file_path.value);
 289 }
 290
 291 static
 292 void mark_thread_as_ready(struct thread_notifiers *notifiers)
 293 {
 294         DBG("Marking agent management thread as ready");
 295         sem_post(&notifiers->ready);
 296 }
 297
 298 static
 299 void wait_until_thread_is_ready(struct thread_notifiers *notifiers)
 300 {
 301         DBG("Waiting for agent management thread to be ready");
 302         sem_wait(&notifiers->ready);
 303         DBG("Agent management thread is ready");
 304 }
 305
 306 /*
 307  * This thread manage application notify communication.
 308  */
 309 static void *thread_agent_management(void *data)
 310 {
 311         int i, ret, pollfd;
 312         uint32_t revents, nb_fd;
 313         struct lttng_poll_event events;
 314         struct lttcomm_sock *reg_sock;
 315         struct thread_notifiers *notifiers = data;
 316         const int quit_pipe_read_fd = lttng_pipe_get_readfd(
 317                         notifiers->quit_pipe);
 318
 319         DBG("[agent-thread] Manage agent application registration.");
 320
 321         rcu_register_thread();
 322         rcu_thread_online();
 323
 324         /* Agent initialization call MUST be called before starting the thread. */
 325         assert(agent_apps_ht_by_sock);
 326
 327         /* Create pollset with size 2, quit pipe and registration socket. */
 328         ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
 329         if (ret < 0) {
 330                 goto error_poll_create;
 331         }
 332
 333         ret = lttng_poll_add(&events, quit_pipe_read_fd,
 334                         LPOLLIN | LPOLLERR);
 335         if (ret < 0) {
 336                 goto error_tcp_socket;
 337         }
 338
 339         reg_sock = init_tcp_socket();
 340         if (reg_sock) {
 341                 uint16_t port;
 342
 343                 assert(lttcomm_sock_get_port(reg_sock, &port) == 0);
 344
 345                 ret = write_agent_port(port);
 346                 if (ret) {
 347                         ERR("[agent-thread] Failed to create agent port file: agent tracing will be unavailable");
 348                         /* Don't prevent the launch of the sessiond on error. */
 349                         mark_thread_as_ready(notifiers);
 350                         goto error;
 351                 }
 352         } else {
 353                 /* Don't prevent the launch of the sessiond on error. */
 354                 mark_thread_as_ready(notifiers);
 355                 goto error_tcp_socket;
 356         }
 357
 358         /*
 359          * Signal that the agent thread is ready. The command thread
 360          * may start to query whether or not agent tracing is enabled.
 361          */
 362         uatomic_set(&agent_tracing_enabled, 1);
 363         mark_thread_as_ready(notifiers);
 364
 365         /* Add TCP socket to poll set. */
 366         ret = lttng_poll_add(&events, reg_sock->fd,
 367                         LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP);
 368         if (ret < 0) {
 369                 goto error;
 370         }
 371
 372         while (1) {
 373                 DBG3("[agent-thread] Manage agent polling");
 374
 375                 /* Inifinite blocking call, waiting for transmission */
 376 restart:
 377                 ret = lttng_poll_wait(&events, -1);
 378                 DBG3("[agent-thread] Manage agent return from poll on %d fds",
 379                                 LTTNG_POLL_GETNB(&events));
 380                 if (ret < 0) {
 381                         /*
 382                          * Restart interrupted system call.
 383                          */
 384                         if (errno == EINTR) {
 385                                 goto restart;
 386                         }
 387                         goto error;
 388                 }
 389                 nb_fd = ret;
 390                 DBG3("[agent-thread] %d fd ready", nb_fd);
 391
 392                 for (i = 0; i < nb_fd; i++) {
 393                         /* Fetch once the poll data */
 394                         revents = LTTNG_POLL_GETEV(&events, i);
 395                         pollfd = LTTNG_POLL_GETFD(&events, i);
 396
 397                         /* Thread quit pipe has been closed. Killing thread. */
 398                         if (pollfd == quit_pipe_read_fd) {
 399                                 goto exit;
 400                         }
 401
 402                         if (revents & LPOLLIN) {
 403                                 int new_fd;
 404                                 struct agent_app *app = NULL;
 405
 406                                 assert(pollfd == reg_sock->fd);
 407                                 new_fd = handle_registration(reg_sock, &app);
 408                                 if (new_fd < 0) {
 409                                         continue;
 410                                 }
 411                                 /* Should not have a NULL app on success. */
 412                                 assert(app);
 413
 414                                 /*
 415                                  * Since this is a command socket (write then read),
 416                                  * only add poll error event to only detect shutdown.
 417                                  */
 418                                 ret = lttng_poll_add(&events, new_fd,
 419                                                 LPOLLERR | LPOLLHUP | LPOLLRDHUP);
 420                                 if (ret < 0) {
 421                                         agent_destroy_app_by_sock(new_fd);
 422                                         continue;
 423                                 }
 424
 425                                 /* Update newly registered app. */
 426                                 update_agent_app(app);
 427
 428                                 /* On failure, the poll will detect it and clean it up. */
 429                                 ret = agent_send_registration_done(app);
 430                                 if (ret < 0) {
 431                                         /* Removing from the poll set */
 432                                         ret = lttng_poll_del(&events, new_fd);
 433                                         if (ret < 0) {
 434                                                 goto error;
 435                                         }
 436                                         agent_destroy_app_by_sock(new_fd);
 437                                         continue;
 438                                 }
 439                         } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
 440                                 /* Removing from the poll set */
 441                                 ret = lttng_poll_del(&events, pollfd);
 442                                 if (ret < 0) {
 443                                         goto error;
 444                                 }
 445                                 agent_destroy_app_by_sock(pollfd);
 446                         } else {
 447                                 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
 448                                 goto error;
 449                         }
 450                 }
 451         }
 452
 453 exit:
 454         /* Whatever happens, try to delete it and exit. */
 455         (void) lttng_poll_del(&events, reg_sock->fd);
 456 error:
 457         destroy_tcp_socket(reg_sock);
 458 error_tcp_socket:
 459         lttng_poll_clean(&events);
 460 error_poll_create:
 461         uatomic_set(&agent_tracing_enabled, 0);
 462         DBG("[agent-thread] Cleaning up and stopping.");
 463         rcu_thread_offline();
 464         rcu_unregister_thread();
 465         return NULL;
 466 }
 467
 468 static bool shutdown_agent_management_thread(void *data)
 469 {
 470         struct thread_notifiers *notifiers = data;
 471         const int write_fd = lttng_pipe_get_writefd(notifiers->quit_pipe);
 472
 473         return notify_thread_pipe(write_fd) == 1;
 474 }
 475
 476 static void cleanup_agent_management_thread(void *data)
 477 {
 478         struct thread_notifiers *notifiers = data;
 479
 480         lttng_pipe_destroy(notifiers->quit_pipe);
 481         sem_destroy(&notifiers->ready);
 482         free(notifiers);
 483 }
 484
 485 bool launch_agent_management_thread(void)
 486 {
 487         struct thread_notifiers *notifiers;
 488         struct lttng_thread *thread;
 489
 490         notifiers = zmalloc(sizeof(*notifiers));
 491         if (!notifiers) {
 492                 goto error_alloc;
 493         }
 494
 495         sem_init(&notifiers->ready, 0, 0);
 496         notifiers->quit_pipe = lttng_pipe_open(FD_CLOEXEC);
 497         if (!notifiers->quit_pipe) {
 498                 goto error;
 499         }
 500         thread = lttng_thread_create("Agent management",
 501                         thread_agent_management,
 502                         shutdown_agent_management_thread,
 503                         cleanup_agent_management_thread,
 504                         notifiers);
 505         if (!thread) {
 506                 goto error;
 507         }
 508         wait_until_thread_is_ready(notifiers);
 509         lttng_thread_put(thread);
 510         return true;
 511 error:
 512         cleanup_agent_management_thread(notifiers);
 513 error_alloc:
 514         return false;
 515 }