[babeltrace.git] / converter / babeltrace-lib.c

/*
 * babeltrace-lib.c
 *
 * Babeltrace Trace Converter 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 <errno.h>
#include <stdio.h>
#include <inttypes.h>
#include <babeltrace/babeltrace-internal.h>
#include <babeltrace/format.h>
#include <babeltrace/ctf/types.h>
#include <babeltrace/ctf/metadata.h>
#include <babeltrace/ctf-text/types.h>
#include <babeltrace/prio_heap.h>
#include <babeltrace/babeltrace.h>
#include <babeltrace/types.h>
#include <babeltrace/ctf/types.h>
#include <babeltrace/ctf-ir/metadata.h>
#include <stdarg.h>

int babeltrace_verbose, babeltrace_debug;

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

struct bt_callback {
	int prio;		/* Callback order priority. Lower first. Dynamically assigned from dependency graph. */
	void *private_data;
	int flags;
	struct bt_dependencies *depends;
	struct bt_dependencies *weak_depends;
	struct bt_dependencies *provides;
	enum bt_cb_ret (*callback)(struct bt_ctf_data *ctf_data,
				   void *private_data);
};

struct bt_callback_chain {
	GArray *callback;	/* Array of struct bt_callback, ordered by priority */
};

/*
 * per id callbacks need to be per stream class because event ID vs
 * event name mapping can vary from stream to stream.
 */
struct bt_stream_callbacks {
	GArray *per_id_callbacks;	/* Array of struct bt_callback_chain */
};

/*
 * struct babeltrace_iter: data structure representing an iterator on a trace
 * collection.
 */
struct babeltrace_iter {
	struct ptr_heap *stream_heap;
	struct trace_collection *tc;
	struct trace_collection_pos *end_pos;
	GArray *callbacks;				/* Array of struct bt_stream_callbacks */
	struct bt_callback_chain main_callbacks;	/* For all events */
	/*
	 * Flag indicating if dependency graph needs to be recalculated.
	 * Set by babeltrace_iter_add_callback(), and checked (and
	 * cleared) by upon entry into babeltrace_iter_read_event().
	 * babeltrace_iter_read_event() is responsible for calling dep
	 * graph calculation if it sees this flag set.
	 */
	int recalculate_dep_graph;
	/*
	 * Array of pointers to struct bt_dependencies, for garbage
	 * collection. We're not using a linked list here because each
	 * struct bt_dependencies can belong to more than one
	 * babeltrace_iter.
	 */
	GPtrArray *dep_gc;
};

struct bt_dependencies {
	GArray *deps;			/* Array of GQuarks */
	int refcount;			/* free when decremented to 0 */
};

static
struct bt_dependencies *_babeltrace_dependencies_create(const char *first,
							va_list ap)
{
	const char *iter;
	struct bt_dependencies *dep;

	dep = g_new0(struct bt_dependencies, 1);
	dep->refcount = 1;
	dep->deps = g_array_new(FALSE, TRUE, sizeof(GQuark));
	iter = first;
	while (iter) {
		GQuark q = g_quark_from_string(iter);
		g_array_append_val(dep->deps, q);
		iter = va_arg(ap, const char *);
	}
	return dep;
}

struct bt_dependencies *babeltrace_dependencies_create(const char *first, ...)
{
	va_list ap;
	struct bt_dependencies *deps;

	va_start(ap, first);
	deps = _babeltrace_dependencies_create(first, ap);
	va_end(ap);
	return deps;
}

/*
 * babeltrace_iter_add_callback: Add a callback to iterator.
 */
