X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=cbe1a607c36add6e8b320f8f57c13a816a9b3dc5;hp=92f9a5600ac28b7280dcea9f1041b7725b13b748;hb=116ffe568c1f1caec896970a2db2a2e30400747b;hpb=7a45c7e6401baebe3715b317a3d871ee49921057

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index 92f9a5600..cbe1a607c 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1083,6 +1083,11 @@ static int set_index_control_data(struct relay_index *index,
 	return relay_index_set_data(index, &index_data);
 }
 
+static bool session_streams_have_index(const struct relay_session *session)
+{
+	return session->minor >= 4 && !session->snapshot;
+}
+
 /*
  * Handle the RELAYD_CREATE_SESSION command.
  *
@@ -1741,18 +1746,27 @@ static
 int try_rotate_stream(struct relay_stream *stream)
 {
 	int ret = 0;
+	uint64_t trace_seq;
 
 	/* No rotation expected. */
 	if (stream->rotate_at_seq_num == -1ULL) {
 		goto end;
 	}
 
-	if (stream->prev_seq < stream->rotate_at_seq_num ||
-			stream->prev_seq == -1ULL) {
-		DBG("Stream %" PRIu64 " no yet ready for rotation",
-				stream->stream_handle);
+	trace_seq = min(stream->prev_data_seq, stream->prev_index_seq);
+	if (stream->prev_data_seq == -1ULL || stream->prev_index_seq == -1ULL ||
+			trace_seq < stream->rotate_at_seq_num) {
+		DBG("Stream %" PRIu64 " not yet ready for rotation (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
+				stream->stream_handle,
+				stream->rotate_at_seq_num,
+				stream->prev_data_seq,
+				stream->prev_index_seq);
 		goto end;
-	} else if (stream->prev_seq > stream->rotate_at_seq_num) {
+	} else if (stream->prev_data_seq > stream->rotate_at_seq_num) {
+		/*
+		 * prev_data_seq is checked here since indexes and rotation
+		 * commands are serialized with respect to each other.
+		 */
 		DBG("Rotation after too much data has been written in tracefile "
 				"for stream %" PRIu64 ", need to truncate before "
 				"rotating", stream->stream_handle);
@@ -1762,7 +1776,20 @@ int try_rotate_stream(struct relay_stream *stream)
 			goto end;
 		}
 	} else {
-		/* stream->prev_seq == stream->rotate_at_seq_num */
+		if (trace_seq != stream->rotate_at_seq_num) {
+			/*
+			 * Unexpected, protocol error/bug.
+			 * It could mean that we received a rotation position
+			 * that is in the past.
+			 */
+			ERR("Stream %" PRIu64 " is in an inconsistent state (rotate_at_seq_num = %" PRIu64 ", prev_data_seq = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
+				stream->stream_handle,
+				stream->rotate_at_seq_num,
+				stream->prev_data_seq,
+				stream->prev_index_seq);
+			ret = -1;
+			goto end;
+		}
 		DBG("Stream %" PRIu64 " ready for rotation",
 				stream->stream_handle);
 		ret = do_rotate_stream(stream);
@@ -1931,6 +1958,7 @@ static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
 	struct relay_stream *stream;
 	ssize_t send_ret;
 	int ret;
+	uint64_t stream_seq;
 
 	DBG("Data pending command received");
 
@@ -1958,12 +1986,23 @@ static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
 
 	pthread_mutex_lock(&stream->lock);
 
-	DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
-			" and last_seq %" PRIu64, msg.stream_id,
-			stream->prev_seq, msg.last_net_seq_num);
+	if (session_streams_have_index(session)) {
+		/*
+		 * Ensure that both the index and stream data have been
+		 * flushed up to the requested point.
+		 */
+		stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
+	} else {
+		stream_seq = stream->prev_data_seq;
+	}
+	DBG("Data pending for stream id %" PRIu64 ": prev_data_seq %" PRIu64
+			", prev_index_seq %" PRIu64
+			", and last_seq %" PRIu64, msg.stream_id,
+			stream->prev_data_seq, stream->prev_index_seq,
+			msg.last_net_seq_num);
 
 	/* Avoid wrapping issue */
