X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=d1ee84eae836739c3addd70cb7ec63a337af682c;hb=0d3cdb1c0cd27ac6255e31eb891b1cdf7c8ec80f;hp=b44255779460289557b7f10640bf14da9f2a190a;hpb=cdd566a28c7058fc32864b8b8a2ff95d82d0e6a8;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index b44255779..d1ee84eae 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -34,11 +34,13 @@ #include #include #include +#include #include #include #include #include #include +#include #include #include @@ -57,6 +59,7 @@ #include #include #include +#include #include #include "version.h" @@ -148,6 +151,9 @@ static uint64_t last_relay_stream_id; */ static struct relay_conn_queue relay_conn_queue; +/* Cap of file desriptors to be in simultaneous use by the relay daemon. */ +static unsigned int lttng_opt_fd_cap; + /* Global relay stream hash table. */ struct lttng_ht *relay_streams_ht; @@ -160,6 +166,9 @@ struct lttng_ht *sessions_ht; /* Relayd health monitoring */ struct health_app *health_relayd; +/* Global fd tracker. */ +struct fd_tracker *the_fd_tracker; + static struct option long_options[] = { { "control-port", 1, 0, 'C', }, { "data-port", 1, 0, 'D', }, @@ -167,6 +176,7 @@ static struct option long_options[] = { { "daemonize", 0, 0, 'd', }, { "background", 0, 0, 'b', }, { "group", 1, 0, 'g', }, + { "fd-cap", 1, 0, '\0', }, { "help", 0, 0, 'h', }, { "output", 1, 0, 'o', }, { "verbose", 0, 0, 'v', }, @@ -192,6 +202,9 @@ static void relayd_config_log(void) if (EXTRA_VERSION_DESCRIPTION[0] != '\0') { DBG("LTTng-relayd extra version description:\n\t" EXTRA_VERSION_DESCRIPTION "\n"); } + if (EXTRA_VERSION_PATCHES[0] != '\0') { + DBG("LTTng-relayd extra patches:\n\t" EXTRA_VERSION_PATCHES "\n"); + } } /* @@ -206,9 +219,33 @@ static int set_option(int opt, const char *arg, const char *optname) switch (opt) { case 0: - fprintf(stderr, "option %s", optname); - if (arg) { - fprintf(stderr, " with arg %s\n", arg); + if (!strcmp(optname, "fd-cap")) { + unsigned long v; + + errno = 0; + v = strtoul(arg, NULL, 0); + if (errno != 0 || !isdigit(arg[0])) { + ERR("Wrong value in --fd-cap parameter: %s", arg); + ret = -1; + goto end; + } + if (v < DEFAULT_RELAYD_MINIMAL_FD_CAP) { + ERR("File descriptor cap must be set to at least %d", + DEFAULT_RELAYD_MINIMAL_FD_CAP); + } + if (v >= UINT_MAX) { + ERR("File descriptor cap overflow in --fd-cap parameter: %s", arg); + ret = -1; + goto end; + } + lttng_opt_fd_cap = (unsigned int) v; + DBG3("File descriptor cap set to %u", lttng_opt_fd_cap); + + } else { + fprintf(stderr, "unknown option %s", optname); + if (arg) { + fprintf(stderr, " with arg %s\n", arg); + } } break; case 'C': @@ -534,6 +571,18 @@ static int set_options(int argc, char **argv) goto exit; } } + if (lttng_opt_fd_cap == 0) { + int ret; + struct rlimit rlimit; + + ret = getrlimit(RLIMIT_NOFILE, &rlimit); + if (ret) { + PERROR("Failed to get file descriptor limit"); + retval = -1; + } + + lttng_opt_fd_cap = rlimit.rlim_cur; + } if (!opt_group_output_by_session && !opt_group_output_by_host) { /* Group by host by default */ @@ -547,13 +596,9 @@ exit: static void print_global_objects(void) { - rcu_register_thread(); - print_viewer_streams(); print_relay_streams(); print_sessions(); - - rcu_unregister_thread(); } /* @@ -576,7 +621,7 @@ static void relayd_cleanup(void) free(opt_output_path); /* Close thread quit pipes */ - utils_close_pipe(thread_quit_pipe); + (void) fd_tracker_util_pipe_close(the_fd_tracker, thread_quit_pipe); uri_free(control_uri); uri_free(data_uri); @@ -738,17 +783,26 @@ void lttng_relay_notify_ready(void) */ static int init_thread_quit_pipe(void) { - int ret; - - ret = utils_create_pipe_cloexec(thread_quit_pipe); + return fd_tracker_util_pipe_open_cloexec(the_fd_tracker, + "Quit pipe", thread_quit_pipe); +} - return ret; +/* + * 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; @@ -757,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); @@ -788,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) { @@ -802,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) { @@ -828,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 */ @@ -844,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; } @@ -858,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; } @@ -934,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"); @@ -1000,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"); } @@ -1011,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"); } @@ -1462,9 +1593,8 @@ int relay_reset_metadata(const struct lttcomm_relayd_hdr *recv_hdr, goto end_unlock; } - ret = utils_rotate_stream_file(stream->path_name, stream->channel_name, - 0, 0, -1, -1, stream->stream_fd->fd, NULL, - &stream->stream_fd->fd); + ret = stream_fd_rotate(stream->stream_fd, + stream->path_name, stream->channel_name, 0, 0, NULL); if (ret < 0) { ERR("Failed to rotate metadata file %s of channel %s", stream->path_name, stream->channel_name); @@ -1584,6 +1714,7 @@ static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr, struct lttcomm_relayd_metadata_payload metadata_payload_header; struct relay_stream *metadata_stream; uint64_t metadata_payload_size; + int metadata_fd = -1; if (!session) { ERR("Metadata sent before version check"); @@ -1614,20 +1745,24 @@ static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr, pthread_mutex_lock(&metadata_stream->lock); - size_ret = lttng_write(metadata_stream->stream_fd->fd, + metadata_fd = stream_fd_get_fd(metadata_stream->stream_fd); + if (metadata_fd < 0) { + goto end_put; + } + size_ret = lttng_write(metadata_fd, payload->data + sizeof(metadata_payload_header), metadata_payload_size); if (size_ret < metadata_payload_size) { ERR("Relay error writing metadata on file"); ret = -1; - goto end_put; + goto end_put_fd; } - size_ret = write_padding_to_file(metadata_stream->stream_fd->fd, + size_ret = write_padding_to_file(metadata_fd, metadata_payload_header.padding_size); if (size_ret < (int64_t) metadata_payload_header.padding_size) { ret = -1; - goto end_put; + goto end_put_fd; } metadata_stream->metadata_received += @@ -1635,6 +1770,8 @@ static int relay_recv_metadata(const struct lttcomm_relayd_hdr *recv_hdr, DBG2("Relay metadata written. Updated metadata_received %" PRIu64, metadata_stream->metadata_received); +end_put_fd: + stream_fd_put_fd(metadata_stream->stream_fd); end_put: pthread_mutex_unlock(&metadata_stream->lock); stream_put(metadata_stream); @@ -2480,14 +2617,14 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num, /* Put ref on previous index_file. */ if (stream->index_file) { - lttng_index_file_put(stream->index_file); + relay_index_file_put(stream->index_file); stream->index_file = NULL; } major = stream->trace->session->major; minor = stream->trace->session->minor; - stream->index_file = lttng_index_file_create(stream->path_name, + stream->index_file = relay_index_file_create(stream->path_name, stream->channel_name, - -1, -1, stream->tracefile_size, + stream->tracefile_size, tracefile_array_get_file_index_head(stream->tfa), lttng_to_index_major(major, minor), lttng_to_index_minor(major, minor)); @@ -2615,11 +2752,9 @@ static enum relay_connection_status relay_process_data_receive_header( /* new_id is updated by utils_rotate_stream_file. */ new_id = old_id; - ret = utils_rotate_stream_file(stream->path_name, + ret = stream_fd_rotate(stream->stream_fd, stream->path_name, stream->channel_name, stream->tracefile_size, - stream->tracefile_count, -1, - -1, stream->stream_fd->fd, - &new_id, &stream->stream_fd->fd); + stream->tracefile_count, &new_id); if (ret < 0) { ERR("Failed to rotate stream output file"); status = RELAY_CONNECTION_STATUS_ERROR; @@ -2656,6 +2791,7 @@ static enum relay_connection_status relay_process_data_receive_payload( bool new_stream = false, close_requested = false; uint64_t left_to_receive = state->left_to_receive; struct relay_session *session; + int stream_fd = -1; DBG3("Receiving data for stream id %" PRIu64 " seqnum %" PRIu64 ", %" PRIu64" bytes received, %" PRIu64 " bytes left to receive", state->header.stream_id, state->header.net_seq_num, @@ -2680,6 +2816,12 @@ static enum relay_connection_status relay_process_data_receive_payload( } } + stream_fd = stream_fd_get_fd(stream->stream_fd); + if (stream_fd < 0) { + status = RELAY_CONNECTION_STATUS_ERROR; + goto end_stream_unlock; + } + /* * The size of the "chunk" received on any iteration is bounded by: * - the data left to receive, @@ -2697,7 +2839,7 @@ static enum relay_connection_status relay_process_data_receive_payload( PERROR("Socket %d error", conn->sock->fd); status = RELAY_CONNECTION_STATUS_ERROR; } - goto end_stream_unlock; + goto end_put_fd; } else if (ret == 0) { /* No more data ready to be consumed on socket. */ DBG3("No more data ready for consumption on data socket of stream id %" PRIu64, @@ -2715,12 +2857,12 @@ static enum relay_connection_status relay_process_data_receive_payload( recv_size = ret; /* Write data to stream output fd. */ - write_ret = lttng_write(stream->stream_fd->fd, data_buffer, + write_ret = lttng_write(stream_fd, data_buffer, recv_size); if (write_ret < (ssize_t) recv_size) { ERR("Relay error writing data to file"); status = RELAY_CONNECTION_STATUS_ERROR; - goto end_stream_unlock; + goto end_put_fd; } left_to_receive -= recv_size; @@ -2739,17 +2881,17 @@ static enum relay_connection_status relay_process_data_receive_payload( DBG3("Partial receive on data connection of stream id %" PRIu64 ", %" PRIu64 " bytes received, %" PRIu64 " bytes left to receive", state->header.stream_id, state->received, state->left_to_receive); - goto end_stream_unlock; + goto end_put_fd; } - ret = write_padding_to_file(stream->stream_fd->fd, + ret = write_padding_to_file(stream_fd, state->header.padding_size); if ((int64_t) ret < (int64_t) state->header.padding_size) { ERR("write_padding_to_file: fail stream %" PRIu64 " net_seq_num %" PRIu64 " ret %d", stream->stream_handle, state->header.net_seq_num, ret); status = RELAY_CONNECTION_STATUS_ERROR; - goto end_stream_unlock; + goto end_put_fd; } @@ -2761,7 +2903,7 @@ static enum relay_connection_status relay_process_data_receive_payload( stream->stream_handle, state->header.net_seq_num, ret); status = RELAY_CONNECTION_STATUS_ERROR; - goto end_stream_unlock; + goto end_put_fd; } } @@ -2782,6 +2924,8 @@ static enum relay_connection_status relay_process_data_receive_payload( connection_reset_protocol_state(conn); state = NULL; +end_put_fd: + stream_fd_put_fd(stream->stream_fd); end_stream_unlock: close_requested = stream->close_requested; pthread_mutex_unlock(&stream->lock); @@ -2829,7 +2973,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); } @@ -2891,7 +3036,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; } @@ -3163,12 +3308,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"); } @@ -3190,11 +3336,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); } /* @@ -3231,6 +3411,11 @@ int main(int argc, char **argv) goto exit_options; } + ret = fclose(stdin); + if (ret) { + PERROR("Failed to close stdin"); + goto exit_options; + } /* Try to create directory if -o, --output is specified. */ if (opt_output_path) { if (*opt_output_path != '/') { @@ -3250,26 +3435,14 @@ int main(int argc, char **argv) /* Daemonize */ if (opt_daemon || opt_background) { - int i; - ret = lttng_daemonize(&child_ppid, &recv_child_signal, !opt_background); if (ret < 0) { retval = -1; goto exit_options; } - - /* - * We are in the child. Make sure all other file - * descriptors are closed, in case we are called with - * more opened file descriptors than the standard ones. - */ - for (i = 3; i < sysconf(_SC_OPEN_MAX); i++) { - (void) close(i); - } } - if (opt_working_directory) { ret = utils_change_working_dir(opt_working_directory); if (ret) { @@ -3277,6 +3450,25 @@ int main(int argc, char **argv) goto exit_options; } } + /* + * The RCU thread registration (and use, through the fd-tracker's + * creation) is done after the daemonization to allow us to not + * deal with liburcu's fork() management as the call RCU needs to + * be restored. + */ + rcu_register_thread(); + + the_fd_tracker = fd_tracker_create(lttng_opt_fd_cap); + if (!the_fd_tracker) { + retval = -1; + 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); @@ -3326,7 +3518,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; @@ -3422,7 +3614,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: @@ -3440,6 +3632,14 @@ 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(); + if (!retval) { exit(EXIT_SUCCESS); } else {