starpu-max/tools/starpu_fxt_tool.c

/* StarPU --- Runtime system for heterogeneous multicore architectures.
 *
 * Copyright (C) 2011  Centre National de la Recherche Scientifique
 *
 * StarPU is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or (at
 * your option) any later version.
 *
 * StarPU is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 *
 * See the GNU Lesser General Public License in COPYING.LGPL for more details.
 */

#include "fxt_tool.h"
#include <inttypes.h>

/*
 *	Default user options
 */

static unsigned per_task_colour = 0;
static unsigned generate_distrib = 0;
static unsigned no_counter = 0;
static unsigned no_bus = 0;

/* TODO don't make that global ? */
struct fxt_ev_64 ev;
/* In case we are going to gather multiple traces (eg in the case of MPI
 * processes), we may need to prefix the name of the containers. */
char *prefix = "";
uint64_t offset = 0;
int rank = -1;

static uint64_t start_time = 0;
static uint64_t end_time = 0;

static int nworkers = 0;

//static char *filename = NULL;
/* XXX remove the 64 ... */
unsigned ninputfiles = 0;
static char *filenames[64];

static uint64_t last_codelet_hash[STARPU_NMAXWORKERS];
static double last_codelet_start[STARPU_NMAXWORKERS];
static char last_codelet_symbol[128][STARPU_NMAXWORKERS];

/* If more than a period of time has elapsed, we flush the profiling info,
 * otherwise they are accumulated everytime there is a new relevant event. */
#define ACTIVITY_PERIOD	75.0
static double last_activity_flush_timestamp[STARPU_NMAXWORKERS];
static double accumulated_sleep_time[STARPU_NMAXWORKERS];
static double accumulated_exec_time[STARPU_NMAXWORKERS];



LIST_TYPE(symbol_name,
	char *name;
);

static symbol_name_list_t symbol_list;

LIST_TYPE(communication,
	unsigned comid;
	float comm_start;	
	float bandwidth;
	unsigned node;
);

static communication_list_t communication_list;

/*
 * Paje trace file tools
 */

static char *out_paje_path = "paje.trace";
static FILE *out_paje_file;

static char *distrib_time_path = "distrib.data";
static FILE *distrib_time;

static char *activity_path = "activity.data";
static FILE *activity_file;

static void paje_output_file_init(void)
{
	/* create a new file */
	out_paje_file = fopen(out_paje_path, "w+");
	if (!out_paje_file)
	{
		perror("fopen");
		STARPU_ABORT();
	}

	write_paje_header(out_paje_file);

	fprintf(out_paje_file, "                                        \n \
	1       MPIP      0       \"MPI Program\"                      	\n \
	1       P      MPIP       \"Program\"                      	\n \
	1       Mn      P       \"Memory Node\"                         \n \
	1       T      Mn       \"Worker\"                               \n \
	1       Sc       P       \"Scheduler State\"                        \n \
	2       event   T       \"event type\"				\n \
	3       S       T       \"Thread State\"                        \n \
	3       MS       Mn       \"Memory Node State\"                        \n \
	4       ntask    Sc       \"Number of tasks\"                        \n \
	4       bw      Mn       \"Bandwidth\"                        \n \
	6       I       S      Initializing       \"0.0 .7 1.0\"            \n \
	6       D       S      Deinitializing       \"0.0 .1 .7\"            \n \
	6       Fi       S      FetchingInput       \"1.0 .1 1.0\"            \n \
	6       Po       S      PushingOutput       \"0.1 1.0 1.0\"            \n \
	6       E       S       Executing       \".0 .6 .4\"            \n \
	6       C       S       Callback       \".0 .3 .8\"            \n \
	6       B       S       Blocked         \".9 .1 .0\"		\n \
	6       Sl       S      Sleeping         \".9 .1 .0\"		\n \
	6       P       S       Progressing         \".4 .1 .6\"		\n \
	6       A       MS      Allocating         \".4 .1 .0\"		\n \
	6       Ar       MS      AllocatingReuse       \".1 .1 .8\"		\n \
	6       R       MS      Reclaiming         \".0 .1 .4\"		\n \
	6       Co       MS     DriverCopy         \".3 .5 .1\"		\n \
	6       No       MS     Nothing         \".0 .0 .0\"		\n \
	5       MPIL     MPIP	P	P      MPIL\n \
	5       L       P	Mn	Mn      L\n");

	fprintf(out_paje_file, "7      0.0 MPIroot      MPIP      0       root\n");
}

