X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=lib%2Fiterator.c;h=d2464ab7c4f8274e2fadee6e70a6f8a51f76dcc4;hp=f5f413e2ee6038105889c251047619bd152a6383;hb=2654fe9bab8f0eaeb17264ef7abadfd14e245b23;hpb=7f89ddce05f0a5c985d6333fbe70a992da4bd244

diff --git a/lib/iterator.c b/lib/iterator.c
index f5f413e2..d2464ab7 100644
--- a/lib/iterator.c
+++ b/lib/iterator.c
@@ -16,6 +16,14 @@
  *
  * The above copyright notice and this permission notice shall be included in
  * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
  */
 
 #include <stdlib.h>
@@ -57,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;
@@ -65,16 +76,24 @@ static int stream_read_event(struct ctf_file_stream *sin)
 }
 
 /*
- * returns true if a < b, false otherwise.
+ * Return true if a < b, false otherwise.
+ * If time stamps are exactly the same, compare by stream path. This
+ * ensures we get the same result between runs on the same trace
+ * collection on different environments.
+ * The result will be random for memory-mapped traces since there is no
+ * fixed path leading to those (they have empty path string).
  */
 static int stream_compare(void *a, void *b)
 {
 	struct ctf_file_stream *s_a = a, *s_b = b;
 
-	if (s_a->parent.real_timestamp < s_b->parent.real_timestamp)
+	if (s_a->parent.real_timestamp < s_b->parent.real_timestamp) {
 		return 1;
-	else
+	} else if (likely(s_a->parent.real_timestamp > s_b->parent.real_timestamp)) {
 		return 0;
+	} else {
+		return strcmp(s_a->parent.path, s_b->parent.path);
+	}
 }
 
 void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
@@ -103,6 +122,10 @@ void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
  *
  * Return 0 if the seek succeded, EOF if we didn't find any packet
  * containing the timestamp, or a positive integer for error.
+ *
+ * TODO: this should be turned into a binary search! It is currently
+ * doing a linear search in the packets. This is a O(n) operation on a
+ * very frequent code path.
  */
 static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
 		uint64_t timestamp)
@@ -112,10 +135,10 @@ static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
 	int i, ret;
 
 	stream_pos = &cfs->pos;
