X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=7b385b49f5264e2ace8823ffbc0f8fe4747630a5;hb=a44ca2ca85e4b64729f7b88b1919fd6737dfff8a;hp=057ac4046c01e67ae05fa938db49e89c9c913718;hpb=7591bab11eceedc6a0d1e02fd6f85592267a63b5;p=lttng-tools.git

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index 057ac4046..7b385b49f 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -71,6 +71,7 @@
 #include "session.h"
 #include "stream.h"
 #include "connection.h"
+#include "tracefile-array.h"
 
 /* command line options */
 char *opt_output_path;
@@ -1890,7 +1891,7 @@ static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
 		 * Only flag a stream inactive when it has already
 		 * received data and no indexes are in flight.
 		 */
-		if (stream->total_index_received > 0
+		if (stream->index_received_seqcount > 0
 				&& stream->indexes_in_flight == 0) {
 			stream->beacon_ts_end =
 				be64toh(index_info.timestamp_end);
@@ -1918,7 +1919,8 @@ static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
 	}
 	ret = relay_index_try_flush(index);
 	if (ret == 0) {
-		stream->total_index_received++;
+		tracefile_array_commit_seq(stream->tfa);
+		stream->index_received_seqcount++;
 	} else if (ret > 0) {
 		/* no flush. */
 		ret = 0;
@@ -2091,7 +2093,7 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
 
 		fd = index_create_file(stream->path_name, stream->channel_name,
 			        -1, -1, stream->tracefile_size,
-				stream->current_tracefile_id);
+				tracefile_array_get_file_index_head(stream->tfa));
 		if (fd < 0) {
 			ret = -1;
 			/* Put self-ref for this index due to error. */
@@ -2120,7 +2122,8 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
 
 	ret = relay_index_try_flush(index);
 	if (ret == 0) {
-		stream->total_index_received++;
+		tracefile_array_commit_seq(stream->tfa);
+		stream->index_received_seqcount++;
 	} else if (ret > 0) {
 		/* No flush. */
 		ret = 0;
@@ -2146,6 +2149,7 @@ static int relay_process_data(struct relay_connection *conn)
 	uint64_t net_seq_num;
 	uint32_t data_size;
 	struct relay_session *session;
+	bool new_stream = false;
 
 	ret = conn->sock->ops->recvmsg(conn->sock, &data_hdr,
 			sizeof(struct lttcomm_relayd_data_hdr), 0);
@@ -2203,35 +2207,23 @@ static int relay_process_data(struct relay_connection *conn)
 	if (stream->tracefile_size > 0 &&
 			(stream->tracefile_size_current + data_size) >
 			stream->tracefile_size) {
-		uint64_t new_id;
+		uint64_t old_id, new_id;
+
+		old_id = tracefile_array_get_file_index_head(stream->tfa);
+		tracefile_array_file_rotate(stream->tfa);
+
+		/* new_id is updated by utils_rotate_stream_file. */
+		new_id = old_id;
 
-		new_id = (stream->current_tracefile_id + 1) %
-			stream->tracefile_count;
-		/*
-		 * Move viewer oldest available data position forward if
-		 * we are overwriting a tracefile.
-		 */
-		if (new_id == stream->oldest_tracefile_id) {
-			stream->oldest_tracefile_id =
-				(stream->oldest_tracefile_id + 1) %
-				stream->tracefile_count;
-		}
 		ret = utils_rotate_stream_file(stream->path_name,
 				stream->channel_name, stream->tracefile_size,
 				stream->tracefile_count, -1,
 			        -1, stream->stream_fd->fd,
-				&stream->current_tracefile_id,
-				&stream->stream_fd->fd);
+				&new_id, &stream->stream_fd->fd);
 		if (ret < 0) {
 			ERR("Rotating stream output file");
 			goto end_stream_unlock;
 		}
-		stream->current_tracefile_seq++;
-		if (stream->current_tracefile_seq
-			- stream->oldest_tracefile_seq >=
-				stream->tracefile_count) {
-			stream->oldest_tracefile_seq++;
-		}
 		/*
 		 * Reset current size because we just performed a stream
 		 * rotation.
@@ -2269,10 +2261,19 @@ static int relay_process_data(struct relay_connection *conn)
 	}
 	stream->tracefile_size_current +=
 			data_size + be32toh(data_hdr.padding_size);
+	if (stream->prev_seq == -1ULL) {
+		new_stream = true;
+	}
+
 	stream->prev_seq = net_seq_num;
 
 end_stream_unlock:
 	pthread_mutex_unlock(&stream->lock);
+	if (new_stream) {
+		pthread_mutex_lock(&session->lock);
+		uatomic_set(&session->new_streams, 1);
+		pthread_mutex_unlock(&session->lock);
+	}
 end_stream_put:
 	stream_put(stream);
 end: