[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 <babeltrace/ctf/events.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->packet_seek(&stream_pos->parent, i, 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 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;

			stream_pos->packet_seek(&stream_pos->parent,
					saved_pos->cur_index, SEEK_SET);

			/*
			 * the timestamp needs to be restored after
			 * packet_seek, 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.packet_seek(&file_stream->pos.parent,
				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;
}

struct bt_ctf_event *bt_iter_read_ctf_event(struct bt_iter *iter)
{
	struct ctf_file_stream *file_stream;
	struct bt_ctf_event *ret = &iter->current_ctf_event;

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

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

	process_callbacks(iter, ret->stream);

end:
	return ret;
stop:
	return NULL;
}
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>
9843982d	30	#include <babeltrace/ctf/events.h>
90fcbacc	31	#include <inttypes.h>
6f3077a2 JD	32
	33	struct stream_saved_pos {
	34	/*
	35	* Use file_stream pointer to check if the trace collection we
	36	* restore to match the one we saved from, for each stream.
	37	*/
	38	struct ctf_file_stream *file_stream;
	39	size_t cur_index; /* current index in packet index */
	40	ssize_t offset; /* offset from base, in bits. EOF for end of file. */
90fcbacc	41	uint64_t current_timestamp;
6f3077a2 JD	42	};
6f3077a2 JD	43
e8c92a62	44	struct bt_saved_pos {
6f3077a2 JD	45	struct trace_collection *tc;
	46	GArray stream_saved_pos; / Contains struct stream_saved_pos */
	47	};
	48
	49	static int stream_read_event(struct ctf_file_stream *sin)
	50	{
	51	int ret;
	52
	53	ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
	54	if (ret == EOF)
	55	return EOF;
	56	else if (ret) {
3394d22e	57	fprintf(stderr, "[error] Reading event failed.\n");
6f3077a2 JD	58	return ret;
	59	}
	60	return 0;
	61	}
	62
	63	/*
	64	* returns true if a < b, false otherwise.
	65	*/
	66	int stream_compare(void a, void b)
	67	{
	68	struct ctf_file_stream s_a = a, s_b = b;
	69
	70	if (s_a->parent.timestamp < s_b->parent.timestamp)
	71	return 1;
	72	else
	73	return 0;
	74	}
	75
90fcbacc JD	76	void bt_iter_free_pos(struct bt_iter_pos *iter_pos)
90fcbacc JD	77	{
6cba487f MD	78	if (!iter_pos)
	79	return;
	80
	81	if (iter_pos->u.restore) {
	82	if (iter_pos->u.restore->stream_saved_pos) {
	83	g_array_free(
	84	iter_pos->u.restore->stream_saved_pos,
	85	TRUE);
90fcbacc	86	}
6cba487f	87	g_free(iter_pos->u.restore);
90fcbacc	88	}
6cba487f	89	g_free(iter_pos);
90fcbacc JD	90	}
90fcbacc JD	91
dc81689e JD	92	/*
	93	* seek_file_stream_by_timestamp
	94	*
	95	* Browse a filestream by index, if an index contains the timestamp passed in
	96	* argument, seek inside the corresponding packet it until we find the event we
	97	* are looking for (either the exact timestamp or the event just after the
	98	* timestamp).
	99	*
	100	* Return 0 if the seek succeded and EOF if we didn't find any packet
	101	* containing the timestamp.
	102	*/
	103	static int seek_file_stream_by_timestamp(struct ctf_file_stream *cfs,
	104	uint64_t timestamp)
	105	{
	106	struct ctf_stream_pos *stream_pos;
	107	struct packet_index *index;
	108	int i, ret;
	109
	110	stream_pos = &cfs->pos;
	111	for (i = 0; i < stream_pos->packet_index->len; i++) {
	112	index = &g_array_index(stream_pos->packet_index,
	113	struct packet_index, i);
	114	if (index->timestamp_begin >= timestamp \|\|
	115	index->timestamp_end <= timestamp)
	116	continue;
	117
20d0dcf9	118	stream_pos->packet_seek(&stream_pos->parent, i, SEEK_SET);
dc81689e JD	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;
90fcbacc JD	192	struct ctf_stream *stream;
	193
	194	saved_pos = &g_array_index(
	195	iter_pos->u.restore->stream_saved_pos,
	196	struct stream_saved_pos, i);
	197	stream = &saved_pos->file_stream->parent;
	198	stream_pos = &saved_pos->file_stream->pos;
90fcbacc	199
06789ffd	200	stream_pos->packet_seek(&stream_pos->parent,
20d0dcf9	201	saved_pos->cur_index, SEEK_SET);
90fcbacc JD	202
	203	/*
	204	* the timestamp needs to be restored after
06789ffd	205	* packet_seek, because this function resets
90fcbacc JD	206	* the timestamp to the beginning of the packet
	207	*/
	208	stream->timestamp = saved_pos->current_timestamp;
	209	stream_pos->offset = saved_pos->offset;
	210	stream_pos->last_offset = saved_pos->offset;
	211
	212	stream->prev_timestamp = 0;
	213	stream->prev_timestamp_end = 0;
	214	stream->consumed = 0;
	215
	216	printf_debug("restored to cur_index = %zd and "
	217	"offset = %zd, timestamp = %" PRIu64 "\n",
	218	stream_pos->cur_index,
	219	stream_pos->offset, stream->timestamp);
	220
	221	stream_read_event(saved_pos->file_stream);
	222
	223	/* Add to heap */
	224	ret = heap_insert(iter->stream_heap,
	225	saved_pos->file_stream);
	226	if (ret)
	227	goto error;
	228	}
dc81689e JD	229	case BT_SEEK_TIME:
	230	tc = iter->ctx->tc;
	231
	232	if (!iter_pos->u.seek_time)
	233	goto error_arg;
	234
	235	heap_free(iter->stream_heap);
	236	ret = heap_init(iter->stream_heap, 0, stream_compare);
	237	if (ret < 0)
	238	goto error;
	239
	240	/* for each trace in the trace_collection */
	241	for (i = 0; i < tc->array->len; i++) {
	242	struct ctf_trace *tin;
	243	struct trace_descriptor *td_read;
	244
	245	td_read = g_ptr_array_index(tc->array, i);
	246	tin = container_of(td_read, struct ctf_trace, parent);
	247
	248	ret = seek_ctf_trace_by_timestamp(tin,
	249	iter_pos->u.seek_time,
	250	iter->stream_heap);
	251	if (ret < 0)
	252	goto error;
	253	}
	254	return 0;
	255	default:
	256	/* not implemented */
	257	goto error_arg;
90fcbacc JD	258	}
	259
	260	return 0;
dc81689e JD	261
	262	error_arg:
	263	ret = -EINVAL;
90fcbacc JD	264	error:
90fcbacc JD	265	heap_free(iter->stream_heap);
dc81689e JD	266	if (heap_init(iter->stream_heap, 0, stream_compare) < 0) {
	267	heap_free(iter->stream_heap);
	268	g_free(iter->stream_heap);
	269	iter->stream_heap = NULL;
	270	ret = -ENOMEM;
	271	}
	272	return ret;
90fcbacc JD	273	}
	274
	275	struct bt_iter_pos bt_iter_get_pos(struct bt_iter iter)
	276	{
	277	struct bt_iter_pos *pos;
	278	struct trace_collection *tc = iter->ctx->tc;
	279	int i, stream_class_id, stream_id;
	280
	281	pos = g_new0(struct bt_iter_pos, 1);
90fcbacc	282	pos->u.restore = g_new0(struct bt_saved_pos, 1);
90fcbacc JD	283	pos->u.restore->tc = tc;
	284	pos->u.restore->stream_saved_pos = g_array_new(FALSE, TRUE,
	285	sizeof(struct stream_saved_pos));
	286	if (!pos->u.restore->stream_saved_pos)
	287	goto error;
	288
	289	for (i = 0; i < tc->array->len; i++) {
	290	struct ctf_trace *tin;
	291	struct trace_descriptor *td_read;
	292
	293	td_read = g_ptr_array_index(tc->array, i);
	294	tin = container_of(td_read, struct ctf_trace, parent);
	295
	296	for (stream_class_id = 0; stream_class_id < tin->streams->len;
	297	stream_class_id++) {
	298	struct ctf_stream_class *stream_class;
	299
	300	stream_class = g_ptr_array_index(tin->streams,
	301	stream_class_id);
	302	for (stream_id = 0;
	303	stream_id < stream_class->streams->len;
	304	stream_id++) {
	305	struct ctf_stream *stream;
	306	struct ctf_file_stream *cfs;
	307	struct stream_saved_pos saved_pos;
	308
	309	stream = g_ptr_array_index(
	310	stream_class->streams,
	311	stream_id);
	312	cfs = container_of(stream,
	313	struct ctf_file_stream,
	314	parent);
	315
	316	saved_pos.file_stream = cfs;
	317	saved_pos.cur_index = cfs->pos.cur_index;
	318
	319	/*
	320	* It is possible that an event was read during
	321	* the last restore, never consumed and its
	322	* position saved again. For this case, we
	323	* need to check if the event really was
	324	* consumed by the caller otherwise it is lost.
	325	*/
	326	if (stream->consumed)
	327	saved_pos.offset = cfs->pos.offset;
	328	else
	329	saved_pos.offset = cfs->pos.last_offset;
	330
	331	saved_pos.current_timestamp = stream->timestamp;
	332
	333	g_array_append_val(
	334	pos->u.restore->stream_saved_pos,
	335	saved_pos);
	336
363402af	337	printf_debug("stream : %" PRIu64 ", cur_index : %zd, "
90fcbacc JD	338	"offset : %zd, "
	339	"timestamp = %" PRIu64 "\n",
	340	stream->stream_id, saved_pos.cur_index,
	341	saved_pos.offset,
	342	saved_pos.current_timestamp);
	343	}
	344	}
	345	}
	346
	347	return pos;
	348
	349	error:
	350	return NULL;
	351	}
	352
