X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=ae631c70a25afc3407736ea8478f782b2c8793c7;hb=173af62f4804133d4a7f45e34b6f72126f3eca5f;hp=3613281dd03e971a45a1187495e40b1bc0f95f74;hpb=9d1bbf2176ae0e362428b920e34b05158027da26;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 3613281dd..ae631c70a 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 @@ -446,6 +447,19 @@ 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 */ @@ -922,7 +936,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, 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); @@ -979,6 +993,85 @@ 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 %lu 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->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(LTTCOMM_ERR); + } else { + reply.ret_code = htobe32(LTTCOMM_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; +} + /* * relay_unknown_command: send -1 if received unknown command */ @@ -1070,8 +1163,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) { @@ -1083,7 +1182,8 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, } 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); + 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"); @@ -1103,7 +1203,7 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, ret = write(metadata_stream->fd, metadata_struct->payload, payload_size); } while (ret < 0 && errno == EINTR); - if (ret < (payload_size)) { + if (ret < payload_size) { ERR("Relay error writing metadata on file"); ret = -1; goto end_unlock; @@ -1111,7 +1211,7 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, DBG2("Relay metadata written"); end_unlock: - rcu_read_lock(); + rcu_read_unlock(); end: return ret; } @@ -1181,6 +1281,9 @@ 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_UPDATE_SYNC_INFO: default: ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd)); @@ -1203,6 +1306,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, @@ -1234,6 +1338,10 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) } memset(data_buffer, 0, data_size); + net_seq_num = be64toh(data_hdr.net_seq_num); + + DBG3("Receiving data of size %u for stream id %zu 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; @@ -1250,6 +1358,21 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) } DBG2("Relay wrote %d bytes to tracefile for stream id %lu", ret, stream->stream_handle); + 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: @@ -1377,6 +1500,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) { /*