lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

save

Browse source
This commit is contained in:
Rongsong Shen 2026-02-09 18:11:27 +08:00
parent 12a6010f29
commit e7ab1c16db
5 changed files with 188 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

29
src/scheduler/include/sched/types.h
View file

@ -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;

4
src/scheduler/src/plugins/seqjob_allocator.c
View file

@ -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;
}

8
src/scheduler/src/resource.c
View file

@ -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;
}

131
src/scheduler/src/schedobj.c
View file

@ -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) {

107
src/scheduler/src/scheduler.c
View file

@ -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;