/*
 * Generic tools
 */

static float get_event_time_stamp(void)
{
	return (float)((ev.time-offset)/1000000.0);
}

static int register_worker_id(unsigned long tid)
{
	int workerid = nworkers++;

	/* create a new key in the htable */
	char *tidstr = malloc(16*sizeof(char));
	sprintf(tidstr, "%ld", tid);

	ENTRY item;
		item.key = tidstr;
		item.data = (void *)(uintptr_t)workerid;

	ENTRY *res;
	res = hsearch(item, FIND);

	/* only register a thread once */
	STARPU_ASSERT(res == NULL);

	res = hsearch(item, ENTER);
	STARPU_ASSERT(res);

	return workerid;
}

static int find_worker_id(unsigned long tid)
{
	char tidstr[16];
	sprintf(tidstr, "%ld", tid);

	ENTRY item;
		item.key = tidstr;
		item.data = NULL;
	ENTRY *res;
	res = hsearch(item, FIND);
	if (!res)
		return -1;

	int id = (uintptr_t)(res->data);

	return id;
}

static void update_accumulated_time(int worker, double sleep_time, double exec_time, double current_timestamp, int forceflush)
{
	accumulated_sleep_time[worker] += sleep_time;
	accumulated_exec_time[worker] += exec_time;

	/* If sufficient time has elapsed since the last flush, we have a new
	 * point in our graph */
	double elapsed = current_timestamp - last_activity_flush_timestamp[worker];
	if (forceflush || (elapsed > ACTIVITY_PERIOD))
	{		
		fprintf(activity_file, "%d\t%lf\t%lf\t%lf\t%lf\n", worker, current_timestamp, elapsed, accumulated_exec_time[worker], accumulated_sleep_time[worker]);

		/* reset the accumulated times */
		last_activity_flush_timestamp[worker] = current_timestamp;
		accumulated_sleep_time[worker] = 0.0;
		accumulated_exec_time[worker] = 0.0;
	}
}

/*
 *	Initialization
 */

static void handle_new_mem_node(void)
{
	fprintf(out_paje_file, "7       %f	%"PRIu64"      Mn      %sp	%sMEMNODE%"PRIu64"\n", get_event_time_stamp(), ev.param[0], prefix, prefix, ev.param[0]);

	if (!no_bus)
		fprintf(out_paje_file, "13       %f bw %sMEMNODE%"PRIu64" 0.0\n", 0.0f, prefix, ev.param[0]);
}

static void handle_worker_init_start(void)
{
	/* 
	   arg0 : type of worker (cuda, cpu ..)
	   arg1 : memory node
	   arg2 : thread id 
	*/
	fprintf(out_paje_file, "7       %f	%s%"PRIu64"      T      %sMEMNODE%"PRIu64"       %s%"PRIu64"\n",
		get_event_time_stamp(), prefix, ev.param[3], prefix, ev.param[2], prefix, ev.param[3]);

	int devid = ev.param[1];
	int workerid = register_worker_id(ev.param[3]);

	char *kindstr = "";

	switch (ev.param[0]) {
		case STARPU_FUT_APPS_KEY:
			set_next_other_worker_color(workerid);
			kindstr = "apps";
			break;
		case STARPU_FUT_CPU_KEY:
			set_next_cpu_worker_color(workerid);
			kindstr = "cpu";
			break;
		case STARPU_FUT_CUDA_KEY:
			set_next_cuda_worker_color(workerid);
			kindstr = "cuda";
			break;
		case STARPU_FUT_OPENCL_KEY:
			set_next_opencl_worker_color(workerid);
			kindstr = "opencl";
			break;
		default:
			STARPU_ABORT();
	}

	/* start initialization */
	fprintf(out_paje_file, "10       %f     S      %s%"PRIu64"      I\n",
			get_event_time_stamp(), prefix, ev.param[3]);

	fprintf(activity_file, "name\t%d\t%s %d\n", workerid, kindstr, devid);
}

