X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=1ea14f8ca5991af15f4d8b7d094ab9c85168c163;hp=8e6b63b011436a06a80a455aad802d23276a4470;hb=02d02e31d47c091a38154c9c188c08387902d97b;hpb=d73bf3d793ee0b0c5b56cb47cb50c27d1789d3bd

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index 8e6b63b01..1ea14f8ca 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -3392,6 +3392,8 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx)
 {
 	ssize_t ret;
+	int rotate_ret;
+	bool rotated = false;
 
 	pthread_mutex_lock(&stream->lock);
 	if (stream->metadata_flag) {
@@ -3400,11 +3402,11 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 
 	switch (consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
-		ret = lttng_kconsumer_read_subbuffer(stream, ctx);
+		ret = lttng_kconsumer_read_subbuffer(stream, ctx, &rotated);
 		break;
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
-		ret = lttng_ustconsumer_read_subbuffer(stream, ctx);
+		ret = lttng_ustconsumer_read_subbuffer(stream, ctx, &rotated);
 		break;
 	default:
 		ERR("Unknown consumer_data type");
@@ -3418,6 +3420,14 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		pthread_mutex_unlock(&stream->metadata_rdv_lock);
 	}
 	pthread_mutex_unlock(&stream->lock);
+	if (rotated) {
+		rotate_ret = consumer_post_rotation(stream, ctx);
+		if (rotate_ret < 0) {
+			ERR("Failed after a rotation");
+			ret = -1;
+		}
+	}
+
 	return ret;
 }
 
@@ -3903,6 +3913,54 @@ unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
 	return start_pos;
 }
 
+/*
+ * Check if a stream is ready to be rotated after extracting it.
+ *
+ * Return 1 if it is ready for rotation, 0 if it is not, a negative value on
+ * error. Stream lock must be held.
+ */
+int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
+{
+	int ret;
+	unsigned long consumed_pos;
+
+	if (!stream->rotate_position && !stream->rotate_ready) {
+		ret = 0;
+		goto end;
+	}
+
+	if (stream->rotate_ready) {
+		ret = 1;
+		goto end;
+	}
+
+	/*
+	 * If we don't have the rotate_ready flag, check the consumed position
+	 * to determine if we need to rotate.
+	 */
+	ret = lttng_consumer_sample_snapshot_positions(stream);
+	if (ret < 0) {
+		ERR("Taking snapshot positions");
+		goto end;
+	}
+
+	ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos);
+	if (ret < 0) {
+		ERR("Consumed snapshot position");
+		goto end;
+	}
+
+	/* Rotate position not reached yet (with check for overflow). */
+	if ((long) (consumed_pos - stream->rotate_position) < 0) {
+		ret = 0;
+		goto end;
+	}
+	ret = 1;
+
+end:
+	return ret;
+}
+
 /*
  * Reset the state for a stream after a rotation occurred.
  */
@@ -4014,7 +4072,7 @@ end:
  * Return 0 on success, a negative number of error.
  */
 int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *stream)
+		struct lttng_consumer_stream *stream, bool *rotated)
 {
 	int ret;
 
@@ -4058,6 +4116,10 @@ int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
 	}
 	lttng_consumer_reset_stream_rotate_state(stream);
 
+	if (rotated) {
+		*rotated = true;
+	}
+
 	ret = 0;
 
 error: