[babeltrace.git] / lib / iterator.c

/*
 * iterator.c
 *
 * Babeltrace Library
 *
 * Copyright 2010-2011 EfficiOS Inc. and Linux Foundation
 *
 * Author: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 */

#include <stdlib.h>
#include <babeltrace/babeltrace.h>
#include <babeltrace/callbacks-internal.h>
#include <babeltrace/context.h>
#include <babeltrace/context-internal.h>
#include <babeltrace/ctf/metadata.h>
#include <babeltrace/iterator-internal.h>
#include <babeltrace/iterator.h>
#include <babeltrace/prio_heap.h>
#include <inttypes.h>

struct stream_saved_pos {
	/*
	 * Use file_stream pointer to check if the trace collection we
	 * restore to match the one we saved from, for each stream.
	 */
	struct ctf_file_stream *file_stream;
	size_t cur_index;	/* current index in packet index */
	ssize_t offset;		/* offset from base, in bits. EOF for end of file. */
	uint64_t current_timestamp;
};

struct bt_saved_pos {
	struct trace_collection *tc;
	GArray *stream_saved_pos;	/* Contains struct stream_saved_pos */
};

static int stream_read_event(struct ctf_file_stream *sin)
{
	int ret;

	ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
	if (ret == EOF)
		return EOF;
	else if (ret) {
		fprintf(stderr, "[error] Reading event failed.\n");
		return ret;
	}
	return 0;
}

/*
 * returns true if a < b, false otherwise.
 */
int stream_compare(void *a, void *b)
{
	struct ctf_file_stream *s_a = a, *s_b = b;

	if (s_a->parent.timestamp < s_b->parent.timestamp)
		return 1;
	else
		return 0;
}

void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
{
	if (!iter_pos)
		return;

	if (iter_pos->u.restore) {
		if (iter_pos->u.restore->stream_saved_pos) {
			g_array_free(
				iter_pos->u.restore->stream_saved_pos,
				TRUE);
		}
		g_free(iter_pos->u.restore);
	}
	g_free(iter_pos);
}

/*
 * seek_file_stream_by_timestamp
 *
 * Browse a filestream by index, if an index contains the timestamp passed in
 * argument, seek inside the corresponding packet it until we find the event we
 * are looking for (either the exact timestamp or the event just after the
 * timestamp).
 *
 * Return 0 if the seek succeded and EOF if we didn't find any packet
 * containing the timestamp.
 */
static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
		uint64_t timestamp)
{
	struct ctf_stream_pos *stream_pos;
	struct packet_index *index;
	int i, ret;

	stream_pos = &cfs->pos;
	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_begin >= timestamp ||
				index->timestamp_end <= timestamp)
			continue;

		stream_pos->cur_index = i;
		stream_pos->move_pos_slow(stream_pos, index->offset, SEEK_SET);
		while (cfs->parent.timestamp < timestamp) {
			ret = stream_read_event(cfs);
			if (ret < 0)
				break;
		}
		return 0;
	}
	return EOF;
}

/*
 * seek_ctf_trace_by_timestamp : for each file stream, seek to the event with
 * the corresponding timestamp
 *
 * Return 0 on success.
 * If the timestamp is not part of any file stream, return EOF to inform the
 * user the timestamp is out of the scope
 */
static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin,
		uint64_t timestamp, struct ptr_heap *stream_heap)
{
	int i, j, ret;
	int found = EOF;

	/* for each stream_class */
	for (i = 0; i < tin->streams->len; i++) {
		struct ctf_stream_class *stream_class;

		stream_class = g_ptr_array_index(tin->streams, i);
		/* for each file_stream */
		for (j = 0; j < stream_class->streams->len; j++) {
			struct ctf_stream *stream;
			struct ctf_file_stream *cfs;

			stream = g_ptr_array_index(stream_class->streams, j);
			cfs = container_of(stream, struct ctf_file_stream,
					parent);
			ret = seek_file_stream_by_timestamp(cfs, timestamp);
			if (ret == 0) {
				/* Add to heap */
				ret = heap_insert(stream_heap, cfs);
				if (ret)
					goto error;
				found = 0;
			}
		}
	}

	return found;

error:
	return -2;
}

int bt_iter_set_pos(struct bt_iter *iter, const struct bt_iter_pos *iter_pos)
{
	struct trace_collection *tc;
	int i, ret;

	switch (iter_pos->type) {
	case BT_SEEK_RESTORE:
		if (!iter_pos->u.restore)
			goto error_arg;

		heap_free(iter->stream_heap);
		ret = heap_init(iter->stream_heap, 0, stream_compare);
		if (ret < 0)
			goto error;

		for (i = 0; i < iter_pos->u.restore->stream_saved_pos->len;
				i++) {
			struct stream_saved_pos *saved_pos;
			struct ctf_stream_pos *stream_pos;
			struct packet_index *index;
			struct ctf_stream *stream;

			saved_pos = &g_array_index(
					iter_pos->u.restore->stream_saved_pos,
					struct stream_saved_pos, i);
			stream = &saved_pos->file_stream->parent;
			stream_pos = &saved_pos->file_stream->pos;
			index = &g_array_index(stream_pos->packet_index,
					struct packet_index,
					saved_pos->cur_index);

			stream_pos->cur_index = saved_pos->cur_index;
			stream_pos->move_pos_slow(stream_pos, index->offset,
					SEEK_SET);

			/*
			 * the timestamp needs to be restored after
			 * move_pos_slow, because this function resets
			 * the timestamp to the beginning of the packet
			 */
			stream->timestamp = saved_pos->current_timestamp;
			stream_pos->offset = saved_pos->offset;
			stream_pos->last_offset = saved_pos->offset;

			stream->prev_timestamp = 0;
			stream->prev_timestamp_end = 0;
			stream->consumed = 0;

			printf_debug("restored to cur_index = %zd and "
				"offset = %zd, timestamp = %" PRIu64 "\n",
				stream_pos->cur_index,
				stream_pos->offset, stream->timestamp);

			stream_read_event(saved_pos->file_stream);

			/* Add to heap */
			ret = heap_insert(iter->stream_heap,
					saved_pos->file_stream);
			if (ret)
				goto error;
		}
	case BT_SEEK_TIME:
		tc = iter->ctx->tc;

		if (!iter_pos->u.seek_time)
			goto error_arg;

		heap_free(iter->stream_heap);
		ret = heap_init(iter->stream_heap, 0, stream_compare);
		if (ret < 0)
			goto error;

		/* for each trace in the trace_collection */
		for (i = 0; i < tc->array->len; i++) {
			struct ctf_trace *tin;
			struct trace_descriptor *td_read;

			td_read = g_ptr_array_index(tc->array, i);
			tin = container_of(td_read, struct ctf_trace, parent);

			ret = seek_ctf_trace_by_timestamp(tin,
					iter_pos->u.seek_time,
					iter->stream_heap);
			if (ret < 0)
				goto error;
		}
		return 0;
	default:
		/* not implemented */
		goto error_arg;
	}

	return 0;

error_arg:
	ret = -EINVAL;
error:
	heap_free(iter->stream_heap);
	if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
		heap_free(iter->stream_heap);
		g_free(iter->stream_heap);
		iter->stream_heap = NULL;
		ret = -ENOMEM;
	}
	return ret;
}

struct bt_iter_pos *bt_iter_get_pos(struct bt_iter *iter)
{
	struct bt_iter_pos *pos;
	struct trace_collection *tc = iter->ctx->tc;
	int i, stream_class_id, stream_id;

	pos = g_new0(struct bt_iter_pos, 1);
	pos->u.restore = g_new0(struct bt_saved_pos, 1);
	pos->u.restore->tc = tc;
	pos->u.restore->stream_saved_pos = g_array_new(FALSE, TRUE,
			sizeof(struct stream_saved_pos));
	if (!pos->u.restore->stream_saved_pos)
		goto error;

	for (i = 0; i < tc->array->len; i++) {
		struct ctf_trace *tin;
		struct trace_descriptor *td_read;

		td_read = g_ptr_array_index(tc->array, i);
		tin = container_of(td_read, struct ctf_trace, parent);

		for (stream_class_id = 0; stream_class_id < tin->streams->len;
				stream_class_id++) {
			struct ctf_stream_class *stream_class;

			stream_class = g_ptr_array_index(tin->streams,
					stream_class_id);
			for (stream_id = 0;
					stream_id < stream_class->streams->len;
					stream_id++) {
				struct ctf_stream *stream;
				struct ctf_file_stream *cfs;
				struct stream_saved_pos saved_pos;

				stream = g_ptr_array_index(
						stream_class->streams,
						stream_id);
				cfs = container_of(stream,
						struct ctf_file_stream,
						parent);

				saved_pos.file_stream = cfs;
				saved_pos.cur_index = cfs->pos.cur_index;

				/*
				 * It is possible that an event was read during
				 * the last restore, never consumed and its
				 * position saved again.  For this case, we
				 * need to check if the event really was
				 * consumed by the caller otherwise it is lost.
				 */
				if (stream->consumed)
					saved_pos.offset = cfs->pos.offset;
				else
					saved_pos.offset = cfs->pos.last_offset;

				saved_pos.current_timestamp = stream->timestamp;

				g_array_append_val(
					pos->u.restore->stream_saved_pos,
					saved_pos);

				printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
					"offset : %zd, "
					"timestamp = %" PRIu64 "\n",
					stream->stream_id, saved_pos.cur_index,
					saved_pos.offset,
					saved_pos.current_timestamp);
			}
		}
	}

	return pos;

