Implement the clock offset correction

dump.c

@@ -12,16 +12,6 @@
 #include "ovni_trace.h"
 #include "emu.h"
 
-#define ENABLE_DEBUG
-
-#ifdef ENABLE_DEBUG
-#define dbg(...) fprintf(stderr, __VA_ARGS__);
-#else
-#define dbg(...)
-#endif
-
-#define err(...) fprintf(stderr, __VA_ARGS__);
-
 
 static void
 hexdump(uint8_t *buf, size_t size)

emu.c

@@ -8,12 +8,20 @@
 #include <stdatomic.h>
 #include <dirent.h> 
 #include <assert.h> 
+#include <unistd.h>
 
 #include "ovni.h"
 #include "ovni_trace.h"
 #include "emu.h"
 #include "prv.h"
 
+/* Obtains the corrected clock of the given event */
+int64_t
+evclock(struct ovni_stream *stream, struct ovni_ev *ev)
+{
+	return (int64_t) ovni_ev_get_clock(ev) + stream->clock_offset;
+}
+
 static void
 emit_ev(struct ovni_stream *stream, struct ovni_ev *ev)
 {
@@ -24,7 +32,7 @@ emit_ev(struct ovni_stream *stream, struct ovni_ev *ev)
 	//dbg("sizeof(*ev) = %d\n", sizeof(*ev));
 	//hexdump((uint8_t *) ev, sizeof(*ev));
 
-	clock = ovni_ev_get_clock(ev);
+	clock = evclock(stream, ev);
 
 	delta = clock - stream->lastclock;
 
@@ -126,7 +134,8 @@ next_event(struct ovni_emu *emu)
 	struct ovni_ev *ev;
 	struct ovni_stream *stream;
 	struct ovni_trace *trace;
-	static int64_t t0 = -1;
+
+	static int done_first = 0;
 
 	trace = &emu->trace;
 
@@ -144,10 +153,10 @@ next_event(struct ovni_emu *emu)
 			continue;
 
 		ev = stream->cur_ev;
-		if(f < 0 || ovni_ev_get_clock(ev) < minclock)
+		if(f < 0 || evclock(stream, ev) < minclock)
 		{
 			f = i;
-			minclock = ovni_ev_get_clock(ev);
+			minclock = evclock(stream, ev);
 		}
 	}
 
@@ -159,18 +168,21 @@ next_event(struct ovni_emu *emu)
 
 	set_current(emu, stream);
 
-	if(emu->lastclock > ovni_ev_get_clock(stream->cur_ev))
+	if(emu->lastclock > evclock(stream, stream->cur_ev))
 	{
 		fprintf(stdout, "warning: backwards jump in time %lu -> %lu\n",
-				emu->lastclock, ovni_ev_get_clock(stream->cur_ev));
+				emu->lastclock, evclock(stream, stream->cur_ev));
 	}
 
-	emu->lastclock = ovni_ev_get_clock(stream->cur_ev);
+	emu->lastclock = evclock(stream, stream->cur_ev);
 
-	if(t0 < 0)
+	if(!done_first)
-		t0 = emu->lastclock;
+	{
+		done_first = 1;
+		emu->firstclock = emu->lastclock;
+	}
 
-	emu->delta_time = emu->lastclock - t0;
+	emu->delta_time = emu->lastclock - emu->firstclock;
 
 	return 0;
 }
@@ -376,44 +388,162 @@ close_prvs(struct ovni_emu *emu)
 	fclose(emu->prv_cpu);
 }
 