int babeltrace_iter_add_callback(struct babeltrace_iter *iter,
		bt_event_name event, void *private_data, int flags,
		enum bt_cb_ret (*callback)(struct bt_ctf_data *ctf_data,
					   void *private_data),
		struct bt_dependencies *depends,
		struct bt_dependencies *weak_depends,
		struct bt_dependencies *provides)
{
	int i, stream_id;
	gpointer *event_id_ptr;
	unsigned long event_id;
	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);

		for (stream_id = 0; stream_id < tin->streams->len; stream_id++) {
			struct ctf_stream_class *stream;
			struct bt_stream_callbacks *bt_stream_cb = NULL;
			struct bt_callback_chain *bt_chain = NULL;
			struct bt_callback new_callback;

			stream = g_ptr_array_index(tin->streams, stream_id);

			/* find or create the bt_stream_callbacks for this stream */
			if (iter->callbacks->len >= stream_id) {
				bt_stream_cb = &g_array_index(iter->callbacks,
						struct bt_stream_callbacks, stream->stream_id);
			} else {
				g_array_set_size(iter->callbacks, stream->stream_id);
			}
			if (!bt_stream_cb || !bt_stream_cb->per_id_callbacks) {
				struct bt_stream_callbacks new_stream_cb;
				new_stream_cb.per_id_callbacks = g_array_new(1, 1,
						sizeof(struct bt_callback_chain));
				g_array_insert_val(iter->callbacks, stream->stream_id, new_stream_cb);
				bt_stream_cb = &g_array_index(iter->callbacks,
						struct bt_stream_callbacks, stream->stream_id);
			}

			if (event) {
				/* find the event id */
				event_id_ptr = g_hash_table_lookup(stream->event_quark_to_id,
						(gconstpointer) (unsigned long) event);
				/* event not found in this stream class */
				if (!event_id_ptr) {
					printf("event not found\n");
					continue;
				}
				event_id = (uint64_t)(unsigned long)*event_id_ptr;

				/* find or create the bt_callback_chain for this event */
				if (bt_stream_cb->per_id_callbacks->len >= event_id) {
					bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
							struct bt_callback_chain, event_id);
				} else {
					g_array_set_size(bt_stream_cb->per_id_callbacks, event_id);
				}
				if (!bt_chain || !bt_chain->callback) {
					struct bt_callback_chain new_chain;
					new_chain.callback = g_array_new(1, 1, sizeof(struct bt_callback));
					g_array_insert_val(bt_stream_cb->per_id_callbacks, event_id,
							new_chain);
					bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
							struct bt_callback_chain, event_id);
				}
			} else {
				/* callback for all events */
				if (!iter->main_callbacks.callback) {
					iter->main_callbacks.callback = g_array_new(1, 1,
							sizeof(struct bt_callback));
				}
				bt_chain = &iter->main_callbacks;
			}

			new_callback.private_data = private_data;
			new_callback.flags = flags;
			new_callback.callback = callback;
			new_callback.depends = depends;
			new_callback.weak_depends = weak_depends;
			new_callback.provides = provides;

			/* TODO : take care of priority, for now just FIFO */
			g_array_append_val(bt_chain->callback, new_callback);
		}
	}

	return 0;
}

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

static
struct ctf_stream_event *extract_ctf_stream_event(struct ctf_stream *stream)
{
	struct ctf_stream_class *stream_class = stream->stream_class;
	struct ctf_event *event_class;
	struct ctf_stream_event *event;
	uint64_t id = stream->event_id;

	if (id >= stream_class->events_by_id->len) {
		fprintf(stdout, "[error] Event id %" PRIu64 " is outside range.\n", id);
		return NULL;
	}
	event = g_ptr_array_index(stream->events_by_id, id);
	if (!event) {
		fprintf(stdout, "[error] Event id %" PRIu64 " is unknown.\n", id);
		return NULL;
	}
	event_class = g_ptr_array_index(stream_class->events_by_id, id);
	if (!event_class) {
		fprintf(stdout, "[error] Event id %" PRIu64 " is unknown.\n", id);
		return NULL;
	}

	return event;
}

static
void process_callbacks(struct babeltrace_iter *iter,
		       struct ctf_stream *stream)
{
	struct bt_stream_callbacks *bt_stream_cb;
	struct bt_callback_chain *bt_chain;
	struct bt_callback *cb;
	int i;
	enum bt_cb_ret ret;
	struct bt_ctf_data ctf_data;

	ctf_data.event = extract_ctf_stream_event(stream);
	ctf_data.stream = stream;

	/* process all events callback first */
	if (iter->main_callbacks.callback) {
		for (i = 0; i < iter->main_callbacks.callback->len; i++) {
			cb = &g_array_index(iter->main_callbacks.callback, struct bt_callback, i);
			if (!cb)
				goto end;
			ret = cb->callback(&ctf_data, cb->private_data);
			switch (ret) {
			case BT_CB_OK_STOP:
			case BT_CB_ERROR_STOP:
				goto end;
			default:
				break;
			}
		}
	}

	/* process per event callbacks */
	bt_stream_cb = &g_array_index(iter->callbacks,
			struct bt_stream_callbacks, stream->stream_id);
	if (!bt_stream_cb || !bt_stream_cb->per_id_callbacks)
		goto end;