error:
	return NULL;
}

struct bt_iter_pos *bt_iter_create_time_pos(struct bt_iter *iter,
		uint64_t timestamp)
{
	struct bt_iter_pos *pos;

	pos = g_new0(struct bt_iter_pos, 1);
	pos->type = BT_SEEK_TIME;
	pos->u.seek_time = timestamp;
	return pos;
}

/*
 * babeltrace_filestream_seek: seek a filestream to given position.
 *
 * The stream_id parameter is only useful for BT_SEEK_RESTORE.
 */
static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
		const struct bt_iter_pos *begin_pos,
		unsigned long stream_id)
{
	int ret = 0;

	switch (begin_pos->type) {
	case BT_SEEK_CUR:
		/*
		 * just insert into the heap we should already know
		 * the timestamps
		 */
		break;
	case BT_SEEK_BEGIN:
		file_stream->pos.move_pos_slow(&file_stream->pos, 0, SEEK_SET);
		ret = stream_read_event(file_stream);
		break;
	case BT_SEEK_TIME:
	case BT_SEEK_RESTORE:
	case BT_SEEK_END:
	default:
		assert(0); /* Not yet defined */
	}

	return ret;
}

/*
 * bt_iter_seek: seek iterator to given position.
 */
int bt_iter_seek(struct bt_iter *iter,
		const struct bt_iter_pos *begin_pos)
{
	int i, stream_id;
	int ret = 0;
	struct trace_collection *tc = iter->ctx->tc;

	for (i = 0; i < tc->array->len; i++) {
		struct ctf_trace *tin;
		struct trace_descriptor *td_read;

		td_read = g_ptr_array_index(tc->array, i);
		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_class *stream;
			int filenr;

			stream = g_ptr_array_index(tin->streams, stream_id);
			for (filenr = 0; filenr < stream->streams->len;
					filenr++) {
				struct ctf_file_stream *file_stream;

				file_stream = g_ptr_array_index(stream->streams,
						filenr);
				ret = babeltrace_filestream_seek(file_stream, begin_pos,
						stream_id);
				if (ret < 0)
					goto end;
			}
		}
	}
end:
	return ret;
}

struct bt_iter *bt_iter_create(struct bt_context *ctx,
		struct bt_iter_pos *begin_pos,
		struct bt_iter_pos *end_pos)
{
	int i, stream_id;
	int ret = 0;
	struct bt_iter *iter;

	iter = g_new0(struct bt_iter, 1);
	iter->stream_heap = g_new(struct ptr_heap, 1);
	iter->end_pos = end_pos;
	iter->callbacks = g_array_new(0, 1, sizeof(struct bt_stream_callbacks));
	iter->recalculate_dep_graph = 0;
	iter->main_callbacks.callback = NULL;
	iter->dep_gc = g_ptr_array_new();
	bt_context_get(ctx);
	iter->ctx = ctx;

	ret = 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;

		td_read = g_ptr_array_index(ctx->tc->array, i);
		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_class *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 (begin_pos) {
				    ret = babeltrace_filestream_seek(file_stream, begin_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;
			}
		}
	}

	return iter;

error:
	heap_free(iter->stream_heap);
error_heap_init:
	g_free(iter->stream_heap);
	g_free(iter);
	return NULL;
}

void bt_iter_destroy(struct bt_iter *iter)
{
	struct bt_stream_callbacks *bt_stream_cb;
	struct bt_callback_chain *bt_chain;
	int i, j;

	if (iter->stream_heap) {
		heap_free(iter->stream_heap);
		g_free(iter->stream_heap);
	}

	/* free all events callbacks */
	if (iter->main_callbacks.callback)
		g_array_free(iter->main_callbacks.callback, TRUE);

	/* free per-event callbacks */
	for (i = 0; i < iter->callbacks->len; i++) {
		bt_stream_cb = &g_array_index(iter->callbacks,
				struct bt_stream_callbacks, i);
		if (!bt_stream_cb || !bt_stream_cb->per_id_callbacks)
			continue;
		for (j = 0; j < bt_stream_cb->per_id_callbacks->len; j++) {
			bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
					struct bt_callback_chain, j);
			if (bt_chain->callback) {
				g_array_free(bt_chain->callback, TRUE);
			}
		}
		g_array_free(bt_stream_cb->per_id_callbacks, TRUE);
	}

	bt_context_put(iter->ctx);

	g_free(iter);
}

