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

- save tmp works

Browse source 
- save works
This commit is contained in:
Rongsong Shen 2026-02-06 17:30:50 +08:00
parent cd4f3953b2
commit 12dcc2a60d
14 changed files with 386 additions and 110 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/include/resreq.h
View file

@ -6,7 +6,9 @@
#include <stdio.h>
#include <stdlib.h>
#include "kvmgr.h"
#include "xlist.h"
#include "xvec.h"
typedef struct xce_resreq_select_ xce_select_t;
typedef struct xce_resreq_rusages_ xce_rusages_t;
@ -35,7 +37,8 @@ typedef struct {
typedef enum {
PER_TASK = 0,
PER_JOB,
PER_HOST
PER_HOST,
PER_DEFAULT
} rusage_per_type_t;
typedef struct {
@ -124,5 +127,9 @@ extern int parse_resreq_primary(xce_primary_resreq_t*, char*, char**);
extern int parse_resreq(xce_resreq_t*, char*, char**);
extern void kill_resreq(xce_resreq_t*);
extern int format_resreq(char**, xce_resreq_t*);
extern int calc_wanted_global_resvals(xvec_idval_t*, resmgr_t*, xlist_t*,
rusage_per_type_t, int, int);
extern int calc_wanted_host_resvals(xvec_idval_t*, resmgr_t*, xlist_t*,
rusage_per_type_t, int, int);
#endif

107
src/lib/resreq.c
View file

@ -3,12 +3,15 @@
#include <stdlib.h>
#include <string.h>
#include "kvmgr.h"
#include "objmgr.h"
#include "resreq.h"
#include "rusage.h"
#include "section.h"
#include "utils.h"
#include "xce_resreq.h"
#include "xlist.h"
#include "xvec.h"
static int get_parse_errmsg(char** errmsg, const char* data, int pos)
{
@ -431,3 +434,107 @@ int format_resreq(char** buf, xce_resreq_t* resreq)
return rc;
}
int calc_wanted_global_resvals(xvec_idval_t* resvals, resmgr_t* resmgr,
xlist_t* rusages, rusage_per_type_t perflag,
int nhosts, int nslots)
{
int i;
for (i = 0; i < rusages->count; i++) {
rusage_value_t* v = (rusage_value_t*)xlist_get(rusages, i);
xce_resource_t* r = NULL;
float val;
rusage_per_type_t perflag1;
if (v == NULL) {
continue;
}
r = resmgr_lockres(resmgr, v->res, OBJREF_NOP);
if (r == NULL) {
continue;
}
if ((r->flags & XCE_RESF_GLOBAL) == 0) {
continue;
}
perflag1 = v->per_flag;
if (perflag1 == PER_DEFAULT) {
perflag1 = perflag;
}
switch (perflag1) {
case PER_TASK:
val = v->value * nslots;
break;
case PER_JOB:
val = v->value;
break;
case PER_HOST:
val = v->value * nhosts;
break;
default:
/* something wrong */
val = 0.0;
break;
}
xvec_idval_store(resvals, r->idx, val);
}
return 0;
}
int calc_wanted_host_resvals(xvec_idval_t* resvals, resmgr_t* resmgr,
xlist_t* rusages, rusage_per_type_t perfflag,
int nhosts, int nslots)
{
int i;
for (i = 0; i < rusages->count; i++) {
rusage_value_t* v = (rusage_value_t*)xlist_get(rusages, i);
xce_resource_t* r = NULL;
float val;
rusage_per_type_t perflag1;
if (v == NULL) {
continue;
}
r = resmgr_lockres(resmgr, v->res, OBJREF_NOP);
if (r == NULL) {
continue;
}
if ((r->flags & XCE_RESF_GLOBAL) != 0) {
continue;
}
perflag1 = v->per_flag;
if (perflag1 == PER_DEFAULT) {
perflag1 = perflag;
}
if (perflag1 == PER_JOB) {
/* FIXME: PER_JOB should be global? */
continue;
}
switch (perflag1) {
case PER_TASK:
val = v->value * nslots;
break;
case PER_HOST:
val = v->value;
break;
default:
/* something wrong */
val = 0.0;
break;
}
xvec_idval_store(resvals, r->idx, val);
}
return 0;
}

2
src/lib/rusage.peg
View file

@ -68,7 +68,7 @@ empty_decay <- { } { auxil->decay_flag = 0; }
identifier <- [a-zA-Z_][a-zA-Z_0-9]* { auxil->resname = strdup($0); }
resvalue <- number '/' per_type { auxil->resvalue = auxil->tmp_value; }
/ number { auxil->resvalue = auxil->tmp_value; auxil->per_flag = PER_TASK; }
/ number { auxil->resvalue = auxil->tmp_value; auxil->per_flag = PER_DEFAULT; }
per_type <- 'task' { auxil->per_flag = PER_TASK; }
/ 'job' { auxil->per_flag = PER_JOB; }

16
src/lib/testsuite/test_resreq.c
View file

@ -91,14 +91,14 @@ void test_parse_rusage1(void)
TEST_ASSERT_TRUE(strcmp(v->res, "mem") == 0);
TEST_ASSERT_TRUE(fabs(v->value - 20.0) < 0.001);
TEST_ASSERT_TRUE(fabs(v->duration - 18000) < 0.001);
TEST_ASSERT_TRUE(v->per_flag == 0);
TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT);
TEST_ASSERT_TRUE(v->decay_flag == 1);
break;
case 1:
TEST_ASSERT_TRUE(strcmp(v->res, "lic") == 0);
TEST_ASSERT_TRUE(fabs(v->value - 55.0) < 0.001);
TEST_ASSERT_TRUE(fabs(v->duration - 1200) < 0.001);
TEST_ASSERT_TRUE(v->per_flag == 0);
TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT);
TEST_ASSERT_TRUE(v->decay_flag == 0);
break;
default:
@ -140,14 +140,14 @@ void test_parse_rusage2(void)
TEST_ASSERT_TRUE(strcmp(v->res, "mem") == 0);
TEST_ASSERT_TRUE(fabs(v->value - 20.0) < 0.001);
TEST_ASSERT_TRUE(fabs(v->duration - 18000) < 0.001);
TEST_ASSERT_TRUE(v->per_flag == 2);
TEST_ASSERT_TRUE(v->per_flag == PER_HOST);
TEST_ASSERT_TRUE(v->decay_flag == 1);
break;
case 1:
TEST_ASSERT_TRUE(strcmp(v->res, "lic") == 0);
TEST_ASSERT_TRUE(fabs(v->value - 55.0) < 0.001);
TEST_ASSERT_TRUE(fabs(v->duration - 1200) < 0.001);
TEST_ASSERT_TRUE(v->per_flag == 1);
TEST_ASSERT_TRUE(v->per_flag == PER_JOB);
TEST_ASSERT_TRUE(v->decay_flag == 0);
break;
default:
@ -188,14 +188,14 @@ void test_resreq_primary(void)
TEST_ASSERT_TRUE(strcmp(v->res, "mem") == 0);
TEST_ASSERT_TRUE(fabs(v->value - 20.0) < 0.001);
TEST_ASSERT_TRUE(fabs(v->duration - 18000) < 0.001);
TEST_ASSERT_TRUE(v->per_flag == 0);
TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT);
TEST_ASSERT_TRUE(v->decay_flag == 1);
break;
case 1:
TEST_ASSERT_TRUE(strcmp(v->res, "lic") == 0);
TEST_ASSERT_TRUE(fabs(v->value - 55.0) < 0.001);
TEST_ASSERT_TRUE(fabs(v->duration - 1200) < 0.001);
TEST_ASSERT_TRUE(v->per_flag == 0);
TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT);
TEST_ASSERT_TRUE(v->decay_flag == 0);
break;
default:
@ -241,14 +241,14 @@ void test_resreq1(void)
TEST_ASSERT_TRUE(strcmp(v->res, "mem") == 0);
TEST_ASSERT_TRUE(fabs(v->value - 20.0) < 0.001);
TEST_ASSERT_TRUE(fabs(v->duration - 18000) < 0.001);
TEST_ASSERT_TRUE(v->per_flag == 0);
TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT);
TEST_ASSERT_TRUE(v->decay_flag == 1);
break;
case 1:
TEST_ASSERT_TRUE(strcmp(v->res, "lic") == 0);
TEST_ASSERT_TRUE(fabs(v->value - 55.0) < 0.001);
TEST_ASSERT_TRUE(fabs(v->duration - 1200) < 0.001);
TEST_ASSERT_TRUE(v->per_flag == 0);
TEST_ASSERT_TRUE(v->per_flag == PER_DEFAULT);
TEST_ASSERT_TRUE(v->decay_flag == 0);
break;
default:

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