static void handle_worker_init_end(void)
{
	fprintf(out_paje_file, "10       %f     S      %s%"PRIu64"      B\n",
			get_event_time_stamp(), prefix, ev.param[0]);

	/* Initilize the accumulated time counters */
	int worker = find_worker_id(ev.param[0]);
	last_activity_flush_timestamp[worker] = get_event_time_stamp();
	accumulated_sleep_time[worker] = 0.0;
	accumulated_exec_time[worker] = 0.0;
}

static void handle_worker_deinit_start(void)
{
	fprintf(out_paje_file, "10       %f     S      %s%"PRIu64"      D\n",
			get_event_time_stamp(), prefix, ev.param[0]);
}

static void handle_worker_deinit_end(void)
{
	fprintf(out_paje_file, "8       %f	%s%"PRIu64"	T\n",
			get_event_time_stamp(), prefix, ev.param[1]);
}

static void create_paje_state_if_not_found(char *name)
{
	symbol_name_itor_t itor;
	for (itor = symbol_name_list_begin(symbol_list);
		itor != symbol_name_list_end(symbol_list);
		itor = symbol_name_list_next(itor))
	{
		if (!strcmp(name, itor->name))
		{
			/* we found an entry */
			return;
		}
	}

	/* it's the first time ... */
	symbol_name_t entry = symbol_name_new();
		entry->name = malloc(strlen(name));
		strcpy(entry->name, name);

	symbol_name_list_push_front(symbol_list, entry);
	
	/* choose some colour ... that's disguting yes */
	unsigned hash_symbol_red = get_colour_symbol_red(name);
	unsigned hash_symbol_green = get_colour_symbol_green(name);
	unsigned hash_symbol_blue = get_colour_symbol_blue(name);

	fprintf(stderr, "name %s hash red %d green %d blue %d \n", name, hash_symbol_red, hash_symbol_green, hash_symbol_blue);

	uint32_t hash_sum = hash_symbol_red + hash_symbol_green + hash_symbol_blue;

	float red = (1.0f * hash_symbol_red) / hash_sum;
	float green = (1.0f * hash_symbol_green) / hash_sum;
	float blue = (1.0f * hash_symbol_blue) / hash_sum;

	/* create the Paje state */
	fprintf(out_paje_file, "6       %s       S       %s \"%f %f %f\" \n", name, name, red, green, blue);
}


static void handle_start_codelet_body(void)
{
	int worker;
	worker = find_worker_id(ev.param[1]);

	if (worker < 0) return;

	unsigned long has_name = ev.param[2];
	char *name = has_name?(char *)&ev.param[3]:"unknown";

	snprintf(last_codelet_symbol[worker], 128, "%s", name);

	/* TODO */
	last_codelet_hash[worker] = 0;

	float start_codelet_time = get_event_time_stamp();
	last_codelet_start[worker] = start_codelet_time;

	if (per_task_colour)
	{
		create_paje_state_if_not_found(name);

		fprintf(out_paje_file, "10       %f	S      %s%"PRIu64"      %s\n", start_codelet_time, prefix, ev.param[1], name);
	}
	else {
		fprintf(out_paje_file, "10       %f	S      %s%"PRIu64"      E\n", start_codelet_time, prefix, ev.param[1]);
	}

	end_time = STARPU_MAX(end_time, ev.time);
}

static void handle_end_codelet_body(void)
{
	int worker;
	worker = find_worker_id(ev.param[1]);
	if (worker < 0) return;

	float end_codelet_time = get_event_time_stamp();

	fprintf(out_paje_file, "10       %f	S      %s%"PRIu64"      B\n", end_codelet_time, prefix, ev.param[1]);

	float codelet_length = (end_codelet_time - last_codelet_start[worker]);

	update_accumulated_time(worker, 0.0, codelet_length, end_codelet_time, 0);
	
	if (generate_distrib)
	fprintf(distrib_time, "%s\t%s%d\t%"PRIx64"\t%f\n", last_codelet_symbol[worker],
				prefix, worker, last_codelet_hash[worker], codelet_length);

	end_time = STARPU_MAX(end_time, ev.time);
}

static void handle_user_event(void)
{
	int worker;
	unsigned long code = ev.param[0];

	worker = find_worker_id(ev.param[1]);
	if (worker < 0)
	{
		fprintf(out_paje_file, "9       %f     event      %sp      %ld\n", get_event_time_stamp(), prefix, code);
	}
	else {
		fprintf(out_paje_file, "9       %f     event      %s%"PRIu64"      %ld\n", get_event_time_stamp(), prefix, ev.param[1], code);
	}
}