int bt_iter_next(struct bt_iter *iter)
{
	struct ctf_file_stream *file_stream, *removed;
	int ret;

	file_stream = heap_maximum(iter->stream_heap);
	if (!file_stream) {
		/* end of file for all streams */
		ret = 0;
		goto end;
	}

	ret = stream_read_event(file_stream);
	if (ret == EOF) {
		removed = heap_remove(iter->stream_heap);
		assert(removed == file_stream);
		ret = 0;
		goto end;
	} else if (ret) {
		goto end;
	}
	/* Reinsert the file stream into the heap, and rebalance. */
	removed = heap_replace_max(iter->stream_heap, file_stream);
	assert(removed == file_stream);

end:
	return ret;
}

int bt_iter_read_event(struct bt_iter *iter,
		struct ctf_stream **stream,
		struct ctf_stream_event **event)
{
	struct ctf_file_stream *file_stream;
	int ret = 0;

	file_stream = heap_maximum(iter->stream_heap);
	if (!file_stream) {
		/* end of file for all streams */
		ret = EOF;
		goto end;
	}
	*stream = &file_stream->parent;
	*event = g_ptr_array_index((*stream)->events_by_id, (*stream)->event_id);

	if ((*stream)->stream_id > iter->callbacks->len)
		goto end;

	process_callbacks(iter, *stream);

end:
	return ret;
}
Commit	Line	Data
6f3077a2 JD	1	/*
	2	* iterator.c
	3	*
	4	* Babeltrace Library
	5	*
	6	* Copyright 2010-2011 EfficiOS Inc. and Linux Foundation
	7	*
	8	* Author: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	9	*
	10	* Permission is hereby granted, free of charge, to any person obtaining a copy
	11	* of this software and associated documentation files (the "Software"), to deal
	12	* in the Software without restriction, including without limitation the rights
	13	* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
	14	* copies of the Software, and to permit persons to whom the Software is
	15	* furnished to do so, subject to the following conditions:
	16	*
	17	* The above copyright notice and this permission notice shall be included in
	18	* all copies or substantial portions of the Software.
	19	*/
	20
	21	#include <stdlib.h>
	22	#include <babeltrace/babeltrace.h>
	23	#include <babeltrace/callbacks-internal.h>
95d36295	24	#include <babeltrace/context.h>
08c22d05	25	#include <babeltrace/context-internal.h>
6f3077a2 JD	26	#include <babeltrace/ctf/metadata.h>
6f3077a2 JD	27	#include <babeltrace/iterator-internal.h>
6204d33c	28	#include <babeltrace/iterator.h>
6f3077a2	29	#include <babeltrace/prio_heap.h>
90fcbacc	30	#include <inttypes.h>
6f3077a2 JD	31
	32	struct stream_saved_pos {
	33	/*
	34	* Use file_stream pointer to check if the trace collection we
	35	* restore to match the one we saved from, for each stream.
	36	*/
	37	struct ctf_file_stream *file_stream;
	38	size_t cur_index; /* current index in packet index */
	39	ssize_t offset; /* offset from base, in bits. EOF for end of file. */
90fcbacc	40	uint64_t current_timestamp;
6f3077a2 JD	41	};
6f3077a2 JD	42
e8c92a62	43	struct bt_saved_pos {
6f3077a2 JD	44	struct trace_collection *tc;
	45	GArray stream_saved_pos; / Contains struct stream_saved_pos */
	46	};
	47
	48	static int stream_read_event(struct ctf_file_stream *sin)
	49	{
	50	int ret;
	51
	52	ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
	53	if (ret == EOF)
	54	return EOF;
	55	else if (ret) {
3394d22e	56	fprintf(stderr, "[error] Reading event failed.\n");
6f3077a2 JD	57	return ret;
	58	}
	59	return 0;
	60	}
	61
	62	/*
	63	* returns true if a < b, false otherwise.
	64	*/
	65	int stream_compare(void a, void b)
	66	{
	67	struct ctf_file_stream s_a = a, s_b = b;
	68
	69	if (s_a->parent.timestamp < s_b->parent.timestamp)
	70	return 1;
	71	else
	72	return 0;
	73	}
	74
