X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Frelayd%2Frelayd.c;h=43f91082385a951682f6ebaf6f839abf3e123a33;hp=db476080ec207d505938ddcb9b351cd1b4d10ac7;hb=f7079f6790ccfb78ec7115ccb5b1706f5c18ebfe;hpb=8994307fa7ccf9b61cc0157f2c5d34e248c56641 diff --git a/src/common/relayd/relayd.c b/src/common/relayd/relayd.c index db476080e..43f910823 100644 --- a/src/common/relayd/relayd.c +++ b/src/common/relayd/relayd.c @@ -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 = -reply.ret_code; + ERR("Relayd create session replied error %d", ret); + 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. * @@ -163,42 +211,61 @@ int relayd_version_check(struct lttcomm_sock *sock, uint32_t major, uint32_t minor) { int ret; - struct lttcomm_relayd_version reply; + struct lttcomm_relayd_version msg; /* Code flow error. Safety net. */ assert(sock); DBG("Relayd version check for major.minor %u.%u", major, minor); + /* Prepare network byte order before transmission. */ + msg.major = htobe32(major); + msg.minor = htobe32(minor); + /* Send command */ - ret = send_command(sock, RELAYD_VERSION, NULL, 0, 0); + ret = send_command(sock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0); if (ret < 0) { goto error; } /* Recevie response */ - ret = recv_reply(sock, (void *) &reply, sizeof(reply)); + ret = recv_reply(sock, (void *) &msg, sizeof(msg)); if (ret < 0) { goto error; } /* Set back to host bytes order */ - reply.major = be32toh(reply.major); - reply.minor = be32toh(reply.minor); + msg.major = be32toh(msg.major); + msg.minor = be32toh(msg.minor); - /* Validate version */ - if (reply.major <= major) { - if (reply.minor <= minor) { - /* Compatible */ - ret = 0; - DBG2("Relayd version is compatible"); - goto error; - } + /* + * Only validate the major version. If the other side is higher, + * communication is not possible. Only major version equal can talk to each + * other. If the minor version differs, the lowest version is used by both + * sides. + * + * For now, before 2.1.0 stable release, we don't have to check the minor + * because this new mechanism with the relayd will only be available with + * 2.1 and NOT 2.0.x. + */ + if (msg.major == major) { + /* Compatible */ + ret = 0; + DBG2("Relayd version is compatible"); + goto error; } + /* + * After 2.1.0 release, for the 2.2 release, at this point will have to + * check the minor version in order for the session daemon to know which + * structure to use to communicate with the relayd. If the relayd's minor + * version is higher, it will adapt to our version so we can continue to + * use the latest relayd communication data structure. + */ + /* Version number not compatible */ - DBG2("Relayd version is NOT compatible %u.%u > %u.%u", reply.major, - reply.minor, major, minor); + DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)", + msg.major, major); ret = -1; error: @@ -348,25 +415,25 @@ error: /* * Check for data availability for a given stream id. * - * Return 0 if NOT available, 1 if so and a negative value on error. + * Return 0 if NOT pending, 1 if so and a negative value on error. */ -int relayd_data_available(struct lttcomm_sock *sock, uint64_t stream_id, +int relayd_data_pending(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_data_pending msg; struct lttcomm_relayd_generic_reply reply; /* Code flow error. Safety net. */ assert(sock); - DBG("Relayd data available for stream id %" PRIu64, stream_id); + DBG("Relayd data pending 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, + ret = send_command(sock, RELAYD_DATA_PENDING, (void *) &msg, sizeof(msg), 0); if (ret < 0) { goto error; @@ -383,14 +450,14 @@ int relayd_data_available(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 available replied error %d", ret); + ERR("Relayd data pending 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); + DBG("Relayd data is %s pending for stream id %" PRIu64, + ret == 1 ? "NOT" : "", stream_id); error: return ret; @@ -431,7 +498,98 @@ int relayd_quiescent_control(struct lttcomm_sock *sock) } /* Control socket is quiescent */ - return 1; + return 0; + +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 = -reply.ret_code; + ERR("Relayd begin data pending replied error %d", ret); + 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;