+static void
+usage(int argc, char *argv[])
+{
+	err("Usage: emu [-c offsetfile] tracedir\n");
+	err("\n");
+	err("Options:\n");
+	err("  -c offsetfile      Use the given offset file to correct\n");
+	err("                     the clocks among nodes. It can be\n");
+	err("                     generated by the ovnisync program\n");
+	err("\n");
+	err("  tracedir           The output trace dir generated by ovni.\n");
+	err("\n");
+	err("The output PRV files are placed in the tracedir directory.\n");
+
+	exit(EXIT_FAILURE);
+}
+
+static void
+parse_args(struct ovni_emu *emu, int argc, char *argv[])
+{
+	int opt;
+
+	while((opt = getopt(argc, argv, "c:")) != -1)
+	{
+		switch(opt)
+		{
+			case 'c':
+				emu->clock_offset_file = optarg;
+				break;
+			default: /* '?' */
+				usage(argc, argv);
+		}
+	}
+
+	if(optind >= argc)
+	{
+		err("missing tracedir\n");
+		usage(argc, argv);
+	}
+
+	emu->tracedir = argv[optind];
+}
+
+static struct ovni_loom *
+find_loom_by_hostname(struct ovni_emu *emu, char *host)
+{
+	int i;
+	struct ovni_loom *loom;
+
+	for(i=0; i<emu->trace.nlooms; i++)
+	{
+		loom = &emu->trace.loom[i];
+
+		if(strcmp(loom->hostname, host) == 0)
+			return loom;
+	}
+
+	return NULL;
+}
+
+static void
+load_clock_offsets(struct ovni_emu *emu)
+{
+	FILE *f;
+	char buf[1024];
+	int i, rank;
+	double offset, std;
+	char host[HOST_NAME_MAX];
+	struct ovni_loom *loom;
+	struct ovni_trace *trace;
+	struct ovni_stream *stream;
+	
+	f = fopen(emu->clock_offset_file, "r");
+
+	if(f == NULL)
+	{
+		perror("fopen clock offset file failed\n");
+		abort();
+	}
+
+	/* Ignore header line */
+	fgets(buf, 1024, f);
+
+	while(fscanf(f, "%d %s %lf %lf", &rank, host, &offset, &std) == 4)
+	{
+		loom = find_loom_by_hostname(emu, host);
+
+		if(loom == NULL)
+		{
+			err("No loom has hostname %s\n", host);
+			abort();
+		}
+
+		if(loom->clock_offset != 0)
+		{
+			err("warning: loom at host %s already has a clock offset\n",
+					host);
+		}
+
+		loom->clock_offset = (int64_t) offset;
+	}
+
+	/* Then populate the stream offsets */
+
+	trace = &emu->trace;
+
+	for(i=0; i<trace->nstreams; i++)
+	{
+		stream = &trace->stream[i];
+		loom = &trace->loom[stream->loom];
+		stream->clock_offset = loom->clock_offset;
+	}
+
+	fclose(f);
+}
+
+static void
+emu_init(struct ovni_emu *emu, int argc, char *argv[])
+{
+	memset(emu, 0, sizeof(emu));
+
+	parse_args(emu, argc, argv);
+
+	if(ovni_load_trace(&emu->trace, emu->tracedir))
+		abort();
+
+	if(ovni_load_streams(&emu->trace))
+		abort();
+
+	if(load_metadata(emu) != 0)
+		abort();
+
+	if(emu->clock_offset_file != NULL)
+		load_clock_offsets(emu);
+
+	open_prvs(emu, emu->tracedir);
+}
+
+static void
+emu_destroy(struct ovni_emu *emu)
+{
+	close_prvs(emu);
+	destroy_metadata(emu);
+	ovni_free_streams(&emu->trace);
+}
 
 int
 main(int argc, char *argv[])
 {
-	char *tracedir;
 	struct ovni_emu emu;
 
-	if(argc != 2)
+	emu_init(&emu, argc, argv);
-	{
-		fprintf(stderr, "missing tracedir\n");
-		exit(EXIT_FAILURE);
-	}
-
-	tracedir = argv[1];
-
-	memset(&emu, 0, sizeof(emu));
-
-	if(ovni_load_trace(&emu.trace, tracedir))
-		abort();
-
-	if(ovni_load_streams(&emu.trace))
-		abort();
-
-	if(load_metadata(&emu) != 0)
-		abort();
-
-	open_prvs(&emu, tracedir);
-
-	printf("#Paraver (19/01/38 at 03:14):00000000000000000000_ns:0:1:1(%d:1)\n", emu.total_cpus);
 
 	emulate(&emu);
 
-	close_prvs(&emu);
+	emu_destroy(&emu);
-
-	destroy_metadata(&emu);
-
-	ovni_free_streams(&emu.trace);
-
 
 	return 0;
 }

emu.h

@@ -8,8 +8,6 @@
 
 /* Debug macros */
 
-#define ENABLE_DEBUG
-
 #ifdef ENABLE_DEBUG
 # define dbg(...) fprintf(stderr, __VA_ARGS__);
 #else