static void handle_start_callback(void)
{
	int worker;
	worker = find_worker_id(ev.param[1]);
	if (worker < 0) return;
	fprintf(out_paje_file, "10       %f	S      %s%"PRIu64"      C\n", get_event_time_stamp(), prefix, ev.param[1] );
}

static void handle_end_callback(void)
{
	int worker;
	worker = find_worker_id(ev.param[1]);
	if (worker < 0) return;
	fprintf(out_paje_file, "10       %f	S      %s%"PRIu64"      B\n", get_event_time_stamp(), prefix, ev.param[1] );
}

static void handle_worker_status(const char *newstatus)
{
	int worker;
	worker = find_worker_id(ev.param[1]);
	if (worker < 0) return;

	fprintf(out_paje_file, "10       %f	S      %s%"PRIu64"      %s\n",
				get_event_time_stamp(), prefix, ev.param[1], newstatus);

	end_time = STARPU_MAX(end_time, ev.time);
}

static double last_sleep_start[STARPU_NMAXWORKERS];

static void handle_start_sleep(void)
{
	int worker;
	worker = find_worker_id(ev.param[0]);
	if (worker < 0) return;

	float start_sleep_time = get_event_time_stamp();
	last_sleep_start[worker] = start_sleep_time;

	fprintf(out_paje_file, "10       %f	S      %s%"PRIu64"      Sl\n",
				get_event_time_stamp(), prefix, ev.param[0]);

	end_time = STARPU_MAX(end_time, ev.time);
}

static void handle_end_sleep(void)
{
	int worker;
	worker = find_worker_id(ev.param[0]);
	if (worker < 0) return;

	float end_sleep_timestamp = get_event_time_stamp();

	fprintf(out_paje_file, "10       %f	S      %s%"PRIu64"      B\n",
				end_sleep_timestamp, prefix, ev.param[0]);

	double sleep_length = end_sleep_timestamp - last_sleep_start[worker];

	update_accumulated_time(worker, sleep_length, 0.0, end_sleep_timestamp, 0);

	end_time = STARPU_MAX(end_time, ev.time);
}



static void handle_data_copy(void)
{
}

static void handle_start_driver_copy(void)
{
	unsigned src = ev.param[0];
	unsigned dst = ev.param[1];
	unsigned size = ev.param[2];
	unsigned comid = ev.param[3];

	if (!no_bus)
	{
		fprintf(out_paje_file, "10       %f     MS      %sMEMNODE%d      Co\n", get_event_time_stamp(), prefix, dst);
		fprintf(out_paje_file, "18       %f	L      %sp	%d	%sMEMNODE%d	com_%d\n", get_event_time_stamp(), prefix, size, prefix, src, comid);

		/* create a structure to store the start of the communication, this will be matched later */
		communication_t com = communication_new();
		com->comid = comid;
		com->comm_start = get_event_time_stamp();

		/* that's a hack: either src or dst is non null */
		com->node = (src + dst);

		communication_list_push_back(communication_list, com);
	}

}

static void handle_end_driver_copy(void)
{
	unsigned dst = ev.param[1];
	unsigned size = ev.param[2];
	unsigned comid = ev.param[3];

	if (!no_bus)
	{
		fprintf(out_paje_file, "10       %f     MS      %sMEMNODE%d      No\n", get_event_time_stamp(), prefix, dst);
		fprintf(out_paje_file, "19       %f	L      %sp	%d	%sMEMNODE%d	com_%d\n", get_event_time_stamp(), prefix, size, prefix, dst, comid);

		/* look for a data transfer to match */
		communication_itor_t itor;
		for (itor = communication_list_begin(communication_list);
			itor != communication_list_end(communication_list);
			itor = communication_list_next(itor))
		{
			if (itor->comid == comid)
			{
				float comm_end = get_event_time_stamp();
				float bandwidth = (float)((0.001*size)/(comm_end - itor->comm_start));

				itor->bandwidth = bandwidth;

				communication_t com = communication_new();
				com->comid = comid;
				com->comm_start = get_event_time_stamp();
				com->bandwidth = -bandwidth;

				com->node = itor->node;

				communication_list_push_back(communication_list, com);

				break;
			}
		}
	}
}

