X-Git-Url: http://git.lttng.org./?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=1017fd47f7201924ca1f7abde2b15f728dbe39f7;hb=c3c841f480fd72c3e9ae7212fb9b48bb2e15e821;hp=f943488be5b94309efd332196bd52e876d2eea1a;hpb=c07d8a78cd9bfe1498ee0f6ed259a903a17d3325;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index f943488be..1017fd47f 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -132,6 +132,7 @@ void usage(void) fprintf(stderr, " -d, --daemonize Start as a daemon.\n"); fprintf(stderr, " -C, --control-port URL Control port listening.\n"); fprintf(stderr, " -D, --data-port URL Data port listening.\n"); + fprintf(stderr, " -L, --live-port URL Live view port listening.\n"); fprintf(stderr, " -o, --output PATH Output path for traces. Must use an absolute path.\n"); fprintf(stderr, " -v, --verbose Verbose mode. Activate DBG() macro.\n"); fprintf(stderr, " -g, --group NAME Specify the tracing group name. (default: tracing)\n"); @@ -157,7 +158,7 @@ int parse_args(int argc, char **argv) while (1) { int option_index = 0; - c = getopt_long(argc, argv, "dhv" "C:D:o:g:", + c = getopt_long(argc, argv, "dhv" "C:D:L:o:g:", long_options, &option_index); if (c == -1) { break; @@ -190,6 +191,16 @@ int parse_args(int argc, char **argv) data_uri->port = DEFAULT_NETWORK_DATA_PORT; } break; + case 'L': + ret = uri_parse(optarg, &live_uri); + if (ret < 0) { + ERR("Invalid live URI specified"); + goto exit; + } + if (live_uri->port == 0) { + live_uri->port = DEFAULT_NETWORK_VIEWER_PORT; + } + break; case 'd': opt_daemon = 1; break; @@ -285,6 +296,7 @@ void cleanup(void) uri_free(control_uri); uri_free(data_uri); + /* Live URI is freed in the live thread. */ } /* @@ -293,12 +305,10 @@ void cleanup(void) static int notify_thread_pipe(int wpipe) { - int ret; + ssize_t ret; - do { - ret = write(wpipe, "!", 1); - } while (ret < 0 && errno == EINTR); - if (ret < 0 || ret != 1) { + ret = lttng_write(wpipe, "!", 1); + if (ret < 1) { PERROR("write poll pipe"); } @@ -307,12 +317,10 @@ int notify_thread_pipe(int wpipe) static void notify_health_quit_pipe(int *pipe) { - int ret; + ssize_t ret; - do { - ret = write(pipe[1], "4", 1); - } while (ret < 0 && errno == EINTR); - if (ret < 0 || ret != 1) { + ret = lttng_write(pipe[1], "4", 1); + if (ret < 1) { PERROR("write relay health quit"); } } @@ -707,7 +715,8 @@ error_sock_control: static void *relay_thread_dispatcher(void *data) { - int ret, err = -1; + int err = -1; + ssize_t ret; struct cds_wfq_node *node; struct relay_command *relay_cmd = NULL; @@ -742,12 +751,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 :) */ - do { - ret = write(relay_cmd_pipe[1], relay_cmd, - sizeof(struct relay_command)); - } while (ret < 0 && errno == EINTR); + ret = lttng_write(relay_cmd_pipe[1], relay_cmd, + sizeof(struct relay_command)); free(relay_cmd); - if (ret < 0 || ret != sizeof(struct relay_command)) { + if (ret < sizeof(struct relay_command)) { PERROR("write cmd pipe"); goto error; } @@ -805,8 +812,6 @@ void deferred_free_stream(struct rcu_head *head) struct relay_stream *stream = caa_container_of(head, struct relay_stream, rcu_node); - ctf_trace_try_destroy(stream->ctf_trace); - free(stream->path_name); free(stream->channel_name); free(stream); @@ -854,7 +859,11 @@ static void destroy_stream(struct relay_stream *stream) * lookup failure on the live thread side of a stream indicates * that the viewer stream index received value should be used. */ + pthread_mutex_lock(&stream->viewer_stream_rotation_lock); vstream->total_index_received = stream->total_index_received; + vstream->tracefile_count_last = stream->tracefile_count_current; + vstream->close_write_flag = 1; + pthread_mutex_unlock(&stream->viewer_stream_rotation_lock); } /* Cleanup index of that stream. */ @@ -866,6 +875,11 @@ static void destroy_stream(struct relay_stream *stream) iter.iter.node = &stream->ctf_trace_node.node; delret = lttng_ht_del(stream->ctf_traces_ht, &iter); assert(!delret); + + if (stream->ctf_trace) { + ctf_trace_try_destroy(stream->ctf_trace); + } + call_rcu(&stream->rcu_node, deferred_free_stream); DBG("Closed tracefile %d from close stream", stream->fd); } @@ -898,6 +912,8 @@ void relay_delete_session(struct relay_command *cmd, stream = caa_container_of(node, struct relay_stream, stream_n); if (stream->session == cmd->session) { destroy_stream(stream); + cmd->session->stream_count--; + assert(cmd->session->stream_count >= 0); } } @@ -995,6 +1011,61 @@ error: return ret; } +/* + * When we have received all the streams and the metadata for a channel, + * we make them visible to the viewer threads. + */ +static +void set_viewer_ready_flag(struct relay_command *cmd) +{ + struct relay_stream_recv_handle *node, *tmp_node; + + cds_list_for_each_entry_safe(node, tmp_node, &cmd->recv_head, node) { + struct relay_stream *stream; + + rcu_read_lock(); + stream = relay_stream_find_by_id(node->id); + if (!stream) { + /* + * Stream is most probably being cleaned up by the data thread thus + * simply continue to the next one. + */ + rcu_read_unlock(); + continue; + } + + stream->viewer_ready = 1; + rcu_read_unlock(); + + /* Clean stream handle node. */ + cds_list_del(&node->node); + free(node); + } + + return; +} + +/* + * Add a recv handle node to the connection recv list with the given stream + * handle. A new node is allocated thus must be freed when the node is deleted + * from the list. + */ +static void queue_stream_handle(uint64_t handle, struct relay_command *cmd) +{ + struct relay_stream_recv_handle *node; + + assert(cmd); + + node = zmalloc(sizeof(*node)); + if (!node) { + PERROR("zmalloc queue stream handle"); + return; + } + + node->id = handle; + cds_list_add(&node->node, &cmd->recv_head); +} + /* * relay_add_stream: allocate a new stream for a session */ @@ -1087,6 +1158,17 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, ctf_trace_assign(cmd->ctf_traces_ht, stream); stream->ctf_traces_ht = cmd->ctf_traces_ht; + /* + * Add the stream handle in the recv list of the connection. Once the end + * stream message is received, this list is emptied and streams are set + * with the viewer ready flag. + */ + if (stream->metadata_flag) { + stream->viewer_ready = 1; + } else { + queue_stream_handle(stream->stream_handle, cmd); + } + lttng_ht_node_init_ulong(&stream->stream_n, (unsigned long) stream->stream_handle); lttng_ht_add_unique_ulong(relay_streams_ht, @@ -1094,6 +1176,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, lttng_ht_node_init_str(&stream->ctf_trace_node, stream->path_name); lttng_ht_add_str(cmd->ctf_traces_ht, &stream->ctf_trace_node); + session->stream_count++; DBG("Relay new stream added %s with ID %" PRIu64, stream->channel_name, stream->stream_handle); @@ -1170,6 +1253,8 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num); stream->close_flag = 1; + session->stream_count--; + assert(session->stream_count >= 0); if (close_stream_check(stream)) { destroy_stream(stream); @@ -1243,7 +1328,7 @@ int relay_start(struct lttcomm_relayd_hdr *recv_hdr, */ static int write_padding_to_file(int fd, uint32_t size) { - int ret = 0; + ssize_t ret = 0; char *zeros; if (size == 0) { @@ -1257,10 +1342,8 @@ static int write_padding_to_file(int fd, uint32_t size) goto end; } - do { - ret = write(fd, zeros, size); - } while (ret < 0 && errno == EINTR); - if (ret < 0 || ret != size) { + ret = lttng_write(fd, zeros, size); + if (ret < size) { PERROR("write padding to file"); } @@ -1278,6 +1361,7 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, struct relay_command *cmd) { int ret = htobe32(LTTNG_OK); + ssize_t size_ret; struct relay_session *session = cmd->session; struct lttcomm_relayd_metadata_payload *metadata_struct; struct relay_stream *metadata_stream; @@ -1334,11 +1418,9 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, goto end_unlock; } - do { - ret = write(metadata_stream->fd, metadata_struct->payload, - payload_size); - } while (ret < 0 && errno == EINTR); - if (ret < 0 || ret != payload_size) { + size_ret = lttng_write(metadata_stream->fd, metadata_struct->payload, + payload_size); + if (size_ret < payload_size) { ERR("Relay error writing metadata on file"); ret = -1; goto end_unlock; @@ -1835,6 +1917,48 @@ end_no_session: return ret; } +/* + * Receive the streams_sent message. + * + * Return 0 on success else a negative value. + */ +static +int relay_streams_sent(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd) +{ + int ret, send_ret; + struct lttcomm_relayd_generic_reply reply; + + assert(cmd); + + DBG("Relay receiving streams_sent"); + + if (!cmd->session || cmd->version_check_done == 0) { + ERR("Trying to close a stream before version check"); + ret = -1; + goto end_no_session; + } + + /* + * Flag every pending stream in the connection recv list that they are + * ready to be used by the viewer. + */ + set_viewer_ready_flag(cmd); + + reply.ret_code = htobe32(LTTNG_OK); + send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (send_ret < 0) { + ERR("Relay sending sent_stream reply"); + ret = send_ret; + } else { + /* Success. */ + ret = 0; + } + +end_no_session: + return ret; +} + /* * Process the commands received on the control socket */ @@ -1878,6 +2002,9 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, case RELAYD_SEND_INDEX: ret = relay_recv_index(recv_hdr, cmd); break; + case RELAYD_STREAMS_SENT: + ret = relay_streams_sent(recv_hdr, cmd); + break; case RELAYD_UPDATE_SYNC_INFO: default: ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd)); @@ -1979,6 +2106,7 @@ static int relay_process_data(struct relay_command *cmd) { int ret = 0, rotate_index = 0; + ssize_t size_ret; struct relay_stream *stream; struct lttcomm_relayd_data_hdr data_hdr; uint64_t stream_id; @@ -2041,10 +2169,54 @@ int relay_process_data(struct relay_command *cmd) if (stream->tracefile_size > 0 && (stream->tracefile_size_current + data_size) > stream->tracefile_size) { + struct relay_viewer_stream *vstream; + uint64_t new_id; + + new_id = (stream->tracefile_count_current + 1) % + stream->tracefile_count; + /* + * When we wrap-around back to 0, we start overwriting old + * trace data. + */ + if (!stream->tracefile_overwrite && new_id == 0) { + stream->tracefile_overwrite = 1; + } + pthread_mutex_lock(&stream->viewer_stream_rotation_lock); + if (stream->tracefile_overwrite) { + stream->oldest_tracefile_id = + (stream->oldest_tracefile_id + 1) % + stream->tracefile_count; + } + vstream = live_find_viewer_stream_by_id(stream->stream_handle); + if (vstream) { + /* + * The viewer is reading a file about to be + * overwritten. Close the FDs it is + * currently using and let it handle the fault. + */ + if (vstream->tracefile_count_current == new_id) { + pthread_mutex_lock(&vstream->overwrite_lock); + vstream->abort_flag = 1; + pthread_mutex_unlock(&vstream->overwrite_lock); + DBG("Streaming side setting abort_flag on stream %s_%lu\n", + stream->channel_name, new_id); + } else if (vstream->tracefile_count_current == + stream->tracefile_count_current) { + /* + * The reader and writer were in the + * same trace file, inform the viewer + * that no new index will ever be added + * to this file. + */ + vstream->close_write_flag = 1; + } + } ret = utils_rotate_stream_file(stream->path_name, stream->channel_name, stream->tracefile_size, stream->tracefile_count, relayd_uid, relayd_gid, stream->fd, &(stream->tracefile_count_current), &stream->fd); + stream->total_index_received = 0; + pthread_mutex_unlock(&stream->viewer_stream_rotation_lock); if (ret < 0) { ERR("Rotating stream output file"); goto end_rcu_unlock; @@ -2066,10 +2238,8 @@ int relay_process_data(struct relay_command *cmd) } /* Write data to stream output fd. */ - do { - ret = write(stream->fd, data_buffer, data_size); - } while (ret < 0 && errno == EINTR); - if (ret < 0 || ret != data_size) { + size_ret = lttng_write(stream->fd, data_buffer, data_size); + if (size_ret < data_size) { ERR("Relay error writing data to file"); ret = -1; goto end_rcu_unlock; @@ -2115,20 +2285,19 @@ int relay_add_connection(int fd, struct lttng_poll_event *events, struct lttng_ht *relay_connections_ht) { struct relay_command *relay_connection; - int ret; + ssize_t ret; relay_connection = zmalloc(sizeof(struct relay_command)); if (relay_connection == NULL) { PERROR("Relay command zmalloc"); goto error; } - do { - ret = read(fd, relay_connection, sizeof(struct relay_command)); - } while (ret < 0 && errno == EINTR); - if (ret < 0 || ret < sizeof(struct relay_command)) { + ret = lttng_read(fd, relay_connection, sizeof(struct relay_command)); + if (ret < sizeof(struct relay_command)) { PERROR("read relay cmd pipe"); goto error_read; } + CDS_INIT_LIST_HEAD(&relay_connection->recv_head); /* * Only used by the control side and the reference is copied inside each @@ -2180,8 +2349,17 @@ void relay_del_connection(struct lttng_ht *relay_connections_ht, assert(!ret); if (relay_connection->type == RELAY_CONTROL) { + struct relay_stream_recv_handle *node, *tmp_node; + relay_delete_session(relay_connection, sessions_ht); lttng_ht_destroy(relay_connection->ctf_traces_ht); + + /* Clean up recv list. */ + cds_list_for_each_entry_safe(node, tmp_node, + &relay_connection->recv_head, node) { + cds_list_del(&node->node); + free(node); + } } call_rcu(&relay_connection->rcu_node, deferred_free_connection); @@ -2547,7 +2725,8 @@ int main(int argc, char **argv) /* Check if daemon is UID = 0 */ if (relayd_uid == 0) { - if (control_uri->port < 1024 || data_uri->port < 1024) { + if (control_uri->port < 1024 || data_uri->port < 1024 || + live_uri->port < 1024) { ERR("Need to be root to use ports < 1024"); ret = -1; goto exit;