[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/ctf/metadata.h>
#include <babeltrace/iterator-internal.h>
#include <babeltrace/prio_heap.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. */
};

struct babeltrace_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(stdout, "[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;
}

/*
 * 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 trace_collection_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;
}

/*
 * babeltrace_iter_seek: seek iterator to given position.
 */
int babeltrace_iter_seek(struct babeltrace_iter *iter,
		const struct trace_collection_pos *begin_pos)
{
	int i, stream_id;
	int ret = 0;
	struct trace_collection *tc = iter->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 babeltrace_iter *babeltrace_iter_create(struct trace_collection *tc,
		struct trace_collection_pos *begin_pos,
		struct trace_collection_pos *end_pos)
{
	int i, stream_id;
	int ret = 0;
	struct babeltrace_iter *iter;

	iter = malloc(sizeof(struct babeltrace_iter));
	if (!iter)
		goto error_malloc;
	iter->stream_heap = g_new(struct ptr_heap, 1);
	iter->tc = tc;
	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();

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

		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);
			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);
	free(iter);
error_malloc:
	return NULL;
}

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

	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);
	}

	free(iter);
}

int babeltrace_iter_next(struct babeltrace_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 babeltrace_iter_read_event(struct babeltrace_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
	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>
	24	#include <babeltrace/ctf/metadata.h>
	25	#include <babeltrace/iterator-internal.h>
	26	#include <babeltrace/prio_heap.h>
	27
	28	struct stream_saved_pos {
	29	/*
	30	* Use file_stream pointer to check if the trace collection we
	31	* restore to match the one we saved from, for each stream.
	32	*/
	33	struct ctf_file_stream *file_stream;
	34	size_t cur_index; /* current index in packet index */
	35	ssize_t offset; /* offset from base, in bits. EOF for end of file. */
	36	};
	37
	38	struct babeltrace_saved_pos {
	39	struct trace_collection *tc;
	40	GArray stream_saved_pos; / Contains struct stream_saved_pos */
	41	};
	42
	43	static int stream_read_event(struct ctf_file_stream *sin)
	44	{
	45	int ret;
	46
	47	ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
	48	if (ret == EOF)
	49	return EOF;
	50	else if (ret) {
	51	fprintf(stdout, "[error] Reading event failed.\n");
	52	return ret;
	53	}
	54	return 0;
	55	}
	56
	57	/*
	58	* returns true if a < b, false otherwise.
	59	*/
	60	int stream_compare(void a, void b)
	61	{
	62	struct ctf_file_stream s_a = a, s_b = b;
	63
	64	if (s_a->parent.timestamp < s_b->parent.timestamp)
	65	return 1;
	66	else
	67	return 0;
	68	}
	69
	70	/*
	71	* babeltrace_filestream_seek: seek a filestream to given position.
	72	*
	73	* The stream_id parameter is only useful for BT_SEEK_RESTORE.
	74	*/
	75	static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
	76	const struct trace_collection_pos *begin_pos,
	77	unsigned long stream_id)
	78	{
	79	int ret = 0;
	80
	81	switch (begin_pos->type) {
	82	case BT_SEEK_CUR:
	83	/*
	84	* just insert into the heap we should already know
	85	* the timestamps
	86	*/
	87	break;
	88	case BT_SEEK_BEGIN:
	89	file_stream->pos.move_pos_slow(&file_stream->pos, 0, SEEK_SET);
	90	ret = stream_read_event(file_stream);
	91	break;
	92	case BT_SEEK_TIME:
	93	case BT_SEEK_RESTORE:
	94	case BT_SEEK_END:
	95	default:
	96	assert(0); /* Not yet defined */
	97	}
	98
	99	return ret;
	100	}
	101
	102	/*
	103	* babeltrace_iter_seek: seek iterator to given position.
	104	*/
	105	int babeltrace_iter_seek(struct babeltrace_iter *iter,
	106	const struct trace_collection_pos *begin_pos)
	107	{
	108	int i, stream_id;
	109	int ret = 0;
	110	struct trace_collection *tc = iter->tc;
	111
	112	for (i = 0; i < tc->array->len; i++) {
	113	struct ctf_trace *tin;
	114	struct trace_descriptor *td_read;
	115
	116	td_read = g_ptr_array_index(tc->array, i);
	117	tin = container_of(td_read, struct ctf_trace, parent);
	118
	119	/* Populate heap with each stream */
	120	for (stream_id = 0; stream_id < tin->streams->len;
	121	stream_id++) {
	122	struct ctf_stream_class *stream;
	123	int filenr;
	124
	125	stream = g_ptr_array_index(tin->streams, stream_id);
	126	for (filenr = 0; filenr < stream->streams->len;
	127	filenr++) {
	128	struct ctf_file_stream *file_stream;
	129
	130	file_stream = g_ptr_array_index(stream->streams,
	131	filenr);
	132	ret = babeltrace_filestream_seek(file_stream, begin_pos,
	133	stream_id);
	134	if (ret < 0)
	135	goto end;
	136	}
	137	}
	138	}
	139	end:
	140	return ret;
	141	}
	142
	143	struct babeltrace_iter babeltrace_iter_create(struct trace_collection tc,
	144	struct trace_collection_pos *begin_pos,
	145	struct trace_collection_pos *end_pos)
	146	{
	147	int i, stream_id;
	148	int ret = 0;
	149	struct babeltrace_iter *iter;
	150
	151	iter = malloc(sizeof(struct babeltrace_iter));
	152	if (!iter)
	153	goto error_malloc;
	154	iter->stream_heap = g_new(struct ptr_heap, 1);
	155	iter->tc = tc;
	156	iter->end_pos = end_pos;
	157	iter->callbacks = g_array_new(0, 1, sizeof(struct bt_stream_callbacks));
	158	iter->recalculate_dep_graph = 0;
	159	iter->main_callbacks.callback = NULL;
	160	iter->dep_gc = g_ptr_array_new();
	161
	162	ret = heap_init(iter->stream_heap, 0, stream_compare);
	163	if (ret < 0)
	164	goto error_heap_init;
	165
	166	for (i = 0; i < tc->array->len; i++) {
	167	struct ctf_trace *tin;
	168	struct trace_descriptor *td_read;
	169
	170	td_read = g_ptr_array_index(tc->array, i);
	171	tin = container_of(td_read, struct ctf_trace, parent);
	172
	173	/* Populate heap with each stream */
	174	for (stream_id = 0; stream_id < tin->streams->len;
	175	stream_id++) {
	176	struct ctf_stream_class *stream;
	177	int filenr;
	178
	179	stream = g_ptr_array_index(tin->streams, stream_id);
	180	if (!stream)
	181	continue;
	182	for (filenr = 0; filenr < stream->streams->len;
	183	filenr++) {
	184	struct ctf_file_stream *file_stream;
	185
	186	file_stream = g_ptr_array_index(stream->streams,
	187	filenr);
	188
	189	if (begin_pos) {
	190	ret = babeltrace_filestream_seek(file_stream, begin_pos,
	191	stream_id);
	192	if (ret == EOF) {
	193	ret = 0;
	194	continue;
	195	} else if (ret) {
	196	goto error;
	197	}
	198	}
	199	/* Add to heap */
	200	ret = heap_insert(iter->stream_heap, file_stream);
	201	if (ret)
	202	goto error;
	203	}
	204	}
	205	}
	206
	207	return iter;
	208
	209	error:
	210	heap_free(iter->stream_heap);
	211	error_heap_init:
	212	g_free(iter->stream_heap);
	213	free(iter);
	214	error_malloc:
	215	return NULL;
	216	}
	217
	218	void babeltrace_iter_destroy(struct babeltrace_iter *iter)
	219	{
	220	struct bt_stream_callbacks *bt_stream_cb;
	221	struct bt_callback_chain *bt_chain;
	222	int i, j;
	223
	224	heap_free(iter->stream_heap);
	225	g_free(iter->stream_heap);
	226
	227	/* free all events callbacks */
	228	if (iter->main_callbacks.callback)
	229	g_array_free(iter->main_callbacks.callback, TRUE);
	230
	231	/* free per-event callbacks */
	232	for (i = 0; i < iter->callbacks->len; i++) {
	233	bt_stream_cb = &g_array_index(iter->callbacks,
	234	struct bt_stream_callbacks, i);
	235	if (!bt_stream_cb \|\| !bt_stream_cb->per_id_callbacks)
	236	continue;
	237	for (j = 0; j < bt_stream_cb->per_id_callbacks->len; j++) {
	238	bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
	239	struct bt_callback_chain, j);
	240	if (bt_chain->callback) {
	241	g_array_free(bt_chain->callback, TRUE);
	242	}
	243	}
	244	g_array_free(bt_stream_cb->per_id_callbacks, TRUE);
	245	}
	246
	247	free(iter);
	248	}
	249
	250	int babeltrace_iter_next(struct babeltrace_iter *iter)
	251	{
	252	struct ctf_file_stream file_stream, removed;
	253	int ret;
	254
	255	file_stream = heap_maximum(iter->stream_heap);
	256	if (!file_stream) {
	257	/* end of file for all streams */
	258	ret = 0;
	259	goto end;
	260	}
	261
	262	ret = stream_read_event(file_stream);
	263	if (ret == EOF) {
	264	removed = heap_remove(iter->stream_heap);
	265	assert(removed == file_stream);
	266	ret = 0;
	267	goto end;
	268	} else if (ret) {
	269	goto end;
	270	}
	271	/* Reinsert the file stream into the heap, and rebalance. */
	272	removed = heap_replace_max(iter->stream_heap, file_stream);
	273	assert(removed == file_stream);
	274
	275	end:
	276	return ret;
	277	}
	278
	279	int babeltrace_iter_read_event(struct babeltrace_iter *iter,
	280	struct ctf_stream **stream,
	281	struct ctf_stream_event **event)
	282	{
	283	struct ctf_file_stream *file_stream;
	284	int ret = 0;
	285
	286	file_stream = heap_maximum(iter->stream_heap);
	287	if (!file_stream) {
	288	/* end of file for all streams */
	289	ret = EOF;
	290	goto end;
	291	}
	292	*stream = &file_stream->parent;
	293	event = g_ptr_array_index((stream)->events_by_id, (*stream)->event_id);
	294
	295	if ((*stream)->stream_id > iter->callbacks->len)
	296	goto end;
	297
	298	process_callbacks(iter, *stream);
	299
	300	end:
	301	return ret;
	302	}