	if (stream->event_id > bt_stream_cb->per_id_callbacks->len)
		goto end;
	bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
			struct bt_callback_chain, stream->event_id);
	if (!bt_chain || !bt_chain->callback)
		goto end;

	for (i = 0; i < bt_chain->callback->len; i++) {
		cb = &g_array_index(bt_chain->callback, struct bt_callback, i);
		if (!cb)
			goto end;
		ret = cb->callback(&ctf_data, cb->private_data);
		switch (ret) {
		case BT_CB_OK_STOP:
		case BT_CB_ERROR_STOP:
			goto end;
		default:
			break;
		}
	}

end:
	return;
}

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

int convert_trace(struct trace_descriptor *td_write,
		  struct trace_collection *trace_collection_read)
{
	struct babeltrace_iter *iter;
	struct ctf_stream *stream;
	struct ctf_stream_event *event;
	struct ctf_text_stream_pos *sout;
	struct trace_collection_pos begin_pos;
	int ret = 0;

	sout = container_of(td_write, struct ctf_text_stream_pos,
			trace_descriptor);

	begin_pos.type = BT_SEEK_BEGIN;
	iter = babeltrace_iter_create(trace_collection_read, &begin_pos, NULL);
	while (babeltrace_iter_read_event(iter, &stream, &event) == 0) {
		ret = sout->parent.event_cb(&sout->parent, stream);
		if (ret) {
			fprintf(stdout, "[error] Writing event failed.\n");
			goto end;
		}
		ret = babeltrace_iter_next(iter);
		if (ret < 0)
			goto end;
	}
end:
	babeltrace_iter_destroy(iter);
	return ret;
}

static
void __attribute__((constructor)) init_babeltrace_lib(void)
{
	if (getenv("BABELTRACE_VERBOSE"))
		babeltrace_verbose = 1;
	if (getenv("BABELTRACE_DEBUG"))
		babeltrace_debug = 1;
}
Commit	Line	Data
46322b33 MD	1	/*
	2	* babeltrace-lib.c
	3	*
	4	* Babeltrace Trace Converter Library
	5	*
64fa3fec MD	6	* Copyright 2010-2011 EfficiOS Inc. and Linux Foundation
	7	*
	8	* Author: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
46322b33 MD	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 <errno.h>
	23	#include <stdio.h>
847bf71a	24	#include <inttypes.h>
70bd0a12	25	#include <babeltrace/babeltrace-internal.h>
46322b33 MD	26	#include <babeltrace/format.h>
	27	#include <babeltrace/ctf/types.h>
	28	#include <babeltrace/ctf/metadata.h>
	29	#include <babeltrace/ctf-text/types.h>
0d0f5149	30	#include <babeltrace/prio_heap.h>
70bd0a12 JD	31	#include <babeltrace/babeltrace.h>
	32	#include <babeltrace/types.h>
	33	#include <babeltrace/ctf/types.h>
	34	#include <babeltrace/ctf-ir/metadata.h>
c34ccddd	35	#include <stdarg.h>
70bd0a12	36
b88d6e85 MD	37	int babeltrace_verbose, babeltrace_debug;
b88d6e85 MD	38
063f7048 MD	39	struct stream_saved_pos {
	40	/*
	41	* Use file_stream pointer to check if the trace collection we
	42	* restore to match the one we saved from, for each stream.
	43	*/
	44	struct ctf_file_stream *file_stream;
	45	size_t cur_index; /* current index in packet index */
	46	ssize_t offset; /* offset from base, in bits. EOF for end of file. */
	47	};
	48
	49	struct babeltrace_saved_pos {
	50	struct trace_collection *tc;
	51	GArray stream_saved_pos; / Contains struct stream_saved_pos */
	52	};
	53
c34ccddd MD	54	struct bt_callback {
	55	int prio; /* Callback order priority. Lower first. Dynamically assigned from dependency graph. */
	56	void *private_data;
8ee2fac4 JD	57	int flags;
	58	struct bt_dependencies *depends;
	59	struct bt_dependencies *weak_depends;
	60	struct bt_dependencies *provides;
e73cc54c JD	61	enum bt_cb_ret (callback)(struct bt_ctf_data ctf_data,
e73cc54c JD	62	void *private_data);
c34ccddd MD	63	};
	64
	65	struct bt_callback_chain {
	66	GArray callback; / Array of struct bt_callback, ordered by priority */
	67	};
	68
	69	/*
	70	* per id callbacks need to be per stream class because event ID vs
	71	* event name mapping can vary from stream to stream.
	72	*/
	73	struct bt_stream_callbacks {
	74	GArray per_id_callbacks; / Array of struct bt_callback_chain */
	75	};
	76
