Add emulator player to read events

src/emu/CMakeLists.txt

@@ -17,6 +17,7 @@ add_library(emu STATIC
   emu_args.c
   emu_stream.c
   emu_trace.c
+  emu_player.c
   chan.c
   bay.c
   mux.c

src/emu/emu.c

@@ -43,7 +43,13 @@ emu_init(struct emu *emu, int argc, char *argv[])
 
 	/* Parse the trace and build the emu_system */
 	if (emu_system_init(&emu->system, &emu->args, &emu->trace) != 0) {
-		err("emu_init: cannot parse trace '%s'\n",
+		err("emu_init: cannot init system for trace '%s'\n",
+				emu->args.tracedir);
+		return -1;
+	}
+
+	if (emu_player_init(&emu->player, &emu->trace) != 0) {
+		err("emu_init: cannot init player for trace '%s'\n",
 				emu->args.tracedir);
 		return -1;
 	}
@@ -53,3 +59,22 @@ emu_init(struct emu *emu, int argc, char *argv[])
 
 	return 0;
 }
+
+int
+emu_step(struct emu *emu)
+{
+	int ret = emu_player_step(&emu->player);
+
+	/* No more events */
+	if (ret > 0)
+		return +1;
+
+	/* Error happened */
+	if (ret < 0) {
+		err("emu_step: emu_player_step failed\n");
+		return -1;
+	}
+
+	/* Otherwise progress */
+	return 0;
+}

src/emu/emu.h

@@ -9,6 +9,7 @@
 #include "emu_trace.h"
 #include "emu_args.h"
 #include "emu_system.h"
+#include "emu_player.h"
 
 enum error_values {
 	ST_BAD = 666,
@@ -36,12 +37,14 @@ struct emu {
 	struct emu_args args;
 	struct emu_trace trace;
 	struct emu_system system;
+	struct emu_player player;
 
 	struct model_spec *model[256];
 	void *model_ctx[256];
 };
 
 int emu_init(struct emu *emu, int argc, char *argv[]);
+int emu_step(struct emu *emu);
 int emu_model_register(struct emu *emu, struct model_spec *spec, void *ctx);
 void *emu_model_get_context(struct emu *emu, struct model_spec *spec, int model);
 

src/emu/emu_player.c

@@ -0,0 +1,139 @@
+#include "emu_player.h"
+
+#include "heap.h"
+#include "utlist.h"
+
+/*
+ * heap_node_compare_t - comparison function, returns:
+
+ * 	> 0 if a > b
+ * 	< 0 if a < b
+ * 	= 0 if a == b
+ *
+ * Invert the comparison function to get a min-heap instead
+ */
+static inline int
+stream_cmp(heap_node_t *a, heap_node_t *b)
+{
+	struct emu_stream *sa, *sb;
+
+	sa = heap_elem(a, struct emu_stream, hh);
+	sb = heap_elem(b, struct emu_stream, hh);
+
+	int64_t ca = emu_stream_lastclock(sa);
+	int64_t cb = emu_stream_lastclock(sb);
+
+	/* Return the opposite, so we have min-heap */
+	if (ca < cb)
+		return +1;
+	else if (ca > cb)
+		return -1;
+	else
+		return 0;
+}
+
+static int
+step_stream(struct emu_player *player, struct emu_stream *stream)
+{
+	if (!stream->active)
+		return +1;
+
+	int ret = emu_stream_step(stream);
+
+	if (ret < 0) {
+		err("step_stream: cannot step stream '%s'\n",
+				stream->relpath);
+		return -1;
+	} else if (ret > 0) {
+		return ret;
+	}
+
+	heap_insert(&player->heap, &stream->hh, &stream_cmp);
+
+	return 0;
+}
+
+int
+emu_player_init(struct emu_player *player, struct emu_trace *trace)
+{
+	memset(player, 0, sizeof(struct emu_player));
+
+	heap_init(&player->heap);
+
+	player->first_event = 1;
+	player->stream = NULL;
+
+	/* Load initial streams and events */
+	struct emu_stream *stream;
+	DL_FOREACH(trace->streams, stream) {
+		int ret = step_stream(player, stream);
+		if (ret > 0) {
+			/* No more events */
+			continue;
+		} else if (ret < 0) {
+			err("emu_player_init: step_stream failed\n");
+			return -1;
+		}
+	}
+
+	return 0;
+}
+
+/* Returns -1 on error, +1 if there are no more events and 0 if next event
+ * loaded properly */
+int
+emu_player_step(struct emu_player *player)
+{
+	/* Add the stream back if still active */
+	if (player->stream != NULL && step_stream(player, player->stream) < 0) {
+		err("player_step: step_stream() failed\n");
+		return -1;
+	}
+
+	/* Extract the next stream based on the lastclock */
+	heap_node_t *node = heap_pop_max(&player->heap, stream_cmp);
+
+	/* No more streams */
+	if (node == NULL)
+		return +1;
+
+	struct emu_stream *stream = heap_elem(node, struct emu_stream, hh);
+
+	if (stream == NULL) {
+		err("player_step: heap_elem() returned NULL\n");
+		return -1;
+	}
+
+	/* This can happen if two events are not ordered in the stream, but the
+	 * emulator picks other events in the middle. Example:
+	 *
+	 * Stream A: 10 3 ...
+	 * Stream B: 5 12
+	 *
+	 * emulator output:
+	 * 5
+	 * 10
+	 * 3  -> error!
+	 * 12
+	 * ...
+	 * */
+	int64_t sclock = emu_stream_lastclock(stream);
+
+	if (player->first_event) {
+		player->first_event = 0;
+		player->firstclock = sclock;
+		player->lastclock = sclock;
+	}
+
+	if (sclock < player->lastclock) {
+		err("emu_player_step: backwards jump in time %ld -> %ld in stream '%s'\n",
+				player->lastclock, sclock, stream->relpath);
+		return -1;
+	}
+
+	player->stream = stream;
+	player->lastclock = sclock;
+	player->deltaclock = player->lastclock - player->firstclock;
+
+	return 0;
+}

src/emu/emu_player.h

@@ -0,0 +1,25 @@
+/* Copyright (c) 2021-2023 Barcelona Supercomputing Center (BSC)
+ * SPDX-License-Identifier: GPL-3.0-or-later */
+
+#ifndef EMU_PLAYER_H
+#define EMU_PLAYER_H
+
+#include "emu_trace.h"
+
+#include <linux/limits.h>
+
+struct emu_player {
+	heap_head_t heap;
+	int64_t firstclock;
+	int64_t lastclock;
+	int64_t deltaclock;
+	int first_event;
+	struct emu_stream *stream;
+};
+
+int emu_player_init(struct emu_player *player, struct emu_trace *trace);
+int emu_player_step(struct emu_player *player);
+void emu_player_ev(struct emu_player *player, struct ovni_ev *ev);
+void emu_player_stream(struct emu_player *player, struct emu_stream *stream);
+
+#endif /* EMU_PLAYER_H */

src/emu/emu_stream.c

@@ -14,7 +14,7 @@ check_stream_header(struct emu_stream *stream)
 {
 	int ret = 0;
 
-	if (stream->size < sizeof(struct ovni_stream_header)) {
+	if (stream->size < (int64_t) sizeof(struct ovni_stream_header)) {
 		err("stream '%s': incomplete stream header\n",
 				stream->path);
 		return -1;
@@ -104,11 +104,18 @@ emu_stream_load(struct emu_stream *stream, const char *tracedir, const char *rel
 	}
 
 	stream->offset = sizeof(struct ovni_stream_header);
+	stream->usize = stream->size - stream->offset;
 
-	if (stream->offset == stream->size)
+	if (stream->offset < stream->size) {
-		stream->active = 0;
-	else
 		stream->active = 1;
+	} else if (stream->offset == stream->size) {
+		err("warning: stream '%s' has zero events\n", stream->relpath);
+		stream->active = 0;
+	} else {
+		err("emu_stream_load: impossible, offset %ld bigger than size %ld\n",
+				stream->offset, stream->size);
+		return -1;
+	}
 
 	/* No need to keep the fd open */
 	if (close(fd)) {
@@ -131,8 +138,102 @@ emu_stream_data_get(struct emu_stream *stream)
 	return stream->data;
 }
 
-void
+int
-emu_stream_clkoff(struct emu_stream *stream, int64_t clkoff)
+emu_stream_clkoff_set(struct emu_stream *stream, int64_t clkoff)
 {
+	if (stream->cur_ev) {
+		die("emu_stream_clkoff_set: cannot set clokoff in started stream '%s'\n",
+				stream->relpath);
+		return -1;
+	}
+
+	if (stream->clock_offset != 0) {
+		err("emu_stream_clkoff_set: stream '%s' already has a clock offset\n",
+				stream->relpath);
+		return -1;
+	}
+
 	stream->clock_offset = clkoff;
+
+	return 0;
+}
+
+struct ovni_ev *
+emu_stream_ev(struct emu_stream *stream)
+{
+	return stream->cur_ev;
+}
+
+int64_t
+emu_stream_evclock(struct emu_stream *stream, struct ovni_ev *ev)
+{
+	return (int64_t) ovni_ev_get_clock(ev) + stream->clock_offset;
+}
+
+int64_t
+emu_stream_lastclock(struct emu_stream *stream)
+{
+	return stream->lastclock;
+}
+
+int
+emu_stream_step(struct emu_stream *stream)
+{
+	if (!stream->active) {
+		err("emu_stream_step: stream is inactive, cannot step\n");
+		return -1;
+	}
+
+	/* Only step the offset if we have loaded an event */
+	if (stream->cur_ev != NULL) {
+		stream->offset += ovni_ev_size(stream->cur_ev);
+
+		/* It cannot pass the size, otherwise we are reading garbage */
+		if (stream->offset > stream->size) {
+			err("emu_stream_step: stream offset %ld exceeds size %ld\n",
+					stream->offset, stream->size);
+			return -1;
+		}
+
+		/* We have reached the end */
+		if (stream->offset == stream->size) {
+			stream->active = 0;
+			stream->cur_ev = NULL;
+			return +1;
+		}
+	}
+
+	stream->cur_ev = (struct ovni_ev *) &stream->buf[stream->offset];
+
+	/* Ensure the event fits */
+	if (stream->offset + ovni_ev_size(stream->cur_ev) > stream->size) {
+		err("emu_stream_step: stream '%s' ends with incomplete event\n",
+				stream->relpath);
+		return -1;
+	}
+
+	/* Ensure the clock grows monotonically */
+	int64_t clock = emu_stream_evclock(stream, stream->cur_ev);
+	if (clock < stream->lastclock) {
+		err("clock goes backwards %ld -> %ld in stream '%s' at offset %ld\n",
+				stream->lastclock,
+				clock,
+				stream->relpath,
+				stream->offset);
+		return -1;
+	}
+	stream->lastclock = clock;
+
+	return 0;
+}
+
+double
+emu_stream_progress(struct emu_stream *stream)
+{
+	if (stream->usize == 0)
+		return 1.0;
+
+	int64_t uoffset = stream->offset - sizeof(struct ovni_stream_header);
+	double prog = (double) uoffset / (double) stream->usize;
+	return prog;
 }

src/emu/emu_stream.h

@@ -4,6 +4,7 @@
 #ifndef EMU_STREAM_H
 #define EMU_STREAM_H
 
+#include "ovni.h"
 #include "heap.h"
 #include <stdint.h>
 #include <linux/limits.h>
@@ -15,8 +16,9 @@ struct emu_stream {
 	char relpath[PATH_MAX]; /* To tracedir */
 
 	uint8_t *buf;
-	size_t size;
+	int64_t size;
-	size_t offset;
+	int64_t usize; /* Useful size for events */
+	int64_t offset;
 
 	int active;
 
@@ -29,13 +31,21 @@ struct emu_stream {
 	struct emu_stream *next;
 	struct emu_stream *prev;
 
+	struct ovni_ev *cur_ev;
+
 	void *data; /* To hold system details */
 };
 
 int emu_stream_load(struct emu_stream *stream,
 		const char *tracedir, const char *relpath);
 
-void emu_stream_clkoff(struct emu_stream *stream, int64_t clock_offset);
+int emu_stream_clkoff_set(struct emu_stream *stream, int64_t clock_offset);
+
+double emu_stream_progress(struct emu_stream *stream);
+int emu_stream_step(struct emu_stream *stream);
+struct ovni_ev *emu_stream_ev(struct emu_stream *stream);
+int64_t emu_stream_evclock(struct emu_stream *stream, struct ovni_ev *ev);
+int64_t emu_stream_lastclock(struct emu_stream *stream);
 
 void emu_stream_data_set(struct emu_stream *stream, void *data);
 void *emu_stream_data_get(struct emu_stream *stream);

src/emu/emu_system.c

@@ -850,7 +850,11 @@ init_offsets(struct emu_system *sys)
 	struct emu_thread *thread;
 	DL_FOREACH2(sys->threads, thread, gnext) {
 		struct emu_loom *loom = thread->proc->loom;
-		emu_stream_clkoff(thread->stream, loom->clock_offset);
+		int64_t offset = loom->clock_offset;
+		if (emu_stream_clkoff_set(thread->stream, offset) != 0) {
+			err("init_offsets: cannot set clock offset\n");
+			return -1;
+		}
 	}
 
 	return 0;

test/unit/emu.c

@@ -16,5 +16,14 @@ int main(void)
 	if (emu_init(&emu, argc, argv) != 0)
 		die("emu_init failed\n");
 
+	int ret = 0;
+
+	while ((ret = emu_step(&emu)) == 0) {
+		err("event clock %ld\n", emu.player.deltaclock);
+	}
+
+	if (ret < 0)
+		die("emu_step failed\n");
+
 	return 0;
 }

test/unit/emu_stream.c

@@ -28,6 +28,9 @@ test_ok(char *fname)
 	const char *relpath = &fname[5];
 	if (emu_stream_load(&stream, "/tmp", relpath) != 0)
 		die("emu_stream_load failed");
+
+	if (stream.active)
+		die("stream is active\n");
 }
 
 static void