dc81689e JD	353	struct bt_iter_pos bt_iter_create_time_pos(struct bt_iter iter,
	354	uint64_t timestamp)
	355	{
	356	struct bt_iter_pos *pos;
	357
	358	pos = g_new0(struct bt_iter_pos, 1);
	359	pos->type = BT_SEEK_TIME;
	360	pos->u.seek_time = timestamp;
	361	return pos;
	362	}
	363
6f3077a2 JD	364	/*
	365	* babeltrace_filestream_seek: seek a filestream to given position.
	366	*
	367	* The stream_id parameter is only useful for BT_SEEK_RESTORE.
	368	*/
	369	static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
e8c92a62	370	const struct bt_iter_pos *begin_pos,
6f3077a2 JD	371	unsigned long stream_id)
	372	{
	373	int ret = 0;
	374
	375	switch (begin_pos->type) {
	376	case BT_SEEK_CUR:
	377	/*
	378	* just insert into the heap we should already know
	379	* the timestamps
	380	*/
	381	break;
	382	case BT_SEEK_BEGIN:
06789ffd	383	file_stream->pos.packet_seek(&file_stream->pos.parent,
d6425aaf	384	0, SEEK_SET);
6f3077a2 JD	385	ret = stream_read_event(file_stream);
	386	break;
	387	case BT_SEEK_TIME:
	388	case BT_SEEK_RESTORE:
	389	case BT_SEEK_END:
	390	default:
	391	assert(0); /* Not yet defined */
	392	}
	393
	394	return ret;
	395	}
	396
	397	/*
e8c92a62	398	* bt_iter_seek: seek iterator to given position.
6f3077a2	399	*/
e8c92a62 JD	400	int bt_iter_seek(struct bt_iter *iter,
e8c92a62 JD	401	const struct bt_iter_pos *begin_pos)
6f3077a2 JD	402	{
	403	int i, stream_id;
	404	int ret = 0;
95d36295	405	struct trace_collection *tc = iter->ctx->tc;
6f3077a2 JD	406
	407	for (i = 0; i < tc->array->len; i++) {
	408	struct ctf_trace *tin;
	409	struct trace_descriptor *td_read;
	410
	411	td_read = g_ptr_array_index(tc->array, i);
	412	tin = container_of(td_read, struct ctf_trace, parent);
	413
	414	/* Populate heap with each stream */
	415	for (stream_id = 0; stream_id < tin->streams->len;
	416	stream_id++) {
	417	struct ctf_stream_class *stream;
	418	int filenr;
	419
	420	stream = g_ptr_array_index(tin->streams, stream_id);
	421	for (filenr = 0; filenr < stream->streams->len;
	422	filenr++) {
	423	struct ctf_file_stream *file_stream;
	424
	425	file_stream = g_ptr_array_index(stream->streams,
	426	filenr);
	427	ret = babeltrace_filestream_seek(file_stream, begin_pos,
	428	stream_id);
	429	if (ret < 0)
	430	goto end;
	431	}
	432	}
	433	}
	434	end:
	435	return ret;
	436	}
	437