70bd0a12 JD	77	/*
	78	* struct babeltrace_iter: data structure representing an iterator on a trace
	79	* collection.
	80	*/
	81	struct babeltrace_iter {
	82	struct ptr_heap *stream_heap;
	83	struct trace_collection *tc;
063f7048	84	struct trace_collection_pos *end_pos;
8ee2fac4	85	GArray callbacks; / Array of struct bt_stream_callbacks */
c34ccddd MD	86	struct bt_callback_chain main_callbacks; /* For all events */
	87	/*
	88	* Flag indicating if dependency graph needs to be recalculated.
	89	* Set by babeltrace_iter_add_callback(), and checked (and
	90	* cleared) by upon entry into babeltrace_iter_read_event().
	91	* babeltrace_iter_read_event() is responsible for calling dep
	92	* graph calculation if it sees this flag set.
	93	*/
	94	int recalculate_dep_graph;
	95	/*
	96	* Array of pointers to struct bt_dependencies, for garbage
	97	* collection. We're not using a linked list here because each
	98	* struct bt_dependencies can belong to more than one
	99	* babeltrace_iter.
	100	*/
	101	GPtrArray *dep_gc;
70bd0a12 JD	102	};
70bd0a12 JD	103
c34ccddd MD	104	struct bt_dependencies {
	105	GArray deps; / Array of GQuarks */
	106	int refcount; /* free when decremented to 0 */
	107	};
	108
	109	static
	110	struct bt_dependencies _babeltrace_dependencies_create(const char first,
	111	va_list ap)
	112	{
	113	const char *iter;
	114	struct bt_dependencies *dep;
	115
	116	dep = g_new0(struct bt_dependencies, 1);
	117	dep->refcount = 1;
	118	dep->deps = g_array_new(FALSE, TRUE, sizeof(GQuark));
	119	iter = first;
	120	while (iter) {
	121	GQuark q = g_quark_from_string(iter);
	122	g_array_append_val(dep->deps, q);
	123	iter = va_arg(ap, const char *);
	124	}
	125	return dep;
	126	}
	127
	128	struct bt_dependencies babeltrace_dependencies_create(const char first, ...)
	129	{
	130	va_list ap;
	131	struct bt_dependencies *deps;
	132
	133	va_start(ap, first);
	134	deps = _babeltrace_dependencies_create(first, ap);
	135	va_end(ap);
	136	return deps;
	137	}
	138