@ -77,6 +77,6 @@ extern int plugin_prepare_allocation(void*, xce_sched_ctx_t*);
extern int plugin_job_election(void*, xce_sched_ctx_t*);
extern int register_fcfs(void*);
extern int register_limit(void*);
extern int register_simple_allocator(void*);
extern int register_seqjob_allocator(void*);
#endif

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

@ -7,6 +7,7 @@
#include <stdlib.h>
#include "kvmgr.h"
#include "resreq.h"
#include "typemap.h"
#include "xce/types.h"
#include "xlist.h"
@ -14,23 +15,12 @@
#include "zarray.h"
/* FIXME: improve later */
typedef void* host_selection_t;
typedef struct {
xvec_idval_t shared;
xvec_idval_t host;
} xce_sched_alloc_req_t;
typedef enum {
RESREQ_OP_SIMPLE
= 0, /* simple just a list of resources and require values */
RESREQ_OP_AFFINITY, /* with affinity allocation */
RESREQ_OP_GPU, /* with GPU allocation */
RESREQ_OP_PARALL, /* parall job nhosts, ptile and other constraints */
RESREQ_OP_AND, /* success for all */
RESREQ_OP_OR /* success for any */
} resreq_op_t;
/* simple resource allocation just have numeric resource values */
typedef struct {
char* res;
@ -38,17 +28,8 @@ typedef struct {
} request_res_t;
typedef struct {
host_selection_t selection;
xlist_t res_shared;
xlist_t res_host;
} resreq_simple_t;
typedef struct {
resreq_op_t op;
union {
resreq_simple_t simple;
} u;
} resreq_t;
xce_resreq_t* resreq;
} xce_sched_resreq_t;
typedef struct {
char* name;
@ -78,7 +59,7 @@ typedef struct {
xce_sched_queue_t* queue;
xce_sched_user_t* user;
xce_sched_project_t* project;
resreq_t* resreq;
xce_sched_resreq_t* resreq;
uint32_t mintasks;
uint32_t maxtasks;
xce_job_alloc_t* allocs;
@ -96,7 +77,7 @@ typedef struct {
typedef struct {
xce_sched_job_t* job;
xce_sched_worker_t* worker;
resreq_t* resreq;
xce_sched_resreq_t* resreq;
xce_job_alloc_t* alloc;
xce_pendreason_t* reason;
struct {

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

@ -59,10 +59,10 @@ typedef struct {
uint64_t nextid;
xce_conf4sched_t* config;
kvmgr_t plugins;
void* simple_allocator;
void* seqjob_allocator;
void* parjob_allocator;
void* affinity_allocator;
void* gpu_allocator;
void* parallel_allocator;
dataq_t* pend_updates[2];
resmgr_t resmgr;
groupmgr_t ugmgr;
@ -107,7 +107,7 @@ extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*);
extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);
extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*,
host_selection_t*);
xce_select_t*);
extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_t*, xvec_idval_t*,
xvec_idval_t*);
extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*,

3
src/scheduler/meson.build
View file

@ -7,7 +7,8 @@ scheduler_srcs = [
'src/plugin.c',
'src/resource.c',
'src/plugins/fcfs.c',
'src/plugins/simple_allocator.c',
'src/plugins/seqjob_allocator.c',
'src/plugins/parjob_allocator.c',
'src/plugins/limit.c',
]

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

@ -91,9 +91,10 @@ static int update_limit_res_value(zarray_t* array, xce_vres_t* vres)
static float get_vres_value(xce_sched_ctx_t* ctx, xce_vres_t* vres)
{
int i;
resreq_t* resreq = ctx->resreq;
int i;
xce_sched_resreq_t* resreq = ctx->resreq;
#if 0
if (resreq->op != RESREQ_OP_SIMPLE) {
return 0.0;
}
@ -114,6 +115,7 @@ static float get_vres_value(xce_sched_ctx_t* ctx, xce_vres_t* vres)
return rq->value;
}
}
#endif
return 0.0;
}