e8c92a62 JD	438	struct bt_iter bt_iter_create(struct bt_context ctx,
	439	struct bt_iter_pos *begin_pos,
	440	struct bt_iter_pos *end_pos)
6f3077a2 JD	441	{
	442	int i, stream_id;
	443	int ret = 0;
e8c92a62	444	struct bt_iter *iter;
6f3077a2	445
90fcbacc	446	iter = g_new0(struct bt_iter, 1);
6f3077a2	447	iter->stream_heap = g_new(struct ptr_heap, 1);
6f3077a2 JD	448	iter->end_pos = end_pos;
	449	iter->callbacks = g_array_new(0, 1, sizeof(struct bt_stream_callbacks));
	450	iter->recalculate_dep_graph = 0;
	451	iter->main_callbacks.callback = NULL;
	452	iter->dep_gc = g_ptr_array_new();
6cba487f	453	bt_context_get(ctx);
95d36295	454	iter->ctx = ctx;
6f3077a2 JD	455
	456	ret = heap_init(iter->stream_heap, 0, stream_compare);
	457	if (ret < 0)
	458	goto error_heap_init;
	459
95d36295	460	for (i = 0; i < ctx->tc->array->len; i++) {
6f3077a2 JD	461	struct ctf_trace *tin;
	462	struct trace_descriptor *td_read;
	463
95d36295	464	td_read = g_ptr_array_index(ctx->tc->array, i);
6f3077a2 JD	465	tin = container_of(td_read, struct ctf_trace, parent);
	466
	467	/* Populate heap with each stream */
	468	for (stream_id = 0; stream_id < tin->streams->len;
	469	stream_id++) {
	470	struct ctf_stream_class *stream;
	471	int filenr;
	472
	473	stream = g_ptr_array_index(tin->streams, stream_id);
	474	if (!stream)
	475	continue;
	476	for (filenr = 0; filenr < stream->streams->len;
	477	filenr++) {
	478	struct ctf_file_stream *file_stream;
	479
	480	file_stream = g_ptr_array_index(stream->streams,
	481	filenr);
	482
	483	if (begin_pos) {
	484	ret = babeltrace_filestream_seek(file_stream, begin_pos,
	485	stream_id);
	486	if (ret == EOF) {
	487	ret = 0;
	488	continue;
	489	} else if (ret) {
	490	goto error;
	491	}
	492	}
	493	/* Add to heap */
	494	ret = heap_insert(iter->stream_heap, file_stream);
	495	if (ret)
	496	goto error;
	497	}
	498	}
	499	}
	500
	501	return iter;
	502
	503	error:
	504	heap_free(iter->stream_heap);
	505	error_heap_init:
	506	g_free(iter->stream_heap);
90fcbacc	507	g_free(iter);
6f3077a2 JD	508	return NULL;
	509	}
	510
e8c92a62	511	void bt_iter_destroy(struct bt_iter *iter)
6f3077a2 JD	512	{
	513	struct bt_stream_callbacks *bt_stream_cb;
	514	struct bt_callback_chain *bt_chain;
	515	int i, j;
	516
dc81689e JD	517	if (iter->stream_heap) {
	518	heap_free(iter->stream_heap);
	519	g_free(iter->stream_heap);
	520	}
6f3077a2 JD	521
	522	/* free all events callbacks */
	523	if (iter->main_callbacks.callback)
	524	g_array_free(iter->main_callbacks.callback, TRUE);
	525
	526	/* free per-event callbacks */
	527	for (i = 0; i < iter->callbacks->len; i++) {
	528	bt_stream_cb = &g_array_index(iter->callbacks,
	529	struct bt_stream_callbacks, i);
	530	if (!bt_stream_cb \|\| !bt_stream_cb->per_id_callbacks)
	531	continue;
	532	for (j = 0; j < bt_stream_cb->per_id_callbacks->len; j++) {
	533	bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
	534	struct bt_callback_chain, j);
	535	if (bt_chain->callback) {
	536	g_array_free(bt_chain->callback, TRUE);
	537	}
	538	}
	539	g_array_free(bt_stream_cb->per_id_callbacks, TRUE);
	540	}
	541
95d36295 JD	542	bt_context_put(iter->ctx);
95d36295 JD	543
90fcbacc	544	g_free(iter);
6f3077a2 JD	545	}
6f3077a2 JD	546
e8c92a62	547	int bt_iter_next(struct bt_iter *iter)
6f3077a2 JD	548	{
	549	struct ctf_file_stream file_stream, removed;
	550	int ret;
	551
	552	file_stream = heap_maximum(iter->stream_heap);
	553	if (!file_stream) {
	554	/* end of file for all streams */
	555	ret = 0;
	556	goto end;
	557	}
	558
	559	ret = stream_read_event(file_stream);
	560	if (ret == EOF) {
	561	removed = heap_remove(iter->stream_heap);
	562	assert(removed == file_stream);
	563	ret = 0;
	564	goto end;
	565	} else if (ret) {
	566	goto end;
	567	}
	568	/* Reinsert the file stream into the heap, and rebalance. */
	569	removed = heap_replace_max(iter->stream_heap, file_stream);
	570	assert(removed == file_stream);
	571
	572	end:
	573	return ret;
	574	}
	575
9843982d	576	struct bt_ctf_event bt_iter_read_ctf_event(struct bt_iter iter)
6f3077a2 JD	577	{
6f3077a2 JD	578	struct ctf_file_stream *file_stream;
9843982d	579	struct bt_ctf_event *ret = &iter->current_ctf_event;
6f3077a2 JD	580
	581	file_stream = heap_maximum(iter->stream_heap);
	582	if (!file_stream) {
	583	/* end of file for all streams */
9843982d	584	goto stop;
6f3077a2	585	}
9843982d JD	586	ret->stream = &file_stream->parent;
	587	ret->event = g_ptr_array_index(ret->stream->events_by_id,
	588	ret->stream->event_id);
6f3077a2	589
9843982d	590	if (ret->stream->stream_id > iter->callbacks->len)
6f3077a2 JD	591	goto end;
6f3077a2 JD	592
9843982d	593	process_callbacks(iter, ret->stream);
6f3077a2 JD	594
	595	end:
	596	return ret;
9843982d JD	597	stop:
9843982d JD	598	return NULL;
6f3077a2	599	}