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

using xvec related api in scheduler

Browse source
This commit is contained in:
Rongsong Shen 2026-01-16 18:21:37 +08:00
parent 06209930f9
commit 924bc1517b
11 changed files with 689 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -86,17 +86,22 @@ typedef struct {
typedef struct {
char* name;
int idx;
uint32_t status;
zarray_t resvals;
zarray_t resInUse;
} xce_sched_worker_t;
typedef struct {
xce_sched_job_t* job;
xce_sched_worker_t* worker;
resreq_t* resreq;
xce_job_alloc_t* alloc;
xce_pendreason_t* reason;
xce_sched_job_t* job;
xce_sched_worker_t* worker;
resreq_t* resreq;
xce_job_alloc_t* alloc;
xce_pendreason_t* reason;
struct {
xvec_t* shared;
zarray_t hosts;
} uncomitted;
xvec_idval_t unmeet;
xlist_t extra_res_requests;
xlist_t extra_host_res_requests;
@ -121,4 +126,9 @@ 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);
#endif

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

@ -97,32 +97,30 @@ extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*);
extern xce_sched_user_t* sched_get_user(xce_scheduler_t*, char*);
extern xce_sched_project_t* sched_get_project(xce_scheduler_t*, char*);
extern int sched_add_job(xce_scheduler_t*, xce_sched_job_t*);
extern int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*);
extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*);
extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*);
extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);
extern int sched_job_is_ready(xce_sched_job_t*);
extern int sched_queue_is_ready(xce_sched_queue_t*);
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*);
extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_idval_t*,
xvec_idval_t*);
extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*,
xvec_idval_t*, xvec_idval_t*);
extern int sched_collect_workers(xce_scheduler_t*, xlist_t*,
zarray_iter_skip_fn_t,
sched_group_worker_fn_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 int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*);
extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*);
extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*);
extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);
extern int sched_job_is_ready(xce_sched_job_t*);
extern int sched_queue_is_ready(xce_sched_queue_t*);
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*);
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*,
xvec_t*, xvec_idval_t*, xvec_idval_t*);
extern int sched_collect_workers(xce_scheduler_t*, xlist_t*,
zarray_iter_skip_fn_t,
sched_group_worker_fn_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_ctx_init(xce_sched_ctx_t*, xce_sched_job_t*);
extern void sched_ctx_fini(xce_sched_ctx_t*);
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*);

17
src/scheduler/src/plugins/simple_allocator.c
View file