78
src/scheduler/src/plugins/parjob_allocator.c Normal file
View file

@ -0,0 +1,78 @@
#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "decision.h"
#include "kvmgr.h"
#include "objmgr.h"
#include "plugin.h"
#include "preason.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xlist.h"
#include "xvec.h"
static int parjob_allocator_init(xce_sched_plugin_t* plugin)
{
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
sched->parjob_allocator = plugin;
return 1;
}
static void parjob_allocator_fini(xce_sched_plugin_t* plugin)
{
if (plugin == NULL) {
return;
}
FREEUP(plugin);
}
static int skip_bad_worker_(xce_sched_worker_t* w)
{
if (w == NULL) {
return 1;
}
/* add more check here */
return 0;
}
static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
{
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
xce_sched_job_t* job = ctx->job;
xce_sched_resreq_t* resreq = ctx->resreq;
return 0;
}
int register_parjob_allocator(void* sched)
{
static xce_sched_plugin_t parjob_allocator_plugin = {
.name = "xce_sched_parjob_allocator",
.hooks = { .config_hook = NULL, .commit_hook = NULL },
.init_fn = (plugin_init_fn_t)parjob_allocator_init,
.fini_fn = (plugin_fini_fn_t)parjob_allocator_fini,
.begin_session_fn = NULL,
.end_session_fn = NULL,
.job_elect_fn = NULL,
.job_alloc_fn = (plugin_job_allocate_fn_t)my_alloc,
.data = NULL
};
xce_sched_plugin_t* plugin = NULL;
plugin = (xce_sched_plugin_t*)calloc(1, sizeof(*plugin));
if (plugin == NULL) {
return -ENOMEM;
}
memcpy(plugin, &parjob_allocator_plugin, sizeof(*plugin));
return plugin_register(sched, plugin);
}

