X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=0f81d556dd4df73a910471756b75cf6567581a36;hp=bdefd16763850b7e75a5a9f294ebfb434aca41b3;hb=c8f59ee5fc11492ef472dc5cfd2fd2c4926b1787;hpb=25672d0260dce634e1a249dd97d3cc0b592f34a6 diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index bdefd1676..0f81d556d 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -32,6 +32,7 @@ #include #include #include +#include #include #include #include @@ -49,14 +50,15 @@ #include #include #include +#include #include "lttng-relayd.h" /* command line options */ static int opt_daemon; static char *opt_output_path; -static struct lttng_uri *control_uri = NULL; -static struct lttng_uri *data_uri = NULL; +static struct lttng_uri *control_uri; +static struct lttng_uri *data_uri; const char *progname; static int is_root; /* Set to 1 if the daemon is running as root */ @@ -73,14 +75,15 @@ static int thread_quit_pipe[2] = { -1, -1 }; */ static int relay_cmd_pipe[2] = { -1, -1 }; +/* Shared between threads */ static int dispatch_thread_exit; static pthread_t listener_thread; static pthread_t dispatcher_thread; static pthread_t worker_thread; -static uint64_t last_relay_stream_id = 0; -static uint64_t last_relay_session_id = 0; +static uint64_t last_relay_stream_id; +static uint64_t last_relay_session_id; /* * Relay command queue. @@ -91,8 +94,8 @@ static uint64_t last_relay_session_id = 0; static struct relay_cmd_queue relay_cmd_queue; /* buffer allocated at startup, used to store the trace data */ -static char *data_buffer = NULL; -static unsigned int data_buffer_size = 0; +static char *data_buffer; +static unsigned int data_buffer_size; /* * usage function on stderr @@ -117,13 +120,13 @@ int parse_args(int argc, char **argv) char *default_address; static struct option long_options[] = { - { "control-port", 1, 0, 'C' }, - { "data-port", 1, 0, 'D' }, - { "daemonize", 0, 0, 'd' }, - { "help", 0, 0, 'h' }, - { "output", 1, 0, 'o' }, - { "verbose", 0, 0, 'v' }, - { NULL, 0, 0, 0 } + { "control-port", 1, 0, 'C', }, + { "data-port", 1, 0, 'D', }, + { "daemonize", 0, 0, 'd', }, + { "help", 0, 0, 'h', }, + { "output", 1, 0, 'o', }, + { "verbose", 0, 0, 'v', }, + { NULL, 0, 0, 0, }, }; while (1) { @@ -228,18 +231,16 @@ exit: static void cleanup(void) { - int i, ret; - DBG("Cleaning up"); - for (i = 0; i < 2; i++) { - if (thread_quit_pipe[i] >= 0) { - ret = close(thread_quit_pipe[i]); - if (ret) { - PERROR("close"); - } - } - } + /* free the dynamically allocated opt_output_path */ + free(opt_output_path); + + /* Close thread quit pipes */ + utils_close_pipe(thread_quit_pipe); + + /* Close relay cmd pipes */ + utils_close_pipe(relay_cmd_pipe); } /* @@ -250,7 +251,9 @@ int notify_thread_pipe(int wpipe) { int ret; - ret = write(wpipe, "!", 1); + do { + ret = write(wpipe, "!", 1); + } while (ret < 0 && errno == EINTR); if (ret < 0) { PERROR("write poll pipe"); } @@ -274,7 +277,7 @@ void stop_threads(void) } /* Dispatch thread */ - dispatch_thread_exit = 1; + CMM_STORE_SHARED(dispatch_thread_exit, 1); futex_nto1_wake(&relay_cmd_queue.futex); } @@ -351,23 +354,10 @@ int set_signal_handler(void) static int init_thread_quit_pipe(void) { - int ret, i; - - ret = pipe(thread_quit_pipe); - if (ret < 0) { - PERROR("thread quit pipe"); - goto error; - } + int ret; - for (i = 0; i < 2; i++) { - ret = fcntl(thread_quit_pipe[i], F_SETFD, FD_CLOEXEC); - if (ret < 0) { - PERROR("fcntl"); - goto error; - } - } + ret = utils_create_pipe_cloexec(thread_quit_pipe); -error: return ret; } @@ -457,34 +447,41 @@ error: return NULL; } +/* + * Return nonzero if stream needs to be closed. + */ +static +int close_stream_check(struct relay_stream *stream) +{ + + if (stream->close_flag && stream->prev_seq == stream->last_net_seq_num) { + return 1; + } + return 0; +} + /* * This thread manages the listening for new connections on the network */ static void *relay_thread_listener(void *data) { - int i, ret, pollfd; + int i, ret, pollfd, err = -1; int val = 1; uint32_t revents, nb_fd; struct lttng_poll_event events; struct lttcomm_sock *control_sock, *data_sock; - /* - * Get allocated in this thread, enqueued to a global queue, dequeued and - * freed in the worker thread. - */ - struct relay_command *relay_cmd = NULL; - DBG("[thread] Relay listener started"); control_sock = relay_init_sock(control_uri); if (!control_sock) { - goto error_sock; + goto error_sock_control; } data_sock = relay_init_sock(data_uri); if (!data_sock) { - goto error_sock; + goto error_sock_relay; } /* @@ -533,14 +530,21 @@ restart: /* Thread quit pipe has been closed. Killing thread. */ ret = check_thread_quit_pipe(pollfd, revents); if (ret) { - goto error; + err = 0; + goto exit; } if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) { ERR("socket poll error"); goto error; } else if (revents & LPOLLIN) { - struct lttcomm_sock *newsock = NULL; + /* + * Get allocated in this thread, + * enqueued to a global queue, dequeued + * and freed in the worker thread. + */ + struct relay_command *relay_cmd; + struct lttcomm_sock *newsock; relay_cmd = zmalloc(sizeof(struct relay_command)); if (relay_cmd == NULL) { @@ -550,16 +554,19 @@ restart: if (pollfd == data_sock->fd) { newsock = data_sock->ops->accept(data_sock); - if (newsock < 0) { + if (!newsock) { PERROR("accepting data sock"); + free(relay_cmd); goto error; } relay_cmd->type = RELAY_DATA; DBG("Relay data connection accepted, socket %d", newsock->fd); - } else if (pollfd == control_sock->fd) { + } else { + assert(pollfd == control_sock->fd); newsock = control_sock->ops->accept(control_sock); - if (newsock < 0) { + if (!newsock) { PERROR("accepting control sock"); + free(relay_cmd); goto error; } relay_cmd->type = RELAY_CONTROL; @@ -569,6 +576,8 @@ restart: &val, sizeof(int)); if (ret < 0) { PERROR("setsockopt inet"); + lttcomm_destroy_sock(newsock); + free(relay_cmd); goto error; } relay_cmd->sock = newsock; @@ -586,28 +595,32 @@ restart: } } +exit: error: error_poll_add: lttng_poll_clean(&events); error_create_poll: - if (control_sock->fd >= 0) { - ret = control_sock->ops->close(control_sock); + if (data_sock->fd >= 0) { + ret = data_sock->ops->close(data_sock); if (ret) { PERROR("close"); } - lttcomm_destroy_sock(control_sock); } - if (data_sock->fd >= 0) { - ret = data_sock->ops->close(data_sock); + lttcomm_destroy_sock(data_sock); +error_sock_relay: + if (control_sock->fd >= 0) { + ret = control_sock->ops->close(control_sock); if (ret) { PERROR("close"); } - lttcomm_destroy_sock(data_sock); } - + lttcomm_destroy_sock(control_sock); +error_sock_control: + if (err) { + DBG("Thread exited with error"); + } DBG("Relay listener thread cleanup complete"); stop_threads(); -error_sock: return NULL; } @@ -623,7 +636,7 @@ void *relay_thread_dispatcher(void *data) DBG("[thread] Relay dispatcher started"); - while (!dispatch_thread_exit) { + while (!CMM_LOAD_SHARED(dispatch_thread_exit)) { /* Atomically prepare the queue futex */ futex_nto1_prepare(&relay_cmd_queue.futex); @@ -644,8 +657,10 @@ void *relay_thread_dispatcher(void *data) * call is blocking so we can be assured that the data will be read * at some point in time or wait to the end of the world :) */ - ret = write(relay_cmd_pipe[1], relay_cmd, - sizeof(struct relay_command)); + do { + ret = write(relay_cmd_pipe[1], relay_cmd, + sizeof(struct relay_command)); + } while (ret < 0 && errno == EINTR); free(relay_cmd); if (ret < 0) { PERROR("write cmd pipe"); @@ -673,7 +688,7 @@ static char *expand_full_path(const char *path) { const char *end_path = path; - char *next, *cut_path, *expanded_path; + char *next, *cut_path, *expanded_path, *respath; /* Find last token delimited by '/' */ while ((next = strpbrk(end_path + 1, "/"))) { @@ -685,11 +700,12 @@ char *expand_full_path(const char *path) expanded_path = malloc(PATH_MAX); if (expanded_path == NULL) { - goto error; + respath = NULL; + goto end; } - expanded_path = realpath((char *)cut_path, expanded_path); - if (expanded_path == NULL) { + respath = realpath(cut_path, expanded_path); + if (respath == NULL) { switch (errno) { case ENOENT: ERR("%s: No such file or directory", cut_path); @@ -698,18 +714,14 @@ char *expand_full_path(const char *path) PERROR("realpath"); break; } - goto error; + free(expanded_path); + } else { + /* Add end part to expanded path */ + strcat(respath, end_path); } - - /* Add end part to expanded path */ - strcat(expanded_path, end_path); - - free(cut_path); - return expanded_path; - -error: +end: free(cut_path); - return NULL; + return respath; } @@ -761,7 +773,7 @@ int mkdir_recursive(char *path, mode_t mode) if (ret < 0) { ret = mkdir(tmp, mode); if (ret < 0) { - if (!(errno == EEXIST)) { + if (errno != EEXIST) { PERROR("mkdir recursive"); ret = -errno; goto error; @@ -774,7 +786,7 @@ int mkdir_recursive(char *path, mode_t mode) ret = mkdir(tmp, mode); if (ret < 0) { - if (!(errno == EEXIST)) { + if (errno != EEXIST) { PERROR("mkdir recursive last piece"); ret = -errno; } else { @@ -786,46 +798,73 @@ error: return ret; } +static +char *create_output_path_auto(char *path_name) +{ + int ret; + char *traces_path = NULL; + char *alloc_path = NULL; + char *default_path; + + default_path = config_get_default_path(); + if (default_path == NULL) { + ERR("Home path not found.\n \ + Please specify an output path using -o, --output PATH"); + goto exit; + } + alloc_path = strdup(default_path); + if (alloc_path == NULL) { + PERROR("Path allocation"); + goto exit; + } + ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME + "/%s", alloc_path, path_name); + if (ret < 0) { + PERROR("asprintf trace dir name"); + goto exit; + } +exit: + free(alloc_path); + return traces_path; +} + +static +char *create_output_path_noauto(char *path_name) +{ + int ret; + char *traces_path = NULL; + char *full_path; + + full_path = expand_full_path(opt_output_path); + ret = asprintf(&traces_path, "%s/%s", full_path, path_name); + if (ret < 0) { + PERROR("asprintf trace dir name"); + goto exit; + } +exit: + free(full_path); + return traces_path; +} + /* * create_output_path: create the output trace directory */ static char *create_output_path(char *path_name) { - int ret = 0; - char *alloc_path = NULL; - char *traces_path = NULL; - char *full_path = NULL; - - /* Auto output path */ if (opt_output_path == NULL) { - alloc_path = strdup(config_get_default_path()); - if (alloc_path == NULL) { - ERR("Home path not found.\n \ - Please specify an output path using -o, --output PATH"); - ret = -1; - goto exit; - } - - ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME - "/%s", alloc_path, path_name); - if (ret < 0) { - PERROR("asprintf trace dir name"); - goto exit; - } + return create_output_path_auto(path_name); } else { - full_path = expand_full_path(opt_output_path); - ret = asprintf(&traces_path, "%s/%s", full_path, path_name); - if (ret < 0) { - PERROR("asprintf trace dir name"); - goto exit; - } + return create_output_path_noauto(path_name); } - free(alloc_path); - free(full_path); +} -exit: - return traces_path; +static +void deferred_free_stream(struct rcu_head *head) +{ + struct relay_stream *stream = + caa_container_of(head, struct relay_stream, rcu_node); + free(stream); } /* @@ -840,12 +879,14 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht struct relay_stream *stream; int ret; - if (!cmd->session) + if (!cmd->session) { return; + } - DBG("Relay deleting session %lu", cmd->session->id); + DBG("Relay deleting session %" PRIu64, cmd->session->id); free(cmd->session->sock); + rcu_read_lock(); cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) { node = lttng_ht_iter_get_node_ulong(&iter); if (node) { @@ -855,10 +896,12 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht close(stream->fd); ret = lttng_ht_del(streams_ht, &iter); assert(!ret); - free(stream); + call_rcu(&stream->rcu_node, + deferred_free_stream); } } } + rcu_read_unlock(); } /* @@ -896,8 +939,9 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, goto end_no_session; } + rcu_read_lock(); stream->stream_handle = ++last_relay_stream_id; - stream->seq = 0; + stream->prev_seq = -1ULL; stream->session = session; root_path = create_output_path(stream_info.pathname); @@ -907,7 +951,6 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, } ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG); if (ret < 0) { - free(root_path); ERR("relay creating output directory"); goto end; } @@ -939,9 +982,9 @@ end: free(root_path); /* send the session id to the client or a negative return code on error */ if (ret < 0) { - reply.ret_code = htobe32(LTTCOMM_ERR); + reply.ret_code = htobe32(LTTNG_ERR_UNK); } else { - reply.ret_code = htobe32(LTTCOMM_OK); + reply.ret_code = htobe32(LTTNG_OK); } reply.handle = htobe64(stream->stream_handle); send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, @@ -949,6 +992,87 @@ end: if (send_ret < 0) { ERR("Relay sending stream id"); } + rcu_read_unlock(); + +end_no_session: + return ret; +} + +/* + * relay_close_stream: close a specific stream + */ +static +int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd, struct lttng_ht *streams_ht) +{ + struct relay_session *session = cmd->session; + struct lttcomm_relayd_close_stream stream_info; + struct lttcomm_relayd_generic_reply reply; + struct relay_stream *stream; + int ret, send_ret; + struct lttng_ht_node_ulong *node; + struct lttng_ht_iter iter; + + DBG("Close stream received"); + + if (!session || session->version_check_done == 0) { + ERR("Trying to close a stream before version check"); + ret = -1; + goto end_no_session; + } + + ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info, + sizeof(struct lttcomm_relayd_close_stream), MSG_WAITALL); + if (ret < sizeof(struct lttcomm_relayd_close_stream)) { + ERR("Relay didn't receive valid add_stream struct size : %d", ret); + ret = -1; + goto end_no_session; + } + + rcu_read_lock(); + lttng_ht_lookup(streams_ht, + (void *)((unsigned long) be64toh(stream_info.stream_id)), + &iter); + node = lttng_ht_iter_get_node_ulong(&iter); + if (node == NULL) { + DBG("Relay stream %" PRIu64 " not found", be64toh(stream_info.stream_id)); + ret = -1; + goto end_unlock; + } + + stream = caa_container_of(node, struct relay_stream, stream_n); + if (!stream) { + ret = -1; + goto end_unlock; + } + + stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num); + stream->close_flag = 1; + + if (close_stream_check(stream)) { + int delret; + + close(stream->fd); + delret = lttng_ht_del(streams_ht, &iter); + assert(!delret); + call_rcu(&stream->rcu_node, + deferred_free_stream); + DBG("Closed tracefile %d from close stream", stream->fd); + } + +end_unlock: + rcu_read_unlock(); + + if (ret < 0) { + reply.ret_code = htobe32(LTTNG_ERR_UNK); + } else { + reply.ret_code = htobe32(LTTNG_OK); + } + send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, + sizeof(struct lttcomm_relayd_generic_reply), 0); + if (send_ret < 0) { + ERR("Relay sending stream id"); + } end_no_session: return ret; @@ -963,7 +1087,7 @@ void relay_unknown_command(struct relay_command *cmd) struct lttcomm_relayd_generic_reply reply; int ret; - reply.ret_code = htobe32(LTTCOMM_ERR); + reply.ret_code = htobe32(LTTNG_ERR_UNK); ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(struct lttcomm_relayd_generic_reply), 0); if (ret < 0) { @@ -979,13 +1103,13 @@ static int relay_start(struct lttcomm_relayd_hdr *recv_hdr, struct relay_command *cmd) { - int ret = htobe32(LTTCOMM_OK); + int ret = htobe32(LTTNG_OK); struct lttcomm_relayd_generic_reply reply; struct relay_session *session = cmd->session; if (!session) { DBG("Trying to start the streaming without a session established"); - ret = htobe32(LTTCOMM_ERR); + ret = htobe32(LTTNG_ERR_UNK); } reply.ret_code = ret; @@ -1000,6 +1124,7 @@ int relay_start(struct lttcomm_relayd_hdr *recv_hdr, /* * Get stream from stream id. + * Need to be called with RCU read-side lock held. */ static struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id, @@ -1014,7 +1139,7 @@ struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id, &iter); node = lttng_ht_iter_get_node_ulong(&iter); if (node == NULL) { - DBG("Relay stream %lu not found", stream_id); + DBG("Relay stream %" PRIu64 " not found", stream_id); ret = NULL; goto end; } @@ -1025,6 +1150,36 @@ end: return ret; } +/* + * Append padding to the file pointed by the file descriptor fd. + */ +static int write_padding_to_file(int fd, uint32_t size) +{ + int ret = 0; + char *zeros; + + if (size == 0) { + goto end; + } + + zeros = zmalloc(size); + if (zeros == NULL) { + PERROR("zmalloc zeros for padding"); + ret = -1; + goto end; + } + + do { + ret = write(fd, zeros, size); + } while (ret < 0 && errno == EINTR); + if (ret < 0) { + PERROR("write padding to file"); + } + +end: + return ret; +} + /* * relay_recv_metadata: receive the metada for the session. */ @@ -1032,7 +1187,7 @@ static int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, struct relay_command *cmd, struct lttng_ht *streams_ht) { - int ret = htobe32(LTTCOMM_OK); + int ret = htobe32(LTTNG_OK); struct relay_session *session = cmd->session; struct lttcomm_relayd_metadata_payload *metadata_struct; struct relay_stream *metadata_stream; @@ -1044,8 +1199,14 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, goto end; } - data_size = be64toh(recv_hdr->data_size); - payload_size = data_size - sizeof(uint64_t); + data_size = payload_size = be64toh(recv_hdr->data_size); + if (data_size < sizeof(struct lttcomm_relayd_metadata_payload)) { + ERR("Incorrect data size"); + ret = -1; + goto end; + } + payload_size -= sizeof(struct lttcomm_relayd_metadata_payload); + if (data_buffer_size < data_size) { data_buffer = realloc(data_buffer, data_size); if (!data_buffer) { @@ -1056,30 +1217,44 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, data_buffer_size = data_size; } memset(data_buffer, 0, data_size); - DBG2("Relay receiving metadata, waiting for %lu bytes", data_size); - ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL); + DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size); + ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, + MSG_WAITALL); if (ret < 0 || ret != data_size) { ret = -1; ERR("Relay didn't receive the whole metadata"); goto end; } metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer; + + rcu_read_lock(); metadata_stream = relay_stream_from_stream_id( be64toh(metadata_struct->stream_id), streams_ht); if (!metadata_stream) { ret = -1; - goto end; + goto end_unlock; } - ret = write(metadata_stream->fd, metadata_struct->payload, - payload_size); - if (ret < (payload_size)) { + do { + ret = write(metadata_stream->fd, metadata_struct->payload, + payload_size); + } while (ret < 0 && errno == EINTR); + if (ret < payload_size) { ERR("Relay error writing metadata on file"); ret = -1; - goto end; + goto end_unlock; } + + ret = write_padding_to_file(metadata_stream->fd, + be32toh(metadata_struct->padding_size)); + if (ret < 0) { + goto end_unlock; + } + DBG2("Relay metadata written"); +end_unlock: + rcu_read_unlock(); end: return ret; } @@ -1091,9 +1266,9 @@ static int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr, struct relay_command *cmd) { - int ret = htobe32(LTTCOMM_OK); + int ret; struct lttcomm_relayd_version reply; - struct relay_session *session = NULL; + struct relay_session *session; if (cmd->session == NULL) { session = zmalloc(sizeof(struct relay_session)); @@ -1103,12 +1278,19 @@ int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr, goto end; } session->id = ++last_relay_session_id; - DBG("Created session %lu", session->id); + DBG("Created session %" PRIu64, session->id); cmd->session = session; + } else { + session = cmd->session; } session->version_check_done = 1; - sscanf(VERSION, "%u.%u", &reply.major, &reply.minor); + ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor); + if (ret < 2) { + ERR("Error in scanning version"); + ret = -1; + goto end; + } reply.major = htobe32(reply.major); reply.minor = htobe32(reply.minor); ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, @@ -1116,12 +1298,109 @@ int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr, if (ret < 0) { ERR("Relay sending version"); } - DBG("Version check done"); + DBG("Version check done (%u.%u)", be32toh(reply.major), + be32toh(reply.minor)); end: return ret; } +/* + * Check for data availability for a given stream id from the session daemon. + */ +static +int relay_data_available(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd, struct lttng_ht *streams_ht) +{ + struct relay_session *session = cmd->session; + struct lttcomm_relayd_data_available msg; + struct lttcomm_relayd_generic_reply reply; + struct relay_stream *stream; + int ret; + struct lttng_ht_node_ulong *node; + struct lttng_ht_iter iter; + uint64_t last_net_seq_num, stream_id; + + DBG("Data available command received"); + + if (!session || session->version_check_done == 0) { + ERR("Trying to check for data before version check"); + ret = -1; + goto end_no_session; + } + + ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), MSG_WAITALL); + if (ret < sizeof(msg)) { + ERR("Relay didn't receive valid data_available struct size : %d", ret); + ret = -1; + goto end_no_session; + } + + stream_id = be64toh(msg.stream_id); + last_net_seq_num = be64toh(msg.last_net_seq_num); + + rcu_read_lock(); + lttng_ht_lookup(streams_ht, (void *)((unsigned long) stream_id), &iter); + node = lttng_ht_iter_get_node_ulong(&iter); + if (node == NULL) { + DBG("Relay stream %" PRIu64 " not found", stream_id); + ret = -1; + goto end_unlock; + } + + stream = caa_container_of(node, struct relay_stream, stream_n); + assert(stream); + + DBG("Data available for stream id %" PRIu64 " prev_seq %" PRIu64 + " and last_seq %" PRIu64, stream_id, stream->prev_seq, + last_net_seq_num); + + if (stream->prev_seq == -1UL || stream->prev_seq <= last_net_seq_num) { + /* Data has in fact been written and is available */ + ret = 1; + } else { + /* Data still being streamed. */ + ret = 0; + } + +end_unlock: + rcu_read_unlock(); + + reply.ret_code = htobe32(ret); + ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (ret < 0) { + ERR("Relay data available ret code failed"); + } + +end_no_session: + return ret; +} + +/* + * Wait for the control socket to reach a quiescent state. + * + * Note that for now, when receiving this command from the session daemon, this + * means that every subsequent commands or data received on the control socket + * has been handled. So, this is why we simply return OK here. + */ +static +int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd) +{ + int ret; + struct lttcomm_relayd_generic_reply reply; + + DBG("Checking quiescent state on control socket"); + + reply.ret_code = htobe32(LTTNG_OK); + ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (ret < 0) { + ERR("Relay data available ret code failed"); + } + + return ret; +} + /* * relay_process_control: Process the commands received on the control socket */ @@ -1149,6 +1428,15 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, case RELAYD_VERSION: ret = relay_send_version(recv_hdr, cmd); break; + case RELAYD_CLOSE_STREAM: + ret = relay_close_stream(recv_hdr, cmd, streams_ht); + break; + case RELAYD_DATA_AVAILABLE: + ret = relay_data_available(recv_hdr, cmd, streams_ht); + break; + case RELAYD_QUIESCENT_CONTROL: + ret = relay_quiescent_control(recv_hdr, cmd); + break; case RELAYD_UPDATE_SYNC_INFO: default: ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd)); @@ -1171,6 +1459,7 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) struct relay_stream *stream; struct lttcomm_relayd_data_hdr data_hdr; uint64_t stream_id; + uint64_t net_seq_num; uint32_t data_size; ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr, @@ -1182,10 +1471,12 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) } stream_id = be64toh(data_hdr.stream_id); + + rcu_read_lock(); stream = relay_stream_from_stream_id(stream_id, streams_ht); if (!stream) { ret = -1; - goto end; + goto end_unlock; } data_size = be32toh(data_hdr.data_size); @@ -1194,38 +1485,65 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) if (!data_buffer) { ERR("Allocating data buffer"); ret = -1; - goto end; + goto end_unlock; } data_buffer_size = data_size; } memset(data_buffer, 0, data_size); + net_seq_num = be64toh(data_hdr.net_seq_num); + + DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64, + data_size, stream_id, net_seq_num); ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL); if (ret <= 0) { ret = -1; - goto end; + goto end_unlock; } - ret = write(stream->fd, data_buffer, data_size); + do { + ret = write(stream->fd, data_buffer, data_size); + } while (ret < 0 && errno == EINTR); if (ret < data_size) { ERR("Relay error writing data to file"); ret = -1; - goto end; + goto end_unlock; } - DBG2("Relay wrote %d bytes to tracefile for stream id %lu", ret, stream->stream_handle); + DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64, + ret, stream->stream_handle); + + ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size)); + if (ret < 0) { + goto end_unlock; + } + + stream->prev_seq = net_seq_num; + + /* Check if we need to close the FD */ + if (close_stream_check(stream)) { + struct lttng_ht_iter iter; + + close(stream->fd); + iter.iter.node = &stream->stream_n.node; + ret = lttng_ht_del(streams_ht, &iter); + assert(!ret); + call_rcu(&stream->rcu_node, + deferred_free_stream); + DBG("Closed tracefile %d after recv data", stream->fd); + } + +end_unlock: + rcu_read_unlock(); end: return ret; } static -void relay_cleanup_connection(struct lttng_ht *relay_connections_ht, struct lttng_poll_event *events, - struct lttng_ht *streams_ht, int pollfd, struct lttng_ht_iter *iter) +void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd) { int ret; - ret = lttng_ht_del(relay_connections_ht, iter); - assert(!ret); lttng_poll_del(events, pollfd); ret = close(pollfd); @@ -1238,32 +1556,58 @@ static int relay_add_connection(int fd, struct lttng_poll_event *events, struct lttng_ht *relay_connections_ht) { - int ret; struct relay_command *relay_connection; + int ret; relay_connection = zmalloc(sizeof(struct relay_command)); if (relay_connection == NULL) { PERROR("Relay command zmalloc"); - ret = -1; - goto end; + goto error; } ret = read(fd, relay_connection, sizeof(struct relay_command)); - if (ret < 0 || ret < sizeof(relay_connection)) { + if (ret < 0 || ret < sizeof(struct relay_command)) { PERROR("read relay cmd pipe"); - ret = -1; - goto end; + goto error_read; } lttng_ht_node_init_ulong(&relay_connection->sock_n, (unsigned long) relay_connection->sock->fd); + rcu_read_lock(); lttng_ht_add_unique_ulong(relay_connections_ht, &relay_connection->sock_n); - ret = lttng_poll_add(events, + rcu_read_unlock(); + return lttng_poll_add(events, relay_connection->sock->fd, LPOLLIN | LPOLLRDHUP); -end: - return ret; +error_read: + free(relay_connection); +error: + return -1; +} + +static +void deferred_free_connection(struct rcu_head *head) +{ + struct relay_command *relay_connection = + caa_container_of(head, struct relay_command, rcu_node); + free(relay_connection); +} + +static +void relay_del_connection(struct lttng_ht *relay_connections_ht, + struct lttng_ht *streams_ht, struct lttng_ht_iter *iter, + struct relay_command *relay_connection) +{ + int ret; + + ret = lttng_ht_del(relay_connections_ht, iter); + assert(!ret); + if (relay_connection->type == RELAY_CONTROL) { + relay_delete_session(relay_connection, streams_ht); + } + call_rcu(&relay_connection->rcu_node, + deferred_free_connection); } /* @@ -1272,7 +1616,7 @@ end: static void *relay_thread_worker(void *data) { - int i, ret, pollfd; + int i, ret, pollfd, err = -1; uint32_t revents, nb_fd; struct relay_command *relay_connection; struct lttng_poll_event events; @@ -1284,11 +1628,19 @@ void *relay_thread_worker(void *data) DBG("[thread] Relay worker started"); + rcu_register_thread(); + /* table of connections indexed on socket */ relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG); + if (!relay_connections_ht) { + goto relay_connections_ht_error; + } /* tables of streams indexed by stream ID */ streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG); + if (!streams_ht) { + goto streams_ht_error; + } ret = create_thread_poll_set(&events, 2); if (ret < 0) { @@ -1308,6 +1660,7 @@ void *relay_thread_worker(void *data) /* Infinite blocking call, waiting for transmission */ restart: + DBG3("Relayd worker thread polling..."); ret = lttng_poll_wait(&events, -1); if (ret < 0) { /* @@ -1327,7 +1680,8 @@ void *relay_thread_worker(void *data) /* Thread quit pipe has been closed. Killing thread. */ ret = check_thread_quit_pipe(pollfd, revents); if (ret) { - goto error; + err = 0; + goto exit; } /* Inspect the relay cmd pipe for new connection */ @@ -1344,12 +1698,14 @@ void *relay_thread_worker(void *data) } } } else if (revents > 0) { + rcu_read_lock(); lttng_ht_lookup(relay_connections_ht, (void *)((unsigned long) pollfd), &iter); node = lttng_ht_iter_get_node_ulong(&iter); if (node == NULL) { DBG2("Relay sock %d not found", pollfd); + rcu_read_unlock(); goto error; } relay_connection = caa_container_of(node, @@ -1357,17 +1713,16 @@ void *relay_thread_worker(void *data) if (revents & (LPOLLERR)) { ERR("POLL ERROR"); - relay_cleanup_connection(relay_connections_ht, - &events, streams_ht, pollfd, &iter); - free(relay_connection); + relay_cleanup_poll_connection(&events, pollfd); + relay_del_connection(relay_connections_ht, + streams_ht, &iter, + relay_connection); } else if (revents & (LPOLLHUP | LPOLLRDHUP)) { DBG("Socket %d hung up", pollfd); - relay_cleanup_connection(relay_connections_ht, - &events, streams_ht, pollfd, &iter); - if (relay_connection->type == RELAY_CONTROL) { - relay_delete_session(relay_connection, streams_ht); - } - free(relay_connection); + relay_cleanup_poll_connection(&events, pollfd); + relay_del_connection(relay_connections_ht, + streams_ht, &iter, + relay_connection); } else if (revents & LPOLLIN) { /* control socket */ if (relay_connection->type == RELAY_CONTROL) { @@ -1376,14 +1731,14 @@ void *relay_thread_worker(void *data) sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL); /* connection closed */ if (ret <= 0) { - relay_cleanup_connection(relay_connections_ht, - &events, streams_ht, pollfd, &iter); - relay_delete_session(relay_connection, streams_ht); - free(relay_connection); + relay_cleanup_poll_connection(&events, pollfd); + relay_del_connection(relay_connections_ht, + streams_ht, &iter, + relay_connection); DBG("Control connection closed with %d", pollfd); } else { if (relay_connection->session) { - DBG2("Relay worker receiving data for session : %lu", + DBG2("Relay worker receiving data for session : %" PRIu64, relay_connection->session->id); } ret = relay_process_control(&recv_hdr, @@ -1394,9 +1749,10 @@ void *relay_thread_worker(void *data) * command: clear the session * */ if (ret < 0) { - relay_cleanup_connection(relay_connections_ht, - &events, streams_ht, pollfd, &iter); - free(relay_connection); + relay_cleanup_poll_connection(&events, pollfd); + relay_del_connection(relay_connections_ht, + streams_ht, &iter, + relay_connection); DBG("Connection closed with %d", pollfd); } } @@ -1405,36 +1761,48 @@ void *relay_thread_worker(void *data) ret = relay_process_data(relay_connection, streams_ht); /* connection closed */ if (ret < 0) { - relay_cleanup_connection(relay_connections_ht, - &events, streams_ht, pollfd, &iter); - relay_delete_session(relay_connection, streams_ht); + relay_cleanup_poll_connection(&events, pollfd); + relay_del_connection(relay_connections_ht, + streams_ht, &iter, + relay_connection); DBG("Data connection closed with %d", pollfd); } } } + rcu_read_unlock(); } } } +exit: error: lttng_poll_clean(&events); /* empty the hash table and free the memory */ + rcu_read_lock(); cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) { node = lttng_ht_iter_get_node_ulong(&iter); if (node) { relay_connection = caa_container_of(node, struct relay_command, sock_n); - free(relay_connection); + relay_del_connection(relay_connections_ht, + streams_ht, &iter, + relay_connection); } - ret = lttng_ht_del(relay_connections_ht, &iter); - assert(!ret); } + rcu_read_unlock(); error_poll_create: - free(data_buffer); + lttng_ht_destroy(streams_ht); +streams_ht_error: lttng_ht_destroy(relay_connections_ht); +relay_connections_ht_error: + if (err) { + DBG("Thread exited with error"); + } DBG("Worker thread cleanup complete"); + free(data_buffer); stop_threads(); + rcu_unregister_thread(); return NULL; } @@ -1444,23 +1812,10 @@ error_poll_create: */ static int create_relay_cmd_pipe(void) { - int ret, i; - - ret = pipe(relay_cmd_pipe); - if (ret < 0) { - PERROR("relay cmd pipe"); - goto error; - } + int ret; - for (i = 0; i < 2; i++) { - ret = fcntl(relay_cmd_pipe[i], F_SETFD, FD_CLOEXEC); - if (ret < 0) { - PERROR("fcntl relay_cmd_pipe"); - goto error; - } - } + ret = utils_create_pipe_cloexec(relay_cmd_pipe); -error: return ret; } @@ -1480,7 +1835,7 @@ int main(int argc, char **argv) /* Parse arguments */ progname = argv[0]; if ((ret = parse_args(argc, argv) < 0)) { - goto error; + goto exit; } if ((ret = set_signal_handler()) < 0) { @@ -1492,7 +1847,7 @@ int main(int argc, char **argv) ret = daemon(0, 0); if (ret < 0) { PERROR("daemon"); - goto error; + goto exit; } } @@ -1503,7 +1858,7 @@ int main(int argc, char **argv) if (control_uri->port < 1024 || data_uri->port < 1024) { ERR("Need to be root to use ports < 1024"); ret = -1; - goto error; + goto exit; } } @@ -1568,6 +1923,7 @@ exit: if (!ret) { exit(EXIT_SUCCESS); } + error: exit(EXIT_FAILURE); }