From 12dcc2a60dcec52e495c64da41675c6aabd81e64 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 6 Feb 2026 17:30:50 +0800 Subject: [PATCH] - save tmp works - save works --- src/include/resreq.h | 9 +- src/lib/resreq.c | 107 ++++++++++++++++++ src/lib/rusage.peg | 2 +- src/lib/testsuite/test_resreq.c | 16 +-- src/scheduler/include/plugin.h | 2 +- src/scheduler/include/sched/types.h | 29 +---- src/scheduler/include/scheduler.h | 6 +- src/scheduler/meson.build | 3 +- src/scheduler/src/plugins/limit.c | 6 +- src/scheduler/src/plugins/parjob_allocator.c | 78 +++++++++++++ ...{simple_allocator.c => seqjob_allocator.c} | 34 +++--- src/scheduler/src/resource.c | 39 ++++--- src/scheduler/src/schedobj.c | 59 +++++----- src/scheduler/src/scheduler.c | 106 +++++++++++++++-- 14 files changed, 386 insertions(+), 110 deletions(-) create mode 100644 src/scheduler/src/plugins/parjob_allocator.c rename src/scheduler/src/plugins/{simple_allocator.c => seqjob_allocator.c} (78%) diff --git a/src/include/resreq.h b/src/include/resreq.h index 4cfd95a..eeee202 100644 --- a/src/include/resreq.h +++ b/src/include/resreq.h @@ -6,7 +6,9 @@ #include #include +#include "kvmgr.h" #include "xlist.h" +#include "xvec.h" typedef struct xce_resreq_select_ xce_select_t; typedef struct xce_resreq_rusages_ xce_rusages_t; @@ -35,7 +37,8 @@ typedef struct { typedef enum { PER_TASK = 0, PER_JOB, - PER_HOST + PER_HOST, + PER_DEFAULT } rusage_per_type_t; typedef struct { @@ -124,5 +127,9 @@ extern int parse_resreq_primary(xce_primary_resreq_t*, char*, char**); extern int parse_resreq(xce_resreq_t*, char*, char**); extern void kill_resreq(xce_resreq_t*); extern int format_resreq(char**, xce_resreq_t*); +extern int calc_wanted_global_resvals(xvec_idval_t*, resmgr_t*, xlist_t*, + rusage_per_type_t, int, int); +extern int calc_wanted_host_resvals(xvec_idval_t*, resmgr_t*, xlist_t*, + rusage_per_type_t, int, int); #endif diff --git a/src/lib/resreq.c b/src/lib/resreq.c index 1b4fc2e..312bc6d 100644 --- a/src/lib/resreq.c +++ b/src/lib/resreq.c @@ -3,12 +3,15 @@ #include #include +#include "kvmgr.h" +#include "objmgr.h" #include "resreq.h" #include "rusage.h" #include "section.h" #include "utils.h" #include "xce_resreq.h" #include "xlist.h" +#include "xvec.h" static int get_parse_errmsg(char** errmsg, const char* data, int pos) { @@ -431,3 +434,107 @@ int format_resreq(char** buf, xce_resreq_t* resreq) return rc; } + +int calc_wanted_global_resvals(xvec_idval_t* resvals, resmgr_t* resmgr, + xlist_t* rusages, rusage_per_type_t perflag, + int nhosts, int nslots) +{ + int i; + + for (i = 0; i < rusages->count; i++) { + rusage_value_t* v = (rusage_value_t*)xlist_get(rusages, i); + xce_resource_t* r = NULL; + float val; + rusage_per_type_t perflag1; + + if (v == NULL) { + continue; + } + + r = resmgr_lockres(resmgr, v->res, OBJREF_NOP); + if (r == NULL) { + continue; + } + + if ((r->flags & XCE_RESF_GLOBAL) == 0) { + continue; + } + + perflag1 = v->per_flag; + if (perflag1 == PER_DEFAULT) { + perflag1 = perflag; + } + + switch (perflag1) { + case PER_TASK: + val = v->value * nslots; + break; + case PER_JOB: + val = v->value; + break; + case PER_HOST: + val = v->value * nhosts; + break; + default: + /* something wrong */ + val = 0.0; + break; + } + + xvec_idval_store(resvals, r->idx, val); + } + return 0; +} + +int calc_wanted_host_resvals(xvec_idval_t* resvals, resmgr_t* resmgr, + xlist_t* rusages, rusage_per_type_t perfflag, + int nhosts, int nslots) +{ + int i; + + for (i = 0; i < rusages->count; i++) { + rusage_value_t* v = (rusage_value_t*)xlist_get(rusages, i); + xce_resource_t* r = NULL; + float val; + rusage_per_type_t perflag1; + + if (v == NULL) { + continue; + } + + r = resmgr_lockres(resmgr, v->res, OBJREF_NOP); + if (r == NULL) { + continue; + } + + if ((r->flags & XCE_RESF_GLOBAL) != 0) { + continue; + } + + perflag1 = v->per_flag; + if (perflag1 == PER_DEFAULT) { + perflag1 = perflag; + } + + if (perflag1 == PER_JOB) { + /* FIXME: PER_JOB should be global? */ + continue; + } + switch (perflag1) { + case PER_TASK: + val = v->value * nslots; + break; + case PER_HOST: + val = v->value; + break; + default: + /* something wrong */ + val = 0.0; + break; + } + + xvec_idval_store(resvals, r->idx, val); + } + + return 0; +} diff --git a/src/lib/rusage.peg b/src/lib/rusage.peg index e2b45f8..7593bb7 100644 --- a/src/lib/rusage.peg +++ b/src/lib/rusage.peg @@ -68,7 +68,7 @@ empty_decay <- { } { auxil->decay_flag = 0; } identifier <- [a-zA-Z_][a-zA-Z_0-9]* { auxil->resname = strdup($0); } resvalue <- number '/' per_type { auxil->resvalue = auxil->tmp_value; } - / number { auxil->resvalue = auxil->tmp_value; auxil->per_flag = PER_TASK; } + / number { auxil->resvalue = auxil->tmp_value; auxil->per_flag = PER_DEFAULT; } per_type <- 'task' { auxil->per_flag = PER_TASK; } / 'job' { auxil->per_flag = PER_JOB; } diff --git a/src/lib/testsuite/test_resreq.c b/src/lib/testsuite/test_resreq.c index e92cbf2..08b515d 100644 --- a/src/lib/testsuite/test_resreq.c +++ b/src/lib/testsuite/test_resreq.c @@ -91,14 +91,14 @@ void test_parse_rusage1(void) TEST_ASSERT_TRUE(strcmp(v->res, "mem") == 0); TEST_ASSERT_TRUE(fabs(v->value - 20.0) < 0.001); TEST_ASSERT_TRUE(fabs(v->duration - 18000) < 0.001); - TEST_ASSERT_TRUE(v->per_flag == 0); + TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT); TEST_ASSERT_TRUE(v->decay_flag == 1); break; case 1: TEST_ASSERT_TRUE(strcmp(v->res, "lic") == 0); TEST_ASSERT_TRUE(fabs(v->value - 55.0) < 0.001); TEST_ASSERT_TRUE(fabs(v->duration - 1200) < 0.001); - TEST_ASSERT_TRUE(v->per_flag == 0); + TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT); TEST_ASSERT_TRUE(v->decay_flag == 0); break; default: @@ -140,14 +140,14 @@ void test_parse_rusage2(void) TEST_ASSERT_TRUE(strcmp(v->res, "mem") == 0); TEST_ASSERT_TRUE(fabs(v->value - 20.0) < 0.001); TEST_ASSERT_TRUE(fabs(v->duration - 18000) < 0.001); - TEST_ASSERT_TRUE(v->per_flag == 2); + TEST_ASSERT_TRUE(v->per_flag == PER_HOST); TEST_ASSERT_TRUE(v->decay_flag == 1); break; case 1: TEST_ASSERT_TRUE(strcmp(v->res, "lic") == 0); TEST_ASSERT_TRUE(fabs(v->value - 55.0) < 0.001); TEST_ASSERT_TRUE(fabs(v->duration - 1200) < 0.001); - TEST_ASSERT_TRUE(v->per_flag == 1); + TEST_ASSERT_TRUE(v->per_flag == PER_JOB); TEST_ASSERT_TRUE(v->decay_flag == 0); break; default: @@ -188,14 +188,14 @@ void test_resreq_primary(void) TEST_ASSERT_TRUE(strcmp(v->res, "mem") == 0); TEST_ASSERT_TRUE(fabs(v->value - 20.0) < 0.001); TEST_ASSERT_TRUE(fabs(v->duration - 18000) < 0.001); - TEST_ASSERT_TRUE(v->per_flag == 0); + TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT); TEST_ASSERT_TRUE(v->decay_flag == 1); break; case 1: TEST_ASSERT_TRUE(strcmp(v->res, "lic") == 0); TEST_ASSERT_TRUE(fabs(v->value - 55.0) < 0.001); TEST_ASSERT_TRUE(fabs(v->duration - 1200) < 0.001); - TEST_ASSERT_TRUE(v->per_flag == 0); + TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT); TEST_ASSERT_TRUE(v->decay_flag == 0); break; default: @@ -241,14 +241,14 @@ void test_resreq1(void) TEST_ASSERT_TRUE(strcmp(v->res, "mem") == 0); TEST_ASSERT_TRUE(fabs(v->value - 20.0) < 0.001); TEST_ASSERT_TRUE(fabs(v->duration - 18000) < 0.001); - TEST_ASSERT_TRUE(v->per_flag == 0); + TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT); TEST_ASSERT_TRUE(v->decay_flag == 1); break; case 1: TEST_ASSERT_TRUE(strcmp(v->res, "lic") == 0); TEST_ASSERT_TRUE(fabs(v->value - 55.0) < 0.001); TEST_ASSERT_TRUE(fabs(v->duration - 1200) < 0.001); - TEST_ASSERT_TRUE(v->per_flag == 0); + TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT); TEST_ASSERT_TRUE(v->decay_flag == 0); break; default: diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index e0d6449..1902ba3 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -77,6 +77,6 @@ extern int plugin_prepare_allocation(void*, xce_sched_ctx_t*); extern int plugin_job_election(void*, xce_sched_ctx_t*); extern int register_fcfs(void*); extern int register_limit(void*); -extern int register_simple_allocator(void*); +extern int register_seqjob_allocator(void*); #endif diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 5fb81e2..eadfd32 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -7,6 +7,7 @@ #include #include "kvmgr.h" +#include "resreq.h" #include "typemap.h" #include "xce/types.h" #include "xlist.h" @@ -14,23 +15,12 @@ #include "zarray.h" /* FIXME: improve later */ -typedef void* host_selection_t; typedef struct { xvec_idval_t shared; xvec_idval_t host; } xce_sched_alloc_req_t; -typedef enum { - RESREQ_OP_SIMPLE - = 0, /* simple just a list of resources and require values */ - RESREQ_OP_AFFINITY, /* with affinity allocation */ - RESREQ_OP_GPU, /* with GPU allocation */ - RESREQ_OP_PARALL, /* parall job nhosts, ptile and other constraints */ - RESREQ_OP_AND, /* success for all */ - RESREQ_OP_OR /* success for any */ -} resreq_op_t; - /* simple resource allocation just have numeric resource values */ typedef struct { char* res; @@ -38,17 +28,8 @@ typedef struct { } request_res_t; typedef struct { - host_selection_t selection; - xlist_t res_shared; - xlist_t res_host; -} resreq_simple_t; - -typedef struct { - resreq_op_t op; - union { - resreq_simple_t simple; - } u; -} resreq_t; + xce_resreq_t* resreq; +} xce_sched_resreq_t; typedef struct { char* name; @@ -78,7 +59,7 @@ typedef struct { xce_sched_queue_t* queue; xce_sched_user_t* user; xce_sched_project_t* project; - resreq_t* resreq; + xce_sched_resreq_t* resreq; uint32_t mintasks; uint32_t maxtasks; xce_job_alloc_t* allocs; @@ -96,7 +77,7 @@ typedef struct { typedef struct { xce_sched_job_t* job; xce_sched_worker_t* worker; - resreq_t* resreq; + xce_sched_resreq_t* resreq; xce_job_alloc_t* alloc; xce_pendreason_t* reason; struct { diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index f31461f..71d035d 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -59,10 +59,10 @@ typedef struct { uint64_t nextid; xce_conf4sched_t* config; kvmgr_t plugins; - void* simple_allocator; + void* seqjob_allocator; + void* parjob_allocator; void* affinity_allocator; void* gpu_allocator; - void* parallel_allocator; dataq_t* pend_updates[2]; resmgr_t resmgr; groupmgr_t ugmgr; @@ -107,7 +107,7 @@ extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*); extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*); extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*); extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*, - host_selection_t*); + xce_select_t*); extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_t*, xvec_idval_t*, xvec_idval_t*); extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*, diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index a1b21d7..0053cb8 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -7,7 +7,8 @@ scheduler_srcs = [ 'src/plugin.c', 'src/resource.c', 'src/plugins/fcfs.c', - 'src/plugins/simple_allocator.c', + 'src/plugins/seqjob_allocator.c', + 'src/plugins/parjob_allocator.c', 'src/plugins/limit.c', ] diff --git a/src/scheduler/src/plugins/limit.c b/src/scheduler/src/plugins/limit.c index 91c5881..792139d 100644 --- a/src/scheduler/src/plugins/limit.c +++ b/src/scheduler/src/plugins/limit.c @@ -91,9 +91,10 @@ static int update_limit_res_value(zarray_t* array, xce_vres_t* vres) static float get_vres_value(xce_sched_ctx_t* ctx, xce_vres_t* vres) { - int i; - resreq_t* resreq = ctx->resreq; + int i; + xce_sched_resreq_t* resreq = ctx->resreq; +#if 0 if (resreq->op != RESREQ_OP_SIMPLE) { return 0.0; } @@ -114,6 +115,7 @@ static float get_vres_value(xce_sched_ctx_t* ctx, xce_vres_t* vres) return rq->value; } } +#endif return 0.0; } diff --git a/src/scheduler/src/plugins/parjob_allocator.c b/src/scheduler/src/plugins/parjob_allocator.c new file mode 100644 index 0000000..93830ae --- /dev/null +++ b/src/scheduler/src/plugins/parjob_allocator.c @@ -0,0 +1,78 @@ +#include +#include +#include +#include +#include + +#include "decision.h" +#include "kvmgr.h" +#include "objmgr.h" +#include "plugin.h" +#include "preason.h" +#include "scheduler.h" +#include "utils.h" +#include "xce/types.h" +#include "xcealloc.h" +#include "xlist.h" +#include "xvec.h" + +static int parjob_allocator_init(xce_sched_plugin_t* plugin) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; + + sched->parjob_allocator = plugin; + return 1; +} + +static void parjob_allocator_fini(xce_sched_plugin_t* plugin) +{ + if (plugin == NULL) { + return; + } + FREEUP(plugin); +} + +static int skip_bad_worker_(xce_sched_worker_t* w) +{ + if (w == NULL) { + return 1; + } + + /* add more check here */ + return 0; +} + +static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; + xce_sched_job_t* job = ctx->job; + xce_sched_resreq_t* resreq = ctx->resreq; + + return 0; +} + +int register_parjob_allocator(void* sched) +{ + static xce_sched_plugin_t parjob_allocator_plugin = { + .name = "xce_sched_parjob_allocator", + .hooks = { .config_hook = NULL, .commit_hook = NULL }, + .init_fn = (plugin_init_fn_t)parjob_allocator_init, + .fini_fn = (plugin_fini_fn_t)parjob_allocator_fini, + .begin_session_fn = NULL, + .end_session_fn = NULL, + .job_elect_fn = NULL, + .job_alloc_fn = (plugin_job_allocate_fn_t)my_alloc, + .data = NULL + }; + + xce_sched_plugin_t* plugin = NULL; + + plugin = (xce_sched_plugin_t*)calloc(1, sizeof(*plugin)); + if (plugin == NULL) { + return -ENOMEM; + } + + memcpy(plugin, &parjob_allocator_plugin, sizeof(*plugin)); + + return plugin_register(sched, plugin); +} diff --git a/src/scheduler/src/plugins/simple_allocator.c b/src/scheduler/src/plugins/seqjob_allocator.c similarity index 78% rename from src/scheduler/src/plugins/simple_allocator.c rename to src/scheduler/src/plugins/seqjob_allocator.c index 837f20e..5d5cc5a 100644 --- a/src/scheduler/src/plugins/simple_allocator.c +++ b/src/scheduler/src/plugins/seqjob_allocator.c @@ -16,15 +16,15 @@ #include "xlist.h" #include "xvec.h" -static int simple_allocator_init(xce_sched_plugin_t* plugin) +static int seqjob_allocator_init(xce_sched_plugin_t* plugin) { xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; - sched->simple_allocator = plugin; + sched->seqjob_allocator = plugin; return 1; } -static void simple_allocator_fini(xce_sched_plugin_t* plugin) +static void seqjob_allocator_fini(xce_sched_plugin_t* plugin) { if (plugin == NULL) { return; @@ -44,13 +44,13 @@ static int skip_bad_worker_(xce_sched_worker_t* w) static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) { - xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; - xce_sched_job_t* job = ctx->job; - resreq_t* resreq = ctx->resreq; - int rc; - xlist_t workers; - xce_pendreason_t* reason = NULL; - xvec_t* uncomit = NULL; + xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; + xce_sched_job_t* job = ctx->job; + xce_sched_resreq_t* resreq = ctx->resreq; + int rc; + xlist_t workers; + xce_pendreason_t* reason = NULL; + xvec_t* uncomit = NULL; reason = make_pendreason(job->jobid); ctx->reason = reason; @@ -87,7 +87,7 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) if (rc < 0) { return rc; } - + xlist_init(&workers, 128); rc = sched_collect_workers(sched, &workers, @@ -117,13 +117,13 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) return 0; } -int register_simple_allocator(void* sched) +int register_seqjob_allocator(void* sched) { - static xce_sched_plugin_t simple_allocator_plugin = { - .name = "xce_sched_simple_allocator", + static xce_sched_plugin_t seqjob_allocator_plugin = { + .name = "xce_sched_seqjob_allocator", .hooks = { .config_hook = NULL, .commit_hook = NULL }, - .init_fn = (plugin_init_fn_t)simple_allocator_init, - .fini_fn = (plugin_fini_fn_t)simple_allocator_fini, + .init_fn = (plugin_init_fn_t)seqjob_allocator_init, + .fini_fn = (plugin_fini_fn_t)seqjob_allocator_fini, .begin_session_fn = NULL, .end_session_fn = NULL, .job_elect_fn = NULL, @@ -138,7 +138,7 @@ int register_simple_allocator(void* sched) return -ENOMEM; } - memcpy(plugin, &simple_allocator_plugin, sizeof(*plugin)); + memcpy(plugin, &seqjob_allocator_plugin, sizeof(*plugin)); return plugin_register(sched, plugin); } diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index 778da70..4816fdd 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -124,7 +124,7 @@ static int unsatisfy_(int nres, int* idxs, float* total, float* inuse, } int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w, - host_selection_t* s) + xce_select_t* s) { /* FIXME: add real work later */ return 1; @@ -259,9 +259,9 @@ int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w, xvec_t* uncomitted, xvec_idval_t* wanted, xvec_idval_t* unmeet) { - xvec_t total; - xvec_t inuse; - int rc; + xvec_t total; + xvec_t inuse; + int rc; if (wanted->count == 0) { return 1; @@ -343,16 +343,18 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) { - xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*); - xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*); - xce_sched_job_t* job = ctx->job; - resreq_t* resreq = ctx->resreq; - int rc; - xce_host_alloc_t** hallocs = NULL; - xce_host_alloc_t* halloc = NULL; - xce_alloc_t** allocs = NULL; - xvec_t* uncomit = NULL; + xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*); + xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*); + xce_sched_job_t* job = ctx->job; + xce_sched_resreq_t* resreq = ctx->resreq; + int rc; + xce_host_alloc_t** hallocs = NULL; + xce_host_alloc_t* halloc = NULL; + xce_alloc_t** allocs = NULL; + xvec_t* uncomit = NULL; + /* FIXME: */ +#if 0 rc = sched_meet_host_selection(sched, w, resreq->u.simple.selection); if (rc == 0) { add_job_preason(ctx->reason->reasons, PREASON_MATCH_FAIL, w->name, @@ -360,6 +362,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) return 0; } +#endif rc = sched_ctx_get_uncomit_host(&uncomit, ctx, w->idx); if (rc < 0) { @@ -401,7 +404,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) if (rc < 0) { return rc; } - + allocs = sched_make_allocs( sched, "", ctx->alloc_req.host.count, ctx->alloc_req.host.idxs.data, ctx->alloc_req.host.vals.data, /* FIXME: no duration now */ NULL); @@ -514,7 +517,8 @@ static int fill_ctx_idval_vec_(xce_scheduler_t* sched, xvec_idval_t* v, int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) { - int rc = 0; + int rc = 0; +#if 0 xlist_t* lists[] = { &(ctx->resreq->u.simple.res_shared), &(ctx->extra_res_requests) }; @@ -522,13 +526,15 @@ int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) if (rc < 0) { xvec_idval_fini(&(ctx->alloc_req.shared)); } +#endif return rc; } int sched_ctx_fill_host_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) { - int rc; + int rc = 0; +#if 0 xlist_t* lists[] = { &(ctx->resreq->u.simple.res_host), &(ctx->extra_host_res_requests) }; @@ -536,6 +542,7 @@ int sched_ctx_fill_host_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) if (rc < 0) { xvec_idval_fini(&(ctx->alloc_req.host)); } +#endif return rc; } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index b338bdf..3468d9d 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -1,7 +1,7 @@ #include +#include #include #include -#include #include "controller.capnp.h" #include "kvmgr.h" @@ -36,39 +36,37 @@ request_res_t* make_resource_request(char* res, float value) return req; } -static resreq_t* make_sched_resreq(xce_scheduler_t* sched, xce_job_t* j) +static int make_sched_resreq(xce_sched_resreq_t* resreq, xce_scheduler_t* sched, + xce_job_t* j) { - resreq_t* resreq = NULL; - request_res_t* req = NULL; + int rc; + char* errmsg; - resreq = (resreq_t*)calloc(1, sizeof(*resreq)); - if (resreq == NULL) { - return NULL; + memset(resreq, 0x0, sizeof(*resreq)); + resreq->resreq = (xce_resreq_t*)calloc(1, sizeof(*(resreq->resreq))); + if (resreq->resreq == NULL) { + return -EINVAL; } - resreq->op = RESREQ_OP_SIMPLE; - resreq->u.simple.selection = NULL; + rc = parse_resreq(resreq->resreq, j->submit->resreq, &errmsg); + if (rc < 0) { + ERROR("%s parse_resreq with error code %d", __func__, rc); + if (errmsg != NULL) { + ERROR("%s", errmsg); + free(errmsg); + } - xlist_init(&(resreq->u.simple.res_shared), 4); - xlist_init(&(resreq->u.simple.res_host), 4); - - req = make_resource_request("slots", 1.0); - if (req != NULL) { - xlist_store(&(resreq->u.simple.res_host), (void*)req); + return rc; } - req = make_resource_request("mem", 1024.0); - if (req != NULL) { - xlist_store(&(resreq->u.simple.res_host), (void*)req); - } - - return resreq; + return 1; } xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j) { xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); xce_scheduler_t* sched = (xce_scheduler_t*)arg; + int rc = 0; if ((sched == NULL) || (job == NULL)) { return NULL; @@ -85,9 +83,18 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j) job->mintasks = j->submit->mintasks; job->maxtasks = j->submit->maxtasks; job->priority = 0; /* FIXME: support job level priority */ - job->resreq = make_sched_resreq(sched, j); - job->allocs = NULL; - job->accounts = NULL; + + job->resreq = (xce_sched_resreq_t*)calloc(1, sizeof(*(job->resreq))); + if (job->resreq == NULL) { + /* FIXME: */ + free(job); + + return NULL; + } + + make_sched_resreq(job->resreq, sched, j); + job->allocs = NULL; + job->accounts = NULL; if (j->allocs != NULL) { MOVE_PTR(job->allocs, j->allocs); job->accounts = make_job_account(job->allocs); @@ -329,7 +336,7 @@ int sched_ctx_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx) v = zarray_get(&(ctx->uncomitted.hosts), idx); if (v == NULL) { int rc = 0; - + v = (xvec_t*)calloc(1, sizeof(*v)); if (v == NULL) { return -ENOMEM; @@ -342,7 +349,7 @@ int sched_ctx_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx) return rc; } - + zarray_put(&(ctx->uncomitted.hosts), idx, v); } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 8271b03..208503f 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -32,12 +32,12 @@ static void* sched_job_picker(void* arg) return job_picker->job_elect_fn(job_picker); } -static int sched_make_simple_allocation(xce_scheduler_t* sched, +static int sched_make_seqjob_allocation(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) { xce_sched_plugin_t* plugin = NULL; - plugin = sched->simple_allocator; + plugin = sched->seqjob_allocator; if (plugin->job_alloc_fn == NULL) { ERROR("%s plugin %s has no allocation function", __func__, plugin->name); @@ -51,10 +51,90 @@ static int sched_make_simple_allocation(xce_scheduler_t* sched, return plugin->job_alloc_fn(plugin, ctx); } +static int sched_make_parjob_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) +{ + xce_sched_plugin_t* plugin = NULL; + + plugin = sched->parjob_allocator; + if (plugin->job_alloc_fn == NULL) { + ERROR("%s plugin %s has no allocation function", __func__, + plugin->name); + return -EINVAL; + } + + plugin_prepare_allocation(sched, ctx); + + sched_ctx_fill_shared_requests(sched, ctx); + + return plugin->job_alloc_fn(plugin, ctx); +} + +static int sched_make_primary_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) +{ + xce_primary_resreq_t *primary = ctx->resreq->resreq->u.primary; + + if (primary->maxTasks == 1) { + return sched_make_seqjob_allocation(sched, ctx); + } + + return sched_make_parjob_allocation(sched, ctx); +} + +static int sched_make_compound_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) +{ + xce_compound_resreq_t *resreq = ctx->resreq->resreq->u.compound; + int i; + + for(i = 0; i < resreq->primaries.count; i ++) { + xce_primary_resreq_t *resreq1 = (xce_primary_resreq_t *)xlist_get(&(resreq->primaries), i); + int rc; + xce_sched_ctx_t ctx1 = {0}; + + /* FIXME: initialize ctx1 */ + + rc = sched_make_primary_allocation(sched, &ctx1); + if (rc <= 0) { + return rc; + } + + } + return 1; +} + +static int sched_make_alter_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) +{ + xce_alter_resreq_t *resreq = ctx->resreq->resreq->u.alter; + int i; + + for(i = 0; i < resreq->compounds.count; i ++) { + xce_compound_resreq_t *resreq1 = (xce_compound_resreq_t *)xlist_get(&(resreq->compounds), i); + int rc; + xce_sched_ctx_t ctx1 = {0}; + + if (resreq1 == NULL) { + continue; + } + + /* FIXME: we need to initialize ctx1 */ + + rc = sched_make_compound_allocation(sched, &ctx1); + if (rc > 0) { + return rc; + } + } + + return 0; +} + static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) { - int rc = 0; - xce_sched_job_t* job = ctx->job; + int rc = 0; + xce_sched_job_t* job = ctx->job; + xce_resreq_t* resreq = job->resreq->resreq; if (job->resreq == NULL) { xce_pendreason_t* reason = NULL; @@ -70,13 +150,19 @@ static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) return 0; } - switch (job->resreq->op) { - case RESREQ_OP_SIMPLE: - ctx->resreq = job->resreq; - rc = sched_make_simple_allocation(sched, ctx); + ctx->resreq = job->resreq; + switch (resreq->tag) { + case XCE_RESREQ_PRIMARY: + rc = sched_make_primary_allocation(sched, ctx); + break; + case XCE_RESREQ_COMPOUND: + rc = sched_make_compound_allocation(sched, ctx); + break; + case XCE_RESREQ_ALTER: + rc = sched_make_alter_allocation(sched, ctx); break; default: - rc = 0; + rc = -EINVAL; break; } return rc; @@ -189,7 +275,7 @@ static int sched_register_plugins(xce_scheduler_t* sched) { register_fcfs(sched); register_limit(sched); - register_simple_allocator(sched); + register_seqjob_allocator(sched); return 0; }