90fcbacc JD	75	void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
90fcbacc JD	76	{
6cba487f MD	77	if (!iter_pos)
	78	return;
	79
	80	if (iter_pos->u.restore) {
	81	if (iter_pos->u.restore->stream_saved_pos) {
	82	g_array_free(
	83	iter_pos->u.restore->stream_saved_pos,
	84	TRUE);
90fcbacc	85	}
6cba487f	86	g_free(iter_pos->u.restore);
90fcbacc	87	}
6cba487f	88	g_free(iter_pos);
90fcbacc JD	89	}
90fcbacc JD	90
dc81689e JD	91	/*
	92	* seek_file_stream_by_timestamp
	93	*
	94	* Browse a filestream by index, if an index contains the timestamp passed in
	95	* argument, seek inside the corresponding packet it until we find the event we
	96	* are looking for (either the exact timestamp or the event just after the
	97	* timestamp).
	98	*
	99	* Return 0 if the seek succeded and EOF if we didn't find any packet
	100	* containing the timestamp.
	101	*/
	102	static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
	103	uint64_t timestamp)
	104	{
	105	struct ctf_stream_pos *stream_pos;
	106	struct packet_index *index;
	107	int i, ret;
	108
	109	stream_pos = &cfs->pos;
	110	for (i = 0; i < stream_pos->packet_index->len; i++) {
	111	index = &g_array_index(stream_pos->packet_index,
	112	struct packet_index, i);
	113	if (index->timestamp_begin >= timestamp \|\|
	114	index->timestamp_end <= timestamp)
	115	continue;
	116
	117	stream_pos->cur_index = i;
	118	stream_pos->move_pos_slow(stream_pos, index->offset, SEEK_SET);
	119	while (cfs->parent.timestamp < timestamp) {
	120	ret = stream_read_event(cfs);
	121	if (ret < 0)
	122	break;
	123	}
	124	return 0;
	125	}
	126	return EOF;
	127	}
	128
	129	/*
	130	* seek_ctf_trace_by_timestamp : for each file stream, seek to the event with
	131	* the corresponding timestamp
	132	*
	133	* Return 0 on success.
	134	* If the timestamp is not part of any file stream, return EOF to inform the
	135	* user the timestamp is out of the scope
	136	*/
	137	static int seek_ctf_trace_by_timestamp(struct ctf_trace *tin,
	138	uint64_t timestamp, struct ptr_heap *stream_heap)
	139	{
	140	int i, j, ret;
	141	int found = EOF;
	142
	143	/* for each stream_class */
	144	for (i = 0; i < tin->streams->len; i++) {
	145	struct ctf_stream_class *stream_class;
	146
	147	stream_class = g_ptr_array_index(tin->streams, i);
	148	/* for each file_stream */
	149	for (j = 0; j < stream_class->streams->len; j++) {
	150	struct ctf_stream *stream;
	151	struct ctf_file_stream *cfs;
	152
	153	stream = g_ptr_array_index(stream_class->streams, j);
	154	cfs = container_of(stream, struct ctf_file_stream,
155	parent);
156	ret = seek_file_stream_by_timestamp(cfs, timestamp);
157	if (ret == 0) {
158	/* Add to heap */
159	ret = heap_insert(stream_heap, cfs);
160	if (ret)
161	goto error;
162	found = 0;
163	}
164	}
165	}
166
167	return found;
168
169	error:
170	return -2;
171	}
172
90fcbacc JD	173	int bt_iter_set_pos(struct bt_iter iter, const struct bt_iter_pos iter_pos)
90fcbacc JD	174	{
dc81689e	175	struct trace_collection *tc;
90fcbacc JD	176	int i, ret;
90fcbacc JD	177
dc81689e JD	178	switch (iter_pos->type) {
	179	case BT_SEEK_RESTORE:
	180	if (!iter_pos->u.restore)
	181	goto error_arg;
	182
90fcbacc	183	heap_free(iter->stream_heap);
90fcbacc JD	184	ret = heap_init(iter->stream_heap, 0, stream_compare);
90fcbacc JD	185	if (ret < 0)
dc81689e	186	goto error;
90fcbacc JD	187
	188	for (i = 0; i < iter_pos->u.restore->stream_saved_pos->len;
	189	i++) {
	190	struct stream_saved_pos *saved_pos;
	191	struct ctf_stream_pos *stream_pos;
	192	struct packet_index *index;
	193	struct ctf_stream *stream;
	194
	195	saved_pos = &g_array_index(
	196	iter_pos->u.restore->stream_saved_pos,
	197	struct stream_saved_pos, i);
	198	stream = &saved_pos->file_stream->parent;
	199	stream_pos = &saved_pos->file_stream->pos;
	200	index = &g_array_index(stream_pos->packet_index,
	201	struct packet_index,
	202	saved_pos->cur_index);
	203
	204	stream_pos->cur_index = saved_pos->cur_index;
	205	stream_pos->move_pos_slow(stream_pos, index->offset,
	206	SEEK_SET);
	207
	208	/*
	209	* the timestamp needs to be restored after
	210	* move_pos_slow, because this function resets
	211	* the timestamp to the beginning of the packet
	212	*/
	213	stream->timestamp = saved_pos->current_timestamp;
	214	stream_pos->offset = saved_pos->offset;
	215	stream_pos->last_offset = saved_pos->offset;
	216
	217	stream->prev_timestamp = 0;
	218	stream->prev_timestamp_end = 0;
	219	stream->consumed = 0;
	220
	221	printf_debug("restored to cur_index = %zd and "
	222	"offset = %zd, timestamp = %" PRIu64 "\n",
	223	stream_pos->cur_index,
	224	stream_pos->offset, stream->timestamp);
	225
	226	stream_read_event(saved_pos->file_stream);
	227
	228	/* Add to heap */
	229	ret = heap_insert(iter->stream_heap,
	230	saved_pos->file_stream);
	231	if (ret)
	232	goto error;
	233	}
dc81689e JD	234	case BT_SEEK_TIME:
	235	tc = iter->ctx->tc;
	236
	237	if (!iter_pos->u.seek_time)
	238	goto error_arg;
	239
	240	heap_free(iter->stream_heap);
	241	ret = heap_init(iter->stream_heap, 0, stream_compare);
	242	if (ret < 0)
	243	goto error;
	244
	245	/* for each trace in the trace_collection */
	246	for (i = 0; i < tc->array->len; i++) {
	247	struct ctf_trace *tin;
	248	struct trace_descriptor *td_read;
	249
	250	td_read = g_ptr_array_index(tc->array, i);
	251	tin = container_of(td_read, struct ctf_trace, parent);
	252
	253	ret = seek_ctf_trace_by_timestamp(tin,
	254	iter_pos->u.seek_time,
	255	iter->stream_heap);
	256	if (ret < 0)
	257	goto error;
	258	}
	259	return 0;
	260	default:
	261	/* not implemented */
	262	goto error_arg;
90fcbacc JD	263	}
	264
	265	return 0;
dc81689e JD	266
	267	error_arg:
	268	ret = -EINVAL;
90fcbacc JD	269	error:
90fcbacc JD	270	heap_free(iter->stream_heap);
dc81689e JD	271	if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
	272	heap_free(iter->stream_heap);
	273	g_free(iter->stream_heap);
	274	iter->stream_heap = NULL;
	275	ret = -ENOMEM;
	276	}
	277	return ret;
