X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=e426469a30ebe4125b70a0bec537af3d1c8a553c;hb=0a6518b0685cb9e07cf156a4c882dc6ec40db35a;hp=2208144dc4a3fd53858164c8e8b39211673030df;hpb=92c6ca54f70068b7cd440d0ccc58610fe9b9d010;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 2208144dc..e426469a3 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -101,9 +101,6 @@ static struct relay_cmd_queue relay_cmd_queue; static char *data_buffer; static unsigned int data_buffer_size; -/* Global hash table that stores relay index object. */ -static struct lttng_ht *indexes_ht; - /* We need those values for the file/dir creation. */ static uid_t relayd_uid; static gid_t relayd_gid; @@ -114,6 +111,9 @@ struct lttng_ht *relay_streams_ht; /* Global relay viewer stream hash table. */ struct lttng_ht *viewer_streams_ht; +/* Global hash table that stores relay index object. */ +struct lttng_ht *indexes_ht; + /* * usage function on stderr */ @@ -764,7 +764,13 @@ void deferred_free_session(struct rcu_head *head) free(session); } -static void close_stream(struct relay_stream *stream, +/* + * Close a given stream. The stream is freed using a call RCU. + * + * 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) { int delret; @@ -828,29 +834,22 @@ void relay_delete_session(struct relay_command *cmd, rcu_read_lock(); cds_lfht_for_each_entry(relay_streams_ht->ht, &iter.iter, node, node) { node = lttng_ht_iter_get_node_ulong(&iter); - if (node) { - stream = caa_container_of(node, - struct relay_stream, stream_n); - if (stream->session == cmd->session) { - ret = close(stream->fd); - if (ret < 0) { - PERROR("close stream fd on delete session"); - } - ret = lttng_ht_del(relay_streams_ht, &iter); - assert(!ret); - call_rcu(&stream->rcu_node, - deferred_free_stream); - } - /* Cleanup index of that stream. */ - relay_index_destroy_by_stream_id(stream->stream_handle, - indexes_ht); + if (!node) { + continue; } + stream = caa_container_of(node, struct relay_stream, stream_n); + if (stream->session == cmd->session) { + destroy_stream(stream, cmd->ctf_traces_ht); + } + /* Cleanup index of that stream. */ + relay_index_destroy_by_stream_id(stream->stream_handle); } + + /* Make this session not visible anymore. */ iter.iter.node = &cmd->session->session_n.node; ret = lttng_ht_del(sessions_ht, &iter); assert(!ret); - call_rcu(&cmd->session->rcu_node, - deferred_free_session); + call_rcu(&cmd->session->rcu_node, deferred_free_session); rcu_read_unlock(); } @@ -1114,7 +1113,7 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, stream->close_flag = 1; if (close_stream_check(stream)) { - close_stream(stream, cmd->ctf_traces_ht); + destroy_stream(stream, cmd->ctf_traces_ht); } end_unlock: @@ -1653,7 +1652,7 @@ end_no_session: */ static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, - struct relay_command *cmd, struct lttng_ht *indexes_ht) + struct relay_command *cmd) { int ret, send_ret, index_created = 0; struct relay_session *session = cmd->session; @@ -1664,7 +1663,6 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, uint64_t net_seq_num; assert(cmd); - assert(indexes_ht); DBG("Relay receiving index"); @@ -1712,7 +1710,7 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, stream->beacon_ts_end = -1ULL; } - index = relay_index_find(stream->stream_handle, net_seq_num, indexes_ht); + index = relay_index_find(stream->stream_handle, net_seq_num); if (!index) { /* A successful creation will add the object to the HT. */ index = relay_index_create(stream->stream_handle, net_seq_num); @@ -1730,7 +1728,7 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, * already exist, destroy back the index created, set the data in this * object and write it on disk. */ - relay_index_add(index, indexes_ht, &wr_index); + relay_index_add(index, &wr_index); if (wr_index) { copy_index_control_data(wr_index, &index_info); free(index); @@ -1753,7 +1751,7 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr, stream->index_fd = ret; } - ret = relay_index_write(wr_index->fd, wr_index, indexes_ht); + ret = relay_index_write(wr_index->fd, wr_index); if (ret < 0) { goto end_rcu_unlock; } @@ -1819,7 +1817,7 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, ret = relay_end_data_pending(recv_hdr, cmd); break; case RELAYD_SEND_INDEX: - ret = relay_recv_index(recv_hdr, cmd, indexes_ht); + ret = relay_recv_index(recv_hdr, cmd); break; case RELAYD_UPDATE_SYNC_INFO: default: @@ -1837,8 +1835,7 @@ end: * relay_process_data: Process the data received on the data socket */ static -int relay_process_data(struct relay_command *cmd, - struct lttng_ht *indexes_ht) +int relay_process_data(struct relay_command *cmd) { int ret = 0, rotate_index = 0, index_created = 0; struct relay_stream *stream; @@ -1925,7 +1922,7 @@ int relay_process_data(struct relay_command *cmd, * exists, the control thread already received the data for it thus we need * to write it on disk. */ - index = relay_index_find(stream_id, net_seq_num, indexes_ht); + index = relay_index_find(stream_id, net_seq_num); if (!index) { /* A successful creation will add the object to the HT. */ index = relay_index_create(stream->stream_handle, net_seq_num); @@ -1955,7 +1952,7 @@ int relay_process_data(struct relay_command *cmd, * Try to add the relay index object to the hash table. If an object * already exist, destroy back the index created and set the data. */ - relay_index_add(index, indexes_ht, &wr_index); + relay_index_add(index, &wr_index); if (wr_index) { /* Copy back data from the created index. */ wr_index->fd = index->fd; @@ -1981,7 +1978,7 @@ int relay_process_data(struct relay_command *cmd, stream->index_fd = ret; } - ret = relay_index_write(wr_index->fd, wr_index, indexes_ht); + ret = relay_index_write(wr_index->fd, wr_index); if (ret < 0) { goto end_rcu_unlock; } @@ -2010,7 +2007,7 @@ int relay_process_data(struct relay_command *cmd, /* Check if we need to close the FD */ if (close_stream_check(stream)) { - close_stream(stream, cmd->ctf_traces_ht); + destroy_stream(stream, cmd->ctf_traces_ht); } end_rcu_unlock: @@ -2312,7 +2309,7 @@ restart: continue; } - ret = relay_process_data(relay_connection, indexes_ht); + ret = relay_process_data(relay_connection); /* connection closed */ if (ret < 0) { relay_cleanup_poll_connection(&events, pollfd); @@ -2356,7 +2353,7 @@ error_poll_create: { struct relay_index *index; cds_lfht_for_each_entry(indexes_ht->ht, &iter.iter, index, index_n.node) { - relay_index_delete(index, indexes_ht); + relay_index_delete(index); } lttng_ht_destroy(indexes_ht); }