diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index db641d59..9265dd81 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -28,10 +28,28 @@ typedef struct { } request_res_t; typedef struct { - char* user; - char* queue; - char* project; - xce_resreq_t* resreq; + char* user; + char* queue; + char* project; + xce_primary_resreq_t* resreq; +} xce_primary_schedobj_t; + +typedef struct { + char* user; + char* queue; + char* project; + xce_compound_resreq_t* resreq; +} xce_compound_schedobj_t; + +typedef struct { + xce_resreq_tag_t tag; + union { + xce_primary_schedobj_t* primary; + xce_compound_schedobj_t* compound; + xlist_t alters; + } u; + xce_job_alloc_t* alloc; + xce_pendreason_t* reason; } xce_sched_obj_t; typedef struct { @@ -62,6 +80,7 @@ typedef struct { xce_sched_queue_t* queue; xce_sched_user_t* user; xce_sched_project_t* project; + xce_resreq_t* resreq; xce_sched_obj_t* schedobj; uint32_t mintasks; uint32_t maxtasks; @@ -81,9 +100,7 @@ typedef struct { xce_sched_job_t* job; xce_sched_worker_t* worker; xce_sched_obj_t* schedobj; - xce_resreq_t* resreq; xce_job_alloc_t* alloc; - xce_pendreason_t* reason; struct { xvec_t* shared; zarray_t hosts; diff --git a/src/scheduler/src/plugins/seqjob_allocator.c b/src/scheduler/src/plugins/seqjob_allocator.c index f4bc7a9a..c4238859 100644 --- a/src/scheduler/src/plugins/seqjob_allocator.c +++ b/src/scheduler/src/plugins/seqjob_allocator.c @@ -52,8 +52,8 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) xce_pendreason_t* reason = NULL; xvec_t* uncomit = NULL; - reason = make_pendreason(job->jobid); - ctx->reason = reason; + reason = make_pendreason(0); + schedobj->reason = reason; if (reason == NULL) { return -ENOMEM; } diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index e3fb7782..712be867 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -356,13 +356,13 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) /* FIXME: */ #if 0 rc = sched_meet_host_selection(sched, w, resreq->u.simple.selection); +#endif if (rc == 0) { - add_job_preason(ctx->reason->reasons, PREASON_MATCH_FAIL, w->name, + add_job_preason(schedobj->reason->reasons, PREASON_MATCH_FAIL, w->name, NULL); return 0; } -#endif rc = sched_ctx_get_uncomit_host(&uncomit, ctx, w->idx); if (rc < 0) { @@ -394,8 +394,8 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) xce_resource_t* r = resmgr_lock_byid( &(sched->resmgr), ((int*)ctx->unmeet.idxs.data)[i], OBJREF_NOP); - add_job_preason(ctx->reason->reasons, PREASON_NO_RESOURCE, w->name, - r->name); + add_job_preason(schedobj->reason->reasons, PREASON_NO_RESOURCE, + w->name, r->name); } return 0; } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index ef184f78..475230a3 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -36,27 +36,103 @@ request_res_t* make_resource_request(char* res, float value) return req; } -static int make_sched_obj(xce_sched_obj_t* obj, xce_scheduler_t* sched, - xce_job_t* j) +static int make_primary_schedobj(xce_sched_obj_t* obj, char* user, char* queue, + char* project, xce_primary_resreq_t* resreq) { - int rc; - char* errmsg; + xce_primary_schedobj_t* primary + = (xce_primary_schedobj_t*)calloc(1, sizeof(*primary)); - memset(obj, 0x0, sizeof(*obj)); - obj->resreq = (xce_resreq_t*)calloc(1, sizeof(*(obj->resreq))); - if (obj->resreq == NULL) { + obj->tag = XCE_RESREQ_PRIMARY; + if (primary == NULL) { + return -ENOMEM; + } + + primary->user = user; + primary->queue = queue; + primary->project = project; + primary->resreq = resreq; + + 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) +{ + xce_compound_schedobj_t* compound + = (xce_compound_schedobj_t*)calloc(1, sizeof(*compound)); + + obj->tag = XCE_RESREQ_COMPOUND; + + if (compound == NULL) { return -EINVAL; } - rc = parse_resreq(obj->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); + compound->user = user; + compound->queue = queue; + compound->project = project; + compound->resreq = resreq; + + 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) +{ + int i; + int rc = 0; + + obj->tag = XCE_RESREQ_ALTER; + 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_resreq_t*)xlist_get(&(resreq->compounds), i); + + if (cresreq == NULL) { + continue; } - return rc; + cobj = (xce_sched_obj_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; + } + + xlist_store(&(obj->u.alters), cobj); + } + + return 1; +} + +static int make_sched_obj(xce_sched_obj_t* obj, xce_scheduler_t* sched, + xce_job_t* j, xce_resreq_t* resreq) +{ + int rc = 0; + + 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); + break; + case XCE_RESREQ_COMPOUND: + rc = make_compound_schedobj(obj, j->submit->user, j->submit->queue, + j->submit->project_name, + 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); + break; } return 1; @@ -64,9 +140,10 @@ static int make_sched_obj(xce_sched_obj_t* obj, xce_scheduler_t* sched, 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; + xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + int rc = 0; + char* errmsg = NULL; if ((sched == NULL) || (job == NULL)) { return NULL; @@ -80,9 +157,21 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j) job->user = sched_get_user(sched, j->submit->user); job->queue = sched_get_queue(sched, j->submit->queue); job->project = sched_get_project(sched, j->submit->project_name); - job->mintasks = j->submit->mintasks; - job->maxtasks = j->submit->maxtasks; - job->priority = 0; /* FIXME: support job level priority */ + job->resreq = (xce_resreq_t*)calloc(1, sizeof(*(job->resreq))); + if (job->resreq == NULL) { + return NULL; + } + + rc = parse_resreq(job->resreq, j->submit->resreq, &errmsg); + if (rc < 0) { + ERROR("%s parse_resreq failed, %s", __func__, errmsg); + FREEUP(errmsg); + + return NULL; + } + job->mintasks = j->submit->mintasks; + job->maxtasks = j->submit->maxtasks; + job->priority = 0; /* FIXME: support job level priority */ job->schedobj = (xce_sched_obj_t*)calloc(1, sizeof(*(job->schedobj))); if (job->schedobj == NULL) { @@ -92,7 +181,7 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j) return NULL; } - make_sched_obj(job->schedobj, sched, j); + make_sched_obj(job->schedobj, sched, j, job->resreq); job->allocs = NULL; job->accounts = NULL; if (j->allocs != NULL) { diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 6c61adf6..ab6c98ed 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -32,8 +32,9 @@ 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) +static int sched_make_seqjob_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx, + xce_primary_resreq_t* primary) { xce_sched_plugin_t* plugin = NULL; @@ -51,8 +52,9 @@ static int sched_make_seqjob_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) +static int sched_make_parjob_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx, + xce_primary_resreq_t* primary) { xce_sched_plugin_t* plugin = NULL; @@ -70,33 +72,44 @@ static int sched_make_parjob_allocation(xce_scheduler_t* sched, return plugin->job_alloc_fn(plugin, ctx); } -static int sched_make_primary_allocation(xce_scheduler_t* sched, - xce_sched_ctx_t* ctx) +static int sched_make_primary_alloc_(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx, + xce_primary_resreq_t* primary) { - xce_primary_resreq_t* primary = ctx->schedobj->resreq->u.primary; if (primary->maxTasks == 1) { - return sched_make_seqjob_allocation(sched, ctx); + return sched_make_seqjob_allocation(sched, ctx, primary); } - return sched_make_parjob_allocation(sched, ctx); + return sched_make_parjob_allocation(sched, ctx, primary); +} + +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 + }; + + return sched_make_primary_alloc_(sched, &ctx, primary); } static int sched_make_compound_allocation(xce_scheduler_t* sched, - xce_sched_ctx_t* ctx) + xce_sched_obj_t* obj) { - xce_compound_resreq_t* resreq = ctx->schedobj->resreq->u.compound; int i; + xce_compound_resreq_t* resreq = obj->u.compound->resreq; + xce_sched_ctx_t ctx = { + .schedobj = obj + }; 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); + rc = sched_make_primary_alloc_(sched, &ctx, resreq1); if (rc <= 0) { return rc; } @@ -105,24 +118,22 @@ static int sched_make_compound_allocation(xce_scheduler_t* sched, } static int sched_make_alter_allocation(xce_scheduler_t* sched, - xce_sched_ctx_t* ctx) + xce_sched_obj_t* obj) { - xce_alter_resreq_t* resreq = ctx->schedobj->resreq->u.alter; - int i; + 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 }; + for (i = 0; i < obj->u.alters.count; i++) { + xce_sched_obj_t* obj1 + = (xce_sched_obj_t*)xlist_get(&(obj->u.alters), i); + int rc; - if (resreq1 == NULL) { + if (obj1 == NULL) { continue; } /* FIXME: we need to initialize ctx1 */ - rc = sched_make_compound_allocation(sched, &ctx1); + rc = sched_make_compound_allocation(sched, obj1); if (rc > 0) { return rc; } @@ -131,36 +142,19 @@ static int sched_make_alter_allocation(xce_scheduler_t* sched, return 0; } -static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) +static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_obj_t* obj) { - int rc = 0; - xce_sched_job_t* job = ctx->job; - xce_resreq_t* resreq = job->schedobj->resreq; + int rc = 0; - if (job->schedobj == NULL) { - xce_pendreason_t* reason = NULL; - - ERROR("%s job %lu has invalid resource requirement", __func__, - job->jobid); - reason = make_pendreason(job->jobid); - - add_job_preason(reason->reasons, PREASON_BAD_RESREQ, NULL, NULL); - - ctx->reason = reason; - - return 0; - } - - ctx->schedobj = job->schedobj; - switch (resreq->tag) { + switch (obj->tag) { case XCE_RESREQ_PRIMARY: - rc = sched_make_primary_allocation(sched, ctx); + rc = sched_make_primary_allocation(sched, obj); break; case XCE_RESREQ_COMPOUND: - rc = sched_make_compound_allocation(sched, ctx); + rc = sched_make_compound_allocation(sched, obj); break; case XCE_RESREQ_ALTER: - rc = sched_make_alter_allocation(sched, ctx); + rc = sched_make_alter_allocation(sched, obj); break; default: rc = -EINVAL; @@ -170,20 +164,20 @@ static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) } static xce_decision_t* calc_decision(xce_scheduler_t* sched, - xce_sched_ctx_t* ctx) + xce_sched_obj_t* obj) { int rc = 0; - rc = sched_make_allocation(sched, ctx); + rc = sched_make_allocation(sched, obj); if (rc < 0) { return NULL; } if (rc == 0) { - return make_pend_decision(ctx->reason); + return make_pend_decision(obj->reason); } - return make_dispatch_decision(ctx->alloc); + return make_dispatch_decision(obj->alloc); } static int commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) @@ -200,7 +194,6 @@ static void* sched_allocator(void* arg) uint64_t jobid; xce_decision_t* decision = NULL; xce_sched_job_t* job = NULL; - xce_sched_ctx_t schedctx; pthread_testcancel(); @@ -218,18 +211,16 @@ static void* sched_allocator(void* arg) continue; } - sched_ctx_init(&schedctx, job); - +#if 0 plugin_job_election(sched, &schedctx); +#endif /* calculate decision for jobid */ - decision = calc_decision(sched, &schedctx); + decision = calc_decision(sched, job->schedobj); if (decision != NULL) { commit_decision(sched, decision); cbuf_put(&(sched->decisions), &decision); } - - sched_ctx_fini(&schedctx); } return NULL;