static void display_bandwidth_evolution(void)
{
	float current_bandwidth = 0.0;
	float current_bandwidth_per_node[32] = {0.0};

	communication_itor_t itor;
	for (itor = communication_list_begin(communication_list);
		itor != communication_list_end(communication_list);
		itor = communication_list_next(itor))
	{
		current_bandwidth += itor->bandwidth;
		fprintf(out_paje_file, "13  %f bw %sMEMNODE0 %f\n",
				itor->comm_start, prefix, current_bandwidth);

		current_bandwidth_per_node[itor->node] +=  itor->bandwidth;
		fprintf(out_paje_file, "13  %f bw %sMEMNODE%d %f\n",
				itor->comm_start, prefix, itor->node, current_bandwidth_per_node[itor->node]);
	}
}

static void handle_memnode_event(const char *eventstr)
{
	unsigned memnode = ev.param[0];

	fprintf(out_paje_file, "10       %f     MS      %sMEMNODE%d      %s\n",
		get_event_time_stamp(), prefix, memnode, eventstr);
}

/*
 *	Number of task submitted to the scheduler
 */
static int curq_size = 0;

static void handle_job_push(void)
{
	float current_timestamp = get_event_time_stamp();

	curq_size++;

	if (!no_counter)
		fprintf(out_paje_file, "13       %f ntask %ssched %f\n", current_timestamp, prefix, (float)curq_size);


	fprintf(activity_file, "cnt_ready\t%lf\t%ld\n", current_timestamp, curq_size);
}

static void handle_job_pop(void)
{
	float current_timestamp = get_event_time_stamp();

	curq_size--;

	if (!no_counter)
		fprintf(out_paje_file, "13       %f ntask %ssched %f\n", current_timestamp, prefix, (float)curq_size);

	fprintf(activity_file, "cnt_ready\t%lf\t%ld\n", current_timestamp, curq_size);
}

void handle_update_task_cnt(void)
{
	float current_timestamp = get_event_time_stamp();
	unsigned long nsubmitted = ev.param[0]; 
	fprintf(activity_file, "cnt_submitted\t%lf\t%ld\n", current_timestamp, nsubmitted);
}

static void handle_codelet_tag_deps(void)
{
	uint64_t child;
	uint64_t father;

	child = ev.param[0]; 
	father = ev.param[1]; 

	add_deps(child, father);
}

static void handle_task_deps(void)
{
	unsigned long dep_prev = ev.param[0];
	unsigned long dep_succ = ev.param[1];

	/* There is a dependency between both job id : dep_prev -> dep_succ */
	add_task_deps(dep_prev, dep_succ);
}

static void handle_task_done(void)
{
	unsigned long job_id;
	job_id = ev.param[0];

	unsigned long has_name = ev.param[3];
	char *name = has_name?(char *)&ev.param[4]:"unknown";

        int worker;
        worker = find_worker_id(ev.param[1]);

	const char *colour;
	char buffer[32];
	if (per_task_colour) {
		snprintf(buffer, 32, "#%x%x%x",
			get_colour_symbol_red(name)/4,
			get_colour_symbol_green(name)/4,
			get_colour_symbol_blue(name)/4);
		colour = &buffer[0];
	}
	else {
		colour= (worker < 0)?"#aaaaaa":get_worker_color(worker);
	}

	unsigned exclude_from_dag = ev.param[2];

	if (!exclude_from_dag)
		dot_set_task_done(job_id, name, colour);
}

static void handle_tag_done(void)
{
	uint64_t tag_id;
	tag_id = ev.param[0];

	unsigned long has_name = ev.param[2];
	char *name = has_name?(char *)&ev.param[3]:"unknown";

        int worker;
        worker = find_worker_id(ev.param[1]);

	const char *colour;
	char buffer[32];
	if (per_task_colour) {
		snprintf(buffer, 32, "%.4f,%.4f,%.4f",
			get_colour_symbol_red(name)/1024.0,
			get_colour_symbol_green(name)/1024.0,
			get_colour_symbol_blue(name)/1024.0);
		colour = &buffer[0];
	}
	else {
		colour= (worker < 0)?"0.0,0.0,0.0":get_worker_color(worker);
	}

	dot_set_tag_done(tag_id, colour);
}