8ee2fac4 JD	139	/*
	140	* babeltrace_iter_add_callback: Add a callback to iterator.
	141	*/
	142	int babeltrace_iter_add_callback(struct babeltrace_iter *iter,
	143	bt_event_name event, void *private_data, int flags,
e73cc54c JD	144	enum bt_cb_ret (callback)(struct bt_ctf_data ctf_data,
e73cc54c JD	145	void *private_data),
8ee2fac4 JD	146	struct bt_dependencies *depends,
	147	struct bt_dependencies *weak_depends,
	148	struct bt_dependencies *provides)
	149	{
	150	int i, stream_id;
	151	gpointer *event_id_ptr;
	152	unsigned long event_id;
	153	struct trace_collection *tc = iter->tc;
	154
	155	for (i = 0; i < tc->array->len; i++) {
	156	struct ctf_trace *tin;
	157	struct trace_descriptor *td_read;
	158
	159	td_read = g_ptr_array_index(tc->array, i);
	160	tin = container_of(td_read, struct ctf_trace, parent);
	161
	162	for (stream_id = 0; stream_id < tin->streams->len; stream_id++) {
	163	struct ctf_stream_class *stream;
	164	struct bt_stream_callbacks *bt_stream_cb = NULL;
	165	struct bt_callback_chain *bt_chain = NULL;
	166	struct bt_callback new_callback;
	167
	168	stream = g_ptr_array_index(tin->streams, stream_id);
	169
	170	/* find or create the bt_stream_callbacks for this stream */
	171	if (iter->callbacks->len >= stream_id) {
	172	bt_stream_cb = &g_array_index(iter->callbacks,
	173	struct bt_stream_callbacks, stream->stream_id);
	174	} else {
	175	g_array_set_size(iter->callbacks, stream->stream_id);
	176	}
	177	if (!bt_stream_cb \|\| !bt_stream_cb->per_id_callbacks) {
	178	struct bt_stream_callbacks new_stream_cb;
	179	new_stream_cb.per_id_callbacks = g_array_new(1, 1,
	180	sizeof(struct bt_callback_chain));
	181	g_array_insert_val(iter->callbacks, stream->stream_id, new_stream_cb);
	182	bt_stream_cb = &g_array_index(iter->callbacks,
	183	struct bt_stream_callbacks, stream->stream_id);
	184	}
	185
	186	if (event) {
	187	/* find the event id */
	188	event_id_ptr = g_hash_table_lookup(stream->event_quark_to_id,
	189	(gconstpointer) (unsigned long) event);
	190	/* event not found in this stream class */
	191	if (!event_id_ptr) {
	192	printf("event not found\n");
	193	continue;
	194	}
294796de	195	event_id = (uint64_t)(unsigned long)*event_id_ptr;
8ee2fac4 JD	196
	197	/* find or create the bt_callback_chain for this event */
	198	if (bt_stream_cb->per_id_callbacks->len >= event_id) {
	199	bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
	200	struct bt_callback_chain, event_id);
	201	} else {
	202	g_array_set_size(bt_stream_cb->per_id_callbacks, event_id);
	203	}
	204	if (!bt_chain \|\| !bt_chain->callback) {
	205	struct bt_callback_chain new_chain;
	206	new_chain.callback = g_array_new(1, 1, sizeof(struct bt_callback));
	207	g_array_insert_val(bt_stream_cb->per_id_callbacks, event_id,
	208	new_chain);
	209	bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
	210	struct bt_callback_chain, event_id);
	211	}
	212	} else {
	213	/* callback for all events */
	214	if (!iter->main_callbacks.callback) {
	215	iter->main_callbacks.callback = g_array_new(1, 1,
	216	sizeof(struct bt_callback));
	217	}
	218	bt_chain = &iter->main_callbacks;
	219	}
	220
	221	new_callback.private_data = private_data;
	222	new_callback.flags = flags;
	223	new_callback.callback = callback;
	224	new_callback.depends = depends;
	225	new_callback.weak_depends = weak_depends;
	226	new_callback.provides = provides;
	227
	228	/* TODO : take care of priority, for now just FIFO */
	229	g_array_append_val(bt_chain->callback, new_callback);
	230	}
	231	}
	232
	233	return 0;
	234	}
	235
9e501292	236	static int stream_read_event(struct ctf_file_stream *sin)
46322b33 MD	237	{
	238	int ret;
	239
2d0bea29	240	ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
0d0f5149 MD	241	if (ret == EOF)
	242	return EOF;
	243	else if (ret) {
	244	fprintf(stdout, "[error] Reading event failed.\n");
	245	return ret;
46322b33	246	}
46322b33	247	return 0;
0d0f5149	248	}
46322b33	249
0d0f5149 MD	250	/*
	251	* returns true if a < b, false otherwise.
	252	*/
	253	int stream_compare(void a, void b)
	254	{
	255	struct ctf_file_stream s_a = a, s_b = b;
	256
2d0bea29	257	if (s_a->parent.timestamp < s_b->parent.timestamp)
0d0f5149 MD	258	return 1;
	259	else
	260	return 0;
46322b33 MD	261	}
46322b33 MD	262
877fb419 JD	263	/*
	264	* babeltrace_filestream_seek: seek a filestream to given position.
	265	*
	266	* The stream_id parameter is only useful for BT_SEEK_RESTORE.
	267	*/
	268	static int babeltrace_filestream_seek(struct ctf_file_stream *file_stream,
	269	const struct trace_collection_pos *begin_pos,
	270	unsigned long stream_id)
	271	{
	272	int ret = 0;
	273
	274	switch (begin_pos->type) {
	275	case BT_SEEK_CUR:
	276	/*
	277	* just insert into the heap we should already know
	278	* the timestamps
	279	*/
	280	break;
	281	case BT_SEEK_BEGIN:
31e67029	282	file_stream->pos.move_pos_slow(&file_stream->pos, 0, SEEK_SET);
877fb419 JD	283	ret = stream_read_event(file_stream);
	284	break;
	285	case BT_SEEK_TIME:
	286	case BT_SEEK_RESTORE:
	287	case BT_SEEK_END:
	288	default:
	289	assert(0); /* Not yet defined */
	290	}
	291
	292	return ret;
	293	}
	294
	295	/*
	296	* babeltrace_iter_seek: seek iterator to given position.
	297	*/
	298	int babeltrace_iter_seek(struct babeltrace_iter *iter,
	299	const struct trace_collection_pos *begin_pos)
	300	{
	301	int i, stream_id;
	302	int ret = 0;
	303	struct trace_collection *tc = iter->tc;
	304
	305	for (i = 0; i < tc->array->len; i++) {
	306	struct ctf_trace *tin;
	307	struct trace_descriptor *td_read;
	308
	309	td_read = g_ptr_array_index(tc->array, i);
	310	tin = container_of(td_read, struct ctf_trace, parent);
	311
	312	/* Populate heap with each stream */
	313	for (stream_id = 0; stream_id < tin->streams->len;
	314	stream_id++) {
	315	struct ctf_stream_class *stream;
	316	int filenr;
	317
	318	stream = g_ptr_array_index(tin->streams, stream_id);
	319	for (filenr = 0; filenr < stream->streams->len;
	320	filenr++) {
	321	struct ctf_file_stream *file_stream;
	322
	323	file_stream = g_ptr_array_index(stream->streams,
	324	filenr);
	325	ret = babeltrace_filestream_seek(file_stream, begin_pos,
	326	stream_id);
	327	if (ret < 0)
	328	goto end;
	329	}
	330	}
	331	}
	332	end:
	333	return ret;
	334	}
	335