@@ -140,13 +138,15 @@ struct ovni_cpu {
 /* State of each loom on post-process */
 struct ovni_loom {
 	size_t nprocs;
-	char name[HOST_NAME_MAX];
+	char hostname[HOST_NAME_MAX];
 
 	int max_ncpus;
 	int max_phyid;
 	int ncpus;
 	struct ovni_cpu cpu[OVNI_MAX_CPU];
 
+	int64_t clock_offset;
+
 	/* Virtual CPU */
 	struct ovni_cpu vcpu;
 
@@ -173,6 +173,7 @@ struct ovni_stream {
 	int active;
 	struct ovni_ev *cur_ev;
 	uint64_t lastclock;
+	int64_t clock_offset;
 };
 
 struct ovni_emu {
@@ -187,12 +188,16 @@ struct ovni_emu {
 
 	struct nosv_task *cur_task;
 
-	uint64_t lastclock;
+	int64_t firstclock;
+	int64_t lastclock;
 	int64_t delta_time;
 
 	FILE *prv_thread;
 	FILE *prv_cpu;
 
+	char *clock_offset_file;
+	char *tracedir;
+
 	/* Total counters */
 	int total_thread;
 	int total_proc;

emu_ovni.c

@@ -252,7 +252,9 @@ hook_pre_ovni(struct ovni_emu *emu)
 	{
 		case 'H': ev_thread(emu); break;
 		case 'A': ev_affinity(emu); break;
-		case 'B': dbg("burst %c\n", emu->cur_ev->header.value); break;
+		case 'B':
+			  //dbg("burst %c\n", emu->cur_ev->header.value);
+			  break;
 		default:
 			dbg("unknown ovni event class %c\n",
 					emu->cur_ev->header.class);

ovni.c

@@ -21,16 +21,6 @@
 #include "ovni_trace.h"
 #include "parson.h"
 
-#define ENABLE_DEBUG
-
-#ifdef ENABLE_DEBUG
-#define dbg(...) fprintf(stderr, __VA_ARGS__);
-#else
-#define dbg(...)
-#endif
-
-#define err(...) fprintf(stderr, __VA_ARGS__);
-
 //#define ENABLE_SLOW_CHECKS
 
 //#define USE_RDTSC
@@ -661,11 +651,7 @@ load_proc(struct ovni_eproc *proc, int index, int pid, char *procdir)
 	while((dirent = readdir(dir)) != NULL)
 	{
 		if(find_dir_prefix_int(dirent, "thread", &tid) != 0)
-		{
-			err("warning: ignoring bogus directory entry %s\n",
-					dirent->d_name);
 			continue;
-		}
 
 		sprintf(path, "%s/%s", procdir, dirent->d_name);
 
@@ -710,11 +696,7 @@ load_loom(struct ovni_loom *loom, int loomid, char *loomdir)
 	while((dirent = readdir(dir)) != NULL)
 	{
 		if(find_dir_prefix_int(dirent, "proc", &pid) != 0)
-		{
-			err("warning: ignoring bogus directory entry %s\n",
-					dirent->d_name);
 			continue;
-		}
 
 		sprintf(path, "%s/%s", loomdir, dirent->d_name);
 
@@ -777,7 +759,7 @@ ovni_load_trace(struct ovni_trace *trace, char *tracedir)
 		loom = &trace->loom[i];
 
 		/* FIXME: Unsafe */
-		strcpy(loom->name, loom_name);
+		strcpy(loom->hostname, loom_name);
 
 		sprintf(path, "%s/%s", tracedir, dirent->d_name);
 
@@ -861,7 +843,7 @@ ovni_load_streams(struct ovni_trace *trace)
 		return -1;
 	}
 
-	fprintf(stderr, "loaded %d streams\n", trace->nstreams);
+	err("loaded %d streams\n", trace->nstreams);
 
 	for(s=0, i=0; i<trace->nlooms; i++)
 	{
@@ -918,7 +900,6 @@ ovni_load_next_event(struct ovni_stream *stream)
 	{
 		stream->cur_ev = (struct ovni_ev *) stream->buf;
 		stream->offset = 0;
-		dbg("first event\n");
 		goto out;
 	}