[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>

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;
	enum bt_cb_ret (*callback)(void *private_data, void *caller_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_hooks */
	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;
}

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

/* TODO: use begin_pos/end_pos */
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;

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

				ret = stream_read_event(file_stream);
				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)
{
	heap_free(iter->stream_heap);
	g_free(iter->stream_heap);
	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);
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;
	int ret = 0;

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

	iter = babeltrace_iter_create(trace_collection_read, NULL, 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;
}
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
063f7048 MD	37	struct stream_saved_pos {
	38	/*
	39	* Use file_stream pointer to check if the trace collection we
	40	* restore to match the one we saved from, for each stream.
	41	*/
	42	struct ctf_file_stream *file_stream;
	43	size_t cur_index; /* current index in packet index */
	44	ssize_t offset; /* offset from base, in bits. EOF for end of file. */
	45	};
	46
	47	struct babeltrace_saved_pos {
	48	struct trace_collection *tc;
	49	GArray stream_saved_pos; / Contains struct stream_saved_pos */
	50	};
	51
c34ccddd MD	52	struct bt_callback {
	53	int prio; /* Callback order priority. Lower first. Dynamically assigned from dependency graph. */
	54	void *private_data;
	55	enum bt_cb_ret (callback)(void private_data, void *caller_data);
	56	};
	57
	58	struct bt_callback_chain {
	59	GArray callback; / Array of struct bt_callback, ordered by priority */
	60	};
	61
	62	/*
	63	* per id callbacks need to be per stream class because event ID vs
	64	* event name mapping can vary from stream to stream.
	65	*/
	66	struct bt_stream_callbacks {
	67	GArray per_id_callbacks; / Array of struct bt_callback_chain */
	68	};
	69
70bd0a12 JD	70	/*
	71	* struct babeltrace_iter: data structure representing an iterator on a trace
	72	* collection.
	73	*/
	74	struct babeltrace_iter {
	75	struct ptr_heap *stream_heap;
	76	struct trace_collection *tc;
063f7048	77	struct trace_collection_pos *end_pos;
c34ccddd MD	78	GArray callbacks; / Array of struct bt_stream_hooks */
	79	struct bt_callback_chain main_callbacks; /* For all events */
	80	/*
	81	* Flag indicating if dependency graph needs to be recalculated.
	82	* Set by babeltrace_iter_add_callback(), and checked (and
	83	* cleared) by upon entry into babeltrace_iter_read_event().
	84	* babeltrace_iter_read_event() is responsible for calling dep
	85	* graph calculation if it sees this flag set.
	86	*/
	87	int recalculate_dep_graph;
	88	/*
	89	* Array of pointers to struct bt_dependencies, for garbage
	90	* collection. We're not using a linked list here because each
	91	* struct bt_dependencies can belong to more than one
	92	* babeltrace_iter.
	93	*/
	94	GPtrArray *dep_gc;
70bd0a12 JD	95	};
70bd0a12 JD	96
c34ccddd MD	97	struct bt_dependencies {
	98	GArray deps; / Array of GQuarks */
	99	int refcount; /* free when decremented to 0 */
	100	};
	101
	102	static
	103	struct bt_dependencies _babeltrace_dependencies_create(const char first,
	104	va_list ap)
	105	{
	106	const char *iter;
	107	struct bt_dependencies *dep;
	108
	109	dep = g_new0(struct bt_dependencies, 1);
	110	dep->refcount = 1;
	111	dep->deps = g_array_new(FALSE, TRUE, sizeof(GQuark));
	112	iter = first;
	113	while (iter) {
	114	GQuark q = g_quark_from_string(iter);
	115	g_array_append_val(dep->deps, q);
	116	iter = va_arg(ap, const char *);
	117	}
	118	return dep;
	119	}
	120
	121	struct bt_dependencies babeltrace_dependencies_create(const char first, ...)
	122	{
	123	va_list ap;
	124	struct bt_dependencies *deps;
	125
	126	va_start(ap, first);
	127	deps = _babeltrace_dependencies_create(first, ap);
	128	va_end(ap);
	129	return deps;
	130	}
	131
9e501292	132	static int stream_read_event(struct ctf_file_stream *sin)
46322b33 MD	133	{
	134	int ret;
	135
2d0bea29	136	ret = sin->pos.parent.event_cb(&sin->pos.parent, &sin->parent);
0d0f5149 MD	137	if (ret == EOF)
	138	return EOF;
	139	else if (ret) {
	140	fprintf(stdout, "[error] Reading event failed.\n");
	141	return ret;
46322b33	142	}
46322b33	143	return 0;
0d0f5149	144	}
46322b33	145
0d0f5149 MD	146	/*
	147	* returns true if a < b, false otherwise.
	148	*/
	149	int stream_compare(void a, void b)
	150	{
	151	struct ctf_file_stream s_a = a, s_b = b;
	152
2d0bea29	153	if (s_a->parent.timestamp < s_b->parent.timestamp)
0d0f5149 MD	154	return 1;
	155	else
	156	return 0;
46322b33 MD	157	}
46322b33 MD	158
063f7048 MD	159	/* TODO: use begin_pos/end_pos */
	160	struct babeltrace_iter babeltrace_iter_create(struct trace_collection tc,
	161	struct trace_collection_pos *begin_pos,
	162	struct trace_collection_pos *end_pos)
