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

- make new scheduler work for sequence jobs

Browse source
This commit is contained in:
Rongsong Shen 2026-02-12 11:19:47 +08:00
parent e7ab1c16db
commit 6cf9aa0979
10 changed files with 461 additions and 319 deletions
Download patch file Download diff file Expand all files Collapse all files

5
src/scheduler/include/plugin.h
View file

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

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

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

16
src/scheduler/include/scheduler.h
View file

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

50
src/scheduler/src/plugin.c
View file

@ -4,6 +4,7 @@
#include <stdlib.h>
#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;
}

182
src/scheduler/src/plugins/limit.c
View file

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

38
src/scheduler/src/plugins/parjob_allocator.c
View file

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

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

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

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

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

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

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

129
src/scheduler/src/scheduler.c
View file

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