X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Frelayd%2Frelayd.c;h=785d3dc584bd66d2d825a0df09f9e63d0e49cc02;hp=6f877c96813d8a78b6d5f2f82235b43058d8ed43;hb=c8f59ee5fc11492ef472dc5cfd2fd2c4926b1787;hpb=173af62f4804133d4a7f45e34b6f72126f3eca5f diff --git a/src/common/relayd/relayd.c b/src/common/relayd/relayd.c index 6f877c968..785d3dc58 100644 --- a/src/common/relayd/relayd.c +++ b/src/common/relayd/relayd.c @@ -21,6 +21,7 @@ #include #include #include +#include #include #include @@ -69,7 +70,7 @@ static int send_command(struct lttcomm_sock *sock, goto error; } - DBG3("Relayd sending command %d", cmd); + DBG3("Relayd sending command %d of size %" PRIu64, cmd, buf_size); error: free(buf); @@ -85,7 +86,7 @@ static int recv_reply(struct lttcomm_sock *sock, void *data, size_t size) { int ret; - DBG3("Relayd waiting for reply..."); + DBG3("Relayd waiting for reply of size %ld", size); ret = sock->ops->recvmsg(sock, data, size, 0); if (ret < 0) { @@ -96,58 +97,6 @@ error: return ret; } -#if 0 -/* - * Create session on the relayd. - * - * On error, return ret_code negative value else return 0. - */ -int relayd_create_session(struct lttcomm_sock *sock, const char *hostname, - const char *session_name) -{ - int ret; - struct lttcomm_relayd_create_session msg; - struct lttcomm_relayd_generic_reply reply; - - /* Code flow error. Safety net. */ - assert(sock); - assert(hostname); - assert(session_name); - - DBG("Relayd creating session for hostname %s and session name %s", - hostname, session_name); - - strncpy(msg.hostname, hostname, sizeof(msg.hostname)); - strncpy(msg.session_name, session_name, sizeof(msg.session_name)); - - /* Send command */ - ret = send_command(sock, RELAYD_CREATE_SESSION, (void *) &msg, - sizeof(msg), 0); - if (ret < 0) { - goto error; - } - - /* Recevie response */ - ret = recv_reply(sock, (void *) &reply, sizeof(reply)); - if (ret < 0) { - goto error; - } - - /* Return session id or negative ret code. */ - if (reply.ret_code != LTTCOMM_OK) { - ret = -reply.ret_code; - } else { - /* Success */ - ret = 0; - } - - DBG2("Relayd created session for %s", session_name); - -error: - return ret; -} -#endif - /* * Add stream on the relayd and assign stream handle to the stream_id argument. * @@ -176,7 +125,7 @@ int relayd_add_stream(struct lttcomm_sock *sock, const char *channel_name, goto error; } - /* Recevie response */ + /* Waiting for reply */ ret = recv_reply(sock, (void *) &reply, sizeof(reply)); if (ret < 0) { goto error; @@ -187,7 +136,7 @@ int relayd_add_stream(struct lttcomm_sock *sock, const char *channel_name, reply.ret_code = be32toh(reply.ret_code); /* Return session id or negative ret code. */ - if (reply.ret_code != LTTCOMM_OK) { + if (reply.ret_code != LTTNG_OK) { ret = -reply.ret_code; ERR("Relayd add stream replied error %d", ret); } else { @@ -196,7 +145,8 @@ int relayd_add_stream(struct lttcomm_sock *sock, const char *channel_name, *stream_id = reply.handle; } - DBG("Relayd stream added successfully with handle %zu", reply.handle); + DBG("Relayd stream added successfully with handle %" PRIu64, + reply.handle); error: return ret; @@ -253,47 +203,6 @@ error: return ret; } -#if 0 -/* - * Start data command on the relayd. - * - * On success return 0 else return ret_code negative value. - */ -int relayd_start_data(struct lttcomm_sock *sock) -{ - int ret; - struct lttcomm_relayd_generic_reply reply; - - /* Code flow error. Safety net. */ - assert(sock); - - DBG("Relayd start data command"); - - /* Send command */ - ret = send_command(sock, RELAYD_START_DATA, NULL, 0, 0); - if (ret < 0) { - goto error; - } - - /* Recevie response */ - ret = recv_reply(sock, (void *) &reply, sizeof(reply)); - if (ret < 0) { - goto error; - } - - /* Return session id or negative ret code. */ - if (reply.ret_code != LTTCOMM_OK) { - ret = -reply.ret_code; - } else { - /* Success */ - ret = 0; - } - -error: - return ret; -} -#endif - /* * Add stream on the relayd and assign stream handle to the stream_id argument. * @@ -306,7 +215,7 @@ int relayd_send_metadata(struct lttcomm_sock *sock, size_t len) /* Code flow error. Safety net. */ assert(sock); - DBG("Relayd sending metadata of size %lu", len); + DBG("Relayd sending metadata of size %zu", len); /* Send command */ ret = send_command(sock, RELAYD_SEND_METADATA, NULL, len, 0); @@ -319,7 +228,7 @@ int relayd_send_metadata(struct lttcomm_sock *sock, size_t len) /* * After that call, the metadata data MUST be sent to the relayd so the * receive size on the other end matches the len of the metadata packet - * header. + * header. This is why we don't wait for a reply here. */ error: @@ -364,7 +273,7 @@ int relayd_send_data_hdr(struct lttcomm_sock *sock, assert(sock); assert(hdr); - DBG3("Relayd sending data header..."); + DBG3("Relayd sending data header of size %ld", size); /* Again, safety net */ if (size == 0) { @@ -399,7 +308,7 @@ int relayd_send_close_stream(struct lttcomm_sock *sock, uint64_t stream_id, /* Code flow error. Safety net. */ assert(sock); - DBG("Relayd closing stream id %zu", stream_id); + DBG("Relayd closing stream id %" PRIu64, stream_id); msg.stream_id = htobe64(stream_id); msg.last_net_seq_num = htobe64(last_net_seq_num); @@ -419,7 +328,7 @@ int relayd_send_close_stream(struct lttcomm_sock *sock, uint64_t stream_id, reply.ret_code = be32toh(reply.ret_code); /* Return session id or negative ret code. */ - if (reply.ret_code != LTTCOMM_OK) { + if (reply.ret_code != LTTNG_OK) { ret = -reply.ret_code; ERR("Relayd close stream replied error %d", ret); } else { @@ -427,7 +336,99 @@ int relayd_send_close_stream(struct lttcomm_sock *sock, uint64_t stream_id, ret = 0; } - DBG("Relayd close stream id %zu successfully", stream_id); + DBG("Relayd close stream id %" PRIu64 " successfully", stream_id); + +error: + return ret; +} + +/* + * Check for data availability for a given stream id. + * + * Return 0 if NOT available, 1 if so and a negative value on error. + */ +int relayd_data_available(struct lttcomm_sock *sock, uint64_t stream_id, + uint64_t last_net_seq_num) +{ + int ret; + struct lttcomm_relayd_data_available msg; + struct lttcomm_relayd_generic_reply reply; + + /* Code flow error. Safety net. */ + assert(sock); + + DBG("Relayd data available for stream id %" PRIu64, stream_id); + + msg.stream_id = htobe64(stream_id); + msg.last_net_seq_num = htobe64(last_net_seq_num); + + /* Send command */ + ret = send_command(sock, RELAYD_DATA_AVAILABLE, (void *) &msg, + sizeof(msg), 0); + if (ret < 0) { + goto error; + } + + /* Recevie response */ + ret = recv_reply(sock, (void *) &reply, sizeof(reply)); + if (ret < 0) { + goto error; + } + + reply.ret_code = be32toh(reply.ret_code); + + /* Return session id or negative ret code. */ + if (reply.ret_code >= LTTNG_OK) { + ret = -reply.ret_code; + ERR("Relayd data available replied error %d", ret); + } + + /* At this point, the ret code is either 1 or 0 */ + ret = reply.ret_code; + + DBG("Relayd data is %s available for stream id %" PRIu64, + ret == 1 ? "" : "NOT", stream_id); + +error: + return ret; +} + +/* + * Check on the relayd side for a quiescent state on the control socket. + */ +int relayd_quiescent_control(struct lttcomm_sock *sock) +{ + int ret; + struct lttcomm_relayd_generic_reply reply; + + /* Code flow error. Safety net. */ + assert(sock); + + DBG("Relayd checking quiescent control state"); + + /* Send command */ + ret = send_command(sock, RELAYD_QUIESCENT_CONTROL, NULL, 0, 0); + if (ret < 0) { + goto error; + } + + /* Recevie response */ + ret = recv_reply(sock, (void *) &reply, sizeof(reply)); + if (ret < 0) { + goto error; + } + + reply.ret_code = be32toh(reply.ret_code); + + /* Return session id or negative ret code. */ + if (reply.ret_code != LTTNG_OK) { + ret = -reply.ret_code; + ERR("Relayd quiescent control replied error %d", ret); + goto error; + } + + /* Control socket is quiescent */ + return 1; error: return ret;