X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=0005714c55abf9bdd9a3347bcb96f99304b2ccca;hb=0bb7724acebb12e986b55e04f13233710d8c71fa;hp=33aad88ee3e5d0d274790dd4edef675ffeeb06db;hpb=bd11e201a9d7e2f2f71a841a6617aa53074b030a;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 33aad88ee..0005714c5 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -293,12 +293,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 +305,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 +703,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 +739,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; } @@ -826,8 +821,7 @@ void deferred_free_session(struct rcu_head *head) * RCU read side lock MUST be acquired. If NO close_stream_check() was called * BEFORE the stream lock MUST be acquired. */ -static void destroy_stream(struct relay_stream *stream, - struct lttng_ht *ctf_traces_ht) +static void destroy_stream(struct relay_stream *stream) { int delret; struct relay_viewer_stream *vstream; @@ -865,7 +859,7 @@ static void destroy_stream(struct relay_stream *stream, delret = lttng_ht_del(relay_streams_ht, &iter); assert(!delret); iter.iter.node = &stream->ctf_trace_node.node; - delret = lttng_ht_del(ctf_traces_ht, &iter); + delret = lttng_ht_del(stream->ctf_traces_ht, &iter); assert(!delret); call_rcu(&stream->rcu_node, deferred_free_stream); DBG("Closed tracefile %d from close stream", stream->fd); @@ -898,7 +892,9 @@ 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->ctf_traces_ht); + destroy_stream(stream); + cmd->session->stream_count--; + assert(cmd->session->stream_count >= 0); } } @@ -1086,6 +1082,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, stream->ctf_trace->metadata_stream = stream; } ctf_trace_assign(cmd->ctf_traces_ht, stream); + stream->ctf_traces_ht = cmd->ctf_traces_ht; lttng_ht_node_init_ulong(&stream->stream_n, (unsigned long) stream->stream_handle); @@ -1094,6 +1091,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,9 +1168,11 @@ 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, cmd->ctf_traces_ht); + destroy_stream(stream); } end_unlock: @@ -1243,7 +1243,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 +1257,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 +1276,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 +1333,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; @@ -1979,6 +1976,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; @@ -2066,10 +2064,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; @@ -2088,7 +2084,7 @@ int relay_process_data(struct relay_command *cmd) /* Check if we need to close the FD */ if (close_stream_check(stream)) { - destroy_stream(stream, cmd->ctf_traces_ht); + destroy_stream(stream); } end_rcu_unlock: @@ -2115,24 +2111,30 @@ 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; } - relay_connection->ctf_traces_ht = lttng_ht_new(0, LTTNG_HT_TYPE_STRING); - if (!relay_connection->ctf_traces_ht) { - goto error_read; + /* + * Only used by the control side and the reference is copied inside each + * stream from that connection. Thus a destroy HT must be done after every + * stream has been destroyed. + */ + if (relay_connection->type == RELAY_CONTROL) { + relay_connection->ctf_traces_ht = lttng_ht_new(0, + LTTNG_HT_TYPE_STRING); + if (!relay_connection->ctf_traces_ht) { + goto error_read; + } } lttng_ht_node_init_ulong(&relay_connection->sock_n, @@ -2157,7 +2159,6 @@ void deferred_free_connection(struct rcu_head *head) struct relay_command *relay_connection = caa_container_of(head, struct relay_command, rcu_node); - lttng_ht_destroy(relay_connection->ctf_traces_ht); lttcomm_destroy_sock(relay_connection->sock); free(relay_connection); } @@ -2171,12 +2172,13 @@ void relay_del_connection(struct lttng_ht *relay_connections_ht, ret = lttng_ht_del(relay_connections_ht, iter); assert(!ret); + if (relay_connection->type == RELAY_CONTROL) { relay_delete_session(relay_connection, sessions_ht); + lttng_ht_destroy(relay_connection->ctf_traces_ht); } - call_rcu(&relay_connection->rcu_node, - deferred_free_connection); + call_rcu(&relay_connection->rcu_node, deferred_free_connection); } /*