static void handle_mpi_barrier(void)
{
	rank = ev.param[0];

	/* Add an event in the trace */
	fprintf(out_paje_file, "9       %f     event      %sp      %d\n", get_event_time_stamp(), prefix, rank);
}

static void handle_mpi_isend(void)
{
	int dest = ev.param[0];
	int mpi_tag = ev.param[1];
	size_t size = ev.param[2];
	float date = get_event_time_stamp();

	add_mpi_send_transfer(rank, dest, mpi_tag, size, date);
}

static void handle_mpi_irecv_end(void)
{
	int src = ev.param[0];
	int mpi_tag = ev.param[1];
	float date = get_event_time_stamp();

	add_mpi_recv_transfer(src, rank, mpi_tag, date);
}

static void handle_set_profiling(void)
{
	int status = ev.param[0];

	fprintf(activity_file, "set_profiling\t%lf\t%d\n", get_event_time_stamp(), status);
}

static void handle_task_wait_for_all(void)
{
	dot_add_sync_point();
}

static void parse_args(int argc, char **argv)
{
	/* We want to support arguments such as "fxt_tool -i trace_*" */
	unsigned reading_input_filenames = 0;

	int i;
	for (i = 1; i < argc; i++) {
		if (strcmp(argv[i], "-c") == 0) {
			per_task_colour = 1;
			reading_input_filenames = 0;
			continue;
		}

		if (strcmp(argv[i], "-o") == 0) {
			out_paje_path = argv[++i];
			reading_input_filenames = 0;
			continue;
		}

		if (strcmp(argv[i], "-i") == 0) {
			filenames[ninputfiles++] = argv[++i];
			reading_input_filenames = 1;
			continue;
		}

		if (strcmp(argv[i], "-no-counter") == 0) {
			no_counter = 1;
			reading_input_filenames = 0;
			continue;
		}

		if (strcmp(argv[i], "-no-bus") == 0) {
			no_bus = 1;
			reading_input_filenames = 0;
			continue;
		}

		if (strcmp(argv[i], "-d") == 0) {
			generate_distrib = 1;
			reading_input_filenames = 0;
			continue;
		}

		if (strcmp(argv[i], "-h") == 0) {
		        fprintf(stderr, "Usage : %s [-c] [-no-counter] [-no-bus] [-i input_filename] [-o output_filename]\n", argv[0]);
			fprintf(stderr, "\t-c: use a different colour for every type of task.\n");
		        exit(-1);
		}

		/* That's pretty dirty: if the reading_input_filenames flag is
		 * set, and that the argument does not match an option, we
		 * assume this may be another filename */
		if (reading_input_filenames)
		{
			filenames[ninputfiles++] = argv[i];
			continue;
		}
	}
}

