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

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