X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=123731e171c662cef4e69f4bacedbd42681be508;hb=5cdb6cb8c097f45141f766a5c6df796f038c48ec;hp=39bb32f1b0e198325c8213e04f782b1f6de118ad;hpb=e5e4e2742b1dbc9c55c2fe6cbffe2aaa8cdc6bdc;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 39bb32f1b..123731e17 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -59,6 +59,7 @@ #include #include #include +#include #include #include "version.h" @@ -629,7 +630,6 @@ static void relayd_cleanup(void) if (tracing_group_name_override) { free((void *) tracing_group_name); } - fd_tracker_log(the_fd_tracker); } /* @@ -787,10 +787,22 @@ static int init_thread_quit_pipe(void) "Quit pipe", thread_quit_pipe); } +/* + * Init health quit pipe. + * + * Return -1 on error or 0 if all pipes are created. + */ +static int init_health_quit_pipe(void) +{ + return fd_tracker_util_pipe_open_cloexec(the_fd_tracker, + "Health quit pipe", health_quit_pipe); +} + /* * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set. */ -static int create_thread_poll_set(struct lttng_poll_event *events, int size) +static int create_named_thread_poll_set(struct lttng_poll_event *events, + int size, const char *name) { int ret; @@ -799,10 +811,8 @@ static int create_thread_poll_set(struct lttng_poll_event *events, int size) goto error; } - ret = lttng_poll_create(events, size, LTTNG_CLOEXEC); - if (ret < 0) { - goto error; - } + ret = fd_tracker_util_poll_create(the_fd_tracker, + name, events, 1, LTTNG_CLOEXEC); /* Add quit pipe */ ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR); @@ -830,13 +840,55 @@ static int check_thread_quit_pipe(int fd, uint32_t events) return 0; } +static int create_sock(void *data, int *out_fd) +{ + int ret; + struct lttcomm_sock *sock = data; + + ret = lttcomm_create_sock(sock); + if (ret < 0) { + goto end; + } + + *out_fd = sock->fd; +end: + return ret; +} + +static int close_sock(void *data, int *in_fd) +{ + struct lttcomm_sock *sock = data; + + return sock->ops->close(sock); +} + +static int accept_sock(void *data, int *out_fd) +{ + int ret = 0; + /* Socks is an array of in_sock, out_sock. */ + struct lttcomm_sock **socks = data; + struct lttcomm_sock *in_sock = socks[0]; + + socks[1] = in_sock->ops->accept(in_sock); + if (!socks[1]) { + ret = -1; + goto end; + } + *out_fd = socks[1]->fd; +end: + return ret; +} + /* * Create and init socket from uri. */ -static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri) +static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri, + const char *name) { - int ret; + int ret, sock_fd; struct lttcomm_sock *sock = NULL; + char uri_str[PATH_MAX]; + char *formated_name = NULL; sock = lttcomm_alloc_sock_from_uri(uri); if (sock == NULL) { @@ -844,11 +896,25 @@ static struct lttcomm_sock *relay_socket_create(struct lttng_uri *uri) goto error; } - ret = lttcomm_create_sock(sock); - if (ret < 0) { - goto error; + /* + * Don't fail to create the socket if the name can't be built as it is + * only used for debugging purposes. + */ + ret = uri_to_str_url(uri, uri_str, sizeof(uri_str)); + uri_str[sizeof(uri_str) - 1] = '\0'; + if (ret >= 0) { + ret = asprintf(&formated_name, "%s socket @ %s", name, + uri_str); + if (ret < 0) { + formated_name = NULL; + } } - DBG("Listening on sock %d", sock->fd); + + ret = fd_tracker_open_unsuspendable_fd(the_fd_tracker, &sock_fd, + (const char **) (formated_name ? &formated_name : NULL), + 1, create_sock, sock); + free(formated_name); + DBG("Listening on %s socket %d", name, sock->fd); ret = sock->ops->bind(sock); if (ret < 0) { @@ -870,6 +936,27 @@ error: return NULL; } +static +struct lttcomm_sock *accept_relayd_sock(struct lttcomm_sock *listening_sock, + const char *name) +{ + int out_fd, ret; + struct lttcomm_sock *socks[2] = { listening_sock, NULL }; + struct lttcomm_sock *new_sock = NULL; + + ret = fd_tracker_open_unsuspendable_fd( + the_fd_tracker, &out_fd, + (const char **) &name, + 1, accept_sock, &socks); + if (ret) { + goto end; + } + new_sock = socks[1]; + DBG("%s accepted, socket %d", name, new_sock->fd); +end: + return new_sock; +} + /* * This thread manages the listening for new connections on the network */ @@ -886,12 +973,12 @@ static void *relay_thread_listener(void *data) health_code_update(); - control_sock = relay_socket_create(control_uri); + control_sock = relay_socket_create(control_uri, "Control listener"); if (!control_sock) { goto error_sock_control; } - data_sock = relay_socket_create(data_uri); + data_sock = relay_socket_create(data_uri, "Data listener"); if (!data_sock) { goto error_sock_relay; } @@ -900,7 +987,7 @@ static void *relay_thread_listener(void *data) * Pass 3 as size here for the thread quit pipe, control and * data socket. */ - ret = create_thread_poll_set(&events, 3); + ret = create_named_thread_poll_set(&events, 3, "Listener thread epoll"); if (ret < 0) { goto error_create_poll; } @@ -976,20 +1063,18 @@ restart: */ int val = 1; struct relay_connection *new_conn; - struct lttcomm_sock *newsock; + struct lttcomm_sock *newsock = NULL; enum connection_type type; if (pollfd == data_sock->fd) { type = RELAY_DATA; - newsock = data_sock->ops->accept(data_sock); - DBG("Relay data connection accepted, socket %d", - newsock->fd); + newsock = accept_relayd_sock(data_sock, + "Data socket to relayd"); } else { assert(pollfd == control_sock->fd); type = RELAY_CONTROL; - newsock = control_sock->ops->accept(control_sock); - DBG("Relay control connection accepted, socket %d", - newsock->fd); + newsock = accept_relayd_sock(control_sock, + "Control socket to relayd"); } if (!newsock) { PERROR("accepting sock"); @@ -1042,10 +1127,12 @@ exit: error: error_poll_add: error_testpoint: - lttng_poll_clean(&events); + (void) fd_tracker_util_poll_clean(the_fd_tracker, &events); error_create_poll: if (data_sock->fd >= 0) { - ret = data_sock->ops->close(data_sock); + ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, + &data_sock->fd, 1, close_sock, + data_sock); if (ret) { PERROR("close"); } @@ -1053,7 +1140,9 @@ error_create_poll: lttcomm_destroy_sock(data_sock); error_sock_relay: if (control_sock->fd >= 0) { - ret = control_sock->ops->close(control_sock); + ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, + &control_sock->fd, 1, close_sock, + control_sock); if (ret) { PERROR("close"); } @@ -2871,7 +2960,8 @@ static void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollf (void) lttng_poll_del(events, pollfd); - ret = close(pollfd); + ret = fd_tracker_close_unsuspendable_fd(the_fd_tracker, &pollfd, 1, + fd_tracker_util_close_fd, NULL); if (ret < 0) { ERR("Closing pollfd %d", pollfd); } @@ -2933,7 +3023,7 @@ static void *relay_thread_worker(void *data) goto relay_connections_ht_error; } - ret = create_thread_poll_set(&events, 2); + ret = create_named_thread_poll_set(&events, 2, "Worker thread epoll"); if (ret < 0) { goto error_poll_create; } @@ -3205,12 +3295,13 @@ error: } rcu_read_unlock(); - lttng_poll_clean(&events); + (void) fd_tracker_util_poll_clean(the_fd_tracker, &events); error_poll_create: lttng_ht_destroy(relay_connections_ht); relay_connections_ht_error: /* Close relay conn pipes */ - utils_close_pipe(relay_conn_pipe); + (void) fd_tracker_util_pipe_close(the_fd_tracker, + relay_conn_pipe); if (err) { DBG("Thread exited with error"); } @@ -3232,11 +3323,45 @@ error_testpoint: */ static int create_relay_conn_pipe(void) { - int ret; + return fd_tracker_util_pipe_open_cloexec(the_fd_tracker, + "Relayd connection pipe", relay_conn_pipe); +} - ret = utils_create_pipe_cloexec(relay_conn_pipe); +static +int stdio_open(void *data, int *fds) +{ + fds[0] = fileno(stdout); + fds[1] = fileno(stderr); + return 0; +} - return ret; +static +int noop_close(void *data, int *fds) +{ + return 0; +} + +static +int track_stdio(void) +{ + int fds[2]; + const char *names[] = { "stdout", "stderr" }; + + return fd_tracker_open_unsuspendable_fd(the_fd_tracker, fds, + names, 2, stdio_open, NULL); +} + +static +void untrack_stdio(void) +{ + int fds[] = { fileno(stdout), fileno(stderr) }; + + /* + * noop_close is used since we don't really want to close + * the stdio output fds; we merely want to stop tracking them. + */ + (void) fd_tracker_close_unsuspendable_fd(the_fd_tracker, + fds, 2, noop_close, NULL); } /* @@ -3326,6 +3451,12 @@ int main(int argc, char **argv) goto exit_options; } + ret = track_stdio(); + if (ret) { + retval = -1; + goto exit_options; + } + /* Initialize thread health monitoring */ health_relayd = health_app_create(NR_HEALTH_RELAYD_TYPES); if (!health_relayd) { @@ -3374,7 +3505,7 @@ int main(int argc, char **argv) goto exit_init_data; } - ret = utils_create_pipe(health_quit_pipe); + ret = init_health_quit_pipe(); if (ret) { retval = -1; goto exit_health_quit_pipe; @@ -3470,7 +3601,7 @@ exit_dispatcher_thread: } exit_health_thread: - utils_close_pipe(health_quit_pipe); + (void) fd_tracker_util_pipe_close(the_fd_tracker, health_quit_pipe); exit_health_quit_pipe: exit_init_data: @@ -3488,6 +3619,11 @@ exit_options: /* Ensure all prior call_rcu are done. */ rcu_barrier(); + untrack_stdio(); + /* + * fd_tracker_destroy() will log the contents of the fd-tracker + * if a leak is detected. + */ fd_tracker_destroy(the_fd_tracker); rcu_unregister_thread();