void parse_new_file(char *filename_in, char *file_prefix, uint64_t file_offset)
{
	prefix = file_prefix;
	offset = file_offset;

	/* Open the trace file */
	int fd_in;
	fd_in = open(filename_in, O_RDONLY);
	if (fd_in < 0) {
	        perror("open failed :");
	        exit(-1);
	}

	static fxt_t fut;
	fut = fxt_fdopen(fd_in);
	if (!fut) {
	        perror("fxt_fdopen :");
	        exit(-1);
	}
	
	fxt_blockev_t block;
	block = fxt_blockev_enter(fut);

	/* create a htable to identify each worker(tid) */
	hcreate(STARPU_NMAXWORKERS);

	symbol_list = symbol_name_list_new(); 
	communication_list = communication_list_new();

	/* TODO starttime ...*/
	/* create the "program" container */
	fprintf(out_paje_file, "7      0.0 %sp      P      MPIroot       program%s \n", prefix, prefix);
	/* create a variable with the number of tasks */
	if (!no_counter)
	{
		fprintf(out_paje_file, "7     %f    %ssched   Sc    %sp     scheduler \n", 0.0, prefix, prefix);
		fprintf(out_paje_file, "13    0.0    ntask %ssched 0.0\n", prefix);
	}

	unsigned first_event = 1;

	while(1) {
		int ret = fxt_next_ev(block, FXT_EV_TYPE_64, (struct fxt_ev *)&ev);
		if (ret != FXT_EV_OK) {
			fprintf(stderr, "no more block ...\n");
			break;
		}

		__attribute__ ((unused)) int nbparam = ev.nb_params;

		if (first_event)
		{
			first_event = 0;
			start_time = ev.time;
		}

		switch (ev.code) {
			case STARPU_FUT_WORKER_INIT_START:
				handle_worker_init_start();
				break;

			case STARPU_FUT_WORKER_INIT_END:
				handle_worker_init_end();
				break;

			case STARPU_FUT_NEW_MEM_NODE:
				handle_new_mem_node();
				break;

			/* detect when the workers were idling or not */
			case STARPU_FUT_START_CODELET_BODY:
				handle_start_codelet_body();
				break;
			case STARPU_FUT_END_CODELET_BODY:
				handle_end_codelet_body();
				break;

			case STARPU_FUT_START_CALLBACK:
				handle_start_callback();
				break;
			case STARPU_FUT_END_CALLBACK:
				handle_end_callback();
				break;

			case STARPU_FUT_UPDATE_TASK_CNT:
				handle_update_task_cnt();
				break;

			/* monitor stack size */
			case STARPU_FUT_JOB_PUSH:
				handle_job_push();
				break;
			case STARPU_FUT_JOB_POP:
				handle_job_pop();
				break;

			/* check the memory transfer overhead */
			case STARPU_FUT_START_FETCH_INPUT:
				handle_worker_status("Fi");
				break;

			case STARPU_FUT_START_PUSH_OUTPUT:
				handle_worker_status("Po");
				break;

			case STARPU_FUT_START_PROGRESS:
				handle_worker_status("P");
				break;

			case STARPU_FUT_END_FETCH_INPUT:
			case STARPU_FUT_END_PROGRESS:
			case STARPU_FUT_END_PUSH_OUTPUT:
				handle_worker_status("B");
				break;

			case STARPU_FUT_WORKER_SLEEP_START:
				handle_start_sleep();
				break;

			case STARPU_FUT_WORKER_SLEEP_END:
				handle_end_sleep();
				break;

			case STARPU_FUT_TAG:
				/* XXX */
				break;

			case STARPU_FUT_TAG_DEPS:
				handle_codelet_tag_deps();
				break;

			case STARPU_FUT_TASK_DEPS:
				handle_task_deps();
				break;

			case STARPU_FUT_TASK_DONE:
				handle_task_done();
				break;

			case STARPU_FUT_TAG_DONE:
				handle_tag_done();
				break;

			case STARPU_FUT_DATA_COPY:
				if (!no_bus)
				handle_data_copy();
				break;

			case STARPU_FUT_START_DRIVER_COPY:
				if (!no_bus)
				handle_start_driver_copy();
				break;

			case STARPU_FUT_END_DRIVER_COPY:
				if (!no_bus)
				handle_end_driver_copy();
				break;

			case STARPU_FUT_WORK_STEALING:
				/* XXX */
				break;

			case STARPU_FUT_WORKER_DEINIT_START:
				handle_worker_deinit_start();
				break;

			case STARPU_FUT_WORKER_DEINIT_END:
				handle_worker_deinit_end();
				break;

			case STARPU_FUT_START_ALLOC:
				if (!no_bus)
				handle_memnode_event("A");
				break;

			case STARPU_FUT_START_ALLOC_REUSE:
				if (!no_bus)
				handle_memnode_event("Ar");
				break;

			case STARPU_FUT_START_MEMRECLAIM:
				handle_memnode_event("R");
				break;

			case STARPU_FUT_END_ALLOC:
			case STARPU_FUT_END_ALLOC_REUSE:
			case STARPU_FUT_END_MEMRECLAIM:
				if (!no_bus)
				handle_memnode_event("No");
				break;

			case STARPU_FUT_USER_EVENT:
				handle_user_event();
				break;

			case FUT_MPI_BARRIER:
				handle_mpi_barrier();
				break;

			case FUT_MPI_ISEND:
				handle_mpi_isend();
				break;

			case FUT_MPI_IRECV_END:
				handle_mpi_irecv_end();
				break;

			case STARPU_FUT_SET_PROFILING:
				handle_set_profiling();
				break;

			case STARPU_FUT_TASK_WAIT_FOR_ALL:
				handle_task_wait_for_all();
				break;

			default:
				fprintf(stderr, "unknown event.. %x at time %llx WITH OFFSET %llx\n",
					(unsigned)ev.code, (long long unsigned)ev.time, (long long unsigned)(ev.time-offset));
				break;
		}
	}

	hdestroy();

	/* Close the trace file */
	if (close(fd_in))
	{
	        perror("close failed :");
	        exit(-1);
	}
}