90fcbacc JD	278	}
	279
	280	struct bt_iter_pos bt_iter_get_pos(struct bt_iter iter)
	281	{
	282	struct bt_iter_pos *pos;
	283	struct trace_collection *tc = iter->ctx->tc;
	284	int i, stream_class_id, stream_id;
	285
	286	pos = g_new0(struct bt_iter_pos, 1);
90fcbacc	287	pos->u.restore = g_new0(struct bt_saved_pos, 1);
90fcbacc JD	288	pos->u.restore->tc = tc;
	289	pos->u.restore->stream_saved_pos = g_array_new(FALSE, TRUE,
	290	sizeof(struct stream_saved_pos));
	291	if (!pos->u.restore->stream_saved_pos)
	292	goto error;
	293
	294	for (i = 0; i < tc->array->len; i++) {
	295	struct ctf_trace *tin;
	296	struct trace_descriptor *td_read;
	297
	298	td_read = g_ptr_array_index(tc->array, i);
	299	tin = container_of(td_read, struct ctf_trace, parent);
	300
	301	for (stream_class_id = 0; stream_class_id < tin->streams->len;
	302	stream_class_id++) {
	303	struct ctf_stream_class *stream_class;
	304
	305	stream_class = g_ptr_array_index(tin->streams,
	306	stream_class_id);
	307	for (stream_id = 0;
	308	stream_id < stream_class->streams->len;
	309	stream_id++) {
	310	struct ctf_stream *stream;
	311	struct ctf_file_stream *cfs;
	312	struct stream_saved_pos saved_pos;
	313
	314	stream = g_ptr_array_index(
	315	stream_class->streams,
	316	stream_id);
	317	cfs = container_of(stream,
	318	struct ctf_file_stream,
	319	parent);
	320
	321	saved_pos.file_stream = cfs;
	322	saved_pos.cur_index = cfs->pos.cur_index;
	323
	324	/*
	325	* It is possible that an event was read during
	326	* the last restore, never consumed and its
	327	* position saved again. For this case, we
	328	* need to check if the event really was
	329	* consumed by the caller otherwise it is lost.
	330	*/
	331	if (stream->consumed)
	332	saved_pos.offset = cfs->pos.offset;
	333	else
	334	saved_pos.offset = cfs->pos.last_offset;
	335
	336	saved_pos.current_timestamp = stream->timestamp;
	337
	338	g_array_append_val(
	339	pos->u.restore->stream_saved_pos,
	340	saved_pos);
	341
363402af	342	printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
90fcbacc JD	343	"offset : %zd, "
	344	"timestamp = %" PRIu64 "\n",
	345	stream->stream_id, saved_pos.cur_index,
	346	saved_pos.offset,
	347	saved_pos.current_timestamp);
	348	}
	349	}
	350	}
	351
	352	return pos;
	353
	354	error:
	355	return NULL;
	356	}
	357