46322b33	163	{
afb48eae	164	int i, stream_id;
0d0f5149	165	int ret = 0;
9e501292	166	struct babeltrace_iter *iter;
0d0f5149	167
9e501292 JD	168	iter = malloc(sizeof(struct babeltrace_iter));
	169	if (!iter)
	170	goto error_malloc;
	171	iter->stream_heap = g_new(struct ptr_heap, 1);
	172	iter->tc = tc;
	173
	174	ret = heap_init(iter->stream_heap, 0, stream_compare);
	175	if (ret < 0)
	176	goto error_heap_init;
46322b33	177
9e501292	178	for (i = 0; i < tc->array->len; i++) {
afb48eae AA	179	struct ctf_trace *tin;
afb48eae AA	180	struct trace_descriptor *td_read;
46322b33	181
9e501292	182	td_read = g_ptr_array_index(tc->array, i);
afb48eae AA	183	tin = container_of(td_read, struct ctf_trace, parent);
	184
	185	/* Populate heap with each stream */
	186	for (stream_id = 0; stream_id < tin->streams->len;
	187	stream_id++) {
	188	struct ctf_stream_class *stream;
	189	int filenr;
	190
	191	stream = g_ptr_array_index(tin->streams, stream_id);
	192	if (!stream)
0d0f5149	193	continue;
afb48eae AA	194	for (filenr = 0; filenr < stream->streams->len;
	195	filenr++) {
	196	struct ctf_file_stream *file_stream;
	197
	198	file_stream = g_ptr_array_index(stream->streams,
	199	filenr);
	200
9e501292	201	ret = stream_read_event(file_stream);
afb48eae AA	202	if (ret == EOF) {
	203	ret = 0;
	204	continue;
	205	} else if (ret) {
9e501292	206	goto error;
afb48eae AA	207	}
afb48eae AA	208	/* Add to heap */
9e501292 JD	209	ret = heap_insert(iter->stream_heap, file_stream);
	210	if (ret)
	211	goto error;
46322b33 MD	212	}
	213	}
	214	}
	215
9e501292	216	return iter;
0d0f5149	217
9e501292 JD	218	error:
	219	heap_free(iter->stream_heap);
	220	error_heap_init:
	221	g_free(iter->stream_heap);
	222	free(iter);
	223	error_malloc:
	224	return NULL;
	225	}
	226
	227	void babeltrace_iter_destroy(struct babeltrace_iter *iter)
	228	{
	229	heap_free(iter->stream_heap);
	230	g_free(iter->stream_heap);
	231	free(iter);
	232	}
	233
	234	int babeltrace_iter_next(struct babeltrace_iter *iter)
	235	{
	236	struct ctf_file_stream file_stream, removed;
	237	int ret;
	238
	239	file_stream = heap_maximum(iter->stream_heap);
	240	if (!file_stream) {
	241	/* end of file for all streams */
	242	ret = 0;
	243	goto end;
	244	}
	245
	246	ret = stream_read_event(file_stream);
	247	if (ret == EOF) {
	248	removed = heap_remove(iter->stream_heap);
	249	assert(removed == file_stream);
	250	ret = 0;
	251	goto end;
	252	} else if (ret) {
	253	goto end;
	254	}
	255	/* Reinsert the file stream into the heap, and rebalance. */
	256	removed = heap_replace_max(iter->stream_heap, file_stream);
	257	assert(removed == file_stream);
	258
	259	end:
	260	return ret;
	261	}
	262
	263	int babeltrace_iter_read_event(struct babeltrace_iter *iter,
	264	struct ctf_stream **stream,
	265	struct ctf_stream_event **event)
	266	{
	267	struct ctf_file_stream *file_stream;
	268	int ret = 0;
	269
	270	file_stream = heap_maximum(iter->stream_heap);
	271	if (!file_stream) {
	272	/* end of file for all streams */
	273	ret = EOF;
	274	goto end;
	275	}
	276	*stream = &file_stream->parent;
	277	event = g_ptr_array_index((stream)->events_by_id, (*stream)->event_id);
	278	end:
	279	return ret;
	280	}
	281
282	int convert_trace(struct trace_descriptor *td_write,
283	struct trace_collection *trace_collection_read)
284	{
285	struct babeltrace_iter *iter;
286	struct ctf_stream *stream;
287	struct ctf_stream_event *event;
288	struct ctf_text_stream_pos *sout;
289	int ret = 0;
290
291	sout = container_of(td_write, struct ctf_text_stream_pos,
292	trace_descriptor);
293
063f7048	294	iter = babeltrace_iter_create(trace_collection_read, NULL, NULL);
9e501292 JD	295	while (babeltrace_iter_read_event(iter, &stream, &event) == 0) {
9e501292 JD	296	ret = sout->parent.event_cb(&sout->parent, stream);
0d0f5149 MD	297	if (ret) {
	298	fprintf(stdout, "[error] Writing event failed.\n");
	299	goto end;
	300	}
9e501292 JD	301	ret = babeltrace_iter_next(iter);
9e501292 JD	302	if (ret < 0)
0d0f5149	303	goto end;
0d0f5149	304	}
0d0f5149	305	end:
9e501292	306	babeltrace_iter_destroy(iter);
46322b33 MD	307	return ret;
46322b33 MD	308	}