-	if (((int64_t) (stream->prev_seq - msg.last_net_seq_num)) >= 0) {
+	if (((int64_t) (stream_seq - msg.last_net_seq_num)) >= 0) {
 		/* Data has in fact been written and is NOT pending */
 		ret = 0;
 	} else {
@@ -2183,7 +2222,18 @@ static int relay_end_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
 		}
 		pthread_mutex_lock(&stream->lock);
 		if (!stream->data_pending_check_done) {
-			if (!stream->closed || !(((int64_t) (stream->prev_seq - stream->last_net_seq_num)) >= 0)) {
+			uint64_t stream_seq;
+
+			if (session_streams_have_index(conn->session)) {
+				/*
+				 * Ensure that both the index and stream data have been
+				 * flushed up to the requested point.
+				 */
+				stream_seq = min(stream->prev_data_seq, stream->prev_index_seq);
+			} else {
+				stream_seq = stream->prev_data_seq;
+			}
+			if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
 				is_data_inflight = 1;
 				DBG("Data is still in flight for stream %" PRIu64,
 						stream->stream_handle);
@@ -2600,6 +2650,7 @@ static int relay_mkdir(const struct lttcomm_relayd_hdr *recv_hdr,
 		goto end;
 	}
 
+	DBG("MKDIR command has path \"%s\", changed to \"%s\"", path_view.data, path);
 	ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, -1, -1);
 	if (ret < 0) {
 		ERR("relay creating output directory");
@@ -2723,6 +2774,8 @@ static int relay_rotate_rename(const struct lttcomm_relayd_hdr *recv_hdr,
 		goto end;
 	}
 
+	DBG("ROTATE_RENAME command has argument old path = \"%s\", new_path = \"%s\"",
+			old_path_view.data, new_path_view.data);
 	complete_old_path = create_output_path(old_path_view.data);
 	if (!complete_old_path) {
 		ERR("Failed to build old output path in rotate_rename command");
@@ -2736,6 +2789,8 @@ static int relay_rotate_rename(const struct lttcomm_relayd_hdr *recv_hdr,
 		ret = -1;
 		goto end;
 	}
+	DBG("Expanded ROTATE_RENAME arguments to old path = \"%s\", new_path = \"%s\"",
+			complete_old_path, complete_new_path);
 
 	ret = utils_mkdir_recursive(complete_new_path, S_IRWXU | S_IRWXG,
 			-1, -1);
@@ -3462,7 +3517,7 @@ static enum relay_connection_status relay_process_data_receive_payload(
 	}
 
 
-	if (session->minor >= 4 && !session->snapshot) {
+	if (session_streams_have_index(session)) {
 		ret = handle_index_data(stream, state->header.net_seq_num,
 				state->rotate_index, &index_flushed, state->header.data_size + state->header.padding_size);
 		if (ret < 0) {
@@ -3477,7 +3532,7 @@ static enum relay_connection_status relay_process_data_receive_payload(
 	stream->tracefile_size_current += state->header.data_size +
 			state->header.padding_size;
 
-	if (stream->prev_seq == -1ULL) {
+	if (stream->prev_data_seq == -1ULL) {
 		new_stream = true;
 	}
 	if (index_flushed) {
@@ -3486,7 +3541,7 @@ static enum relay_connection_status relay_process_data_receive_payload(
 		stream->prev_index_seq = state->header.net_seq_num;
 	}
 
-	stream->prev_seq = state->header.net_seq_num;
+	stream->prev_data_seq = state->header.net_seq_num;
 
 	/*
 	 * Resetting the protocol state (to RECEIVE_HEADER) will trash the