dc81689e JD	358	struct bt_iter_pos bt_iter_create_time_pos(struct bt_iter iter,
	359	uint64_t timestamp)
	360	{
	361	struct bt_iter_pos *pos;
	362
	363	pos = g_new0(struct bt_iter_pos, 1);
	364	pos->type = BT_SEEK_TIME;
	365	pos->u.seek_time = timestamp;
	366	return pos;
	367	}
	368
6f3077a2 JD	369	/*
	370	* babeltrace_filestream_seek: seek a filestream to given position.
	371	*
	372	* The stream_id parameter is only useful for BT_SEEK_RESTORE.
	373	*/
	374	static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
e8c92a62	375	const struct bt_iter_pos *begin_pos,
6f3077a2 JD	376	unsigned long stream_id)
	377	{
	378	int ret = 0;
	379
	380	switch (begin_pos->type) {
	381	case BT_SEEK_CUR:
	382	/*
	383	* just insert into the heap we should already know
	384	* the timestamps
	385	*/
	386	break;
	387	case BT_SEEK_BEGIN:
	388	file_stream->pos.move_pos_slow(&file_stream->pos, 0, SEEK_SET);
	389	ret = stream_read_event(file_stream);
	390	break;
	391	case BT_SEEK_TIME:
	392	case BT_SEEK_RESTORE:
	393	case BT_SEEK_END:
	394	default:
	395	assert(0); /* Not yet defined */
	396	}
	397
	398	return ret;
	399	}
	400
	401	/*
e8c92a62	402	* bt_iter_seek: seek iterator to given position.
6f3077a2	403	*/
e8c92a62 JD	404	int bt_iter_seek(struct bt_iter *iter,
e8c92a62 JD	405	const struct bt_iter_pos *begin_pos)
6f3077a2 JD	406	{
	407	int i, stream_id;
	408	int ret = 0;
95d36295	409	struct trace_collection *tc = iter->ctx->tc;
6f3077a2 JD	410
	411	for (i = 0; i < tc->array->len; i++) {
	412	struct ctf_trace *tin;
	413	struct trace_descriptor *td_read;
	414
	415	td_read = g_ptr_array_index(tc->array, i);
	416	tin = container_of(td_read, struct ctf_trace, parent);
	417
	418	/* Populate heap with each stream */
	419	for (stream_id = 0; stream_id < tin->streams->len;
	420	stream_id++) {
	421	struct ctf_stream_class *stream;
	422	int filenr;
	423
	424	stream = g_ptr_array_index(tin->streams, stream_id);
	425	for (filenr = 0; filenr < stream->streams->len;
	426	filenr++) {
	427	struct ctf_file_stream *file_stream;
	428
	429	file_stream = g_ptr_array_index(stream->streams,
	430	filenr);
	431	ret = babeltrace_filestream_seek(file_stream, begin_pos,
	432	stream_id);
	433	if (ret < 0)
	434	goto end;
	435	}
	436	}
	437	}
	438	end:
	439	return ret;
	440	}
	441
e8c92a62 JD	442	struct bt_iter bt_iter_create(struct bt_context ctx,
	443	struct bt_iter_pos *begin_pos,
	444	struct bt_iter_pos *end_pos)
