Add partial support for the breakdown model

For now only Runtime and Task breakdown trees are implemented. The Idle
state is not decomposed.

src/emu/CMakeLists.txt

@@ -48,6 +48,7 @@ add_library(emu STATIC
   ovni/setup.c
   nanos6/setup.c
   nanos6/event.c
+  nanos6/breakdown.c
   nosv/setup.c
   nosv/event.c
   nodes/setup.c

src/emu/emu_prv.h

@@ -25,6 +25,7 @@ enum emu_prv_types {
 	PRV_NANOS6_RANK      = 38,
 	PRV_NANOS6_THREAD    = 39,
 	PRV_NANOS6_IDLE      = 40,
+	PRV_NANOS6_BREAKDOWN = 41,
 	PRV_KERNEL_CS        = 45,
 	PRV_RESERVED         = 100,
 };

src/emu/nanos6/breakdown.c

@@ -0,0 +1,213 @@
+/* Copyright (c) 2021-2023 Barcelona Supercomputing Center (BSC)
+ * SPDX-License-Identifier: GPL-3.0-or-later */
+
+//#define ENABLE_DEBUG
+
+#include "nanos6_priv.h"
+
+#include "emu_prv.h"
+
+static int
+create_cpu(struct bay *bay, struct breakdown_cpu *bcpu, int64_t gindex)
+{
+	enum chan_type t = CHAN_SINGLE;
+	chan_init(&bcpu->tr,  t, "nanos6.cpu%ld.breakdown.tr",  gindex);
+	chan_init(&bcpu->tri, t, "nanos6.cpu%ld.breakdown.tri", gindex);
+
+	/* Register all channels in the bay */
+	if (bay_register(bay, &bcpu->tr) != 0) {
+		err("bay_register tr failed");
+		return -1;
+	}
+	if (bay_register(bay, &bcpu->tri) != 0) {
+		err("bay_register tri failed");
+		return -1;
+	}
+
+	return 0;
+}
+
+int
+model_nanos6_breakdown_create(struct emu *emu)
+{
+	if (emu->args.breakdown == 0)
+		return 0;
+
+	struct nanos6_emu *memu = EXT(emu, '6');
+	struct breakdown_emu *bemu = &memu->brk;
+
+	/* Count phy cpus */
+	struct system *sys = &emu->system;
+	int nphycpus = sys->ncpus - sys->nlooms;
+
+	/* Create a new Paraver trace */
+	struct recorder *rec = &emu->recorder;
+	bemu->pvt = recorder_add_pvt(rec, "nanos6-breakdown", nphycpus);
+	if (bemu->pvt == NULL) {
+		err("recorder_add_pvt failed");
+		return -1;
+	}
+
+	if (sort_init(&bemu->sort, &emu->bay, nphycpus, "nanos6.breakdown.sort") != 0) {
+		err("sort_init failed");
+		return -1;
+	}
+
+	int64_t i = 0;
+	for (struct cpu *cpu = sys->cpus; cpu; cpu = cpu->next) {
+		if (cpu->is_virtual)
+			continue;
+
+		struct nanos6_cpu *mcpu = EXT(cpu, '6');
+		struct breakdown_cpu *bcpu = &mcpu->brk;
+
+		if (create_cpu(&emu->bay, bcpu, cpu->gindex) != 0) {
+			err("create_cpu failed");
+			return -1;
+		}
+
+		i++;
+	}
+
+	return 0;
+}
+
+static int
+select_tr(struct mux *mux, struct value value, struct mux_input **input)
+{
+	/* Only select the task if we are in ST_TASK_BODY and the task_type has
+	 * a non-null value */
+
+	int64_t in_body = (value.type == VALUE_INT64 && value.i == ST_TASK_BODY);
+
+	if (in_body) {
+		struct value tt;
+		struct mux_input *ttinput = mux_get_input(mux, 1);
+		if (chan_read(ttinput->chan, &tt) != 0) {
+			err("chan_read failed");
+			return -1;
+		}
+
+		/* Only show task type if we have a task */
+		if (tt.type == VALUE_NULL)
+			in_body = 0;
+	}
+	
+	int64_t i = in_body;
+	char *inputs[] = { "subsystem", "task_type" };
+	dbg("selecting input %ld (%s)", i, inputs[i]);
+	*input = mux_get_input(mux, i);
+
+	return 0;
+}
+
+static int
+select_idle(struct mux *mux, struct value value, struct mux_input **input)
+{
+	char buf[128];
+	dbg("value is %s", value_str(value, buf));
+	if (value.type == VALUE_INT64 && value.i == ST_WORKER_IDLE) {
+		dbg("selecting input 1 (idle)");
+		*input = mux_get_input(mux, 1);
+	} else {
+		dbg("selecting input 0 (tr)");
+		*input = mux_get_input(mux, 0);
+	}
+
+	return 0;
+}
+
+static int
+connect_cpu(struct bay *bay, struct nanos6_cpu *mcpu)
+{
+	struct breakdown_cpu *bcpu = &mcpu->brk;
+
+	/* Channel aliases */
+	struct chan *ss = &mcpu->m.track[CH_SUBSYSTEM].ch;
+	struct chan *tt = &mcpu->m.track[CH_TYPE].ch;
+	struct chan *idle = &mcpu->m.track[CH_IDLE].ch;
+	struct chan *tr = &bcpu->tr;
+	struct chan *tri = &bcpu->tri;
+
+	/* Connect mux0 using ss as select */
+	if (mux_init(&bcpu->mux0, bay, ss, tr, select_tr, 2) != 0) {
+		err("mux_init failed");
+		return -1;
+	}
+
+	if (mux_set_input(&bcpu->mux0, 0, ss) != 0) {
+		err("mux_set_input ss failed");
+		return -1;
+	}
+
+	if (mux_set_input(&bcpu->mux0, 1, tt) != 0) {
+		err("mux_set_input tt failed");
+		return -1;
+	}
+
+	/* Connect mux 1 using idle as select */
+	if (mux_init(&bcpu->mux1, bay, idle, tri, select_idle, 2) != 0) {
+		err("mux_init failed");
+		return -1;
+	}
+
+	if (mux_set_input(&bcpu->mux1, 0, tr) != 0) {
+		err("mux_set_input tr failed");
+		return -1;
+	}
+
+	if (mux_set_input(&bcpu->mux1, 1, idle) != 0) {
+		err("mux_set_input idle failed");
+		return -1;
+	}
+
+	return 0;
+}
+
+int
+model_nanos6_breakdown_connect(struct emu *emu)
+{
+	if (emu->args.breakdown == 0)
+		return 0;
+
+	struct nanos6_emu *memu = EXT(emu, '6');
+	struct breakdown_emu *bemu = &memu->brk;
+	struct bay *bay = &emu->bay;
+	struct system *sys = &emu->system;
+
+	int64_t i = 0;
+	for (struct cpu *cpu = sys->cpus; cpu; cpu = cpu->next) {
+		if (cpu->is_virtual)
+			continue;
+
+		struct nanos6_cpu *mcpu = EXT(cpu, '6');
+		struct breakdown_cpu *bcpu = &mcpu->brk;
+
+		/* Connect tr and tri channels and muxes */
+		if (connect_cpu(bay, mcpu) != 0) {
+			err("connect_cpu failed");
+			return -1;
+		}
+
+		/* Connect tri to sort */
+		if (sort_set_input(&bemu->sort, i, &bcpu->tri) != 0) {
+			err("sort_set_input failed");
+			return -1;
+		}
+
+		/* Connect out to PRV */
+		struct prv *prv = pvt_get_prv(bemu->pvt);
+		long type = PRV_NANOS6_BREAKDOWN;
+		long flags = PRV_SKIPDUP;
+
+		struct chan *out = sort_get_output(&bemu->sort, i);
+		if (prv_register(prv, i, type, bay, out, flags)) {
+			err("prv_register failed");
+			return -1;
+		}
+
+		i++;
+	}
+
+	return 0;
+}

src/emu/nanos6/breakdown.h

@@ -0,0 +1,52 @@
+#ifndef BREAKDOWN_H
+#define BREAKDOWN_H
+
+/* 
+ * The breakdown model is implemented on top of the CPU subsystem, task_type and
+ * idle channels. The first mux0 selects the task type when the subsystem
+ * matches "Task body" otherwise forwards the subsystem as-is to tr. The second
+ * mux1 selects tr only when the CPU is not Idle, otherwise sets the output tri
+ * as Idle.
+ *
+ *            +--------+
+ *            |        |
+ *            |        v
+ *            |     +------+
+ * subsystem -+-->--|      |
+ *                  | mux0 |              +------+
+ * task_type ---->--|      |-->-- tr -->--|      |
+ *                  +------+              | mux1 |-->-- tri
+ * idle --------->-------------------+->--|      |
+ *                                   |    +------+
+ *                                   |        ^
+ *                                   |        |
+ *                                   +--------+
+ *
+ * Then the sort module takes the output tri of each CPU and sorts the values
+ * which are propagated to the PRV directly.
+ *
+ *                 +------+       +-----+
+ * cpu0.tri --->---|      |--->---|     |
+ * ...             | sort |  ...  | PRV |
+ * cpuN.tri --->---|      |--->---|     |
+ *                 +------+       +-----+
+ */
+
+#include "mux.h"
+#include "chan.h"
+#include "sort.h"
+#include "pv/pvt.h"
+
+struct breakdown_cpu {
+	struct mux  mux0;
+	struct chan tr;
+	struct mux  mux1;
+	struct chan tri;
+};
+
+struct breakdown_emu {
+	struct sort sort;
+	struct pvt *pvt;
+};
+
+#endif /* BREAKDOWN_H */

src/emu/nanos6/event.c

@@ -60,8 +60,8 @@ static const int ss_table[256][256][3] = {
 		[']'] = { CHSS, POP,  ST_TASK_FOR },
 	},
 	['t'] = {
-		['['] = { CHSS, PUSH, ST_TASK_BODY },
+		['['] = { CHSS, IGN, -1 },
-		[']'] = { CHSS, POP,  ST_TASK_BODY },
+		[']'] = { CHSS, IGN, -1 },
 	},
 	['M'] = {
 		['a'] = { CHSS, PUSH, ST_ALLOCATING },
@@ -377,6 +377,27 @@ enforce_task_rules(struct emu *emu, char tr, struct task *next)
 	return 0;
 }
 
+static int
+update_task_ss_channel(struct emu *emu, char tr)
+{
+	struct nanos6_thread *th = EXT(emu->thread, '6');
+	struct chan *ss = &th->m.ch[CH_SUBSYSTEM];
+
+	if (tr == 'x') {
+		if (chan_push(ss, value_int64(ST_TASK_BODY)) != 0) {
+			err("chan_push subsystem failed");
+			return -1;
+		}
+	} else if (tr == 'e') {
+		if (chan_pop(ss, value_int64(ST_TASK_BODY)) != 0) {
+			err("chan_pop subsystem failed");
+			return -1;
+		}
+	}
+
+	return 0;
+}
+
 static int
 update_task(struct emu *emu)
 {
@@ -393,6 +414,12 @@ update_task(struct emu *emu)
 
 	struct task *next = task_get_running(stack);
 
+	/* Update the subsystem channel */
+	if (update_task_ss_channel(emu, emu->ev->v) != 0) {
+		err("update_task_ss_channel failed");
+		return -1;
+	}
+
 	int was_running = (prev != NULL);
 	int runs_now = (next != NULL);
 	char tr;
@@ -402,7 +429,10 @@ update_task(struct emu *emu)
 	}
 
 	/* Update the task related channels now */
-	update_task_channels(emu, tr, prev, next);
+	if (update_task_channels(emu, tr, prev, next) != 0) {
+		err("update_task_channels failed");
+		return -1;
+	}
 
 	if (enforce_task_rules(emu, tr, next) != 0) {
 		err("enforce_task_rules failed");

src/emu/nanos6/nanos6_priv.h

@@ -6,8 +6,10 @@
 
 #include "emu.h"
 #include "task.h"
+#include "sort.h"
 #include "model_cpu.h"
 #include "model_thread.h"
+#include "breakdown.h"
 
 /* Private enums */
 
@@ -40,6 +42,7 @@ enum nanos6_ss_state {
 	ST_FREEING,
 	ST_HANDLING_TASK,
 	ST_WORKER_LOOP,
+	ST_WORKER_INIT,
 	ST_SWITCH_TO,
 	ST_MIGRATE,
 	ST_SUSPEND,
@@ -62,7 +65,9 @@ enum nanos6_thread_type {
 };
 
 enum nanos6_worker_idle {
-	ST_WORKER_IDLE = 1,
+	/* Can mix with subsystem values */
+	ST_WORKER_IDLE = 100,
+	ST_WORKER_BUSY = 101,
 };
 
 struct nanos6_thread {
@@ -72,16 +77,24 @@ struct nanos6_thread {
 
 struct nanos6_cpu {
 	struct model_cpu m;
+	struct breakdown_cpu brk;
 };
 
 struct nanos6_proc {
 	struct task_info task_info;
 };
 
+struct nanos6_emu {
+	struct breakdown_emu brk;
+};
+
 int model_nanos6_probe(struct emu *emu);
 int model_nanos6_create(struct emu *emu);
 int model_nanos6_connect(struct emu *emu);
 int model_nanos6_event(struct emu *emu);
 int model_nanos6_finish(struct emu *emu);
 
+int model_nanos6_breakdown_create(struct emu *emu);
+int model_nanos6_breakdown_connect(struct emu *emu);
+
 #endif /* NANOS6_PRIV_H */

src/emu/nanos6/setup.c

@@ -99,6 +99,7 @@ static const struct pcf_value_label nanos6_ss_values[] = {
 	{ ST_BLK_WAITFOR,      "Blocking: Wait for deadline" },
 	{ ST_HANDLING_TASK,    "Worker: Handling task" },
 	{ ST_WORKER_LOOP,      "Worker: Looking for work" },
+	{ ST_WORKER_INIT,      "Worker: Starting" },
 	{ ST_SWITCH_TO,        "Worker: Switching to another thread" },
 	{ ST_MIGRATE,          "Worker: Migrating CPU" },
 	{ ST_SUSPEND,          "Worker: Suspending thread" },
@@ -158,7 +159,7 @@ static const int th_track[CH_MAX] = {
 	[CH_SUBSYSTEM] = TRACK_TH_ACT,
 	[CH_RANK]      = TRACK_TH_RUN,
 	[CH_THREAD]    = TRACK_TH_ANY,
-	[CH_IDLE]      = TRACK_TH_ANY,
+	[CH_IDLE]      = TRACK_TH_RUN,
 };
 
 static const int cpu_track[CH_MAX] = {
@@ -259,6 +260,19 @@ model_nanos6_create(struct emu *emu)
 		}
 	}
 
+	struct nanos6_emu *e = calloc(1, sizeof(struct nanos6_emu));
+	if (e == NULL) {
+		err("calloc failed:");
+		return -1;
+	}
+
+	extend_set(&emu->ext, model_id, e);
+
+	if (model_nanos6_breakdown_create(emu) != 0) {
+		err("model_nanos6_breakdown_connect failed");
+		return -1;
+	}
+
 	return 0;
 }
 
@@ -275,6 +289,34 @@ model_nanos6_connect(struct emu *emu)
 		return -1;
 	}
 
+	if (model_nanos6_breakdown_connect(emu) != 0) {
+		err("model_nanos6_breakdown_connect failed");
+		return -1;
+	}
+
+	for (struct thread *th = emu->system.threads; th; th = th->gnext) {
+		struct nanos6_thread *mth = EXT(th, model_id);
+		struct chan *idle = &mth->m.ch[CH_IDLE];
+		/* By default set all threads as Busy */
+		if (chan_push(idle, value_int64(ST_WORKER_BUSY)) != 0) {
+			err("chan_push idle failed");
+			return -1;
+		}
+		struct chan *ss = &mth->m.ch[CH_SUBSYSTEM];
+		/* And push initial subsystem to worker init */
+		if (chan_push(ss, value_int64(ST_WORKER_INIT)) != 0) {
+			err("chan_push idle failed");
+			return -1;
+		}
+	}
+
+	for (struct cpu *cpu = emu->system.cpus; cpu; cpu = cpu->next) {
+		struct nanos6_cpu *mcpu = EXT(cpu, model_id);
+		struct mux *mux = &mcpu->m.track[CH_IDLE].mux;
+		/* Emit Idle when a CPU has no idle threads */
+		mux_set_default(mux, value_int64(ST_WORKER_IDLE));
+	}
+
 	return 0;
 }
 
@@ -293,7 +335,7 @@ end_lint(struct emu *emu)
 		struct nanos6_thread *th = EXT(t, model_id);
 		struct chan *ch = &th->m.ch[CH_SUBSYSTEM];
 		int stacked = ch->data.stack.n;
-		if (stacked > 0) {
+		if (stacked > 1) {
 			struct value top;
 			if (chan_read(ch, &top) != 0) {
 				err("chan_read failed for subsystem");