X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=41afccabdfa33b7e045b38a57a825d4d6246ab4b;hp=bd19a323715309697ba3b3ff61c6d29a615ad379;hb=0d9c5d7750202aa5f46cd7c6f642fb31fe1556c4;hpb=092b625914723e40ccb3345826a2280cbd01f8b8 diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index bd19a3237..41afccabd 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -455,6 +455,13 @@ int close_stream_check(struct relay_stream *stream) { if (stream->close_flag && stream->prev_seq == stream->last_net_seq_num) { + /* + * We are about to close the stream so set the data pending flag to 1 + * which will make the end data pending command skip the stream which + * is now closed and ready. Note that after proceeding to a file close, + * the written file is ready for reading. + */ + stream->data_pending_check_done = 1; return 1; } return 0; @@ -507,8 +514,6 @@ void *relay_thread_listener(void *data) while (1) { DBG("Listener accepting connections"); - nb_fd = LTTNG_POLL_GETNB(&events); - restart: ret = lttng_poll_wait(&events, -1); if (ret < 0) { @@ -521,6 +526,8 @@ restart: goto error; } + nb_fd = ret; + DBG("Relay new connection received"); for (i = 0; i < nb_fd; i++) { /* Fetch once the poll data */ @@ -885,8 +892,6 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht DBG("Relay deleting session %" PRIu64, cmd->session->id); - lttcomm_destroy_sock(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); @@ -910,6 +915,54 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht free(cmd->session); } +/* + * Handle the RELAYD_CREATE_SESSION command. + * + * On success, send back the session id or else return a negative value. + */ +static +int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd) +{ + int ret = 0, send_ret; + struct relay_session *session; + struct lttcomm_relayd_status_session reply; + + assert(recv_hdr); + assert(cmd); + + memset(&reply, 0, sizeof(reply)); + + session = zmalloc(sizeof(struct relay_session)); + if (session == NULL) { + PERROR("relay session zmalloc"); + ret = -1; + goto error; + } + + session->id = ++last_relay_session_id; + session->sock = cmd->sock; + cmd->session = session; + + reply.session_id = htobe64(session->id); + + DBG("Created session %" PRIu64, session->id); + +error: + if (ret < 0) { + reply.ret_code = htobe32(LTTNG_ERR_FATAL); + } else { + reply.ret_code = htobe32(LTTNG_OK); + } + + send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (send_ret < 0) { + ERR("Relayd sending session id"); + } + + return ret; +} + /* * relay_add_stream: allocate a new stream for a session */ @@ -924,7 +977,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, char *path = NULL, *root_path = NULL; int ret, send_ret; - if (!session || session->version_check_done == 0) { + if (!session || cmd->version_check_done == 0) { ERR("Trying to add a stream before version check"); ret = -1; goto end_no_session; @@ -932,7 +985,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr, /* FIXME : use data_size for something ? */ ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info, - sizeof(struct lttcomm_relayd_add_stream), MSG_WAITALL); + sizeof(struct lttcomm_relayd_add_stream), 0); if (ret < sizeof(struct lttcomm_relayd_add_stream)) { ERR("Relay didn't receive valid add_stream struct size : %d", ret); ret = -1; @@ -1021,14 +1074,14 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, DBG("Close stream received"); - if (!session || session->version_check_done == 0) { + if (!session || cmd->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); + sizeof(struct lttcomm_relayd_close_stream), 0); if (ret < sizeof(struct lttcomm_relayd_close_stream)) { ERR("Relay didn't receive valid add_stream struct size : %d", ret); ret = -1; @@ -1232,8 +1285,7 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr, } memset(data_buffer, 0, data_size); DBG2("Relay receiving metadata, waiting for %" PRIu64 " 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, 0); if (ret < 0 || ret != data_size) { ret = -1; ERR("Relay didn't receive the whole metadata"); @@ -1282,25 +1334,13 @@ int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr, { int ret; struct lttcomm_relayd_version reply, msg; - struct relay_session *session; - if (cmd->session == NULL) { - session = zmalloc(sizeof(struct relay_session)); - if (session == NULL) { - PERROR("relay session zmalloc"); - ret = -1; - goto end; - } - session->id = ++last_relay_session_id; - DBG("Created session %" PRIu64, session->id); - cmd->session = session; - } else { - session = cmd->session; - } - session->version_check_done = 1; + assert(cmd); + + cmd->version_check_done = 1; /* Get version from the other side. */ - ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), MSG_WAITALL); + ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0); if (ret < 0 || ret != sizeof(msg)) { ret = -1; ERR("Relay failed to receive the version values."); @@ -1352,13 +1392,13 @@ int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, DBG("Data pending command received"); - if (!session || session->version_check_done == 0) { + if (!session || cmd->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); + ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0); if (ret < sizeof(msg)) { ERR("Relay didn't receive valid data_pending struct size : %d", ret); ret = -1; @@ -1393,6 +1433,9 @@ int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, ret = 1; } + /* Pending check is now done. */ + stream->data_pending_check_done = 1; + end_unlock: rcu_read_unlock(); @@ -1431,6 +1474,141 @@ int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr, return ret; } +/* + * Initialize a data pending command. This means that a client is about to ask + * for data pending for each stream he/she holds. Simply iterate over all + * streams of a session and set the data_pending_check_done flag. + * + * This command returns to the client a LTTNG_OK code. + */ +static +int relay_begin_data_pending(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd, struct lttng_ht *streams_ht) +{ + int ret; + struct lttng_ht_iter iter; + struct lttcomm_relayd_begin_data_pending msg; + struct lttcomm_relayd_generic_reply reply; + struct relay_stream *stream; + uint64_t session_id; + + assert(recv_hdr); + assert(cmd); + assert(streams_ht); + + DBG("Init streams for data pending"); + + if (!cmd->session || cmd->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), 0); + if (ret < sizeof(msg)) { + ERR("Relay didn't receive valid begin data_pending struct size: %d", + ret); + ret = -1; + goto end_no_session; + } + + session_id = be64toh(msg.session_id); + + /* + * Iterate over all streams to set the begin data pending flag. For now, the + * streams are indexed by stream handle so we have to iterate over all + * streams to find the one associated with the right session_id. + */ + rcu_read_lock(); + cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, stream, stream_n.node) { + if (stream->session->id == session_id) { + stream->data_pending_check_done = 0; + DBG("Set begin data pending flag to stream %" PRIu64, + stream->stream_handle); + } + } + rcu_read_unlock(); + + /* All good, send back reply. */ + reply.ret_code = htobe32(LTTNG_OK); + + ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (ret < 0) { + ERR("Relay begin data pending send reply failed"); + } + +end_no_session: + return ret; +} + +/* + * End data pending command. This will check, for a given session id, if each + * stream associated with it has its data_pending_check_done flag set. If not, + * this means that the client lost track of the stream but the data is still + * being streamed on our side. In this case, we inform the client that data is + * inflight. + * + * Return to the client if there is data in flight or not with a ret_code. + */ +static +int relay_end_data_pending(struct lttcomm_relayd_hdr *recv_hdr, + struct relay_command *cmd, struct lttng_ht *streams_ht) +{ + int ret; + struct lttng_ht_iter iter; + struct lttcomm_relayd_end_data_pending msg; + struct lttcomm_relayd_generic_reply reply; + struct relay_stream *stream; + uint64_t session_id; + uint32_t is_data_inflight = 0; + + assert(recv_hdr); + assert(cmd); + assert(streams_ht); + + DBG("End data pending command"); + + if (!cmd->session || cmd->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), 0); + if (ret < sizeof(msg)) { + ERR("Relay didn't receive valid end data_pending struct size: %d", + ret); + ret = -1; + goto end_no_session; + } + + session_id = be64toh(msg.session_id); + + /* Iterate over all streams to see if the begin data pending flag is set. */ + rcu_read_lock(); + cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, stream, stream_n.node) { + if (stream->session->id == session_id && + !stream->data_pending_check_done) { + is_data_inflight = 1; + DBG("Data is still in flight for stream %" PRIu64, + stream->stream_handle); + break; + } + } + rcu_read_unlock(); + + /* All good, send back reply. */ + reply.ret_code = htobe32(is_data_inflight); + + ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); + if (ret < 0) { + ERR("Relay end data pending send reply failed"); + } + +end_no_session: + return ret; +} + /* * relay_process_control: Process the commands received on the control socket */ @@ -1441,11 +1619,9 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, int ret = 0; switch (be32toh(recv_hdr->cmd)) { - /* case RELAYD_CREATE_SESSION: ret = relay_create_session(recv_hdr, cmd); break; - */ case RELAYD_ADD_STREAM: ret = relay_add_stream(recv_hdr, cmd, streams_ht); break; @@ -1467,6 +1643,12 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, case RELAYD_QUIESCENT_CONTROL: ret = relay_quiescent_control(recv_hdr, cmd); break; + case RELAYD_BEGIN_DATA_PENDING: + ret = relay_begin_data_pending(recv_hdr, cmd, streams_ht); + break; + case RELAYD_END_DATA_PENDING: + ret = relay_end_data_pending(recv_hdr, cmd, streams_ht); + break; case RELAYD_UPDATE_SYNC_INFO: default: ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd)); @@ -1493,7 +1675,7 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) uint32_t data_size; ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr, - sizeof(struct lttcomm_relayd_data_hdr), MSG_WAITALL); + sizeof(struct lttcomm_relayd_data_hdr), 0); if (ret <= 0) { ERR("Connections seems to be closed"); ret = -1; @@ -1527,7 +1709,7 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) 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); + ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0); if (ret <= 0) { ret = -1; goto end_unlock; @@ -1692,11 +1874,6 @@ void *relay_thread_worker(void *data) } while (1) { - /* Zeroed the events structure */ - lttng_poll_reset(&events); - - nb_fd = LTTNG_POLL_GETNB(&events); - /* Infinite blocking call, waiting for transmission */ restart: DBG3("Relayd worker thread polling..."); @@ -1711,6 +1888,8 @@ void *relay_thread_worker(void *data) goto error; } + nb_fd = ret; + for (i = 0; i < nb_fd; i++) { /* Fetch once the poll data */ revents = LTTNG_POLL_GETEV(&events, i); @@ -1767,7 +1946,7 @@ void *relay_thread_worker(void *data) if (relay_connection->type == RELAY_CONTROL) { ret = relay_connection->sock->ops->recvmsg( relay_connection->sock, &recv_hdr, - sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL); + sizeof(struct lttcomm_relayd_hdr), 0); /* connection closed */ if (ret <= 0) { relay_cleanup_poll_connection(&events, pollfd);