X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=0f81d556dd4df73a910471756b75cf6567581a36;hp=15e2b2266f7264d31b3029e7dad0bdac5c2f6576;hb=c8f59ee5fc11492ef472dc5cfd2fd2c4926b1787;hpb=4b7f17b22c323c74554f7026304aa6c00a1ee434

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index 15e2b2266..0f81d556d 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1046,6 +1046,7 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
 		goto end_unlock;
 	}
 
+	stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
 	stream->close_flag = 1;
 
 	if (close_stream_check(stream)) {
@@ -1265,9 +1266,9 @@ static
 int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
 		struct relay_command *cmd)
 {
-	int ret = htobe32(LTTNG_OK);
+	int ret;
 	struct lttcomm_relayd_version reply;
-	struct relay_session *session = NULL;
+	struct relay_session *session;
 
 	if (cmd->session == NULL) {
 		session = zmalloc(sizeof(struct relay_session));
@@ -1279,10 +1280,17 @@ int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
 		session->id = ++last_relay_session_id;
 		DBG("Created session %" PRIu64, session->id);
 		cmd->session = session;
+	} else {
+		session = cmd->session;
 	}
 	session->version_check_done = 1;
 
-	sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
+	ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
+	if (ret < 2) {
+		ERR("Error in scanning version");
+		ret = -1;
+		goto end;
+	}
 	reply.major = htobe32(reply.major);
 	reply.minor = htobe32(reply.minor);
 	ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
@@ -1297,6 +1305,102 @@ end:
 	return ret;
 }
 
+/*
+ * Check for data availability for a given stream id from the session daemon.
+ */
+static
+int relay_data_available(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_generic_reply reply;
+	struct relay_stream *stream;
+	int ret;
+	struct lttng_ht_node_ulong *node;
+	struct lttng_ht_iter iter;
+	uint64_t last_net_seq_num, stream_id;
+
+	DBG("Data available command received");
+
+	if (!session || session->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);
+	if (ret < sizeof(msg)) {
+		ERR("Relay didn't receive valid data_available struct size : %d", ret);
+		ret = -1;
+		goto end_no_session;
+	}
+
+	stream_id = be64toh(msg.stream_id);
+	last_net_seq_num = be64toh(msg.last_net_seq_num);
+
+	rcu_read_lock();
+	lttng_ht_lookup(streams_ht, (void *)((unsigned long) stream_id), &iter);
+	node = lttng_ht_iter_get_node_ulong(&iter);
+	if (node == NULL) {
+		DBG("Relay stream %" PRIu64 " not found", stream_id);
+		ret = -1;
+		goto end_unlock;
+	}
+
+	stream = caa_container_of(node, struct relay_stream, stream_n);
+	assert(stream);
+
+	DBG("Data available for stream id %" PRIu64 " prev_seq %" PRIu64
+			" and last_seq %" PRIu64, stream_id, stream->prev_seq,
+			last_net_seq_num);
+
+	if (stream->prev_seq == -1UL || stream->prev_seq <= last_net_seq_num) {
+		/* Data has in fact been written and is available */
+		ret = 1;
+	} else {
+		/* Data still being streamed. */
+		ret = 0;
+	}
+
+end_unlock:
+	rcu_read_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");
+	}
+
+end_no_session:
+	return ret;
+}
+
+/*
+ * Wait for the control socket to reach a quiescent state.
+ *
+ * Note that for now, when receiving this command from the session daemon, this
+ * means that every subsequent commands or data received on the control socket
+ * has been handled. So, this is why we simply return OK here.
+ */
+static
+int relay_quiescent_control(struct lttcomm_relayd_hdr *recv_hdr,
+		struct relay_command *cmd)
+{
+	int ret;
+	struct lttcomm_relayd_generic_reply reply;
+
+	DBG("Checking quiescent state on control socket");
+
+	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");
+	}
+
+	return ret;
+}
+
 /*
  * relay_process_control: Process the commands received on the control socket
  */
@@ -1327,6 +1431,12 @@ 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);
+		break;
+	case RELAYD_QUIESCENT_CONTROL:
+		ret = relay_quiescent_control(recv_hdr, cmd);
+		break;
 	case RELAYD_UPDATE_SYNC_INFO:
 	default:
 		ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
@@ -1400,14 +1510,14 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
 		goto end_unlock;
 	}
 
+	DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
+			ret, stream->stream_handle);
+
 	ret = write_padding_to_file(stream->fd, be32toh(data_hdr.padding_size));
 	if (ret < 0) {
 		goto end_unlock;
 	}
 
-	DBG2("Relay wrote %d bytes to tracefile for stream id %" PRIu64,
-		ret, stream->stream_handle);
-
 	stream->prev_seq = net_seq_num;
 
 	/* Check if we need to close the FD */