X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Frelayd%2Frelayd.c;h=56ca98223a383bd6d9a36b288decea7c677505ac;hp=27cb58e52ae49113f877c9e31b604d349000ffaf;hb=c5b6f4f08fe8d1abff74c7f6ad3630b7dcf0669d;hpb=f73fabfda365d22e7dd180fb1614e37c446fbd9e

diff --git a/src/common/relayd/relayd.c b/src/common/relayd/relayd.c
index 27cb58e52..56ca98223 100644
--- a/src/common/relayd/relayd.c
+++ b/src/common/relayd/relayd.c
@@ -67,10 +67,11 @@ static int send_command(struct lttcomm_sock *sock,
 
 	ret = sock->ops->sendmsg(sock, buf, buf_size, flags);
 	if (ret < 0) {
+		ret = -errno;
 		goto error;
 	}
 
-	DBG3("Relayd sending command %d", cmd);
+	DBG3("Relayd sending command %d of size %" PRIu64, cmd, buf_size);
 
 error:
 	free(buf);
@@ -86,10 +87,11 @@ 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) {
+		ret = -errno;
 		goto error;
 	}
 
@@ -97,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.
  *
@@ -125,7 +175,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;
@@ -161,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);
-
-	/* Validate version */
-	if (reply.major <= major) {
-		if (reply.minor <= minor) {
-			/* Compatible */
-			ret = 0;
-			DBG2("Relayd version is compatible");
-			goto error;
-		}
+	msg.major = be32toh(msg.major);
+	msg.minor = be32toh(msg.minor);
+
+	/*
+	 * 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:
@@ -228,7 +297,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:
@@ -273,7 +342,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) {
@@ -283,6 +352,7 @@ int relayd_send_data_hdr(struct lttcomm_sock *sock,
 	/* Only send data header. */
 	ret = sock->ops->sendmsg(sock, hdr, size, 0);
 	if (ret < 0) {
+		ret = -errno;
 		goto error;
 	}
 
@@ -341,3 +411,95 @@ int relayd_send_close_stream(struct lttcomm_sock *sock, uint64_t stream_id,
 error:
 	return ret;
 }
+
+/*
+ * Check for data availability for a given stream id.
+ *
+ * Return 0 if NOT pending, 1 if so and a negative value on error.
+ */
+int relayd_data_pending(struct lttcomm_sock *sock, uint64_t stream_id,
+		uint64_t last_net_seq_num)
+{
+	int ret;
+	struct lttcomm_relayd_data_pending msg;
+	struct lttcomm_relayd_generic_reply reply;
+
+	/* Code flow error. Safety net. */
+	assert(sock);
+
+	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_PENDING, (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 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 pending 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 0;
+
+error:
+	return ret;
+}