34
src/scheduler/src/plugins/simple_allocator.c → src/scheduler/src/plugins/seqjob_allocator.c
View file

@ -16,15 +16,15 @@
#include "xlist.h"
#include "xvec.h"
static int simple_allocator_init(xce_sched_plugin_t* plugin)
static int seqjob_allocator_init(xce_sched_plugin_t* plugin)
{
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
sched->simple_allocator = plugin;
sched->seqjob_allocator = plugin;
return 1;
}
static void simple_allocator_fini(xce_sched_plugin_t* plugin)
static void seqjob_allocator_fini(xce_sched_plugin_t* plugin)
{
if (plugin == NULL) {
return;
@ -44,13 +44,13 @@ static int skip_bad_worker_(xce_sched_worker_t* w)
static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
{
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
xce_sched_job_t* job = ctx->job;
resreq_t* resreq = ctx->resreq;
int rc;
xlist_t workers;
xce_pendreason_t* reason = NULL;
xvec_t* uncomit = NULL;
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
xce_sched_job_t* job = ctx->job;
xce_sched_resreq_t* resreq = ctx->resreq;
int rc;
xlist_t workers;
xce_pendreason_t* reason = NULL;
xvec_t* uncomit = NULL;
reason = make_pendreason(job->jobid);
ctx->reason = reason;
@ -87,7 +87,7 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
if (rc < 0) {
return rc;
}
xlist_init(&workers, 128);
rc = sched_collect_workers(sched, &workers,
@ -117,13 +117,13 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
return 0;
}
int register_simple_allocator(void* sched)
int register_seqjob_allocator(void* sched)
{
static xce_sched_plugin_t simple_allocator_plugin = {
.name = "xce_sched_simple_allocator",
static xce_sched_plugin_t seqjob_allocator_plugin = {
.name = "xce_sched_seqjob_allocator",
.hooks = { .config_hook = NULL, .commit_hook = NULL },
.init_fn = (plugin_init_fn_t)simple_allocator_init,
.fini_fn = (plugin_fini_fn_t)simple_allocator_fini,
.init_fn = (plugin_init_fn_t)seqjob_allocator_init,
.fini_fn = (plugin_fini_fn_t)seqjob_allocator_fini,
.begin_session_fn = NULL,
.end_session_fn = NULL,
.job_elect_fn = NULL,
@ -138,7 +138,7 @@ int register_simple_allocator(void* sched)
return -ENOMEM;
}
memcpy(plugin, &simple_allocator_plugin, sizeof(*plugin));
memcpy(plugin, &seqjob_allocator_plugin, sizeof(*plugin));
return plugin_register(sched, plugin);
}

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

@ -124,7 +124,7 @@ static int unsatisfy_(int nres, int* idxs, float* total, float* inuse,
}
int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w,
host_selection_t* s)
xce_select_t* s)
{
/* FIXME: add real work later */
return 1;
@ -259,9 +259,9 @@ int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w,
xvec_t* uncomitted, xvec_idval_t* wanted,
xvec_idval_t* unmeet)
{
xvec_t total;
xvec_t inuse;
int rc;
xvec_t total;
xvec_t inuse;
int rc;
if (wanted->count == 0) {
return 1;
@ -343,16 +343,18 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array,
static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
{
xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*);
xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*);
xce_sched_job_t* job = ctx->job;
resreq_t* resreq = ctx->resreq;
int rc;
xce_host_alloc_t** hallocs = NULL;
xce_host_alloc_t* halloc = NULL;
xce_alloc_t** allocs = NULL;
xvec_t* uncomit = NULL;
xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*);
xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*);
xce_sched_job_t* job = ctx->job;
xce_sched_resreq_t* resreq = ctx->resreq;
int rc;
xce_host_alloc_t** hallocs = NULL;
xce_host_alloc_t* halloc = NULL;
xce_alloc_t** allocs = NULL;
xvec_t* uncomit = NULL;
/* FIXME: */
#if 0
rc = sched_meet_host_selection(sched, w, resreq->u.simple.selection);
if (rc == 0) {
add_job_preason(ctx->reason->reasons, PREASON_MATCH_FAIL, w->name,
@ -360,6 +362,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
return 0;
}
#endif
rc = sched_ctx_get_uncomit_host(&uncomit, ctx, w->idx);
if (rc < 0) {
@ -401,7 +404,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
if (rc < 0) {
return rc;
}
allocs = sched_make_allocs(
sched, "", ctx->alloc_req.host.count, ctx->alloc_req.host.idxs.data,
ctx->alloc_req.host.vals.data, /* FIXME: no duration now */ NULL);
@ -514,7 +517,8 @@ static int fill_ctx_idval_vec_(xce_scheduler_t* sched, xvec_idval_t* v,
int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
{
int rc = 0;
int rc = 0;
#if 0
xlist_t* lists[]
= { &(ctx->resreq->u.simple.res_shared), &(ctx->extra_res_requests) };
@ -522,13 +526,15 @@ int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
if (rc < 0) {
xvec_idval_fini(&(ctx->alloc_req.shared));
}
#endif
return rc;
}
int sched_ctx_fill_host_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
{
int rc;
int rc = 0;
#if 0
xlist_t* lists[] = { &(ctx->resreq->u.simple.res_host),
&(ctx->extra_host_res_requests) };
@ -536,6 +542,7 @@ int sched_ctx_fill_host_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
if (rc < 0) {
xvec_idval_fini(&(ctx->alloc_req.host));
}
#endif
return rc;
}

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

@ -1,7 +1,7 @@
#include <errno.h>
#include <rtti.h>
#include <stdlib.h>
#include <string.h>
#include <rtti.h>
#include "controller.capnp.h"
#include "kvmgr.h"
@ -36,39 +36,37 @@ request_res_t* make_resource_request(char* res, float value)
return req;
}
static resreq_t* make_sched_resreq(xce_scheduler_t* sched, xce_job_t* j)
static int make_sched_resreq(xce_sched_resreq_t* resreq, xce_scheduler_t* sched,
xce_job_t* j)
{
resreq_t* resreq = NULL;
request_res_t* req = NULL;
int rc;
char* errmsg;
resreq = (resreq_t*)calloc(1, sizeof(*resreq));
if (resreq == NULL) {
return NULL;
memset(resreq, 0x0, sizeof(*resreq));
resreq->resreq = (xce_resreq_t*)calloc(1, sizeof(*(resreq->resreq)));
if (resreq->resreq == NULL) {
return -EINVAL;
}
resreq->op = RESREQ_OP_SIMPLE;
resreq->u.simple.selection = NULL;
rc = parse_resreq(resreq->resreq, j->submit->resreq, &errmsg);
if (rc < 0) {
ERROR("%s parse_resreq with error code %d", __func__, rc);
if (errmsg != NULL) {
ERROR("%s", errmsg);
free(errmsg);
}
xlist_init(&(resreq->u.simple.res_shared), 4);
xlist_init(&(resreq->u.simple.res_host), 4);
req = make_resource_request("slots", 1.0);
if (req != NULL) {
xlist_store(&(resreq->u.simple.res_host), (void*)req);
return rc;
}
req = make_resource_request("mem", 1024.0);
if (req != NULL) {
xlist_store(&(resreq->u.simple.res_host), (void*)req);
}
return resreq;
return 1;
}
xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
{
xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job));
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
int rc = 0;
if ((sched == NULL) || (job == NULL)) {
return NULL;
@ -85,9 +83,18 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
job->mintasks = j->submit->mintasks;
job->maxtasks = j->submit->maxtasks;
job->priority = 0; /* FIXME: support job level priority */
job->resreq = make_sched_resreq(sched, j);
job->allocs = NULL;
job->accounts = NULL;
job->resreq = (xce_sched_resreq_t*)calloc(1, sizeof(*(job->resreq)));
if (job->resreq == NULL) {
/* FIXME: */
free(job);
return NULL;
}
make_sched_resreq(job->resreq, sched, j);
job->allocs = NULL;
job->accounts = NULL;
if (j->allocs != NULL) {
MOVE_PTR(job->allocs, j->allocs);
job->accounts = make_job_account(job->allocs);
@ -329,7 +336,7 @@ int sched_ctx_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx)
v = zarray_get(&(ctx->uncomitted.hosts), idx);
if (v == NULL) {
int rc = 0;
v = (xvec_t*)calloc(1, sizeof(*v));
if (v == NULL) {
return -ENOMEM;
@ -342,7 +349,7 @@ int sched_ctx_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx)
return rc;
}
zarray_put(&(ctx->uncomitted.hosts), idx, v);
}

106
src/scheduler/src/scheduler.c
View file

@ -32,12 +32,12 @@ static void* sched_job_picker(void* arg)
return job_picker->job_elect_fn(job_picker);
}
static int sched_make_simple_allocation(xce_scheduler_t* sched,
static int sched_make_seqjob_allocation(xce_scheduler_t* sched,
xce_sched_ctx_t* ctx)
{
xce_sched_plugin_t* plugin = NULL;
plugin = sched->simple_allocator;
plugin = sched->seqjob_allocator;
if (plugin->job_alloc_fn == NULL) {
ERROR("%s plugin %s has no allocation function", __func__,
plugin->name);
@ -51,10 +51,90 @@ static int sched_make_simple_allocation(xce_scheduler_t* sched,
return plugin->job_alloc_fn(plugin, ctx);
}
static int sched_make_parjob_allocation(xce_scheduler_t* sched,
xce_sched_ctx_t* ctx)
{
xce_sched_plugin_t* plugin = NULL;
plugin = sched->parjob_allocator;
if (plugin->job_alloc_fn == NULL) {
ERROR("%s plugin %s has no allocation function", __func__,
plugin->name);
return -EINVAL;
}
plugin_prepare_allocation(sched, ctx);
sched_ctx_fill_shared_requests(sched, ctx);
return plugin->job_alloc_fn(plugin, ctx);
}
static int sched_make_primary_allocation(xce_scheduler_t* sched,
xce_sched_ctx_t* ctx)
{
xce_primary_resreq_t *primary = ctx->resreq->resreq->u.primary;
if (primary->maxTasks == 1) {
return sched_make_seqjob_allocation(sched, ctx);
}
return sched_make_parjob_allocation(sched, ctx);
}
static int sched_make_compound_allocation(xce_scheduler_t* sched,
xce_sched_ctx_t* ctx)
{
xce_compound_resreq_t *resreq = ctx->resreq->resreq->u.compound;
int i;
for(i = 0; i < resreq->primaries.count; i ++) {
xce_primary_resreq_t *resreq1 = (xce_primary_resreq_t *)xlist_get(&(resreq->primaries), i);
int rc;
xce_sched_ctx_t ctx1 = {0};
/* FIXME: initialize ctx1 */
rc = sched_make_primary_allocation(sched, &ctx1);
if (rc <= 0) {
return rc;
}
}
return 1;
}
static int sched_make_alter_allocation(xce_scheduler_t* sched,
xce_sched_ctx_t* ctx)
{
xce_alter_resreq_t *resreq = ctx->resreq->resreq->u.alter;
int i;
for(i = 0; i < resreq->compounds.count; i ++) {
xce_compound_resreq_t *resreq1 = (xce_compound_resreq_t *)xlist_get(&(resreq->compounds), i);
int rc;
xce_sched_ctx_t ctx1 = {0};
if (resreq1 == NULL) {
continue;
}
/* FIXME: we need to initialize ctx1 */
rc = sched_make_compound_allocation(sched, &ctx1);
if (rc > 0) {
return rc;
}
}
return 0;
}
static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
{
int rc = 0;
xce_sched_job_t* job = ctx->job;
int rc = 0;
xce_sched_job_t* job = ctx->job;
xce_resreq_t* resreq = job->resreq->resreq;
if (job->resreq == NULL) {
xce_pendreason_t* reason = NULL;
@ -70,13 +150,19 @@ static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
return 0;
}
switch (job->resreq->op) {
case RESREQ_OP_SIMPLE:
ctx->resreq = job->resreq;
rc = sched_make_simple_allocation(sched, ctx);
ctx->resreq = job->resreq;
switch (resreq->tag) {
case XCE_RESREQ_PRIMARY:
rc = sched_make_primary_allocation(sched, ctx);
break;
case XCE_RESREQ_COMPOUND:
rc = sched_make_compound_allocation(sched, ctx);
break;
case XCE_RESREQ_ALTER:
rc = sched_make_alter_allocation(sched, ctx);
break;
default:
rc = 0;
rc = -EINVAL;
break;
}
return rc;
@ -189,7 +275,7 @@ static int sched_register_plugins(xce_scheduler_t* sched)
{
register_fcfs(sched);
register_limit(sched);
register_simple_allocator(sched);
register_seqjob_allocator(sched);
return 0;
}