063f7048 MD	336	struct babeltrace_iter babeltrace_iter_create(struct trace_collection tc,
	337	struct trace_collection_pos *begin_pos,
	338	struct trace_collection_pos *end_pos)
46322b33	339	{
afb48eae	340	int i, stream_id;
0d0f5149	341	int ret = 0;
9e501292	342	struct babeltrace_iter *iter;
0d0f5149	343
9e501292 JD	344	iter = malloc(sizeof(struct babeltrace_iter));
	345	if (!iter)
	346	goto error_malloc;
	347	iter->stream_heap = g_new(struct ptr_heap, 1);
	348	iter->tc = tc;
877fb419	349	iter->end_pos = end_pos;
8ee2fac4 JD	350	iter->callbacks = g_array_new(0, 1, sizeof(struct bt_stream_callbacks));
	351	iter->recalculate_dep_graph = 0;
	352	iter->main_callbacks.callback = NULL;
	353	iter->dep_gc = g_ptr_array_new();
9e501292 JD	354
	355	ret = heap_init(iter->stream_heap, 0, stream_compare);
	356	if (ret < 0)
	357	goto error_heap_init;
46322b33	358
9e501292	359	for (i = 0; i < tc->array->len; i++) {
afb48eae AA	360	struct ctf_trace *tin;
afb48eae AA	361	struct trace_descriptor *td_read;
46322b33	362
9e501292	363	td_read = g_ptr_array_index(tc->array, i);
afb48eae AA	364	tin = container_of(td_read, struct ctf_trace, parent);
	365
	366	/* Populate heap with each stream */
	367	for (stream_id = 0; stream_id < tin->streams->len;
	368	stream_id++) {
	369	struct ctf_stream_class *stream;
	370	int filenr;
	371
	372	stream = g_ptr_array_index(tin->streams, stream_id);
	373	if (!stream)
0d0f5149	374	continue;
afb48eae AA	375	for (filenr = 0; filenr < stream->streams->len;
	376	filenr++) {
	377	struct ctf_file_stream *file_stream;
	378
	379	file_stream = g_ptr_array_index(stream->streams,
	380	filenr);
	381
877fb419 JD	382	if (begin_pos) {
	383	ret = babeltrace_filestream_seek(file_stream, begin_pos,
	384	stream_id);
	385	if (ret == EOF) {
	386	ret = 0;
	387	continue;
	388	} else if (ret) {
	389	goto error;
	390	}
afb48eae AA	391	}
afb48eae AA	392	/* Add to heap */
9e501292 JD	393	ret = heap_insert(iter->stream_heap, file_stream);
	394	if (ret)
	395	goto error;
46322b33 MD	396	}
	397	}
	398	}
	399
9e501292	400	return iter;
0d0f5149	401
9e501292 JD	402	error:
	403	heap_free(iter->stream_heap);
	404	error_heap_init:
	405	g_free(iter->stream_heap);
	406	free(iter);
	407	error_malloc:
	408	return NULL;
	409	}
	410
	411	void babeltrace_iter_destroy(struct babeltrace_iter *iter)
	412	{
ee126a93 JD	413	struct bt_stream_callbacks *bt_stream_cb;
	414	struct bt_callback_chain *bt_chain;
	415	int i, j;
	416
9e501292 JD	417	heap_free(iter->stream_heap);
9e501292 JD	418	g_free(iter->stream_heap);
ee126a93 JD	419
	420	/* free all events callbacks */
	421	if (iter->main_callbacks.callback)
	422	g_array_free(iter->main_callbacks.callback, TRUE);
	423
	424	/* free per-event callbacks */
	425	for (i = 0; i < iter->callbacks->len; i++) {
	426	bt_stream_cb = &g_array_index(iter->callbacks,
	427	struct bt_stream_callbacks, i);
	428	if (!bt_stream_cb \|\| !bt_stream_cb->per_id_callbacks)
	429	continue;
	430	for (j = 0; j < bt_stream_cb->per_id_callbacks->len; j++) {
	431	bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
	432	struct bt_callback_chain, j);
e73cc54c JD	433	if (bt_chain->callback) {
	434	g_array_free(bt_chain->callback, TRUE);
	435	}
ee126a93 JD	436	}
	437	g_array_free(bt_stream_cb->per_id_callbacks, TRUE);
	438	}
	439
