[lttng-dev] [BABELTRACE PATCH 1/3] Handle the inactive streams

[lttng-dev] [BABELTRACE PATCH 1/3] Handle the inactive streams

[Julien Desfossez]

When content_size == 0, just set the timestamp_end of the packet and
reinsert the stream in the heap.
This is required to handle the beacons of inactivity generated in
LTTng live trace streaming.

Signed-off-by: Julien Desfossez <jdesfossez at efficios.com>
---
 formats/ctf/ctf.c |  5 +++++
 lib/iterator.c    | 11 +++++++++++
 2 files changed, 16 insertions(+)

diff --git a/formats/ctf/ctf.c b/formats/ctf/ctf.c
index 794e428..5f263cd 100644
--- a/formats/ctf/ctf.c
+++ b/formats/ctf/ctf.c
@@ -471,6 +471,11 @@ int ctf_read_event(struct bt_stream_pos *ppos, struct ctf_stream_definition *str
 	 */
 	if (unlikely(pos->offset == EOF))
 		return EOF;
+
+	if (pos->content_size == 0) {
+		/* Stream is inactive for now (live reading). */
+		return EAGAIN;
+	}
 	assert(pos->offset < pos->content_size);
 
 	/* Read event header */
diff --git a/lib/iterator.c b/lib/iterator.c
index 009fcd3..155fcbe 100644
--- a/lib/iterator.c
+++ b/lib/iterator.c
@@ -65,6 +65,9 @@ static int stream_read_event(struct ctf_file_stream *sin)
 	ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
 	if (ret == EOF)
 		return EOF;
+	else if (ret == EAGAIN)
+		/* Stream is inactive for now (live reading). */
+		return EAGAIN;
 	else if (ret) {
 		fprintf(stderr, "[error] Reading event failed.\n");
 		return ret;
@@ -803,9 +806,17 @@ int bt_iter_next(struct bt_iter *iter)
 		assert(removed == file_stream);
 		ret = 0;
 		goto end;
+	} else if (ret == EAGAIN) {
+		/*
+		 * The stream is inactive for now, we just updated the timestamp_end
+		 * to skip over this stream up to a certain point in time.
+		 */
+		goto reinsert;
 	} else if (ret) {
 		goto end;
 	}
+
+reinsert:
 	/* Reinsert the file stream into the heap, and rebalance. */
 	removed = bt_heap_replace_max(iter->stream_heap, file_stream);
 	assert(removed == file_stream);
-- 
1.8.3.2

[lttng-dev] [BABELTRACE PATCH 2/3] Handle packets containing only a header

[Julien Desfossez]

In live streaming with LTTng, if we do a stop and then a destroy, we
receive a packet that contains only a header. With this patch, we will
ask for a new packet when we see this case.

Signed-off-by: Julien Desfossez <jdesfossez at efficios.com>
---
 formats/ctf/ctf.c | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/formats/ctf/ctf.c b/formats/ctf/ctf.c
index 5f263cd..5cf8097 100644
--- a/formats/ctf/ctf.c
+++ b/formats/ctf/ctf.c
@@ -476,6 +476,10 @@ int ctf_read_event(struct bt_stream_pos *ppos, struct ctf_stream_definition *str
 		/* Stream is inactive for now (live reading). */
 		return EAGAIN;
 	}
+	/* Packet only contains headers */
+	if (pos->offset == pos->content_size)
+		return EAGAIN;
+
 	assert(pos->offset < pos->content_size);
 
 	/* Read event header */
-- 
1.8.3.2

[lttng-dev] [BABELTRACE PATCH 3/3] Handle empty streams on iterator init

[Julien Desfossez]

In live trace streaming, we can create an iterator with inactive
streams. This patch allows inactive streams to be considered as valid.

Signed-off-by: Julien Desfossez <jdesfossez at efficios.com>
---
 lib/iterator.c | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/lib/iterator.c b/lib/iterator.c
index 155fcbe..3280f4a 100644
--- a/lib/iterator.c
+++ b/lib/iterator.c
@@ -725,7 +725,7 @@ int bt_iter_init(struct bt_iter *iter,
 				if (ret == EOF) {
 					ret = 0;
 					continue;
-				} else if (ret) {
+				} else if (ret != 0 && ret != EAGAIN) {
 					goto error;
 				}
 				/* Add to heap */
@@ -821,6 +821,13 @@ reinsert:
 	removed = bt_heap_replace_max(iter->stream_heap, file_stream);
 	assert(removed == file_stream);
 
+	file_stream = bt_heap_maximum(iter->stream_heap);
+	if (file_stream->pos.content_size == 0) {
+		ret = EAGAIN;
+	} else {
+		ret = 0;
+	}
+
 end:
 	return ret;
 }
-- 
1.8.3.2

[lttng-dev] [BABELTRACE PATCH 3/3] Handle empty streams on iterator init

[Mathieu Desnoyers]

All 3 patches merged, thanks!

Mathieu

----- Original Message -----
> From: "Julien Desfossez" <jdesfossez@efficios.com>
> To: "mathieu desnoyers" <mathieu.desnoyers at efficios.com>
> Cc: lttng-dev at lists.lttng.org, "Julien Desfossez" <jdesfossez at efficios.com>
> Sent: Wednesday, November 27, 2013 11:40:12 AM
> Subject: [BABELTRACE PATCH 3/3] Handle empty streams on iterator init
> 
> In live trace streaming, we can create an iterator with inactive
> streams. This patch allows inactive streams to be considered as valid.
> 
> Signed-off-by: Julien Desfossez <jdesfossez at efficios.com>
> ---
>  lib/iterator.c | 9 ++++++++-
>  1 file changed, 8 insertions(+), 1 deletion(-)
> 
> diff --git a/lib/iterator.c b/lib/iterator.c
> index 155fcbe..3280f4a 100644
> --- a/lib/iterator.c
> +++ b/lib/iterator.c
> @@ -725,7 +725,7 @@ int bt_iter_init(struct bt_iter *iter,
>  				if (ret == EOF) {
>  					ret = 0;
>  					continue;
> -				} else if (ret) {
> +				} else if (ret != 0 && ret != EAGAIN) {
>  					goto error;
>  				}
>  				/* Add to heap */
> @@ -821,6 +821,13 @@ reinsert:
>  	removed = bt_heap_replace_max(iter->stream_heap, file_stream);
>  	assert(removed == file_stream);
>  
> +	file_stream = bt_heap_maximum(iter->stream_heap);
> +	if (file_stream->pos.content_size == 0) {
> +		ret = EAGAIN;
> +	} else {
> +		ret = 0;
> +	}
> +
>  end:
>  	return ret;
>  }
> --
> 1.8.3.2
> 
> 

-- 
Mathieu Desnoyers
EfficiOS Inc.
http://www.efficios.com

[lttng-dev] [BABELTRACE PATCH 1/3] Handle the inactive streams

[Julien Desfossez]

When content_size == 0, just set the timestamp_end of the packet and
reinsert the stream in the heap.
This is required to handle the beacons of inactivity generated in
LTTng live trace streaming.

Signed-off-by: Julien Desfossez <jdesfossez at efficios.com>
---
 formats/ctf/ctf.c |  5 +++++
 lib/iterator.c    | 11 +++++++++++
 2 files changed, 16 insertions(+)

diff --git a/formats/ctf/ctf.c b/formats/ctf/ctf.c
index 794e428..5f263cd 100644
--- a/formats/ctf/ctf.c
+++ b/formats/ctf/ctf.c
@@ -471,6 +471,11 @@ int ctf_read_event(struct bt_stream_pos *ppos, struct ctf_stream_definition *str
 	 */
 	if (unlikely(pos->offset == EOF))
 		return EOF;
+
+	if (pos->content_size == 0) {
+		/* Stream is inactive for now (live reading). */
+		return EAGAIN;
+	}
 	assert(pos->offset < pos->content_size);
 
 	/* Read event header */
diff --git a/lib/iterator.c b/lib/iterator.c
index 009fcd3..155fcbe 100644
--- a/lib/iterator.c
+++ b/lib/iterator.c
@@ -65,6 +65,9 @@ static int stream_read_event(struct ctf_file_stream *sin)
 	ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
 	if (ret == EOF)
 		return EOF;
+	else if (ret == EAGAIN)
+		/* Stream is inactive for now (live reading). */
+		return EAGAIN;
 	else if (ret) {
 		fprintf(stderr, "[error] Reading event failed.\n");
 		return ret;
@@ -803,9 +806,17 @@ int bt_iter_next(struct bt_iter *iter)
 		assert(removed == file_stream);
 		ret = 0;
 		goto end;
+	} else if (ret == EAGAIN) {
+		/*
+		 * The stream is inactive for now, we just updated the timestamp_end
+		 * to skip over this stream up to a certain point in time.
+		 */
+		goto reinsert;
 	} else if (ret) {
 		goto end;
 	}
+
+reinsert:
 	/* Reinsert the file stream into the heap, and rebalance. */
 	removed = bt_heap_replace_max(iter->stream_heap, file_stream);
 	assert(removed == file_stream);
-- 
1.8.3.2