@ -14,6 +14,7 @@
#include "xce/types.h"
#include "xcealloc.h"
#include "xlist.h"
#include "xvec.h"
static int simple_allocator_init(xce_sched_plugin_t* plugin)
{
@ -48,7 +49,8 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
resreq_t* resreq = ctx->resreq;
int rc;
xlist_t workers;
xce_pendreason_t* reason = NULL;
xce_pendreason_t* reason = NULL;
xvec_t* uncomit = NULL;
reason = make_pendreason(job->jobid);
ctx->reason = reason;
@ -56,7 +58,12 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
return -ENOMEM;
}
rc = sched_check_shared_satisfy(sched, &(ctx->alloc_req.shared),
rc = sched_ctx_get_uncomit_shared(&uncomit, ctx);
if (rc < 0) {
return rc;
}
rc = sched_check_shared_satisfy(sched, uncomit, &(ctx->alloc_req.shared),
&(ctx->unmeet));
if (rc < 0) {
return rc;
@ -75,6 +82,12 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
return 0;
}
/* add uncomitted shared */
rc = xvec_idval(uncomit, &(ctx->alloc_req.shared), XVEC_ADD_FLOAT);
if (rc < 0) {
return rc;
}
xlist_init(&workers, 128);
rc = sched_collect_workers(sched, &workers,

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

@ -94,14 +94,15 @@ int put_resvals(zarray_t* array, int nres, int* idxs, float* vals)
}
static int unsatisfy_(int nres, int* idxs, float* total, float* inuse,
float* wanted, int* unmeet_idxs, float* unmeet_vals)
float* uncomitted, float* wanted, int* unmeet_idxs,
float* unmeet_vals)
{
int i, j;
int count;
count = 0;
for (i = 0; i < nres; i++) {
if (total[i] < inuse[i] + wanted[i]) {
if (total[i] < inuse[i] + uncomitted[i] + wanted[i]) {
count++;
}
}
@ -112,9 +113,9 @@ static int unsatisfy_(int nres, int* idxs, float* total, float* inuse,
j = 0;
for (i = 0; i < nres; i++) {
if (total[i] < inuse[i] + wanted[i]) {
if (total[i] < inuse[i] + uncomitted[i] + wanted[i]) {
unmeet_idxs[j] = idxs[i];
unmeet_vals[j] = inuse[i] + wanted[i] - total[i];
unmeet_vals[j] = inuse[i] + wanted[i] + uncomitted[i] - total[i];
j++;
}
}
@ -130,7 +131,8 @@ int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w,
}
static int sched_check_shared_satisfy_(xce_scheduler_t* sched, xvec_t* total,
xvec_t* inuse, xvec_idval_t* wanted,
xvec_t* inuse, xvec_t* uncomitted,
xvec_idval_t* wanted,
xvec_idval_t* unmeet)
{
int nres = wanted->count;
@ -164,8 +166,9 @@ static int sched_check_shared_satisfy_(xce_scheduler_t* sched, xvec_t* total,
}
rc = unsatisfy_(nres, (int*)wanted->idxs.data, (float*)total->data,
(float*)inuse->data, (float*)wanted->vals.data,
(int*)unmeet->idxs.data, (float*)unmeet->vals.data);
(float*)inuse->data, (float*)uncomitted->data,
(float*)wanted->vals.data, (int*)unmeet->idxs.data,
(float*)unmeet->vals.data);
if (rc < 0) {
return rc;
}
@ -179,8 +182,8 @@ static int sched_check_shared_satisfy_(xce_scheduler_t* sched, xvec_t* total,
return 1;
}
int sched_check_shared_satisfy(xce_scheduler_t* sched, xvec_idval_t* wanted,
xvec_idval_t* unmeet)
int sched_check_shared_satisfy(xce_scheduler_t* sched, xvec_t* uncomitted,
xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
xvec_t total;
xvec_t inuse;
@ -190,7 +193,8 @@ int sched_check_shared_satisfy(xce_scheduler_t* sched, xvec_idval_t* wanted,
return 1;
}
rc = sched_check_shared_satisfy_(sched, &total, &inuse, wanted, unmeet);
rc = sched_check_shared_satisfy_(sched, &total, &inuse, uncomitted, wanted,
unmeet);
if (rc < 0) {
xvec_idval_fini(unmeet);
}
@ -203,8 +207,8 @@ int sched_check_shared_satisfy(xce_scheduler_t* sched, xvec_idval_t* wanted,
static int sched_check_host_satisfy_(xce_scheduler_t* sched,
xce_sched_worker_t* w, xvec_t* total,
xvec_t* inuse, xvec_idval_t* wanted,
xvec_idval_t* unmeet)
xvec_t* inuse, xvec_t* uncomitted,
xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
int nres = wanted->count;
int rc = 0;
@ -237,8 +241,9 @@ static int sched_check_host_satisfy_(xce_scheduler_t* sched,
}
rc = unsatisfy_(nres, (int*)wanted->idxs.data, (float*)total->data,
(float*)inuse->data, (float*)wanted->vals.data,
(int*)unmeet->idxs.data, (float*)unmeet->vals.data);
(float*)inuse->data, (float*)uncomitted->data,
(float*)wanted->vals.data, (int*)unmeet->idxs.data,
(float*)unmeet->vals.data);
if (rc < 0) {
return rc;
}
@ -251,17 +256,19 @@ static int sched_check_host_satisfy_(xce_scheduler_t* sched,
return 1;
}
int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w,
xvec_idval_t* wanted, xvec_idval_t* unmeet)
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;
}
rc = sched_check_host_satisfy_(sched, w, &total, &inuse, wanted, unmeet);
rc = sched_check_host_satisfy_(sched, w, &total, &inuse, uncomitted, wanted,
unmeet);
if (rc < 0) {
xvec_idval_fini(unmeet);
}
@ -344,6 +351,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
xce_host_alloc_t** hallocs = NULL;
xce_host_alloc_t* halloc = NULL;
xce_alloc_t** allocs = NULL;
xvec_t* uncomit = NULL;
rc = sched_meet_host_selection(sched, w, resreq->u.simple.selection);
if (rc == 0) {
@ -353,6 +361,11 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
return 0;
}
rc = sched_ctx_get_uncomit_host(&uncomit, ctx, w->idx);
if (rc < 0) {
return rc;
}
ctx->worker = w;
rc = plugin_prepare_allocation(sched, ctx);
@ -364,7 +377,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
sched_ctx_fill_host_requests(sched, ctx);
rc = sched_check_host_satisfy(sched, w, &(ctx->alloc_req.host),
rc = sched_check_host_satisfy(sched, w, uncomit, &(ctx->alloc_req.host),
&(ctx->unmeet));
if (rc < 0) {
/* FIXME: */
@ -384,6 +397,11 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
return 0;
}
rc = xvec_idval(uncomit, &(ctx->alloc_req.host), XVEC_ADD_FLOAT);
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);

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

@ -7,10 +7,10 @@
#include "objmgr.h"
#include "sched/types.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xlist.h"
#include "utils.h"
#include "xvec.h"
void kill_sched_job(void* arg)
@ -20,15 +20,16 @@ void kill_sched_job(void* arg)
}
}
request_res_t *make_resource_request(char *res, float value) {
request_res_t *req = NULL;
request_res_t* make_resource_request(char* res, float value)
{
request_res_t* req = NULL;
req = (request_res_t *)calloc(1, sizeof(*req));
req = (request_res_t*)calloc(1, sizeof(*req));
if (req == NULL) {
return NULL;
}
req->res = strdup(res);
req->res = strdup(res);
req->value = value;
return req;
@ -57,7 +58,7 @@ static resreq_t* make_sched_resreq(xce_scheduler_t* sched, xce_job_t* j)
req = make_resource_request("mem", 1024.0);
if (req != NULL) {
xlist_store(&(resreq->u.simple.res_host), (void *)req);
xlist_store(&(resreq->u.simple.res_host), (void*)req);
}
return resreq;
@ -158,6 +159,7 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
worker->name = strdup(w->name);
worker->status = w->status;
worker->idx = -1;
zarray_init(&(worker->resvals), 128,
(zarray_item_killer_t)kill_sched_resval);
@ -178,8 +180,8 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
void kill_sched_user(void* arg)
{
xce_sched_user_t *user = (xce_sched_user_t *)arg;
xce_sched_user_t* user = (xce_sched_user_t*)arg;
if (arg == NULL) {
return;
}
@ -193,22 +195,22 @@ xce_sched_user_t* make_sched_user(void* arg, char* name)
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xce_sched_user_t* user = NULL;
user = (xce_sched_user_t *)calloc(1, sizeof(*user));
user = (xce_sched_user_t*)calloc(1, sizeof(*user));
if (user == NULL) {
return NULL;
}
user->name = strdup(name);
sched_add_user(sched, user);
return user;
}
void kill_sched_project(void* arg)
{
xce_sched_project_t *project = (xce_sched_project_t *)arg;
xce_sched_project_t* project = (xce_sched_project_t*)arg;
if (arg == NULL) {
return;
}
@ -222,25 +224,38 @@ xce_sched_project_t* make_sched_project(void* arg, char* name)
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xce_sched_project_t* project = NULL;
project = (xce_sched_project_t *)calloc(1, sizeof(*project));
project = (xce_sched_project_t*)calloc(1, sizeof(*project));
if (project == NULL) {
return NULL;
}
if ((name == NULL) || (strlen(name) == 0)) {
project->name = strdup("default");
}
else{
} else {
project->name = strdup(name);
}
sched_add_project(sched, project);
return project;
}
static void host_uncommit_killer(void* obj)
{
xvec_t* v = (xvec_t*)obj;
if (obj == NULL) {
return;
}
xvec_fini(v);
FREEUP(v);
}
int sched_ctx_init(xce_sched_ctx_t* ctx, xce_sched_job_t* job)
{
int rc = 0;
if ((ctx == NULL) || (job == NULL)) {
return -EINVAL;
}
@ -253,6 +268,18 @@ int sched_ctx_init(xce_sched_ctx_t* ctx, xce_sched_job_t* job)
xlist_init(&(ctx->extra_res_requests), 4);
xlist_init(&(ctx->extra_host_res_requests), 4);
ctx->uncomitted.shared = (xvec_t*)calloc(1, sizeof(xvec_t));
if (ctx->uncomitted.shared == NULL) {
return -ENOMEM;
}
rc = xvec_init(ctx->uncomitted.shared, 1024, sizeof(float));
if (rc < 0) {
return rc;
}
zarray_init(&(ctx->uncomitted.hosts), 32,
(zarray_item_killer_t)host_uncommit_killer);
return 0;
}
@ -262,11 +289,62 @@ void sched_ctx_fini(xce_sched_ctx_t* ctx)
return;
}
if (ctx->uncomitted.shared != NULL) {
xvec_fini(ctx->uncomitted.shared);
FREEUP(ctx->uncomitted.shared);
}
zarray_fini(&(ctx->uncomitted.hosts));
xvec_idval_fini(&(ctx->alloc_req.shared));
xvec_idval_fini(&(ctx->alloc_req.host));
xlist_fini(&(ctx->extra_res_requests));
xlist_fini(&(ctx->extra_host_res_requests));
xlist_fini(&(ctx->limits));
xlist_fini(&(ctx->limits_host));
}
int sched_ctx_get_uncomit_shared(xvec_t** shared, xce_sched_ctx_t* ctx)
{
if ((shared == NULL) || (ctx == NULL)) {
return -EINVAL;
}
*shared = ctx->uncomitted.shared;
return 1;
}
int sched_ctx_get_uncomit_host(xvec_t** host, xce_sched_ctx_t* ctx, int idx)
{
xvec_t* v;
if ((host == NULL) || (ctx == NULL) || (idx < 0)) {
return -EINVAL;
}
*host = NULL;
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;
}
rc = xvec_init(v, 1024, sizeof(float));
if (rc < 0) {
xvec_fini(v);
free(v);
return rc;
}
zarray_put(&(ctx->uncomitted.hosts), idx, v);
}
*host = v;
return 1;
}

9
src/scheduler/src/scheduler.c
View file

@ -597,7 +597,14 @@ xce_sched_queue_t* sched_get_queue(xce_scheduler_t* sched, char* name)
int sched_add_worker(xce_scheduler_t* sched, xce_sched_worker_t* w)
{
return kvmgr_add(&(sched->workers), w->name, w);
int rc;
rc = kvmgr_add(&(sched->workers), w->name, w);
if (rc >= 0) {
w->idx = rc;
}
return rc;
}
xce_sched_worker_t* sched_get_worker(xce_scheduler_t* sched, char* name)