9e501292 JD	440	free(iter);
	441	}
	442
	443	int babeltrace_iter_next(struct babeltrace_iter *iter)
	444	{
	445	struct ctf_file_stream file_stream, removed;
	446	int ret;
	447
	448	file_stream = heap_maximum(iter->stream_heap);
	449	if (!file_stream) {
	450	/* end of file for all streams */
	451	ret = 0;
	452	goto end;
	453	}
	454
	455	ret = stream_read_event(file_stream);
	456	if (ret == EOF) {
	457	removed = heap_remove(iter->stream_heap);
	458	assert(removed == file_stream);
	459	ret = 0;
	460	goto end;
	461	} else if (ret) {
	462	goto end;
	463	}
	464	/* Reinsert the file stream into the heap, and rebalance. */
	465	removed = heap_replace_max(iter->stream_heap, file_stream);
	466	assert(removed == file_stream);
	467
	468	end:
	469	return ret;
	470	}
	471
e73cc54c JD	472	static
	473	struct ctf_stream_event extract_ctf_stream_event(struct ctf_stream stream)
	474	{
	475	struct ctf_stream_class *stream_class = stream->stream_class;
	476	struct ctf_event *event_class;
	477	struct ctf_stream_event *event;
	478	uint64_t id = stream->event_id;
	479
	480	if (id >= stream_class->events_by_id->len) {
	481	fprintf(stdout, "[error] Event id %" PRIu64 " is outside range.\n", id);
	482	return NULL;
	483	}
	484	event = g_ptr_array_index(stream->events_by_id, id);
	485	if (!event) {
	486	fprintf(stdout, "[error] Event id %" PRIu64 " is unknown.\n", id);
	487	return NULL;
	488	}
	489	event_class = g_ptr_array_index(stream_class->events_by_id, id);
	490	if (!event_class) {
	491	fprintf(stdout, "[error] Event id %" PRIu64 " is unknown.\n", id);
	492	return NULL;
	493	}
	494
	495	return event;
	496	}
	497