/*
 * This program should be used to parse the log generated by FxT 
 */
int main(int argc, char **argv)
{
	int fd_out;

	parse_args(argc, argv);

	init_dag_dot();

	if (generate_distrib)
		distrib_time = fopen(distrib_time_path, "w+");

	activity_file = fopen(activity_path, "w+");

	paje_output_file_init();

	if (ninputfiles == 1)
	{
		/* we usually only have a single trace */
		uint64_t file_start_time = find_start_time(filenames[0]);
		parse_new_file(filenames[0], "", file_start_time);
	}
	else {
		unsigned inputfile;

		uint64_t offsets[64];
		uint64_t found_offsets[64];
		uint64_t start_times[64];

		uint64_t max = 0;

		/*
		 * Find the trace offsets:
		 *	- If there is no sync point
		 *		psi_k(x) = x - start_k
		 *	- If there is a sync point sync_k
		 *		psi_k(x) = x - sync_k + M
		 *		where M = max { sync_i - start_i | there exists sync_i}
		 * More generally:
		 *	- psi_k(x) = x - offset_k
		 */
		
		int unique_keys[64];
		int rank_k[64];
		uint64_t start_k[64];
		uint64_t sync_k[64];
		unsigned sync_k_exists[64];
		uint64_t M = 0;

		unsigned found_one_sync_point = 0;
		int key;
		unsigned display_mpi = 0; 

		/* Compute all start_k */
		for (inputfile = 0; inputfile < ninputfiles; inputfile++)
		{
			uint64_t file_start = find_start_time(filenames[inputfile]);
			start_k[inputfile] = file_start; 
		}

		/* Compute all sync_k if they exist */
		for (inputfile = 0; inputfile < ninputfiles; inputfile++)
		{
			int ret = find_sync_point(filenames[inputfile],
							&sync_k[inputfile],
							&unique_keys[inputfile],
							&rank_k[inputfile]);
			if (ret == -1)
			{
				/* There was no sync point, we assume there is no offset */
				sync_k_exists[inputfile] = 0;
			}
			else {
				if (!found_one_sync_point)
				{
					key = unique_keys[inputfile];
					display_mpi = 1;
					found_one_sync_point = 1;
				}
				else {
					if (key != unique_keys[inputfile])
					{
						fprintf(stderr, "Warning: traces are coming from different run so we will not try to display MPI communications.\n");
						display_mpi = 0;
					}
				}


				STARPU_ASSERT(sync_k[inputfile] >= start_k[inputfile]);

				sync_k_exists[inputfile] = 1;

				uint64_t diff = sync_k[inputfile] - start_k[inputfile];
				if (diff > M)
					M = diff;
			}
		}

		/* Compute the offset */
		for (inputfile = 0; inputfile < ninputfiles; inputfile++)
		{
			offsets[inputfile] = sync_k_exists[inputfile]?
						(sync_k[inputfile]-M):start_k[inputfile];
		}

		/* generate the Paje trace for the different files */
		for (inputfile = 0; inputfile < ninputfiles; inputfile++)
		{
			int filerank = rank_k[inputfile];

			fprintf(stderr, "Handle file %s (rank %d)\n", filenames[inputfile], filerank);

			char file_prefix[32];
			snprintf(file_prefix, 32, "mpi_%d_", filerank);
			parse_new_file(filenames[inputfile], file_prefix, offsets[inputfile]);
		}

		/* display the MPI transfers if possible */
		if (display_mpi)
		for (inputfile = 0; inputfile < ninputfiles; inputfile++)
		{
			int filerank = rank_k[inputfile];
			display_all_transfers_from_trace(out_paje_file, filerank);
		}
	}

	display_bandwidth_evolution();

	/* close the different files */
	fclose(out_paje_file);
	
	fclose(activity_file);

	if (generate_distrib)
		fclose(distrib_time);

	terminate_dat_dot();

	return 0;
}