X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fagent-thread.c;h=f8456b428d49882945c2089bf12e3c4121c7420b;hp=0e827a4456d4c0ea9c37b54818bbcb6ba4c5e7a7;hb=2288467f63826a06d25ac361fa04ea92ec7ddfa3;hpb=9474416fd76b18db4642bc60b15451b87d74e5d5 diff --git a/src/bin/lttng-sessiond/agent-thread.c b/src/bin/lttng-sessiond/agent-thread.c index 0e827a445..f8456b428 100644 --- a/src/bin/lttng-sessiond/agent-thread.c +++ b/src/bin/lttng-sessiond/agent-thread.c @@ -15,7 +15,7 @@ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE +#define _LGPL_SOURCE #include #include @@ -27,10 +27,13 @@ #include "fd-limit.h" #include "agent-thread.h" +#include "agent.h" #include "lttng-sessiond.h" #include "session.h" #include "utils.h" +static int agent_tracing_enabled = -1; + /* * Note that there is not port here. It's set after this URI is parsed so we * can let the user define a custom one. However, localhost is ALWAYS the @@ -60,63 +63,18 @@ static void update_agent_app(struct agent_app *app) if (session->ust_session) { struct agent *agt; + rcu_read_lock(); agt = trace_ust_find_agent(session->ust_session, app->domain); if (agt) { agent_update(agt, app->sock->fd); } + rcu_read_unlock(); } session_unlock(session); } session_unlock_list(); } -/* - * Destroy a agent application by socket. - */ -static void destroy_agent_app(int sock) -{ - struct agent_app *app; - - assert(sock >= 0); - - /* - * Not finding an application is a very important error that should NEVER - * happen. The hash table deletion is ONLY done through this call even on - * thread cleanup. - */ - rcu_read_lock(); - app = agent_find_app_by_sock(sock); - assert(app); - rcu_read_unlock(); - - /* RCU read side lock is taken in this function call. */ - agent_delete_app(app); - - /* The application is freed in a RCU call but the socket is closed here. */ - agent_destroy_app(app); -} - -/* - * Cleanup remaining agent apps in the hash table. This should only be called in - * the exit path of the thread. - */ -static void clean_agent_apps_ht(void) -{ - struct lttng_ht_node_ulong *node; - struct lttng_ht_iter iter; - - DBG3("[agent-thread] Cleaning agent apps ht"); - - rcu_read_lock(); - cds_lfht_for_each_entry(agent_apps_ht_by_sock->ht, &iter.iter, node, node) { - struct agent_app *app; - - app = caa_container_of(node, struct agent_app, node); - destroy_agent_app(app->sock->fd); - } - rcu_read_unlock(); -} - /* * Create and init socket from uri. */ @@ -125,6 +83,8 @@ static struct lttcomm_sock *init_tcp_socket(void) int ret; struct lttng_uri *uri = NULL; struct lttcomm_sock *sock = NULL; + unsigned int port; + bool bind_succeeded = false; /* * This should never fail since the URI is hardcoded and the port is set @@ -132,8 +92,8 @@ static struct lttcomm_sock *init_tcp_socket(void) */ ret = uri_parse(default_reg_uri, &uri); assert(ret); - assert(agent_tcp_port); - uri->port = agent_tcp_port; + assert(config.agent_tcp_port.begin > 0); + uri->port = config.agent_tcp_port.begin; sock = lttcomm_alloc_sock_from_uri(uri); uri_free(uri); @@ -147,11 +107,43 @@ static struct lttcomm_sock *init_tcp_socket(void) goto error; } - ret = sock->ops->bind(sock); - if (ret < 0) { - WARN("Another session daemon is using this agent port. Agent support " - "will be deactivated to prevent interfering with the tracing."); - goto error; + for (port = config.agent_tcp_port.begin; + port <= config.agent_tcp_port.end; port++) { + ret = lttcomm_sock_set_port(sock, (uint16_t) port); + if (ret) { + ERR("[agent-thread] Failed to set port %u on socket", + port); + goto error; + } + DBG3("[agent-thread] Trying to bind on port %u", port); + ret = sock->ops->bind(sock); + if (!ret) { + bind_succeeded = true; + break; + } + + if (errno == EADDRINUSE) { + DBG("Failed to bind to port %u since it is already in use", + port); + } else { + PERROR("Failed to bind to port %u", port); + goto error; + } + } + + if (!bind_succeeded) { + if (config.agent_tcp_port.begin == config.agent_tcp_port.end) { + WARN("Another process is already using the agent port %i. " + "Agent support will be deactivated.", + config.agent_tcp_port.begin); + goto error; + } else { + WARN("All ports in the range [%i, %i] are already in use. " + "Agent support will be deactivated.", + config.agent_tcp_port.begin, + config.agent_tcp_port.end); + goto error; + } } ret = sock->ops->listen(sock, -1); @@ -160,7 +152,7 @@ static struct lttcomm_sock *init_tcp_socket(void) } DBG("[agent-thread] Listening on TCP port %u and socket %d", - agent_tcp_port, sock->fd); + port, sock->fd); return sock; @@ -176,9 +168,19 @@ error: */ static void destroy_tcp_socket(struct lttcomm_sock *sock) { + int ret; + uint16_t port; + assert(sock); - DBG3("[agent-thread] Destroy TCP socket on port %u", agent_tcp_port); + ret = lttcomm_sock_get_port(sock, &port); + if (ret) { + ERR("[agent-thread] Failed to get port of agent TCP socket"); + port = 0; + } + + DBG3("[agent-thread] Destroy TCP socket on port %" PRIu16, + port); /* This will return gracefully if fd is invalid. */ sock->ops->close(sock); @@ -215,7 +217,7 @@ static int handle_registration(struct lttcomm_sock *reg_sock, size = new_sock->ops->recvmsg(new_sock, &msg, sizeof(msg), 0); if (size < sizeof(msg)) { - ret = -errno; + ret = -EINVAL; goto error_socket; } domain = be32toh(msg.domain); @@ -267,6 +269,24 @@ error: return ret; } +bool agent_tracing_is_enabled(void) +{ + int enabled; + + enabled = uatomic_read(&agent_tracing_enabled); + assert(enabled != -1); + return enabled == 1; +} + +/* + * Write agent TCP port using the rundir. + */ +static int write_agent_port(uint16_t port) +{ + return utils_create_pid_file((pid_t) port, + config.agent_port_file_path.value); +} + /* * This thread manage application notify communication. */ @@ -292,11 +312,32 @@ void *agent_thread_manage_registration(void *data) } reg_sock = init_tcp_socket(); - if (!reg_sock) { + if (reg_sock) { + uint16_t port; + + assert(lttcomm_sock_get_port(reg_sock, &port) == 0); + + ret = write_agent_port(port); + if (ret) { + ERR("[agent-thread] Failed to create agent port file: agent tracing will be unavailable"); + /* Don't prevent the launch of the sessiond on error. */ + sessiond_notify_ready(); + goto error; + } + } else { + /* Don't prevent the launch of the sessiond on error. */ + sessiond_notify_ready(); goto error_tcp_socket; } - /* Add create valid TCP socket to poll set. */ + /* + * Signal that the agent thread is ready. The command thread + * may start to query whether or not agent tracing is enabled. + */ + uatomic_set(&agent_tracing_enabled, 1); + sessiond_notify_ready(); + + /* Add TCP socket to poll set. */ ret = lttng_poll_add(&events, reg_sock->fd, LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP); if (ret < 0) { @@ -304,12 +345,13 @@ void *agent_thread_manage_registration(void *data) } while (1) { - DBG3("[agent-thread] Manage agent polling on %d fds", - LTTNG_POLL_GETNB(&events)); + DBG3("[agent-thread] Manage agent polling"); /* Inifinite blocking call, waiting for transmission */ restart: ret = lttng_poll_wait(&events, -1); + DBG3("[agent-thread] Manage agent return from poll on %d fds", + LTTNG_POLL_GETNB(&events)); if (ret < 0) { /* * Restart interrupted system call. @@ -327,45 +369,37 @@ restart: revents = LTTNG_POLL_GETEV(&events, i); pollfd = LTTNG_POLL_GETFD(&events, i); + if (!revents) { + /* No activity for this FD (poll implementation). */ + continue; + } + /* Thread quit pipe has been closed. Killing thread. */ ret = sessiond_check_thread_quit_pipe(pollfd, revents); if (ret) { goto exit; } - /* - * Check first if this is a POLLERR since POLLIN is also included - * in an error value thus checking first. - */ - if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { - /* Removing from the poll set */ - ret = lttng_poll_del(&events, pollfd); - if (ret < 0) { - goto error; - } - - destroy_agent_app(pollfd); - } else if (revents & (LPOLLIN)) { + if (revents & LPOLLIN) { int new_fd; struct agent_app *app = NULL; - /* Pollin event of agent app socket should NEVER happen. */ assert(pollfd == reg_sock->fd); - new_fd = handle_registration(reg_sock, &app); if (new_fd < 0) { - WARN("[agent-thread] agent registration failed. Ignoring."); - /* Somehow the communication failed. Just continue. */ continue; } /* Should not have a NULL app on success. */ assert(app); - /* Only add poll error event to only detect shutdown. */ + /* + * Since this is a command socket (write then read), + * only add poll error event to only detect shutdown. + */ ret = lttng_poll_add(&events, new_fd, LPOLLERR | LPOLLHUP | LPOLLRDHUP); if (ret < 0) { - destroy_agent_app(new_fd); + agent_destroy_app_by_sock(new_fd); continue; } @@ -373,10 +407,26 @@ restart: update_agent_app(app); /* On failure, the poll will detect it and clean it up. */ - (void) agent_send_registration_done(app); + ret = agent_send_registration_done(app); + if (ret < 0) { + /* Removing from the poll set */ + ret = lttng_poll_del(&events, new_fd); + if (ret < 0) { + goto error; + } + agent_destroy_app_by_sock(new_fd); + continue; + } + } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { + /* Removing from the poll set */ + ret = lttng_poll_del(&events, pollfd); + if (ret < 0) { + goto error; + } + agent_destroy_app_by_sock(pollfd); } else { - ERR("Unknown poll events %u for sock %d", revents, pollfd); - continue; + ERR("Unexpected poll events %u for sock %d", revents, pollfd); + goto error; } } } @@ -389,13 +439,9 @@ error: error_tcp_socket: lttng_poll_clean(&events); error_poll_create: + uatomic_set(&agent_tracing_enabled, 0); DBG("[agent-thread] is cleaning up and stopping."); - if (agent_apps_ht_by_sock) { - clean_agent_apps_ht(); - lttng_ht_destroy(agent_apps_ht_by_sock); - } - rcu_thread_offline(); rcu_unregister_thread(); return NULL;