8ee2fac4 JD	498	static
8ee2fac4 JD	499	void process_callbacks(struct babeltrace_iter *iter,
e73cc54c	500	struct ctf_stream *stream)
8ee2fac4 JD	501	{
	502	struct bt_stream_callbacks *bt_stream_cb;
	503	struct bt_callback_chain *bt_chain;
	504	struct bt_callback *cb;
	505	int i;
	506	enum bt_cb_ret ret;
e73cc54c JD	507	struct bt_ctf_data ctf_data;
	508
	509	ctf_data.event = extract_ctf_stream_event(stream);
f970ae1d	510	ctf_data.stream = stream;
8ee2fac4 JD	511
	512	/* process all events callback first */
	513	if (iter->main_callbacks.callback) {
	514	for (i = 0; i < iter->main_callbacks.callback->len; i++) {
	515	cb = &g_array_index(iter->main_callbacks.callback, struct bt_callback, i);
	516	if (!cb)
	517	goto end;
e73cc54c	518	ret = cb->callback(&ctf_data, cb->private_data);
8ee2fac4	519	switch (ret) {
e73cc54c JD	520	case BT_CB_OK_STOP:
	521	case BT_CB_ERROR_STOP:
	522	goto end;
	523	default:
	524	break;
8ee2fac4 JD	525	}
	526	}
	527	}
	528
	529	/* process per event callbacks */
	530	bt_stream_cb = &g_array_index(iter->callbacks,
	531	struct bt_stream_callbacks, stream->stream_id);
	532	if (!bt_stream_cb \|\| !bt_stream_cb->per_id_callbacks)
	533	goto end;
	534
	535	if (stream->event_id > bt_stream_cb->per_id_callbacks->len)
	536	goto end;
	537	bt_chain = &g_array_index(bt_stream_cb->per_id_callbacks,
	538	struct bt_callback_chain, stream->event_id);
	539	if (!bt_chain \|\| !bt_chain->callback)
	540	goto end;
	541
	542	for (i = 0; i < bt_chain->callback->len; i++) {
	543	cb = &g_array_index(bt_chain->callback, struct bt_callback, i);
	544	if (!cb)
	545	goto end;
e73cc54c	546	ret = cb->callback(&ctf_data, cb->private_data);
8ee2fac4 JD	547	switch (ret) {
	548	case BT_CB_OK_STOP:
	549	case BT_CB_ERROR_STOP:
	550	goto end;
	551	default:
	552	break;
	553	}
	554	}
	555
	556	end:
	557	return;
	558	}
	559
9e501292 JD	560	int babeltrace_iter_read_event(struct babeltrace_iter *iter,
	561	struct ctf_stream **stream,
	562	struct ctf_stream_event **event)
	563	{
	564	struct ctf_file_stream *file_stream;
	565	int ret = 0;
	566
	567	file_stream = heap_maximum(iter->stream_heap);
	568	if (!file_stream) {
	569	/* end of file for all streams */
	570	ret = EOF;
	571	goto end;
	572	}
	573	*stream = &file_stream->parent;
	574	event = g_ptr_array_index((stream)->events_by_id, (*stream)->event_id);
8ee2fac4 JD	575
	576	if ((*stream)->stream_id > iter->callbacks->len)
	577	goto end;
	578
	579	process_callbacks(iter, *stream);
	580
9e501292 JD	581	end:
	582	return ret;
	583	}
	584
	585	int convert_trace(struct trace_descriptor *td_write,
	586	struct trace_collection *trace_collection_read)
	587	{
	588	struct babeltrace_iter *iter;
	589	struct ctf_stream *stream;
	590	struct ctf_stream_event *event;
	591	struct ctf_text_stream_pos *sout;
877fb419	592	struct trace_collection_pos begin_pos;
9e501292 JD	593	int ret = 0;
	594
	595	sout = container_of(td_write, struct ctf_text_stream_pos,
	596	trace_descriptor);
	597
877fb419 JD	598	begin_pos.type = BT_SEEK_BEGIN;
877fb419 JD	599	iter = babeltrace_iter_create(trace_collection_read, &begin_pos, NULL);
9e501292 JD	600	while (babeltrace_iter_read_event(iter, &stream, &event) == 0) {
9e501292 JD	601	ret = sout->parent.event_cb(&sout->parent, stream);
0d0f5149 MD	602	if (ret) {
	603	fprintf(stdout, "[error] Writing event failed.\n");
	604	goto end;
	605	}
9e501292 JD	606	ret = babeltrace_iter_next(iter);
9e501292 JD	607	if (ret < 0)
0d0f5149	608	goto end;
0d0f5149	609	}
0d0f5149	610	end:
9e501292	611	babeltrace_iter_destroy(iter);
46322b33 MD	612	return ret;
46322b33 MD	613	}
b88d6e85 MD	614
	615	static
	616	void __attribute__((constructor)) init_babeltrace_lib(void)
	617	{
	618	if (getenv("BABELTRACE_VERBOSE"))
	619	babeltrace_verbose = 1;
	620	if (getenv("BABELTRACE_DEBUG"))
	621	babeltrace_debug = 1;
	622	}