From 6cf9aa0979b061e5511ba8d5b13ab6010851a868 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 12 Feb 2026 11:19:47 +0800 Subject: [PATCH] - make new scheduler work for sequence jobs --- src/scheduler/include/plugin.h | 5 +- src/scheduler/include/sched/types.h | 49 +++-- src/scheduler/include/scheduler.h | 16 +- src/scheduler/src/plugin.c | 50 ++++- src/scheduler/src/plugins/limit.c | 182 ++++++++++--------- src/scheduler/src/plugins/parjob_allocator.c | 38 +++- src/scheduler/src/plugins/seqjob_allocator.c | 92 ++++++++-- src/scheduler/src/resource.c | 116 ++++-------- src/scheduler/src/schedobj.c | 103 +++++++---- src/scheduler/src/scheduler.c | 129 +++++++------ 10 files changed, 461 insertions(+), 319 deletions(-) diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 1902ba31..77d8dbd5 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -7,6 +7,7 @@ #include "decision.h" #include "sched/types.h" +#include "resreq.h" typedef void* plugin_data_t; @@ -18,7 +19,7 @@ typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*); typedef int (*plugin_job_state_change_hook_t)(void*, xce_sched_job_t*, uint32_t); typedef int (*plugin_commit_decision_hook_t)(void*, xce_decision_t*); -typedef int (*plugin_job_election_hook_t)(void*, xce_sched_ctx_t*); +typedef int (*plugin_job_election_hook_t)(void*, xce_sched_job_t*); typedef int (*plugin_prepare_allocation_hook_t)(void*, xce_sched_ctx_t*); /* initialize function of plugin */ @@ -74,7 +75,7 @@ extern int plugin_begin_session(void*); extern int plugin_end_session(void*); extern int plugin_commit_decision(void*, xce_decision_t*); extern int plugin_prepare_allocation(void*, xce_sched_ctx_t*); -extern int plugin_job_election(void*, xce_sched_ctx_t*); +extern int plugin_job_election(void*, xce_sched_job_t*); extern int register_fcfs(void*); extern int register_limit(void*); extern int register_seqjob_allocator(void*); diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 9265dd81..d4f4ff63 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -27,22 +27,29 @@ typedef struct { float value; } request_res_t; +struct xce_sched_obj_; +typedef struct xce_sched_obj_ xce_sched_obj_t; + typedef struct { - char* user; - char* queue; - char* project; + xce_sched_obj_t* owner; xce_primary_resreq_t* resreq; + xce_pendreason_t* reason; } xce_primary_schedobj_t; typedef struct { - char* user; - char* queue; - char* project; + xce_sched_obj_t* owner; xce_compound_resreq_t* resreq; + xce_pendreason_t* reason; } xce_compound_schedobj_t; -typedef struct { +struct xce_sched_obj_ { xce_resreq_tag_t tag; + uint64_t configver; + char* user; + char* queue; + char* project; + xlist_t glimits; + zarray_t hlimits; union { xce_primary_schedobj_t* primary; xce_compound_schedobj_t* compound; @@ -50,7 +57,7 @@ typedef struct { } u; xce_job_alloc_t* alloc; xce_pendreason_t* reason; -} xce_sched_obj_t; +}; typedef struct { char* name; @@ -97,20 +104,20 @@ typedef struct { } xce_sched_worker_t; typedef struct { - xce_sched_job_t* job; - xce_sched_worker_t* worker; - xce_sched_obj_t* schedobj; - xce_job_alloc_t* alloc; + void* sched; + xce_sched_worker_t* worker; + xce_sched_obj_t* schedobj; + xce_primary_resreq_t* primary; + int nhosts; + int nslots; struct { xvec_t* shared; zarray_t hosts; } uncomitted; xvec_idval_t unmeet; - xlist_t extra_res_requests; - xlist_t extra_host_res_requests; - xlist_t limits; /* non per host limits */ - xlist_t limits_host; /* PER HOST limits */ xce_sched_alloc_req_t alloc_req; + xce_job_alloc_t* alloc; + xce_pendreason_t* reason; } xce_sched_ctx_t; typedef struct { @@ -129,9 +136,11 @@ extern void kill_sched_worker(void*); extern void kill_sched_user(void*); extern void kill_sched_project(void*); extern void kill_sched_resval(xce_val_t*); -extern int sched_ctx_init(xce_sched_ctx_t*, xce_sched_job_t*); -extern void sched_ctx_fini(xce_sched_ctx_t*); -extern int sched_ctx_get_uncomit_shared(xvec_t**, xce_sched_ctx_t*); -extern int sched_ctx_get_uncomit_host(xvec_t**, xce_sched_ctx_t*, int); +extern int sched_ctx_init(xce_sched_ctx_t*, void*, xce_sched_obj_t*); +extern void sched_ctx_fini(xce_sched_ctx_t*); +extern void sched_ctx_set_primary(xce_sched_ctx_t*, xce_primary_resreq_t*); +extern int sched_ctx_get_uncomit_shared(xvec_t**, xce_sched_ctx_t*); +extern int sched_ctx_get_uncomit_host(xvec_t**, xce_sched_ctx_t*, int); +extern void reset_sched_obj(xce_sched_obj_t*); #endif diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 71d035da..7035646d 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -58,6 +58,7 @@ typedef struct { xcecli_ctx_t apictx; uint64_t nextid; xce_conf4sched_t* config; + uint64_t configver; kvmgr_t plugins; void* seqjob_allocator; void* parjob_allocator; @@ -118,14 +119,13 @@ extern int sched_collect_workers(xce_scheduler_t*, xlist_t*, extern int sched_charge_account(xce_scheduler_t*, xce_job_account_t*); extern int sched_uncharge_account(xce_scheduler_t*, xce_job_account_t*); extern int sched_commit_job_alloc(xce_scheduler_t*, xce_sched_job_t*); -extern uint64_t sched_get_id(xce_scheduler_t*); -extern xce_alloc_t** sched_make_allocs(xce_scheduler_t*, char*, int, int*, - float*, int*); -extern int sched_try_alloc_on_hosts(xce_worker_group_t*, va_list); -extern int sched_ctx_fill_shared_requests(xce_scheduler_t*, xce_sched_ctx_t*); -extern int sched_ctx_fill_host_requests(xce_scheduler_t*, xce_sched_ctx_t*); -extern int get_resvals(zarray_t*, int, int*, float*); -extern int put_resvals(zarray_t*, int, int*, float*); +extern uint64_t sched_get_id(xce_scheduler_t*); +extern xce_alloc_t** sched_make_allocs(xce_scheduler_t*, char*, int, int*, + float*, int*); +extern int sched_try_alloc_on_hosts(xce_worker_group_t*, va_list); +extern int get_resvals(zarray_t*, int, int*, float*); +extern int put_resvals(zarray_t*, int, int*, float*); extern request_res_t* make_resource_request(char*, float); +extern int sched_fill_limit_requests(xce_scheduler_t*, xvec_idval_t*, xlist_t*); #endif diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 28785d4a..270e768c 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -4,6 +4,7 @@ #include #include "plugin.h" +#include "resreq.h" #include "scheduler.h" #include "xlist.h" @@ -334,33 +335,33 @@ int plugin_job_state_change(void* arg, xce_sched_job_t* job, uint32_t state) return rc; } -static int plugin_job_election_(xce_sched_ctx_t* ctx, +static int plugin_job_election_(xce_sched_job_t* job, xce_sched_plugin_t* plugin) { - if ((plugin == NULL) || (ctx == NULL)) { + if ((plugin == NULL) || (job == NULL)) { return 0; } if (plugin->hooks.job_election_hook != NULL) { - return plugin->hooks.job_election_hook(plugin, ctx); + return plugin->hooks.job_election_hook(plugin, job); } return 0; } -int plugin_job_election(void* arg, xce_sched_ctx_t* ctx) +int plugin_job_election(void* arg, xce_sched_job_t* job) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; xlist_t lst; int rc; - if ((sched == NULL) || (ctx == NULL)) { + if ((sched == NULL) || (job == NULL)) { return -EINVAL; } xlist_init(&lst, 3); xlist_store(&lst, plugin_job_election_); - rc = kvmgr_iter(&(sched->plugins), NULL, &lst, ctx); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, job); xlist_fini(&lst); return rc; @@ -373,7 +374,7 @@ static int plugin_prepare_allocation_(xce_sched_ctx_t* ctx, return 0; } - if (plugin->hooks.job_election_hook != NULL) { + if (plugin->hooks.prepare_allocation_hook != NULL) { return plugin->hooks.prepare_allocation_hook(plugin, ctx); } @@ -385,15 +386,50 @@ int plugin_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) xce_scheduler_t* sched = (xce_scheduler_t*)arg; xlist_t lst; int rc; + xce_sched_obj_t* obj = NULL; if ((sched == NULL) || (ctx == NULL)) { return -EINVAL; } + obj = ctx->schedobj; + if (obj == NULL) { + return -EINVAL; + } + + if (ctx->worker == NULL) { + xvec_idval_fini(&(ctx->alloc_req.shared)); + xvec_idval_init(&(ctx->alloc_req.shared), 8); + } else { + xvec_idval_fini(&(ctx->alloc_req.host)); + xvec_idval_init(&(ctx->alloc_req.host), 8); + } + xlist_init(&lst, 3); xlist_store(&lst, plugin_prepare_allocation_); rc = kvmgr_iter(&(sched->plugins), NULL, &lst, ctx); xlist_fini(&lst); + if (rc < 0) { + return rc; + } + + /* fill the extra requests based on limit(global/host) */ + if (ctx->worker == NULL) { + /* sequence/parallel job plugin should fill the basic rusages based on + * resreq */ + sched_fill_limit_requests(sched, &(ctx->alloc_req.shared), + &(obj->glimits)); + } else { + xlist_t* hlimits = NULL; + + /* sequence/parallel job plugin should fill the basic rusages based on + * resreq */ + hlimits = (xlist_t*)zarray_get(&(obj->hlimits), ctx->worker->idx); + if (hlimits != NULL) { + sched_fill_limit_requests(sched, &(ctx->alloc_req.host), hlimits); + } + } + return rc; } diff --git a/src/scheduler/src/plugins/limit.c b/src/scheduler/src/plugins/limit.c index 4459b3ea..32309e8e 100644 --- a/src/scheduler/src/plugins/limit.c +++ b/src/scheduler/src/plugins/limit.c @@ -62,7 +62,7 @@ static int limit_update_config(xce_sched_plugin_t* plugin, typedef struct { xce_sched_plugin_t* plugin; xce_sched_ctx_t* ctx; -} match_job_limit_args_t; +} match_limit_args_t; static int update_limit_res_value(zarray_t* array, xce_vres_t* vres) { @@ -122,31 +122,29 @@ static float get_vres_value(xce_sched_ctx_t* ctx, xce_vres_t* vres) static int match_job_global_limit_(void* arg1, void* arg2) { - xce_limit_t* limit = (xce_limit_t*)arg2; - match_job_limit_args_t* args = (match_job_limit_args_t*)arg1; - xce_sched_ctx_t* ctx = NULL; - xce_sched_plugin_t* plugin = NULL; - xce_sched_job_t* job = NULL; - int rc = 0; - char* vresname = NULL; - xce_vres_t* vres = NULL; - limit_plugin_data_t* privdata = NULL; - xce_scheduler_t* sched = NULL; + xce_limit_t* limit = (xce_limit_t*)arg2; + match_limit_args_t* args = (match_limit_args_t*)arg1; + xce_sched_plugin_t* plugin = NULL; + xce_sched_obj_t* obj = NULL; + int rc = 0; + char* vresname = NULL; + xce_vres_t* vres = NULL; + limit_plugin_data_t* privdata = NULL; + xce_scheduler_t* sched = NULL; if ((limit == NULL) || (args == NULL)) { return -EINVAL; } - ctx = args->ctx; + obj = args->ctx->schedobj; plugin = args->plugin; - if ((ctx == NULL) || (plugin == NULL)) { + if ((plugin == NULL) || (obj == NULL)) { return -EINVAL; } - job = ctx->job; sched = plugin->sched; privdata = (limit_plugin_data_t*)plugin->data; - if ((job == NULL) || (privdata == NULL) || (sched == NULL)) { + if ((privdata == NULL) || (sched == NULL)) { return -EINVAL; } @@ -154,17 +152,14 @@ static int match_job_global_limit_(void* arg1, void* arg2) return 0; } - rc = match_limit(limit, 1, job->user->name, job->queue->name, - job->project->name); + rc = match_limit(limit, 1, obj->user, obj->queue, obj->project); TRACE("%s match_limit limit %s user %s queue %s project %s rc %d", __func__, - limit->name, job->user->name, job->queue->name, job->project->name, - rc); + limit->name, obj->user, obj->queue, obj->project, rc); if (rc <= 0) { return rc; } - rc = build_vres_name(&vresname, limit, job->user->name, job->queue->name, - job->project->name); + rc = build_vres_name(&vresname, limit, obj->user, obj->queue, obj->project); if (rc < 0) { return rc; } @@ -173,8 +168,8 @@ static int match_job_global_limit_(void* arg1, void* arg2) FREEUP(vresname); if (vres == NULL) { - rc = instantiate_limit(&vres, &(sched->resmgr), limit, job->user->name, - job->queue->name, job->project->name); + rc = instantiate_limit(&vres, &(sched->resmgr), limit, obj->user, + obj->queue, obj->project); if (rc < 0) { return rc; } @@ -187,63 +182,78 @@ static int match_job_global_limit_(void* arg1, void* arg2) rc = update_limit_res_value(&(sched->sharedInst), vres); if (rc >= 0) { - rc = xlist_store(&(ctx->limits), vres); + rc = xlist_store(&(obj->glimits), vres); } + return rc; } -static int limit_job_election(void* arg, xce_sched_ctx_t* ctx) +static int limit_job_election(void* arg, xce_sched_job_t* job) { - xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; - limit_plugin_data_t* privdata = (limit_plugin_data_t*)plugin->data; - xce_sched_job_t* job = (xce_sched_job_t*)ctx->job; - int rc = 0; - xlist_t lst; - match_job_limit_args_t args; + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + limit_plugin_data_t* privdata = (limit_plugin_data_t*)plugin->data; + xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; + xce_sched_obj_t* obj = job->schedobj; + int rc = 0; TRACE("%s job %lu plugin %s", __func__, job->jobid, plugin->name); - args.plugin = plugin; - args.ctx = ctx; + if (obj->configver < sched->configver) { + xlist_t lst; + match_limit_args_t args; + xce_sched_ctx_t ctx = { 0 }; - xlist_init(&lst, 3); - xlist_store(&lst, match_job_global_limit_); - rc = kvmgr_iter(&(privdata->limits), NULL, &lst, &args); - xlist_fini(&lst); + reset_sched_obj(obj); + + ctx.sched = plugin->sched; + ctx.worker = NULL; + ctx.schedobj = job->schedobj; + + args.plugin = plugin; + args.ctx = &ctx; + + xlist_init(&lst, 3); + xlist_store(&lst, match_job_global_limit_); + rc = kvmgr_iter(&(privdata->limits), NULL, &lst, &args); + xlist_fini(&lst); + + obj->configver = sched->configver; + } return rc; } static int match_job_host_limit_(void* arg1, void* arg2) { - xce_limit_t* limit = (xce_limit_t*)arg2; - match_job_limit_args_t* args = (match_job_limit_args_t*)arg1; - xce_sched_ctx_t* ctx = NULL; - xce_sched_plugin_t* plugin = NULL; - xce_sched_job_t* job = NULL; - int rc = 0; - char* vresname = NULL; - xce_vres_t* vres = NULL; - limit_plugin_data_t* privdata = NULL; - xce_scheduler_t* sched = NULL; - xce_sched_worker_t* worker = NULL; + xce_limit_t* limit = (xce_limit_t*)arg2; + match_limit_args_t* args = (match_limit_args_t*)arg1; + xce_sched_plugin_t* plugin = NULL; + xce_sched_obj_t* obj = NULL; + int rc = 0; + char* vresname = NULL; + xce_vres_t* vres = NULL; + limit_plugin_data_t* privdata = NULL; + xce_scheduler_t* sched = NULL; + xce_sched_worker_t* worker = NULL; + xce_sched_ctx_t* ctx = NULL; + xlist_t* hlimits = NULL; if ((limit == NULL) || (args == NULL)) { return -EINVAL; } - ctx = args->ctx; + obj = args->ctx->schedobj; plugin = args->plugin; - if ((ctx == NULL) || (plugin == NULL)) { + if ((obj == NULL) || (plugin == NULL)) { return -EINVAL; } - job = ctx->job; sched = plugin->sched; + ctx = args->ctx; privdata = (limit_plugin_data_t*)plugin->data; worker = ctx->worker; - if ((job == NULL) || (sched == NULL) || (privdata == NULL)) { + if ((sched == NULL) || (privdata == NULL)) { return -EINVAL; } @@ -255,17 +265,27 @@ static int match_job_host_limit_(void* arg1, void* arg2) return 0; } - rc = match_limit(limit, 0, job->user->name, job->queue->name, - job->project->name, worker->name); + hlimits = (xlist_t*)zarray_get(&(obj->hlimits), worker->idx); + if (hlimits == NULL) { + hlimits = (xlist_t*)calloc(1, sizeof(*hlimits)); + if (hlimits == NULL) { + return -EINVAL; + } + + xlist_init(hlimits, 8); + + zarray_put(&(obj->hlimits), worker->idx, hlimits); + } + + rc = match_limit(limit, 0, obj->user, obj->queue, obj->project, + worker->name); TRACE("%s match_limit limit %s user %s queue %s project %s rc %d", __func__, - limit->name, job->user->name, job->queue->name, job->project->name, - rc); + limit->name, obj->user, obj->queue, obj->project, rc); if (rc <= 0) { return rc; } - rc = build_vres_name(&vresname, limit, job->user->name, job->queue->name, - job->project->name); + rc = build_vres_name(&vresname, limit, obj->user, obj->queue, obj->project); if (rc <= 0) { return rc; } @@ -274,8 +294,8 @@ static int match_job_host_limit_(void* arg1, void* arg2) FREEUP(vresname); if (vres == NULL) { - rc = instantiate_limit(&vres, &(sched->resmgr), limit, job->user->name, - job->queue->name, job->project->name); + rc = instantiate_limit(&vres, &(sched->resmgr), limit, obj->user, + obj->queue, obj->project); if (rc < 0) { return rc; } @@ -288,20 +308,19 @@ static int match_job_host_limit_(void* arg1, void* arg2) rc = update_limit_res_value(&(worker->resvals), vres); if (rc >= 0) { - rc = xlist_store(&(ctx->limits_host), vres); + rc = xlist_store(hlimits, vres); } return rc; } +#if 0 static int build_limit_vres_requests(xce_scheduler_t* sched, - xce_sched_ctx_t* ctx, xlist_t* lst, - xlist_t* requests) + xce_sched_ctx_t* ctx) { int i; - if ((sched == NULL) || (ctx == NULL) || (lst == NULL) - || (requests == NULL)) { + if ((sched == NULL) || (ctx == NULL)) { return -EINVAL; } @@ -335,17 +354,19 @@ static int build_limit_vres_requests(xce_scheduler_t* sched, xlist_store(requests, rq); } + return 0; } +#endif + static int limit_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) { - xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; - xce_sched_job_t* job = NULL; - xce_sched_worker_t* worker = NULL; - match_job_limit_args_t args; - xlist_t lst; - int rc = 0; - limit_plugin_data_t* privdata = NULL; + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + xce_sched_worker_t* worker = NULL; + match_limit_args_t args; + xlist_t lst; + int rc = 0; + limit_plugin_data_t* privdata = NULL; if ((plugin == NULL) || (ctx == NULL)) { return -EINVAL; @@ -353,34 +374,19 @@ static int limit_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) worker = ctx->worker; if (worker == NULL) { - /* we need to fill global limit requests here */ - return build_limit_vres_requests(plugin->sched, ctx, &(ctx->limits), - &(ctx->extra_res_requests)); + return 0; } - job = ctx->job; privdata = (limit_plugin_data_t*)plugin->data; - TRACE("%s job %lu plugin %s", __func__, job->jobid, plugin->name); - args.plugin = plugin; args.ctx = ctx; - xlist_fini(&(ctx->limits_host)); - xlist_init(&(ctx->limits_host), 32); - xlist_init(&lst, 2); xlist_store(&lst, match_job_host_limit_); rc = kvmgr_iter(&(privdata->limits), NULL, &lst, &args); xlist_fini(&lst); - if (rc < 0) { - return rc; - } - /* fill host based limits here */ - rc = build_limit_vres_requests(plugin->sched, ctx, &(ctx->limits_host), - &(ctx->extra_host_res_requests)); - return rc; } diff --git a/src/scheduler/src/plugins/parjob_allocator.c b/src/scheduler/src/plugins/parjob_allocator.c index 4041fe10..86bc3d73 100644 --- a/src/scheduler/src/plugins/parjob_allocator.c +++ b/src/scheduler/src/plugins/parjob_allocator.c @@ -42,10 +42,38 @@ static int skip_bad_worker_(xce_sched_worker_t* w) return 0; } +static int parjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + xce_sched_worker_t* worker = NULL; + xce_sched_obj_t* obj = NULL; + + if ((plugin == NULL) || (ctx == NULL)) { + return -EINVAL; + } + + obj = ctx->schedobj; + if (obj == NULL) { + return -EINVAL; + } + + if (obj->tag == XCE_RESREQ_PRIMARY) { + xce_primary_resreq_t* primary = obj->u.primary->resreq; + + if (primary->maxTasks == 1) { + /* do not care sequence job */ + + return 0; + } + } + + /* we care parallel jobs */ + 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_obj_t* schedobj = ctx->schedobj; return 0; @@ -54,8 +82,12 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) 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 }, + .name = "xce_sched_parjob_allocator", + .hooks + = { .config_hook = NULL, + .commit_hook = NULL, + .prepare_allocation_hook + = (plugin_prepare_allocation_hook_t)parjob_prepare_allocation }, .init_fn = (plugin_init_fn_t)parjob_allocator_init, .fini_fn = (plugin_fini_fn_t)parjob_allocator_fini, .begin_session_fn = NULL, diff --git a/src/scheduler/src/plugins/seqjob_allocator.c b/src/scheduler/src/plugins/seqjob_allocator.c index c4238859..bea24bb9 100644 --- a/src/scheduler/src/plugins/seqjob_allocator.c +++ b/src/scheduler/src/plugins/seqjob_allocator.c @@ -42,17 +42,67 @@ static int skip_bad_worker_(xce_sched_worker_t* w) return 0; } +static int seqjob_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; + xce_sched_worker_t* worker = NULL; + xce_sched_obj_t* obj = NULL; + xce_resource_t* r = NULL; + xce_primary_resreq_t* primary = NULL; + + if ((plugin == NULL) || (ctx == NULL) || (sched == NULL)) { + return -EINVAL; + } + + obj = ctx->schedobj; + if (obj == NULL) { + return -EINVAL; + } + + if (obj->tag != XCE_RESREQ_PRIMARY) { + /* we just take care of sequence job */ + return 0; + } + + primary = obj->u.primary->resreq; + if (primary == NULL) { + return -EINVAL; + } + + if (ctx->worker == NULL) { + ctx->nhosts = 1; + ctx->nslots = 1; + calc_wanted_global_resvals(&(ctx->alloc_req.shared), &(sched->resmgr), + &(primary->rusages), PER_TASK, 1, 1); + } else { + calc_wanted_host_resvals(&(ctx->alloc_req.host), &(sched->resmgr), + &(primary->rusages), PER_TASK, 1, 1); + } + + /* add slots into requested resources */ + r = resmgr_lockres(&(sched->resmgr), "slots", OBJREF_NOP); + if (r == NULL) { + return -EINVAL; + } + + xvec_idval_store(&(ctx->alloc_req.host), r->idx, 1.0); + + /* as r has been locked with OBJREF_NOP, so no need to unlock it */ + 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_obj_t* schedobj = ctx->schedobj; - 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_obj_t* schedobj = ctx->schedobj; + int rc; + xlist_t workers; + xce_pendreason_t* reason = NULL; + xvec_t* uncomit = NULL; + xce_sched_worker_t* w = NULL; - reason = make_pendreason(0); + reason = make_pendreason(0); schedobj->reason = reason; if (reason == NULL) { return -ENOMEM; @@ -96,9 +146,23 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) return rc; } - xlist_find(&workers, (xlist_find_fn_t)sched_try_alloc_on_hosts, sched, ctx); + w = xlist_find(&workers, (xlist_find_fn_t)sched_try_alloc_on_hosts, sched, + ctx); + + if (w != NULL) { + xce_host_alloc_t** hallocs = NULL; + xce_host_alloc_t* halloc = NULL; + xce_alloc_t** allocs = NULL; + + 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); + + halloc = make_host_alloc(w->name, ctx->alloc_req.host.count, allocs); + hallocs = make_host_allocs(1, halloc); + + ctx->alloc = make_job_alloc(0, 1, hallocs); - if (ctx->alloc != NULL) { if (ctx->alloc_req.shared.count > 0) { ctx->alloc->allocs_shared = sched_make_allocs(sched, "", ctx->alloc_req.shared.count, @@ -120,8 +184,12 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) int register_seqjob_allocator(void* sched) { static xce_sched_plugin_t seqjob_allocator_plugin = { - .name = "xce_sched_seqjob_allocator", - .hooks = { .config_hook = NULL, .commit_hook = NULL }, + .name = "xce_sched_seqjob_allocator", + .hooks + = { .config_hook = NULL, + .commit_hook = NULL, + .prepare_allocation_hook + = (plugin_prepare_allocation_hook_t)seqjob_prepare_allocation }, .init_fn = (plugin_init_fn_t)seqjob_allocator_init, .fini_fn = (plugin_fini_fn_t)seqjob_allocator_fini, .begin_session_fn = NULL, diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index 712be867..34e9dcef 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -343,20 +343,14 @@ 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; - xce_sched_obj_t* schedobj = ctx->schedobj; - 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_obj_t* schedobj = ctx->schedobj; + xce_primary_resreq_t* primary = ctx->primary; + int rc; + xvec_t* uncomit = NULL; - /* FIXME: */ -#if 0 - rc = sched_meet_host_selection(sched, w, resreq->u.simple.selection); -#endif + rc = sched_meet_host_selection(sched, w, primary->select); if (rc == 0) { add_job_preason(schedobj->reason->reasons, PREASON_MATCH_FAIL, w->name, NULL); @@ -378,8 +372,6 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) return rc; } - sched_ctx_fill_host_requests(sched, ctx); - rc = sched_check_host_satisfy(sched, w, uncomit, &(ctx->alloc_req.host), &(ctx->unmeet)); if (rc < 0) { @@ -405,15 +397,6 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) 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); - - halloc = make_host_alloc(w->name, ctx->alloc_req.host.count, allocs); - hallocs = make_host_allocs(1, halloc); - - ctx->alloc = make_job_alloc(job->jobid, 1, hallocs); - /* host is OK */ return 1; } @@ -471,78 +454,43 @@ int sched_commit_job_alloc(xce_scheduler_t* sched, xce_sched_job_t* job) return sched_charge_account(sched, job->accounts); } -static int fill_ctx_idval_vec_(xce_scheduler_t* sched, xvec_idval_t* v, - int nlists, xlist_t** lists) +static float get_resval_by_id(xvec_idval_t* vec, int idx) { - int count; - int i, rc; + int i; - count = 0; - for (i = 0; i < nlists; i++) { - count += lists[i]->count; - } + for (i = 0; i < vec->count; i++) { + int residx; - memset(v, 0x0, sizeof(*v)); - rc = xvec_idval_init(v, count); - if (rc < 0) { - return rc; - } + xvec_get(&(vec->idxs), i, &residx); - for (i = 0; i < nlists; i++) { - xlist_t* lst = lists[i]; - int j; + if (residx == idx) { + float resval; - for (j = 0; j < lst->count; j++) { - request_res_t* rq = (request_res_t*)xlist_get(lst, j); - xce_resource_t* r = NULL; + xvec_get(&(vec->vals), i, &resval); - if (rq == NULL) { - return -EINVAL; - } - - r = resmgr_lockres(&(sched->resmgr), rq->res, OBJREF_NOP); - if (r == NULL) { - return -ENOENT; - } - - rc = xvec_idval_store(v, r->idx, rq->value); - if (rc < 0) { - return rc; - } + return resval; } } - - return v->count; + return 0.0; } -int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) +int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec, + xlist_t* limits) { - int rc = 0; -#if 0 - xlist_t* lists[] - = { &(ctx->resreq->u.simple.res_shared), &(ctx->extra_res_requests) }; + int i; - rc = fill_ctx_idval_vec_(sched, &(ctx->alloc_req.shared), 2, &lists[0]); - if (rc < 0) { - xvec_idval_fini(&(ctx->alloc_req.shared)); + for (i = 0; i < limits->count; i++) { + xce_vres_t* vres = (xce_vres_t*)xlist_get(limits, i); + float val = 0.0; + + if (vres == NULL) { + continue; + } + + val = get_resval_by_id(vec, vres->res->idx); + + xvec_idval_store(vec, vres->vres->idx, val); } -#endif - return rc; -} - -int sched_ctx_fill_host_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) -{ - int rc = 0; -#if 0 - xlist_t* lists[] = { &(ctx->resreq->u.simple.res_host), - &(ctx->extra_host_res_requests) }; - - rc = fill_ctx_idval_vec_(sched, &(ctx->alloc_req.host), 2, &lists[0]); - if (rc < 0) { - xvec_idval_fini(&(ctx->alloc_req.host)); - } -#endif - - return rc; + return 0; } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index 475230a3..62a13f04 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -36,8 +36,8 @@ request_res_t* make_resource_request(char* res, float value) return req; } -static int make_primary_schedobj(xce_sched_obj_t* obj, char* user, char* queue, - char* project, xce_primary_resreq_t* resreq) +static int make_primary_schedobj(xce_sched_obj_t* obj, + xce_primary_resreq_t* resreq) { xce_primary_schedobj_t* primary = (xce_primary_schedobj_t*)calloc(1, sizeof(*primary)); @@ -47,18 +47,17 @@ static int make_primary_schedobj(xce_sched_obj_t* obj, char* user, char* queue, return -ENOMEM; } - primary->user = user; - primary->queue = queue; - primary->project = project; - primary->resreq = resreq; + primary->resreq = resreq; + primary->reason = NULL; + primary->owner = obj; obj->u.primary = primary; return 1; } -static int make_compound_schedobj(xce_sched_obj_t* obj, char* user, char* queue, - char* project, xce_compound_resreq_t* resreq) +static int make_compound_schedobj(xce_sched_obj_t* obj, + xce_compound_resreq_t* resreq) { xce_compound_schedobj_t* compound = (xce_compound_schedobj_t*)calloc(1, sizeof(*compound)); @@ -69,18 +68,15 @@ static int make_compound_schedobj(xce_sched_obj_t* obj, char* user, char* queue, return -EINVAL; } - compound->user = user; - compound->queue = queue; - compound->project = project; - compound->resreq = resreq; + compound->resreq = resreq; + compound->owner = obj; obj->u.compound = compound; return 1; } -static int make_alter_schedobj(xce_sched_obj_t* obj, char* user, char* queue, - char* project, xce_alter_resreq_t* resreq) +static int make_alter_schedobj(xce_sched_obj_t* obj, xce_alter_resreq_t* resreq) { int i; int rc = 0; @@ -89,24 +85,21 @@ static int make_alter_schedobj(xce_sched_obj_t* obj, char* user, char* queue, xlist_init(&(obj->u.alters), 4); for (i = 0; i < resreq->compounds.count; i++) { - xce_sched_obj_t* cobj = NULL; - xce_compound_resreq_t* cresreq + xce_compound_schedobj_t* cobj = NULL; + xce_compound_resreq_t* cresreq = (xce_compound_resreq_t*)xlist_get(&(resreq->compounds), i); if (cresreq == NULL) { continue; } - cobj = (xce_sched_obj_t*)calloc(1, sizeof(*cobj)); + cobj = (xce_compound_schedobj_t*)calloc(1, sizeof(*cobj)); if (cobj == NULL) { return -ENOMEM; } - rc = make_compound_schedobj(cobj, user, queue, project, cresreq); - if (rc < 0) { - FREEUP(cobj); - return rc; - } + cobj->resreq = cresreq; + cobj->owner = obj; xlist_store(&(obj->u.alters), cobj); } @@ -119,25 +112,38 @@ static int make_sched_obj(xce_sched_obj_t* obj, xce_scheduler_t* sched, { int rc = 0; + obj->user = strdup(j->submit->user); + obj->queue = strdup(j->submit->queue); + obj->project = strdup(j->submit->project_name); + + xlist_init(&(obj->glimits), 8); + zarray_init(&(obj->hlimits), 32, NULL); + switch (resreq->tag) { case XCE_RESREQ_PRIMARY: - rc = make_primary_schedobj(obj, j->submit->user, j->submit->queue, - j->submit->project_name, resreq->u.primary); + rc = make_primary_schedobj(obj, resreq->u.primary); break; case XCE_RESREQ_COMPOUND: - rc = make_compound_schedobj(obj, j->submit->user, j->submit->queue, - j->submit->project_name, - resreq->u.compound); + rc = make_compound_schedobj(obj, resreq->u.compound); break; case XCE_RESREQ_ALTER: - rc = make_alter_schedobj(obj, j->submit->user, j->submit->queue, - j->submit->project_name, resreq->u.alter); + rc = make_alter_schedobj(obj, resreq->u.alter); break; } + obj->configver = 0; + return 1; } +void reset_sched_obj(xce_sched_obj_t* obj) +{ + xlist_fini(&(obj->glimits)); + zarray_fini(&(obj->hlimits)); + + return; +} + 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)); @@ -349,21 +355,28 @@ static void host_uncommit_killer(void* obj) FREEUP(v); } -int sched_ctx_init(xce_sched_ctx_t* ctx, xce_sched_job_t* job) +int sched_ctx_init(xce_sched_ctx_t* ctx, void* arg, xce_sched_obj_t* obj) { int rc = 0; - if ((ctx == NULL) || (job == NULL)) { + if ((ctx == NULL) || (obj == NULL)) { return -EINVAL; } memset(ctx, 0x0, sizeof(*ctx)); - ctx->job = job; - xlist_init(&(ctx->limits), 4); - xlist_init(&(ctx->limits_host), 4); - xlist_init(&(ctx->extra_res_requests), 4); - xlist_init(&(ctx->extra_host_res_requests), 4); + switch (obj->tag) { + case XCE_RESREQ_PRIMARY: + break; + case XCE_RESREQ_COMPOUND: + break; + default: + ERROR("%s bad sched obj type %d", __func__, obj->tag); + return -EINVAL; + } + + ctx->schedobj = obj; + ctx->sched = arg; ctx->uncomitted.shared = (xvec_t*)calloc(1, sizeof(xvec_t)); if (ctx->uncomitted.shared == NULL) { @@ -375,8 +388,13 @@ int sched_ctx_init(xce_sched_ctx_t* ctx, xce_sched_job_t* job) return rc; } + xvec_idval_init(&(ctx->unmeet), 8); + xvec_idval_init(&(ctx->alloc_req.shared), 8); + xvec_idval_init(&(ctx->alloc_req.host), 8); + zarray_init(&(ctx->uncomitted.hosts), 32, (zarray_item_killer_t)host_uncommit_killer); + return 0; } @@ -395,11 +413,16 @@ void sched_ctx_fini(xce_sched_ctx_t* ctx) xvec_idval_fini(&(ctx->alloc_req.shared)); xvec_idval_fini(&(ctx->alloc_req.host)); + xvec_idval_fini(&(ctx->unmeet)); +} - xlist_fini(&(ctx->extra_res_requests)); - xlist_fini(&(ctx->extra_host_res_requests)); - xlist_fini(&(ctx->limits)); - xlist_fini(&(ctx->limits_host)); +void sched_ctx_set_primary(xce_sched_ctx_t* ctx, xce_primary_resreq_t* primary) +{ + ctx->primary = primary; + + plugin_prepare_allocation((xce_scheduler_t*)ctx->sched, ctx); + + return; } int sched_ctx_get_uncomit_shared(xvec_t** shared, xce_sched_ctx_t* ctx) diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index ab6c98ed..29e32599 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -32,9 +32,8 @@ static void* sched_job_picker(void* arg) return job_picker->job_elect_fn(job_picker); } -static int sched_make_seqjob_allocation(xce_scheduler_t* sched, - xce_sched_ctx_t* ctx, - xce_primary_resreq_t* primary) +static int sched_make_seqjob_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) { xce_sched_plugin_t* plugin = NULL; @@ -45,16 +44,11 @@ static int sched_make_seqjob_allocation(xce_scheduler_t* sched, 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_parjob_allocation(xce_scheduler_t* sched, - xce_sched_ctx_t* ctx, - xce_primary_resreq_t* primary) +static int sched_make_parjob_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) { xce_sched_plugin_t* plugin = NULL; @@ -65,34 +59,42 @@ static int sched_make_parjob_allocation(xce_scheduler_t* sched, 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_alloc_(xce_scheduler_t* sched, - xce_sched_ctx_t* ctx, - xce_primary_resreq_t* primary) +static int sched_make_primary_alloc_(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) { + xce_primary_resreq_t* primary = ctx->primary; if (primary->maxTasks == 1) { - return sched_make_seqjob_allocation(sched, ctx, primary); + return sched_make_seqjob_allocation(sched, ctx); } - return sched_make_parjob_allocation(sched, ctx, primary); + return sched_make_parjob_allocation(sched, ctx); } static int sched_make_primary_allocation(xce_scheduler_t* sched, xce_sched_obj_t* obj) { xce_primary_resreq_t* primary = obj->u.primary->resreq; - xce_sched_ctx_t ctx = { - .schedobj = obj - }; + xce_sched_ctx_t ctx; + int rc = 0; - return sched_make_primary_alloc_(sched, &ctx, primary); + sched_ctx_init(&ctx, sched, obj); + sched_ctx_set_primary(&ctx, primary); + rc = sched_make_primary_alloc_(sched, &ctx); + if (ctx.alloc != NULL) { + MOVE_PTR(obj->alloc, ctx.alloc); + } + + if (ctx.reason != NULL) { + MOVE_PTR(obj->reason, ctx.reason); + } + + sched_ctx_fini(&ctx); + + return rc; } static int sched_make_compound_allocation(xce_scheduler_t* sched, @@ -100,21 +102,33 @@ static int sched_make_compound_allocation(xce_scheduler_t* sched, { int i; xce_compound_resreq_t* resreq = obj->u.compound->resreq; - xce_sched_ctx_t ctx = { - .schedobj = obj - }; + xce_sched_ctx_t ctx; + int rc = 0; + + sched_ctx_init(&ctx, sched, obj); for (i = 0; i < resreq->primaries.count; i++) { - xce_primary_resreq_t* resreq1 + xce_primary_resreq_t* primary = (xce_primary_resreq_t*)xlist_get(&(resreq->primaries), i); - int rc; - rc = sched_make_primary_alloc_(sched, &ctx, resreq1); + sched_ctx_set_primary(&ctx, primary); + rc = sched_make_primary_alloc_(sched, &ctx); if (rc <= 0) { - return rc; + break; } } - return 1; + + if (ctx.alloc != NULL) { + MOVE_PTR(obj->alloc, ctx.alloc); + } + + if (ctx.reason != NULL) { + MOVE_PTR(obj->reason, ctx.reason); + } + + sched_ctx_fini(&ctx); + + return rc; } static int sched_make_alter_allocation(xce_scheduler_t* sched, @@ -131,14 +145,18 @@ static int sched_make_alter_allocation(xce_scheduler_t* sched, continue; } - /* FIXME: we need to initialize ctx1 */ - rc = sched_make_compound_allocation(sched, obj1); if (rc > 0) { + if (obj1->alloc != NULL) { + MOVE_PTR(obj->alloc, obj1->alloc); + } + return rc; } } + /* FIXME: merge pend reason from all children */ + return 0; } @@ -163,23 +181,6 @@ static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_obj_t* obj) return rc; } -static xce_decision_t* calc_decision(xce_scheduler_t* sched, - xce_sched_obj_t* obj) -{ - int rc = 0; - - rc = sched_make_allocation(sched, obj); - if (rc < 0) { - return NULL; - } - - if (rc == 0) { - return make_pend_decision(obj->reason); - } - - return make_dispatch_decision(obj->alloc); -} - static int commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) { return sched_commit_decision(sched, decision); @@ -194,6 +195,7 @@ static void* sched_allocator(void* arg) uint64_t jobid; xce_decision_t* decision = NULL; xce_sched_job_t* job = NULL; + int rc = 0; pthread_testcancel(); @@ -211,12 +213,26 @@ static void* sched_allocator(void* arg) continue; } -#if 0 - plugin_job_election(sched, &schedctx); -#endif + plugin_job_election(sched, job); /* calculate decision for jobid */ - decision = calc_decision(sched, job->schedobj); + rc = sched_make_allocation(sched, job->schedobj); + if (rc < 0) { + /* something wrong */ + continue; + } else if (rc == 0) { + /* job is pend */ + xce_pendreason_t* reason = job->schedobj->reason; + + reason->jobid = job->jobid; + decision = make_pend_decision(reason); + } else { + MOVE_PTR(job->allocs, job->schedobj->alloc); + job->allocs->jobid = job->jobid; + + decision = make_dispatch_decision(job->allocs); + } + if (decision != NULL) { commit_decision(sched, decision); cbuf_put(&(sched->decisions), &decision); @@ -285,7 +301,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, memset(sched, 0x0, sizeof(*sched)); - sched->nextid = 1; + sched->nextid = 1; + sched->configver = 0; mylock_init(&(sched->lock)); cbuf_init(&(sched->wbuf), 1, sizeof(int)); @@ -444,6 +461,8 @@ int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config) sched_init_sharedInst(sched); sched_init_queues(sched); + sched->configver++; /* increase the version of configuration */ + return 0; }