X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=5c14dd1d80be07eeda87fae696cbde7a450de8ec;hp=5625e59ddd5cb8a2092d21d1145149df63ed66d6;hb=83f4233dde73280d45fcfb799add96cce32098ce;hpb=adf7a0a197a90d6b8ed7a831d6681beb77371a3d

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index 5625e59dd..5c14dd1d8 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -571,6 +571,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
 	stream->monitor = monitor;
 	stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
 	stream->index_fd = -1;
+	stream->last_sequence_number = -1ULL;
 	pthread_mutex_init(&stream->lock, NULL);
 	pthread_mutex_init(&stream->metadata_timer_lock, NULL);
 
@@ -1020,7 +1021,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 
 	CDS_INIT_LIST_HEAD(&channel->streams.head);
 
-	DBG("Allocated channel (key %" PRIu64 ")", channel->key)
+	DBG("Allocated channel (key %" PRIu64 ")", channel->key);
 
 end:
 	return channel;
@@ -1228,9 +1229,15 @@ void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
 	DBG("Consumer flag that it should quit");
 }
 
+
+/*
+ * Flush pending writes to trace output disk file.
+ */
+static
 void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
 		off_t orig_offset)
 {
+	int ret;
 	int outfd = stream->out_fd;
 
 	/*
@@ -1261,8 +1268,12 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
 	 * defined. So it can be expected to lead to lower throughput in
 	 * streaming.
 	 */
-	posix_fadvise(outfd, orig_offset - stream->max_sb_size,
+	ret = posix_fadvise(outfd, orig_offset - stream->max_sb_size,
 			stream->max_sb_size, POSIX_FADV_DONTNEED);
+	if (ret && ret != -ENOSYS) {
+		errno = ret;
+		PERROR("posix_fadvise on fd %i", outfd);
+	}
 }
 
 /*
@@ -1555,6 +1566,16 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		if (stream->metadata_flag) {
 			/* Metadata requires the control socket. */
 			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+			if (stream->reset_metadata_flag) {
+				ret = relayd_reset_metadata(&relayd->control_sock,
+						stream->relayd_stream_id,
+						stream->metadata_version);
+				if (ret < 0) {
+					relayd_hang_up = 1;
+					goto write_error;
+				}
+				stream->reset_metadata_flag = 0;
+			}
 			netlen += sizeof(struct lttcomm_relayd_metadata_payload);
 		}
 
@@ -1578,6 +1599,15 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		/* No streaming, we have to set the len with the full padding */
 		len += padding;
 
+		if (stream->metadata_flag && stream->reset_metadata_flag) {
+			ret = utils_truncate_stream_file(stream->out_fd, 0);
+			if (ret < 0) {
+				ERR("Reset metadata file");
+				goto end;
+			}
+			stream->reset_metadata_flag = 0;
+		}
+
 		/*
 		 * Check if we need to change the tracefile before writing the packet.
 		 */
@@ -1662,8 +1692,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		lttng_sync_file_range(outfd, stream->out_fd_offset, len,
 				SYNC_FILE_RANGE_WRITE);
 		stream->out_fd_offset += len;
+		lttng_consumer_sync_trace_file(stream, orig_offset);
 	}
-	lttng_consumer_sync_trace_file(stream, orig_offset);
 
 write_error:
 	/*
@@ -1743,6 +1773,16 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 			 */
 			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 
+			if (stream->reset_metadata_flag) {
+				ret = relayd_reset_metadata(&relayd->control_sock,
+						stream->relayd_stream_id,
+						stream->metadata_version);
+				if (ret < 0) {
+					relayd_hang_up = 1;
+					goto write_error;
+				}
+				stream->reset_metadata_flag = 0;
+			}
 			ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
 					padding);
 			if (ret < 0) {
@@ -1766,6 +1806,14 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 		/* No streaming, we have to set the len with the full padding */
 		len += padding;
 
+		if (stream->metadata_flag && stream->reset_metadata_flag) {
+			ret = utils_truncate_stream_file(stream->out_fd, 0);
+			if (ret < 0) {
+				ERR("Reset metadata file");
+				goto end;
+			}
+			stream->reset_metadata_flag = 0;
+		}
 		/*
 		 * Check if we need to change the tracefile before writing the packet.
 		 */
@@ -1874,7 +1922,9 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 		stream->output_written += ret_splice;
 		written += ret_splice;
 	}
-	lttng_consumer_sync_trace_file(stream, orig_offset);
+	if (!relayd) {
+		lttng_consumer_sync_trace_file(stream, orig_offset);
+	}
 	goto end;
 
 write_error:
@@ -2223,10 +2273,10 @@ restart:
 		DBG("Metadata poll return from wait with %d fd(s)",
 				LTTNG_POLL_GETNB(&events));
 		health_poll_exit();
-		DBG("Metadata event catched in thread");
+		DBG("Metadata event caught in thread");
 		if (ret < 0) {
 			if (errno == EINTR) {
-				ERR("Poll EINTR catched");
+				ERR("Poll EINTR caught");
 				goto restart;
 			}
 			if (LTTNG_POLL_GETNB(&events) == 0) {
@@ -2324,7 +2374,7 @@ restart:
 					len = ctx->on_buffer_ready(stream, ctx);
 					/*
 					 * We don't check the return value here since if we get
-					 * a negative len, it means an error occured thus we
+					 * a negative len, it means an error occurred thus we
 					 * simply remove it from the poll set and free the
 					 * stream.
 					 */
@@ -2351,7 +2401,7 @@ restart:
 						len = ctx->on_buffer_ready(stream, ctx);
 						/*
 						 * We don't check the return value here since if we get
-						 * a negative len, it means an error occured thus we
+						 * a negative len, it means an error occurred thus we
 						 * simply remove it from the poll set and free the
 						 * stream.
 						 */
@@ -2806,10 +2856,10 @@ restart:
 		DBG("Channel poll return from wait with %d fd(s)",
 				LTTNG_POLL_GETNB(&events));
 		health_poll_exit();
-		DBG("Channel event catched in thread");
+		DBG("Channel event caught in thread");
 		if (ret < 0) {
 			if (errno == EINTR) {
-				ERR("Poll EINTR catched");
+				ERR("Poll EINTR caught");
 				goto restart;
 			}
 			if (LTTNG_POLL_GETNB(&events) == 0) {