X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Frelayd%2Frelayd.c;h=c883ea5d7f44432ead4ecbc614f8a8abc52686b1;hb=bb63afd9d236f8aa96935ce21f415d76452187e2;hp=9cc682713ca7368b6d0b21255bbf84c4740a39b2;hpb=092b625914723e40ccb3345826a2280cbd01f8b8;p=lttng-tools.git diff --git a/src/common/relayd/relayd.c b/src/common/relayd/relayd.c index 9cc682713..c883ea5d7 100644 --- a/src/common/relayd/relayd.c +++ b/src/common/relayd/relayd.c @@ -33,7 +33,7 @@ * Send command. Fill up the header and append the data. */ static int send_command(struct lttcomm_sock *sock, - enum lttcomm_sessiond_command cmd, void *data, size_t size, + enum lttcomm_relayd_command cmd, void *data, size_t size, int flags) { int ret; @@ -99,6 +99,54 @@ error: return ret; } +/* + * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and + * set session_id of the relayd if we have a successful reply from the relayd. + * + * On success, return 0 else a negative value being a lttng_error_code returned + * from the relayd. + */ +int relayd_create_session(struct lttcomm_sock *sock, uint64_t *session_id) +{ + int ret; + struct lttcomm_relayd_status_session reply; + + assert(sock); + assert(session_id); + + DBG("Relayd create session"); + + /* Send command */ + ret = send_command(sock, RELAYD_CREATE_SESSION, NULL, 0, 0); + if (ret < 0) { + goto error; + } + + /* Recevie response */ + ret = recv_reply(sock, (void *) &reply, sizeof(reply)); + if (ret < 0) { + goto error; + } + + reply.session_id = be64toh(reply.session_id); + reply.ret_code = be32toh(reply.ret_code); + + /* Return session id or negative ret code. */ + if (reply.ret_code != LTTNG_OK) { + ret = -1; + ERR("Relayd create session replied error %d", reply.ret_code); + goto error; + } else { + ret = 0; + *session_id = reply.session_id; + } + + DBG("Relayd session created with id %" PRIu64, reply.session_id); + +error: + return ret; +} + /* * Add stream on the relayd and assign stream handle to the stream_id argument. * @@ -139,8 +187,8 @@ int relayd_add_stream(struct lttcomm_sock *sock, const char *channel_name, /* Return session id or negative ret code. */ if (reply.ret_code != LTTNG_OK) { - ret = -reply.ret_code; - ERR("Relayd add stream replied error %d", ret); + ret = -1; + ERR("Relayd add stream replied error %d", reply.ret_code); } else { /* Success */ ret = 0; @@ -351,8 +399,8 @@ int relayd_send_close_stream(struct lttcomm_sock *sock, uint64_t stream_id, /* Return session id or negative ret code. */ if (reply.ret_code != LTTNG_OK) { - ret = -reply.ret_code; - ERR("Relayd close stream replied error %d", ret); + ret = -1; + ERR("Relayd close stream replied error %d", reply.ret_code); } else { /* Success */ ret = 0; @@ -401,15 +449,14 @@ int relayd_data_pending(struct lttcomm_sock *sock, uint64_t stream_id, /* Return session id or negative ret code. */ if (reply.ret_code >= LTTNG_OK) { - ret = -reply.ret_code; - ERR("Relayd data pending replied error %d", ret); + ERR("Relayd data pending replied error %d", reply.ret_code); } /* At this point, the ret code is either 1 or 0 */ ret = reply.ret_code; DBG("Relayd data is %s pending for stream id %" PRIu64, - ret == 1 ? "NOT" : "", stream_id); + ret == 1 ? "" : "NOT", stream_id); error: return ret; @@ -418,9 +465,11 @@ error: /* * Check on the relayd side for a quiescent state on the control socket. */ -int relayd_quiescent_control(struct lttcomm_sock *sock) +int relayd_quiescent_control(struct lttcomm_sock *sock, + uint64_t metadata_stream_id) { int ret; + struct lttcomm_relayd_quiescent_control msg; struct lttcomm_relayd_generic_reply reply; /* Code flow error. Safety net. */ @@ -428,8 +477,10 @@ int relayd_quiescent_control(struct lttcomm_sock *sock) DBG("Relayd checking quiescent control state"); + msg.stream_id = htobe64(metadata_stream_id); + /* Send command */ - ret = send_command(sock, RELAYD_QUIESCENT_CONTROL, NULL, 0, 0); + ret = send_command(sock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0); if (ret < 0) { goto error; } @@ -444,8 +495,8 @@ int relayd_quiescent_control(struct lttcomm_sock *sock) /* 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); + ret = -1; + ERR("Relayd quiescent control replied error %d", reply.ret_code); goto error; } @@ -455,3 +506,94 @@ int relayd_quiescent_control(struct lttcomm_sock *sock) error: return ret; } + +/* + * Begin a data pending command for a specific session id. + */ +int relayd_begin_data_pending(struct lttcomm_sock *sock, uint64_t id) +{ + int ret; + struct lttcomm_relayd_begin_data_pending msg; + struct lttcomm_relayd_generic_reply reply; + + /* Code flow error. Safety net. */ + assert(sock); + + DBG("Relayd begin data pending"); + + msg.session_id = htobe64(id); + + /* Send command */ + ret = send_command(sock, RELAYD_BEGIN_DATA_PENDING, &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 = -1; + ERR("Relayd begin data pending replied error %d", reply.ret_code); + goto error; + } + + return 0; + +error: + return ret; +} + +/* + * End a data pending command for a specific session id. + * + * Return 0 on success and set is_data_inflight to 0 if no data is being + * streamed or 1 if it is the case. + */ +int relayd_end_data_pending(struct lttcomm_sock *sock, uint64_t id, + unsigned int *is_data_inflight) +{ + int ret; + struct lttcomm_relayd_end_data_pending msg; + struct lttcomm_relayd_generic_reply reply; + + /* Code flow error. Safety net. */ + assert(sock); + + DBG("Relayd end data pending"); + + msg.session_id = htobe64(id); + + /* Send command */ + ret = send_command(sock, RELAYD_END_DATA_PENDING, &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); + if (reply.ret_code < 0) { + ret = reply.ret_code; + goto error; + } + + *is_data_inflight = reply.ret_code; + + DBG("Relayd end data pending is data inflight: %d", reply.ret_code); + + return 0; + +error: + return ret; +}