X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=74e785e2f52d62e0afd16fb46db0fbe37e9c7151;hp=152e4bbc86911a00a53f8598e472e7b1545736eb;hb=c5b6f4f08fe8d1abff74c7f6ad3630b7dcf0669d;hpb=5b6d8097dcb979a1e10b5fea7b8dd7a2f1498fe0 diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c index 152e4bbc8..74e785e2f 100644 --- a/src/bin/lttng-relayd/main.c +++ b/src/bin/lttng-relayd/main.c @@ -239,8 +239,8 @@ void cleanup(void) /* Close thread quit pipes */ utils_close_pipe(thread_quit_pipe); - /* Close relay cmd pipes */ - utils_close_pipe(relay_cmd_pipe); + uri_free(control_uri); + uri_free(data_uri); } /* @@ -885,8 +885,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); @@ -894,7 +892,10 @@ void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht stream = caa_container_of(node, struct relay_stream, stream_n); if (stream->session == cmd->session) { - close(stream->fd); + ret = close(stream->fd); + if (ret < 0) { + PERROR("close stream fd on delete session"); + } ret = lttng_ht_del(streams_ht, &iter); assert(!ret); call_rcu(&stream->rcu_node, @@ -907,6 +908,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 */ @@ -921,7 +970,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; @@ -929,7 +978,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; @@ -1018,14 +1067,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; @@ -1055,7 +1104,10 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr, if (close_stream_check(stream)) { int delret; - close(stream->fd); + delret = close(stream->fd); + if (delret < 0) { + PERROR("close stream"); + } delret = lttng_ht_del(streams_ht, &iter); assert(!delret); call_rcu(&stream->rcu_node, @@ -1226,8 +1278,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"); @@ -1275,23 +1326,26 @@ int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr, struct relay_command *cmd) { int ret; - struct lttcomm_relayd_version reply; - struct relay_session *session; + struct lttcomm_relayd_version reply, msg; - 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; + assert(cmd); + + cmd->version_check_done = 1; + + /* Get version from the other side. */ + 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."); + goto end; } - session->version_check_done = 1; + + /* + * For now, we just ignore the received version but after 2.1 stable + * release, a check must be done to see if we either adapt to the other + * side version (which MUST be lower than us) or keep the latest data + * structure considering that the other side will adapt. + */ ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor); if (ret < 2) { @@ -1314,14 +1368,14 @@ end: } /* - * Check for data availability for a given stream id from the session daemon. + * Check for data pending for a given stream id from the session daemon. */ static -int relay_data_available(struct lttcomm_relayd_hdr *recv_hdr, +int relay_data_pending(struct lttcomm_relayd_hdr *recv_hdr, struct relay_command *cmd, struct lttng_ht *streams_ht) { struct relay_session *session = cmd->session; - struct lttcomm_relayd_data_available msg; + struct lttcomm_relayd_data_pending msg; struct lttcomm_relayd_generic_reply reply; struct relay_stream *stream; int ret; @@ -1329,17 +1383,17 @@ int relay_data_available(struct lttcomm_relayd_hdr *recv_hdr, struct lttng_ht_iter iter; uint64_t last_net_seq_num, stream_id; - DBG("Data available command received"); + 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_available struct size : %d", ret); + ERR("Relay didn't receive valid data_pending struct size : %d", ret); ret = -1; goto end_no_session; } @@ -1359,17 +1413,17 @@ int relay_data_available(struct lttcomm_relayd_hdr *recv_hdr, stream = caa_container_of(node, struct relay_stream, stream_n); assert(stream); - DBG("Data available for stream id %" PRIu64 " prev_seq %" PRIu64 + DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64 " and last_seq %" PRIu64, stream_id, stream->prev_seq, last_net_seq_num); /* Avoid wrapping issue */ if (((int64_t) (stream->prev_seq - last_net_seq_num)) <= 0) { - /* Data has in fact been written and is available */ - ret = 1; - } else { - /* Data still being streamed. */ + /* Data has in fact been written and is NOT pending */ ret = 0; + } else { + /* Data still being streamed thus pending */ + ret = 1; } end_unlock: @@ -1378,7 +1432,7 @@ end_unlock: reply.ret_code = htobe32(ret); ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); if (ret < 0) { - ERR("Relay data available ret code failed"); + ERR("Relay data pending ret code failed"); } end_no_session: @@ -1404,7 +1458,7 @@ int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr, reply.ret_code = htobe32(LTTNG_OK); ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0); if (ret < 0) { - ERR("Relay data available ret code failed"); + ERR("Relay data quiescent control ret code failed"); } return ret; @@ -1420,11 +1474,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; @@ -1440,8 +1492,8 @@ int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr, case RELAYD_CLOSE_STREAM: ret = relay_close_stream(recv_hdr, cmd, streams_ht); break; - case RELAYD_DATA_AVAILABLE: - ret = relay_data_available(recv_hdr, cmd, streams_ht); + case RELAYD_DATA_PENDING: + ret = relay_data_pending(recv_hdr, cmd, streams_ht); break; case RELAYD_QUIESCENT_CONTROL: ret = relay_quiescent_control(recv_hdr, cmd); @@ -1472,7 +1524,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; @@ -1506,7 +1558,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; @@ -1533,9 +1585,13 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht) /* Check if we need to close the FD */ if (close_stream_check(stream)) { + int cret; struct lttng_ht_iter iter; - close(stream->fd); + cret = close(stream->fd); + if (cret < 0) { + PERROR("close stream process data"); + } iter.iter.node = &stream->stream_n.node; ret = lttng_ht_del(streams_ht, &iter); assert(!ret); @@ -1742,7 +1798,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); @@ -1810,6 +1866,8 @@ error_poll_create: streams_ht_error: lttng_ht_destroy(relay_connections_ht); relay_connections_ht_error: + /* Close relay cmd pipes */ + utils_close_pipe(relay_cmd_pipe); if (err) { DBG("Thread exited with error"); }