From 259de4df43be8791f4d6459de3d87445532da286 Mon Sep 17 00:00:00 2001 From: Rodrigo Arias Date: Mon, 13 Feb 2023 18:36:54 +0100 Subject: [PATCH] Add track module and port Nanos6 model --- src/emu/CMakeLists.txt | 1 + src/emu/cpu.c | 9 +++ src/emu/cpu.h | 3 + src/emu/nanos6/connect.c | 138 +++++++++++----------------------- src/emu/nanos6/create.c | 82 +++++++------------- src/emu/nanos6/nanos6_priv.h | 21 ++---- src/emu/nanos6/pvt.c | 12 +-- src/emu/thread.c | 23 ++++++ src/emu/thread.h | 1 + src/emu/track.c | 140 +++++++++++++++++++++++++++++++++++ src/emu/track.h | 42 +++++++++++ 11 files changed, 301 insertions(+), 171 deletions(-) create mode 100644 src/emu/track.c create mode 100644 src/emu/track.h diff --git a/src/emu/CMakeLists.txt b/src/emu/CMakeLists.txt index 279de7f..fb3906e 100644 --- a/src/emu/CMakeLists.txt +++ b/src/emu/CMakeLists.txt @@ -34,6 +34,7 @@ add_library(emu STATIC recorder.c system.c task.c + track.c thread.c extend.c ovni/probe.c diff --git a/src/emu/cpu.c b/src/emu/cpu.c index 96a398e..ddccf44 100644 --- a/src/emu/cpu.c +++ b/src/emu/cpu.c @@ -285,3 +285,12 @@ cpu_migrate_thread(struct cpu *cpu, struct thread *thread, struct cpu *newcpu) return 0; } + +struct chan * +cpu_get_th_chan(struct cpu *cpu, enum track_th trackmode) +{ + if (trackmode != TRACK_TH_RUN) + die("cpu tracking must be running"); + + return &cpu->chan[CPU_CHAN_THRUN]; +} diff --git a/src/emu/cpu.h b/src/emu/cpu.h index 81b0e75..b78f7fa 100644 --- a/src/emu/cpu.h +++ b/src/emu/cpu.h @@ -8,6 +8,7 @@ struct cpu; /* Needed for thread */ #include "thread.h" #include "chan.h" +#include "track.h" #include "bay.h" #include "uthash.h" #include "recorder.h" @@ -74,4 +75,6 @@ int cpu_add_thread(struct cpu *cpu, struct thread *thread); int cpu_remove_thread(struct cpu *cpu, struct thread *thread); int cpu_migrate_thread(struct cpu *cpu, struct thread *thread, struct cpu *newcpu); +struct chan *cpu_get_th_chan(struct cpu *cpu, enum track_th mode); + #endif /* CPU_H */ diff --git a/src/emu/nanos6/connect.c b/src/emu/nanos6/connect.c index 91d12ec..e797b84 100644 --- a/src/emu/nanos6/connect.c +++ b/src/emu/nanos6/connect.c @@ -1,119 +1,67 @@ #include "nanos6_priv.h" -const enum nanos6_track nanos6_chan_track[CH_MAX][CT_MAX] = { - /* Thread CPU */ - [CH_TASKID] = { RUN_TH, RUN_TH }, - [CH_TYPE] = { RUN_TH, RUN_TH }, - [CH_SUBSYSTEM] = { ACT_TH, RUN_TH }, - [CH_RANK] = { RUN_TH, RUN_TH }, - [CH_THREAD] = { NONE, RUN_TH }, +const int nanos6_chan_track[CH_MAX][CT_MAX] = { + /* Thread CPU */ + [CH_TASKID] = { TRACK_TH_RUN, TRACK_TH_RUN }, + [CH_TYPE] = { TRACK_TH_RUN, TRACK_TH_RUN }, + [CH_SUBSYSTEM] = { TRACK_TH_ACT, TRACK_TH_RUN }, + [CH_RANK] = { TRACK_TH_RUN, TRACK_TH_RUN }, + [CH_THREAD] = { TRACK_TH_ANY, TRACK_TH_RUN }, }; static int -connect_thread_mux(struct emu *emu, struct thread *thread) +connect_thread(struct thread *sth) { - struct nanos6_thread *th = EXT(thread, '6'); + struct nanos6_thread *th = EXT(sth, '6'); + for (int i = 0; i < CH_MAX; i++) { + struct track *track = &th->track[i]; - /* TODO: Let the thread take the select channel - * and build the mux as a tracking mode */ - struct chan *inp = &th->ch[i]; - struct chan *sel = &thread->chan[TH_CHAN_STATE]; - - struct mux *mux_run = &th->mux_run[i]; - mux_select_func_t selrun = thread_select_running; - if (mux_init(mux_run, &emu->bay, sel, &th->ch_run[i], selrun) != 0) { - err("mux_init failed"); + if (track_thread(track, &sth->chan[TH_CHAN_STATE], &th->ch[i]) != 0) { + err("track_thread failed"); return -1; } - if (mux_add_input(mux_run, value_int64(0), inp) != 0) { - err("mux_add_input failed"); - return -1; - } - - struct mux *mux_act = &th->mux_act[i]; - mux_select_func_t selact = thread_select_active; - if (mux_init(mux_act, &emu->bay, sel, &th->ch_act[i], selact) != 0) { - err("mux_init failed"); - return -1; - } - - if (mux_add_input(mux_act, value_int64(0), inp) != 0) { - err("mux_add_input failed"); - return -1; - } - - if (mux_act->ninputs != 1) - die("expecting one input only"); - - /* The tracking only sets the ch_out, but we keep both tracking - * updated as the CPU tracking channels may use them. */ - enum nanos6_track track = nanos6_chan_track[i][CT_TH]; - if (track == RUN_TH) - th->ch_out[i] = &th->ch_run[i]; - else if (track == ACT_TH) - th->ch_out[i] = &th->ch_act[i]; - else - th->ch_out[i] = &th->ch[i]; - + /* Select the default output to PRV */ + track_set_default(track, nanos6_chan_track[i][CT_TH]); } return 0; } static int -add_inputs_cpu_mux(struct emu *emu, struct mux *mux, int i) -{ - for (struct thread *t = emu->system.threads; t; t = t->gnext) { - struct nanos6_thread *th = EXT(t, '6'); - - /* Choose input thread channel based on tracking mode */ - struct chan *inp = NULL; - enum nanos6_track track = nanos6_chan_track[i][CT_CPU]; - if (track == RUN_TH) - inp = &th->ch_run[i]; - else if (track == ACT_TH) - inp = &th->ch_act[i]; - else - die("cpu tracking must be running or active"); - - if (mux_add_input(mux, value_int64(t->gindex), inp) != 0) { - err("mux_add_input failed"); - return -1; - } - } - - return 0; -} - -static int -connect_cpu_mux(struct emu *emu, struct cpu *scpu) +connect_cpu(struct emu *emu, struct cpu *scpu) { struct nanos6_cpu *cpu = EXT(scpu, '6'); for (int i = 0; i < CH_MAX; i++) { - struct mux *mux = &cpu->mux[i]; - struct chan *out = &cpu->ch[i]; + struct track *cpu_track = &cpu->track[i]; - /* Choose select CPU channel based on tracking mode */ - struct chan *sel = NULL; - enum nanos6_track track = nanos6_chan_track[i][CT_CPU]; - if (track == RUN_TH) - sel = &scpu->chan[CPU_CHAN_THRUN]; - else if (track == ACT_TH) - sel = &scpu->chan[CPU_CHAN_THACT]; - else - die("cpu tracking must be running or active"); - - if (mux_init(mux, &emu->bay, sel, out, NULL) != 0) { - err("mux_init failed"); + /* Choose select CPU channel based on tracking mode (only + * TRACK_TH_RUN allowed, as active may cause collisions) */ + int mode = nanos6_chan_track[i][CT_CPU]; + struct chan *sel = cpu_get_th_chan(scpu, mode); + if (track_set_select(cpu_track, mode, sel, NULL) != 0) { + err("track_select failed"); return -1; } - if (add_inputs_cpu_mux(emu, mux, i) != 0) { - err("add_inputs_cpu_mux failed"); - return -1; + /* Add each thread as input */ + for (struct thread *t = emu->system.threads; t; t = t->gnext) { + struct nanos6_thread *th = EXT(t, '6'); + + /* Choose input channel from the thread output channels + * based on CPU tracking mode */ + struct value key = value_int64(t->gindex); + struct chan *inp = track_get_output(&th->track[i], mode); + + if (track_add_input(cpu_track, mode, key, inp) != 0) { + err("track_add_input failed"); + return -1; + } } + + /* Set the PRV output */ + track_set_default(cpu_track, nanos6_chan_track[i][CT_CPU]); } return 0; @@ -126,16 +74,16 @@ nanos6_connect(struct emu *emu) /* threads */ for (struct thread *t = sys->threads; t; t = t->gnext) { - if (connect_thread_mux(emu, t) != 0) { - err("connect_thread_mux failed"); + if (connect_thread(t) != 0) { + err("connect_thread failed"); return -1; } } /* cpus */ for (struct cpu *c = sys->cpus; c; c = c->next) { - if (connect_cpu_mux(emu, c) != 0) { - err("connect_cpu_mux failed"); + if (connect_cpu(emu, c) != 0) { + err("connect_cpu failed"); return -1; } } diff --git a/src/emu/nanos6/create.c b/src/emu/nanos6/create.c index f1a29a0..3e54b10 100644 --- a/src/emu/nanos6/create.c +++ b/src/emu/nanos6/create.c @@ -1,12 +1,5 @@ #include "nanos6_priv.h" -static const char chan_fmt_cpu_raw[] = "nanos6.cpu%ld.%s"; -//static const char chan_fmt_cpu_run[] = "nanos6.cpu%ld.%s.run"; -//static const char chan_fmt_cpu_act[] = "nanos6.cpu%ld.%s.act"; -static const char chan_fmt_th_raw[] = "nanos6.thread%ld.%s.raw"; -static const char chan_fmt_th_run[] = "nanos6.thread%ld.%s.run"; -static const char chan_fmt_th_act[] = "nanos6.thread%ld.%s.act"; - static const char *chan_name[CH_MAX] = { [CH_TASKID] = "taskid", [CH_TYPE] = "task_type", @@ -21,11 +14,11 @@ static const int chan_stack[CH_MAX] = { }; static int -init_chans(struct bay *bay, struct chan *chans, const char *fmt, int64_t gindex, int filtered) +init_chans(struct bay *bay, struct chan *chans, const char *fmt, int64_t gindex) { for (int i = 0; i < CH_MAX; i++) { struct chan *c = &chans[i]; - int type = (chan_stack[i] && !filtered) ? CHAN_STACK : CHAN_SINGLE; + int type = chan_stack[i] ? CHAN_STACK : CHAN_SINGLE; chan_init(c, type, fmt, gindex, chan_name[i]); if (bay_register(bay, c) != 0) { @@ -37,6 +30,21 @@ init_chans(struct bay *bay, struct chan *chans, const char *fmt, int64_t gindex, return 0; } +static int +init_tracks(struct bay *bay, struct track *tracks, const char *fmt, int64_t gindex) +{ + for (int i = 0; i < CH_MAX; i++) { + struct track *track = &tracks[i]; + + if (track_init(track, bay, TRACK_TYPE_TH, fmt, gindex, chan_name[i]) != 0) { + err("track_init failed"); + return -1; + } + } + + return 0; +} + static int init_cpu(struct bay *bay, struct cpu *syscpu) { @@ -46,20 +54,15 @@ init_cpu(struct bay *bay, struct cpu *syscpu) return -1; } - cpu->ch = calloc(CH_MAX, sizeof(struct chan)); - if (cpu->ch == NULL) { + cpu->track = calloc(CH_MAX, sizeof(struct track)); + if (cpu->track == NULL) { err("calloc failed:"); return -1; } - cpu->mux = calloc(CH_MAX, sizeof(struct mux)); - if (cpu->mux == NULL) { - err("calloc failed:"); - return -1; - } - - if (init_chans(bay, cpu->ch, chan_fmt_cpu_raw, syscpu->gindex, 1) != 0) { - err("init_chans failed"); + char *fmt = "nanos6.cpu%ld.%s"; + if (init_tracks(bay, cpu->track, fmt, syscpu->gindex) != 0) { + err("init_tracks failed"); return -1; } @@ -82,52 +85,23 @@ init_thread(struct bay *bay, struct thread *systh) return -1; } - th->ch_run = calloc(CH_MAX, sizeof(struct chan)); - if (th->ch_run == NULL) { + th->track = calloc(CH_MAX, sizeof(struct track)); + if (th->track == NULL) { err("calloc failed:"); return -1; } - th->ch_act = calloc(CH_MAX, sizeof(struct chan)); - if (th->ch_act == NULL) { - err("calloc failed:"); - return -1; - } - - th->ch_out = calloc(CH_MAX, sizeof(struct chan *)); - if (th->ch_out == NULL) { - err("calloc failed:"); - return -1; - } - - th->mux_run = calloc(CH_MAX, sizeof(struct mux)); - if (th->mux_run == NULL) { - err("calloc failed:"); - return -1; - } - - th->mux_act = calloc(CH_MAX, sizeof(struct mux)); - if (th->mux_act == NULL) { - err("calloc failed:"); - return -1; - } - - if (init_chans(bay, th->ch, chan_fmt_th_raw, systh->gindex, 0) != 0) { + char *fmt = "nanos6.thread%ld.%s"; + if (init_chans(bay, th->ch, fmt, systh->gindex) != 0) { err("init_chans failed"); return -1; } - if (init_chans(bay, th->ch_run, chan_fmt_th_run, systh->gindex, 1) != 0) { - err("init_chans failed"); + if (init_tracks(bay, th->track, fmt, systh->gindex) != 0) { + err("init_tracks failed"); return -1; } - if (init_chans(bay, th->ch_act, chan_fmt_th_act, systh->gindex, 1) != 0) { - err("init_chans failed"); - return -1; - } - - th->task_stack.thread = systh; extend_set(&systh->ext, '6', th); diff --git a/src/emu/nanos6/nanos6_priv.h b/src/emu/nanos6/nanos6_priv.h index 9f7989e..4f3ca87 100644 --- a/src/emu/nanos6/nanos6_priv.h +++ b/src/emu/nanos6/nanos6_priv.h @@ -8,6 +8,7 @@ #include "chan.h" #include "mux.h" #include "task.h" +#include "track.h" /* Private enums */ @@ -26,14 +27,7 @@ enum nanos6_chan { CH_MAX, }; -enum nanos6_track { - NONE = 0, - RUN_TH, - ACT_TH, - TRACK_MAX, -}; - -extern const enum nanos6_track nanos6_chan_track[CH_MAX][CT_MAX]; +extern const int nanos6_chan_track[CH_MAX][CT_MAX]; enum nanos6_ss_state { ST_TASK_BODY = 1, @@ -76,18 +70,13 @@ enum nanos6_thread_type { }; struct nanos6_thread { - struct chan *ch; /* Raw, modified by nanos6 */ - struct chan *ch_run; /* Tracking running thread */ - struct chan *ch_act; /* Tracking active thread */ - struct chan **ch_out; /* Output to PRV */ - struct mux *mux_run; - struct mux *mux_act; + struct chan *ch; + struct track *track; struct task_stack task_stack; }; struct nanos6_cpu { - struct chan *ch; - struct mux *mux; + struct track *track; }; struct nanos6_proc { diff --git a/src/emu/nanos6/pvt.c b/src/emu/nanos6/pvt.c index 4b3db9c..0087631 100644 --- a/src/emu/nanos6/pvt.c +++ b/src/emu/nanos6/pvt.c @@ -23,10 +23,10 @@ static const char *pcf_prefix[CH_MAX] = { [CH_THREAD] = "Nanos6 thread type", }; -static const char *pcf_suffix[TRACK_MAX] = { - [NONE] = "", - [RUN_TH] = "of the RUNNING thread", - [ACT_TH] = "of the ACTIVE thread", +static const char *pcf_suffix[TRACK_TH_MAX] = { + [TRACK_TH_ANY] = "", + [TRACK_TH_RUN] = "of the RUNNING thread", + [TRACK_TH_ACT] = "of the ACTIVE thread", }; static const struct pcf_value_label nanos6_ss_values[] = { @@ -139,7 +139,7 @@ connect_thread_prv(struct emu *emu, struct thread *thread, struct prv *prv) { struct nanos6_thread *th = EXT(thread, '6'); for (int i = 0; i < CH_MAX; i++) { - struct chan *out = th->ch_out[i]; + struct chan *out = track_get_default(&th->track[i]); long type = pvt_type[i]; long row = thread->gindex; if (prv_register(prv, row, type, &emu->bay, out, PRV_DUP)) { @@ -156,7 +156,7 @@ connect_cpu_prv(struct emu *emu, struct cpu *scpu, struct prv *prv) { struct nanos6_cpu *cpu = EXT(scpu, '6'); for (int i = 0; i < CH_MAX; i++) { - struct chan *out = &cpu->ch[i]; + struct chan *out = track_get_default(&cpu->track[i]); long type = pvt_type[i]; long row = scpu->gindex; if (prv_register(prv, row, type, &emu->bay, out, PRV_DUP)) { diff --git a/src/emu/thread.c b/src/emu/thread.c index 80686d6..ec119c0 100644 --- a/src/emu/thread.c +++ b/src/emu/thread.c @@ -254,7 +254,30 @@ thread_select_running(struct mux *mux, return 0; } +int +thread_select_any(struct mux *mux, + struct value value, + struct mux_input **input) +{ + if (value.type == VALUE_NULL) { + *input = NULL; + return 0; + } + if (value.type != VALUE_INT64) { + err("expecting NULL or INT64 channel value"); + return -1; + } + + if (mux->ninputs != 1) { + err("mux doesn't have one input but %d", mux->ninputs); + return -1; + } + + *input = mux->input; + + return 0; +} int thread_set_cpu(struct thread *th, struct cpu *cpu) diff --git a/src/emu/thread.h b/src/emu/thread.h index 9cc7711..0eae38e 100644 --- a/src/emu/thread.h +++ b/src/emu/thread.h @@ -85,5 +85,6 @@ int thread_connect(struct thread *th, struct bay *bay, struct recorder *rec); int thread_select_active(struct mux *mux, struct value value, struct mux_input **input); int thread_select_running(struct mux *mux, struct value value, struct mux_input **input); +int thread_select_any(struct mux *mux, struct value value, struct mux_input **input); #endif /* THREAD_H */ diff --git a/src/emu/track.c b/src/emu/track.c new file mode 100644 index 0000000..5b8f3da --- /dev/null +++ b/src/emu/track.c @@ -0,0 +1,140 @@ +/* Copyright (c) 2021-2023 Barcelona Supercomputing Center (BSC) + * SPDX-License-Identifier: GPL-3.0-or-later */ + +#include "track.h" + +#include "thread.h" + +static const int track_nmodes[TRACK_TYPE_MAX] = { + [TRACK_TYPE_TH] = TRACK_TH_MAX, +}; + +static const char *th_suffix[TRACK_TH_MAX] = { + [TRACK_TH_ANY] = ".any", + [TRACK_TH_RUN] = ".run", + [TRACK_TH_ACT] = ".act", +}; + +static const char **track_suffix[TRACK_TYPE_MAX] = { + [TRACK_TYPE_TH] = th_suffix, +}; + +int +track_init(struct track *track, struct bay *bay, enum track_type type, const char *fmt, ...) +{ + va_list ap; + va_start(ap, fmt); + + int n = ARRAYLEN(track->name); + int ret = vsnprintf(track->name, n, fmt, ap); + if (ret >= n) + die("track name too long\n"); + va_end(ap); + + track->nmodes = track_nmodes[type]; + track->type = type; + track->bay = bay; + + /* Create one output channel per tracking mode */ + track->out = calloc(track->nmodes, sizeof(struct chan)); + if (track->out == NULL) { + err("calloc failed:"); + return -1; + } + + /* Set the channel name by appending the suffix */ + for (int i = 0; i < track->nmodes; i++) { + struct chan *ch = &track->out[i]; + chan_init(ch, CHAN_SINGLE, "%s%s", track->name, track_suffix[type][i]); + + if (bay_register(bay, ch) != 0) { + err("bay_register failed"); + return -1; + } + } + + track->mux = calloc(track->nmodes, sizeof(struct mux)); + if (track->mux == NULL) { + err("calloc failed:"); + return -1; + } + + return 0; +} + +void +track_set_default(struct track *track, int mode) +{ + track->def = &track->out[mode]; +} + +struct chan * +track_get_default(struct track *track) +{ + return track->def; +} + +int +track_set_select(struct track *track, int mode, struct chan *sel, mux_select_func_t fsel) +{ + struct mux *mux = &track->mux[mode]; + struct chan *out = &track->out[mode]; + if (mux_init(mux, track->bay, sel, out, fsel) != 0) { + err("mux_init failed"); + return -1; + } + + return 0; +} + + +struct chan * +track_get_output(struct track *track, int mode) +{ + return &track->out[mode]; +} + +int +track_add_input(struct track *track, int mode, struct value key, struct chan *inp) +{ + struct mux *mux = &track->mux[mode]; + if (mux_add_input(mux, key, inp) != 0) { + err("mux_add_input failed"); + return -1; + } + + return 0; +} + +int +track_thread(struct track *track, struct chan *sel, struct chan *inp) +{ + /* Create all thread tracking modes */ + if (track_set_select(track, TRACK_TH_ANY, sel, thread_select_any) != 0) { + err("track_select failed"); + return -1; + } + if (track_add_input(track, TRACK_TH_ANY, value_int64(0), inp) != 0) { + err("track_add_input failed"); + return -1; + } + if (track_set_select(track, TRACK_TH_RUN, sel, thread_select_running) != 0) { + err("track_select failed"); + return -1; + } + if (track_add_input(track, TRACK_TH_RUN, value_int64(0), inp) != 0) { + err("track_add_input failed"); + return -1; + } + if (track_set_select(track, TRACK_TH_ACT, sel, thread_select_active) != 0) { + err("track_select failed"); + return -1; + } + if (track_add_input(track, TRACK_TH_ACT, value_int64(0), inp) != 0) { + err("track_add_input failed"); + return -1; + } + + return 0; +} + diff --git a/src/emu/track.h b/src/emu/track.h new file mode 100644 index 0000000..b91f245 --- /dev/null +++ b/src/emu/track.h @@ -0,0 +1,42 @@ +/* Copyright (c) 2021-2023 Barcelona Supercomputing Center (BSC) + * SPDX-License-Identifier: GPL-3.0-or-later */ + +#ifndef TRACK_H +#define TRACK_H + +#include "mux.h" + +enum track_type { + TRACK_TYPE_TH = 0, + TRACK_TYPE_MAX +}; + +enum track_th { + TRACK_TH_ANY = 0, + TRACK_TH_RUN, + TRACK_TH_ACT, + TRACK_TH_MAX, +}; + +struct track { + enum track_type type; + char name[MAX_CHAN_NAME]; + int nmodes; + struct bay *bay; + struct chan *out; + struct chan *def; + struct mux *mux; +}; + +int track_init(struct track *track, struct bay *bay, enum track_type type, const char *fmt, ...); +int track_set_select(struct track *track, int mode, struct chan *sel, mux_select_func_t fsel); + +int track_add_input(struct track *track, int mode, struct value key, struct chan *inp); +int track_output(struct track *track, int mode); +void track_set_default(struct track *track, int mode); +struct chan *track_get_default(struct track *track); +struct chan *track_get_output(struct track *track, int mode); + +int track_thread(struct track *track, struct chan *sel, struct chan *inp); + +#endif /* TRACK_H */