6f3077a2 JD	445	{
	446	int i, stream_id;
	447	int ret = 0;
e8c92a62	448	struct bt_iter *iter;
6f3077a2	449
90fcbacc	450	iter = g_new0(struct bt_iter, 1);
6f3077a2	451	iter->stream_heap = g_new(struct ptr_heap, 1);
6f3077a2 JD	452	iter->end_pos = end_pos;
	453	iter->callbacks = g_array_new(0, 1, sizeof(struct bt_stream_callbacks));
	454	iter->recalculate_dep_graph = 0;
	455	iter->main_callbacks.callback = NULL;
	456	iter->dep_gc = g_ptr_array_new();
6cba487f	457	bt_context_get(ctx);
95d36295	458	iter->ctx = ctx;
6f3077a2 JD	459
	460	ret = heap_init(iter->stream_heap, 0, stream_compare);
	461	if (ret < 0)
	462	goto error_heap_init;
	463
95d36295	464	for (i = 0; i < ctx->tc->array->len; i++) {
6f3077a2 JD	465	struct ctf_trace *tin;
	466	struct trace_descriptor *td_read;
	467
95d36295	468	td_read = g_ptr_array_index(ctx->tc->array, i);
6f3077a2 JD	469	tin = container_of(td_read, struct ctf_trace, parent);
	470
	471	/* Populate heap with each stream */
	472	for (stream_id = 0; stream_id < tin->streams->len;
	473	stream_id++) {
	474	struct ctf_stream_class *stream;
	475	int filenr;
	476
	477	stream = g_ptr_array_index(tin->streams, stream_id);
	478	if (!stream)
	479	continue;
	480	for (filenr = 0; filenr < stream->streams->len;
	481	filenr++) {
	482	struct ctf_file_stream *file_stream;
	483
	484	file_stream = g_ptr_array_index(stream->streams,
	485	filenr);
	486
	487	if (begin_pos) {
	488	ret = babeltrace_filestream_seek(file_stream, begin_pos,
	489	stream_id);
	490	if (ret == EOF) {
	491	ret = 0;
	492	continue;
	493	} else if (ret) {
	494	goto error;
	495	}
	496	}
	497	/* Add to heap */
	498	ret = heap_insert(iter->stream_heap, file_stream);
	499	if (ret)
	500	goto error;
	501	}
	502	}
	503	}
	504
	505	return iter;
	506
	507	error:
	508	heap_free(iter->stream_heap);
	509	error_heap_init:
	510	g_free(iter->stream_heap);
90fcbacc	511	g_free(iter);
6f3077a2 JD	512	return NULL;
	513	}
	514
e8c92a62	515	void bt_iter_destroy(struct bt_iter *iter)
6f3077a2 JD	516	{
	517	struct bt_stream_callbacks *bt_stream_cb;
	518	struct bt_callback_chain *bt_chain;
	519	int i, j;
	520
dc81689e JD	521	if (iter->stream_heap) {
	522	heap_free(iter->stream_heap);
	523	g_free(iter->stream_heap);
	524	}
6f3077a2 JD	525
	526	/* free all events callbacks */
	527	if (iter->main_callbacks.callback)
	528	g_array_free(iter->main_callbacks.callback, TRUE);
	529
	530	/* free per-event callbacks */
	531	for (i = 0; i < iter->callbacks->len; i++) {
	532	bt_stream_cb = &g_array_index(iter->callbacks,
	533	struct bt_stream_callbacks, i);
	534	if (!bt_stream_cb \|\| !bt_stream_cb->per_id_callbacks)
	535	continue;
	536	for (j = 0; j < bt_stream_cb->per_id_callbacks->len; j++) {
	537	bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
	538	struct bt_callback_chain, j);
	539	if (bt_chain->callback) {
	540	g_array_free(bt_chain->callback, TRUE);
	541	}
	542	}
	543	g_array_free(bt_stream_cb->per_id_callbacks, TRUE);
	544	}
	545
95d36295 JD	546	bt_context_put(iter->ctx);
95d36295 JD	547
90fcbacc	548	g_free(iter);
6f3077a2 JD	549	}
6f3077a2 JD	550
e8c92a62	551	int bt_iter_next(struct bt_iter *iter)
6f3077a2 JD	552	{
	553	struct ctf_file_stream file_stream, removed;
	554	int ret;
	555
	556	file_stream = heap_maximum(iter->stream_heap);
	557	if (!file_stream) {
	558	/* end of file for all streams */
	559	ret = 0;
	560	goto end;
	561	}
	562
	563	ret = stream_read_event(file_stream);
	564	if (ret == EOF) {
	565	removed = heap_remove(iter->stream_heap);
	566	assert(removed == file_stream);
	567	ret = 0;
	568	goto end;
	569	} else if (ret) {
	570	goto end;
	571	}
	572	/* Reinsert the file stream into the heap, and rebalance. */
	573	removed = heap_replace_max(iter->stream_heap, file_stream);
	574	assert(removed == file_stream);
	575
	576	end:
	577	return ret;
	578	}
	579
e8c92a62	580	int bt_iter_read_event(struct bt_iter *iter,
6f3077a2 JD	581	struct ctf_stream **stream,
	582	struct ctf_stream_event **event)
	583	{
	584	struct ctf_file_stream *file_stream;
	585	int ret = 0;
	586
	587	file_stream = heap_maximum(iter->stream_heap);
	588	if (!file_stream) {
	589	/* end of file for all streams */
	590	ret = EOF;
	591	goto end;
	592	}
	593	*stream = &file_stream->parent;
	594	event = g_ptr_array_index((stream)->events_by_id, (*stream)->event_id);
	595
	596	if ((*stream)->stream_id > iter->callbacks->len)
	597	goto end;
	598
	599	process_callbacks(iter, *stream);
	600
	601	end:
	602	return ret;
	603	}