-	for (i = 0; i < stream_pos->packet_real_index->len; i++) {
-		index = &g_array_index(stream_pos->packet_real_index,
+	for (i = 0; i < stream_pos->packet_index->len; i++) {
+		index = &g_array_index(stream_pos->packet_index,
 				struct packet_index, i);
-		if (index->timestamp_end < timestamp)
+		if (index->ts_real.timestamp_end < timestamp)
 			continue;
 
 		stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
@@ -168,7 +191,7 @@ static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin,
 			ret = seek_file_stream_by_timestamp(cfs, timestamp);
 			if (ret == 0) {
 				/* Add to heap */
-				ret = heap_insert(stream_heap, cfs);
+				ret = bt_heap_insert(stream_heap, cfs);
 				if (ret) {
 					/* Return positive error. */
 					return -ret;
@@ -187,6 +210,152 @@ static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin,
 	return found ? 0 : EOF;
 }
 
+/*
+ * Find timestamp of last event in the stream.
+ *
+ * Return value: 0 if OK, positive error value on error, EOF if no
+ * events were found.
+ */
+static int find_max_timestamp_ctf_file_stream(struct ctf_file_stream *cfs,
+		uint64_t *timestamp_end)
+{
+	int ret, count = 0, i;
+	uint64_t timestamp = 0;
+	struct ctf_stream_pos *stream_pos;
+
+	stream_pos = &cfs->pos;
+	/*
+	 * We start by the last packet, and iterate backwards until we
+	 * either find at least one event, or we reach the first packet
+	 * (some packets can be empty).
+	 */
+	for (i = stream_pos->packet_index->len - 1; i >= 0; i--) {
+		stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
+		count = 0;
+		/* read each event until we reach the end of the stream */
+		do {
+			ret = stream_read_event(cfs);
+			if (ret == 0) {
+				count++;
+				timestamp = cfs->parent.real_timestamp;
+			}
+		} while (ret == 0);
+
+		/* Error */
+		if (ret > 0)
+			goto end;
+		assert(ret == EOF);
+		if (count)
+			break;
+	}
+
+	if (count) {
+		*timestamp_end = timestamp;
+		ret = 0;
+	} else {
+		/* Return EOF if no events were found */
+		ret = EOF;
+	}
+end:
+	return ret;
+}
+
+/*
+ * Find the stream within a stream class that contains the event with
+ * the largest timestamp, and save that timestamp.
+ *
+ * Return 0 if OK, EOF if no events were found in the streams, or
+ * positive value on error.
+ */
+static int find_max_timestamp_ctf_stream_class(
+		struct ctf_stream_declaration *stream_class,
+		struct ctf_file_stream **cfsp,
+		uint64_t *max_timestamp)
+{
+	int ret = EOF, i, found = 0;
+
+	for (i = 0; i < stream_class->streams->len; i++) {
+		struct ctf_stream_definition *stream;
+		struct ctf_file_stream *cfs;
+		uint64_t current_max_ts = 0;
+
+		stream = g_ptr_array_index(stream_class->streams, i);
+		if (!stream)
+			continue;
+		cfs = container_of(stream, struct ctf_file_stream, parent);
+		ret = find_max_timestamp_ctf_file_stream(cfs, &current_max_ts);
+		if (ret == EOF)
+			continue;
+		if (ret != 0)
+			break;
+		if (current_max_ts >= *max_timestamp) {
+			*max_timestamp = current_max_ts;
+			*cfsp = cfs;
+			found = 1;
+		}
+	}
+	assert(ret >= 0 || ret == EOF);
+	if (found) {
+		return 0;
+	}
+	return ret;
+}
+
+/*
+ * seek_last_ctf_trace_collection: seek trace collection to last event.
+ *
+ * Return 0 if OK, EOF if no events were found, or positive error value
+ * on error.
+ */
+static int seek_last_ctf_trace_collection(struct trace_collection *tc,
+		struct ctf_file_stream **cfsp)
+{
+	int i, j, ret;
+	int found = 0;
+	uint64_t max_timestamp = 0;
+
+	if (!tc)
+		return 1;
+
+	/* For each trace in the trace_collection */
+	for (i = 0; i < tc->array->len; i++) {
+		struct ctf_trace *tin;
+		struct bt_trace_descriptor *td_read;
+
+		td_read = g_ptr_array_index(tc->array, i);
+		if (!td_read)
+			continue;
+		tin = container_of(td_read, struct ctf_trace, parent);
+		/* For each stream_class in the trace */
+		for (j = 0; j < tin->streams->len; j++) {
+			struct ctf_stream_declaration *stream_class;
+
+			stream_class = g_ptr_array_index(tin->streams, j);
+			if (!stream_class)
+				continue;
+			ret = find_max_timestamp_ctf_stream_class(stream_class,
+					cfsp, &max_timestamp);
+			if (ret > 0)
+				goto end;
+			if (ret == 0)
+				found = 1;
+			assert(ret == EOF || ret == 0);
+		}
+	}
+	/*
+	 * Now we know in which file stream the last event is located,
+	 * and we know its timestamp.
+	 */
+	if (!found) {
+		ret = EOF;
+	} else {
+		ret = seek_file_stream_by_timestamp(*cfsp, max_timestamp);
+		assert(ret == 0);
+	}
+end:
+	return ret;
+}
+
 int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 {
 	struct trace_collection *tc;
@@ -200,8 +369,8 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 		if (!iter_pos->u.restore)
 			return -EINVAL;
 
-		heap_free(iter->stream_heap);
-		ret = heap_init(iter->stream_heap, 0, stream_compare);
+		bt_heap_free(iter->stream_heap);
+		ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 		if (ret < 0)
 			goto error_heap_init;
 
@@ -230,20 +399,28 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 			stream_pos->offset = saved_pos->offset;
 			stream_pos->last_offset = LAST_OFFSET_POISON;
 
-			stream->prev_real_timestamp = 0;
-			stream->prev_real_timestamp_end = 0;
-			stream->prev_cycles_timestamp = 0;
-			stream->prev_cycles_timestamp_end = 0;
+			stream->current.real.begin = 0;
+			stream->current.real.end = 0;
+			stream->current.cycles.begin = 0;
+			stream->current.cycles.end = 0;
+
+			stream->prev.real.begin = 0;
+			stream->prev.real.end = 0;
+			stream->prev.cycles.begin = 0;
+			stream->prev.cycles.end = 0;
 
-			printf_debug("restored to cur_index = %zd and "
-				"offset = %zd, timestamp = %" PRIu64 "\n",
+			printf_debug("restored to cur_index = %" PRId64 " and "
+				"offset = %" PRId64 ", timestamp = %" PRIu64 "\n",
 				stream_pos->cur_index,
 				stream_pos->offset, stream->real_timestamp);
 
-			stream_read_event(saved_pos->file_stream);
+			ret = stream_read_event(saved_pos->file_stream);
+			if (ret != 0) {
+				goto error;
+			}
 
 			/* Add to heap */
-			ret = heap_insert(iter->stream_heap,
+			ret = bt_heap_insert(iter->stream_heap,
 					saved_pos->file_stream);
 			if (ret)
 				goto error;
@@ -252,15 +429,15 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 	case BT_SEEK_TIME:
 		tc = iter->ctx->tc;
 
-		heap_free(iter->stream_heap);
-		ret = heap_init(iter->stream_heap, 0, stream_compare);
+		bt_heap_free(iter->stream_heap);
+		ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 		if (ret < 0)
 			goto error_heap_init;
 
 		/* for each trace in the trace_collection */
 		for (i = 0; i < tc->array->len; i++) {
 			struct ctf_trace *tin;
-			struct trace_descriptor *td_read;
+			struct bt_trace_descriptor *td_read;
 
 			td_read = g_ptr_array_index(tc->array, i);
 			if (!td_read)
@@ -283,14 +460,14 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 		return 0;
 	case BT_SEEK_BEGIN:
 		tc = iter->ctx->tc;
-		heap_free(iter->stream_heap);
-		ret = heap_init(iter->stream_heap, 0, stream_compare);
+		bt_heap_free(iter->stream_heap);
+		ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 		if (ret < 0)
 			goto error_heap_init;
 
 		for (i = 0; i < tc->array->len; i++) {
 			struct ctf_trace *tin;
-			struct trace_descriptor *td_read;
+			struct bt_trace_descriptor *td_read;
 			int stream_id;
 
 			td_read = g_ptr_array_index(tc->array, i);
@@ -322,13 +499,37 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 					if (ret != 0 && ret != EOF) {
 						goto error;
 					}
-					ret = heap_insert(iter->stream_heap, file_stream);
+					if (ret == EOF) {
+						/* Do not add EOF streams */
+						continue;
+					}
+					ret = bt_heap_insert(iter->stream_heap, file_stream);
 					if (ret)
 						goto error;
 				}
 			}
 		}
 		break;
+	case BT_SEEK_LAST:
+	{
+		struct ctf_file_stream *cfs = NULL;
+
+		tc = iter->ctx->tc;
+		ret = seek_last_ctf_trace_collection(tc, &cfs);
+		if (ret != 0 || !cfs)
+			goto error;
+		/* remove all streams from the heap */
+		bt_heap_free(iter->stream_heap);
+		/* Create a new empty heap */
+		ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
+		if (ret < 0)
+			goto error;
+		/* Insert the stream that contains the last event */
+		ret = bt_heap_insert(iter->stream_heap, cfs);
+		if (ret)
+			goto error;
+		break;
+	}
 	default:
 		/* not implemented */
 		return -EINVAL;
@@ -337,10 +538,10 @@ int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
 	return 0;
 
 error:
-	heap_free(iter->stream_heap);
+	bt_heap_free(iter->stream_heap);
 error_heap_init:
-	if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
-		heap_free(iter->stream_heap);
+	if (bt_heap_init(iter->stream_heap, 0, stream_compare) < 0) {
+		bt_heap_free(iter->stream_heap);
 		g_free(iter->stream_heap);
 		iter->stream_heap = NULL;
 		ret = -ENOMEM;
@@ -370,12 +571,12 @@ struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
 	if (!pos->u.restore->stream_saved_pos)
 		goto error;
 
-	ret = heap_copy(&iter_heap_copy, iter->stream_heap);
+	ret = bt_heap_copy(&iter_heap_copy, iter->stream_heap);
 	if (ret < 0)
 		goto error_heap;
 
 	/* iterate over each stream in the heap */
-	file_stream = heap_maximum(&iter_heap_copy);
+	file_stream = bt_heap_maximum(&iter_heap_copy);
 	while (file_stream != NULL) {
 		struct stream_saved_pos saved_pos;
 
@@ -399,12 +600,12 @@ struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
 				saved_pos.current_real_timestamp);
 
 		/* remove the stream from the heap copy */
-		removed = heap_remove(&iter_heap_copy);
+		removed = bt_heap_remove(&iter_heap_copy);
 		assert(removed == file_stream);
 
-		file_stream = heap_maximum(&iter_heap_copy);
+		file_stream = bt_heap_maximum(&iter_heap_copy);
 	}
-	heap_free(&iter_heap_copy);
+	bt_heap_free(&iter_heap_copy);
 	return pos;
 
 error_heap:
@@ -456,7 +657,6 @@ static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
 		break;
 	case BT_SEEK_TIME:
 	case BT_SEEK_RESTORE:
-	case BT_SEEK_END:
 	default:
 		assert(0); /* Not yet defined */
 	}
@@ -464,12 +664,60 @@ static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
 	return ret;
 }
 
+int bt_iter_add_trace(struct bt_iter *iter,
+		struct bt_trace_descriptor *td_read)
+{
+	struct ctf_trace *tin;
+	int stream_id, ret = 0;
+
+	tin = container_of(td_read, struct ctf_trace, parent);
+
+	/* Populate heap with each stream */
+	for (stream_id = 0; stream_id < tin->streams->len;
+			stream_id++) {
+		struct ctf_stream_declaration *stream;
+		int filenr;
+
+		stream = g_ptr_array_index(tin->streams, stream_id);
+		if (!stream)
+			continue;
+		for (filenr = 0; filenr < stream->streams->len;
+				filenr++) {
+			struct ctf_file_stream *file_stream;
+			struct bt_iter_pos pos;
+
+			file_stream = g_ptr_array_index(stream->streams,
+					filenr);
+			if (!file_stream)
+				continue;
+
+			pos.type = BT_SEEK_BEGIN;
+			ret = babeltrace_filestream_seek(file_stream,
+					&pos, stream_id);
+
+			if (ret == EOF) {
+				ret = 0;
+				continue;
+			} else if (ret != 0 && ret != EAGAIN) {
+				goto error;
+			}
+			/* Add to heap */
+			ret = bt_heap_insert(iter->stream_heap, file_stream);
+			if (ret)
+				goto error;
+		}
+	}
+
+error:
+	return ret;
+}
+
 int bt_iter_init(struct bt_iter *iter,
 		struct bt_context *ctx,
 		const struct bt_iter_pos *begin_pos,
 		const struct bt_iter_pos *end_pos)
 {
-	int i, stream_id;
+	int i;
 	int ret = 0;
 
 	if (!iter || !ctx)
@@ -485,67 +733,30 @@ int bt_iter_init(struct bt_iter *iter,
 	bt_context_get(ctx);
 	iter->ctx = ctx;
 
-	ret = heap_init(iter->stream_heap, 0, stream_compare);
+	ret = bt_heap_init(iter->stream_heap, 0, stream_compare);
 	if (ret < 0)
 		goto error_heap_init;
 
 	for (i = 0; i < ctx->tc->array->len; i++) {
-		struct ctf_trace *tin;
-		struct trace_descriptor *td_read;
+		struct bt_trace_descriptor *td_read;
 
 		td_read = g_ptr_array_index(ctx->tc->array, i);
 		if (!td_read)
 			continue;
-		tin = container_of(td_read, struct ctf_trace, parent);
-
-		/* Populate heap with each stream */
-		for (stream_id = 0; stream_id < tin->streams->len;
-				stream_id++) {
-			struct ctf_stream_declaration *stream;
-			int filenr;
-
-			stream = g_ptr_array_index(tin->streams, stream_id);
-			if (!stream)
-				continue;
-			for (filenr = 0; filenr < stream->streams->len;
-					filenr++) {
-				struct ctf_file_stream *file_stream;
-
-				file_stream = g_ptr_array_index(stream->streams,
-						filenr);
-				if (!file_stream)
-					continue;
-				if (begin_pos) {
-					ret = babeltrace_filestream_seek(
-							file_stream,
-							begin_pos,
-							stream_id);
-				} else {
-					struct bt_iter_pos pos;
-					pos.type = BT_SEEK_BEGIN;
-					ret = babeltrace_filestream_seek(
-							file_stream, &pos,
-							stream_id);
-				}
-				if (ret == EOF) {
-					ret = 0;
-					continue;
-				} else if (ret) {
-					goto error;
-				}
-				/* Add to heap */
-				ret = heap_insert(iter->stream_heap, file_stream);
-				if (ret)
-					goto error;
-			}
-		}
+		ret = bt_iter_add_trace(iter, td_read);
+		if (ret < 0)
+			goto error;
 	}
 
 	ctx->current_iterator = iter;
-	return 0;
+	if (begin_pos && begin_pos->type != BT_SEEK_BEGIN) {
+		ret = bt_iter_set_pos(iter, begin_pos);
+	}
+
+	return ret;
 
 error:
-	heap_free(iter->stream_heap);
+	bt_heap_free(iter->stream_heap);
 error_heap_init:
 	g_free(iter->stream_heap);
 	iter->stream_heap = NULL;
@@ -576,7 +787,7 @@ void bt_iter_fini(struct bt_iter *iter)
 {
 	assert(iter);
 	if (iter->stream_heap) {
-		heap_free(iter->stream_heap);
+		bt_heap_free(iter->stream_heap);
 		g_free(iter->stream_heap);
 	}
 	iter->ctx->current_iterator = NULL;
@@ -598,7 +809,7 @@ int bt_iter_next(struct bt_iter *iter)
 	if (!iter)
 		return -EINVAL;
 
-	file_stream = heap_maximum(iter->stream_heap);
+	file_stream = bt_heap_maximum(iter->stream_heap);
 	if (!file_stream) {
 		/* end of file for all streams */
 		ret = 0;
@@ -607,17 +818,33 @@ int bt_iter_next(struct bt_iter *iter)
 
 	ret = stream_read_event(file_stream);
 	if (ret == EOF) {
-		removed = heap_remove(iter->stream_heap);
+		removed = bt_heap_remove(iter->stream_heap);
 		assert(removed == file_stream);
 		ret = 0;
 		goto end;
+	} else if (ret == EAGAIN) {
+		/*
+		 * Live streaming: the stream is inactive for now, we
+		 * just updated the timestamp_end to skip over this
+		 * stream up to a certain point in time.
+		 *
+		 * Since we can't guarantee that a stream will ever have
+		 * any activity, we can't rely on the fact that
+		 * bt_iter_next will be called for each stream and deal
+		 * with inactive streams. So instead, we return 0 here
+		 * to the caller and let the read API handle the
+		 * retry case.
+		 */
+		ret = 0;
+		goto reinsert;
 	} else if (ret) {
 		goto end;
 	}
+
+reinsert:
 	/* Reinsert the file stream into the heap, and rebalance. */
-	removed = heap_replace_max(iter->stream_heap, file_stream);
+	removed = bt_heap_replace_max(iter->stream_heap, file_stream);
 	assert(removed == file_stream);
-
 end:
 	return ret;
 }