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

improve scheduler

Browse source
This commit is contained in:
Rongsong Shen 2026-02-27 15:15:16 +08:00
parent 638173dec9
commit b1faeb94b6
4 changed files with 303 additions and 156 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/lib/kvmgr.c
View file

@ -459,6 +459,7 @@ static int xkvmgr_init(meta_obj_t* obj, va_list ap)
kvm->nslots = params->nslots;
kvm->key_type = params->key_type;
kvm->dynamic = params->dynamic;
if (params->dynamic) {
xkvmgr_private_t* priv = RTTI_OBJ_TO_PRIV(obj);
@ -543,6 +544,7 @@ static int xkvmgr_find(meta_obj_t* obj, void* key, void** value)
rc = vget(kvm->data, idx, &data);
if (rc >= 0) {
*value = data;
rc = 1;
}
}

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

@ -73,11 +73,11 @@ typedef struct {
groupmgr_t hgmgr;
zarray_t sharedInst;
zarray_t sharedInUse;
kvmgr_t jobs;
kvmgr_t queues;
kvmgr_t workers;
kvmgr_t users;
kvmgr_t projects;
var jobs;
var queues;
var workers;
var users;
var projects;
pthread_t jobpicker;
pthread_t allocator;
cbuf_t wbuf; /* action to trigger scheduling */

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

@ -1,4 +1,5 @@
#include <errno.h>
#include <rtti.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
@ -13,16 +14,37 @@
#include "xlist.h"
typedef struct {
kvmgr_t limits;
kvmgr_t instances;
var limits;
var instances;
} limit_plugin_data_t;
static int limit_priv_data_init(limit_plugin_data_t* data)
{
kvmgr_init(&(data->limits), string_hash_key, 64, 64,
(zarray_item_killer_t)kill_limit);
kvmgr_init(&(data->instances), string_hash_key, 64, 64,
(zarray_item_killer_t)kill_vres);
kvmgr_params_t limits_params = {
.key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = 1021,
.delta = 64,
.datakiller = (destroy_fn_t)kill_limit,
};
kvmgr_params_t instances_params = {
.key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = 1021,
.delta = 64,
.datakiller = (destroy_fn_t)kill_limit,
};
data->limits = makeInstance(&xkvmgr_type, &limits_params);
if (data->limits == NULL) {
return -errno;
}
data->instances = makeInstance(&xkvmgr_type, &instances_params);
if (data->instances == NULL) {
return -errno;
}
return 0;
}
@ -48,7 +70,7 @@ static int limit_update_config(xce_sched_plugin_t* plugin,
return rc;
}
rc = kvmgr_add(&(privdata->limits), limit->name, limit);
rc = key_add(privdata->limits, (void*)limit->name, (void*)limit);
if (rc < 0) {
kill_limit(limit);
@ -120,11 +142,9 @@ static float get_vres_value(xce_sched_ctx_t* ctx, xce_vres_t* vres)
return 0.0;
}
static int match_job_global_limit_(void* arg1, void* arg2)
static int match_job_global_limit_(xce_sched_plugin_t* plugin,
xce_sched_ctx_t* ctx, xce_limit_t* limit)
{
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;
@ -132,13 +152,12 @@ static int match_job_global_limit_(void* arg1, void* arg2)
limit_plugin_data_t* privdata = NULL;
xce_scheduler_t* sched = NULL;
if ((limit == NULL) || (args == NULL)) {
if ((limit == NULL) || (plugin == NULL) || (ctx == NULL)) {
return -EINVAL;
}
obj = args->ctx->schedobj;
plugin = args->plugin;
if ((plugin == NULL) || (obj == NULL)) {
obj = ctx->schedobj;
if (obj == NULL) {
return -EINVAL;
}
@ -164,7 +183,11 @@ static int match_job_global_limit_(void* arg1, void* arg2)
return rc;
}
vres = kvmgr_lock(&(privdata->instances), vresname, OBJREF_NOP);
rc = key_find(privdata->instances, (void*)vresname, (void*)&vres);
if (rc <= 0) {
return rc;
}
FREEUP(vresname);
if (vres == NULL) {
@ -174,7 +197,7 @@ static int match_job_global_limit_(void* arg1, void* arg2)
return rc;
}
rc = kvmgr_add(&(privdata->instances), vres->vres->name, vres);
rc = key_add(privdata->instances, (void*)vres->vres->name, vres);
if (rc < 0) {
return rc;
}
@ -195,13 +218,14 @@ static int limit_job_election(void* arg, xce_sched_job_t* job)
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
xce_sched_obj_t* obj = job->schedobj;
int rc = 0;
iterator_t it;
xce_sched_ctx_t ctx = { 0 };
TRACE("%s job %lu plugin %s", __func__, job->jobid, plugin->name);
if (obj->configver < sched->configver) {
xlist_t lst;
match_limit_args_t args;
xce_sched_ctx_t ctx = { 0 };
if (obj->configver >= sched->configver) {
return 0;
}
reset_sched_obj(obj);
@ -209,25 +233,39 @@ static int limit_job_election(void* arg, xce_sched_job_t* job)
ctx.worker = NULL;
ctx.schedobj = job->schedobj;
args.plugin = plugin;
args.ctx = &ctx;
rc = iter_make(&it, privdata->limits);
if (rc < 0) {
return rc;
}
xlist_init(&lst, 3);
xlist_store(&lst, match_job_global_limit_);
rc = kvmgr_iter(&(privdata->limits), NULL, &lst, &args);
xlist_fini(&lst);
for (;;) {
xce_limit_t* limit = NULL;
rc = iter_next(&it, (void**)&limit);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
rc = match_job_global_limit_(plugin, &ctx, limit);
if (rc < 0) {
break;
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
obj->configver = sched->configver;
}
return rc;
}
static int match_job_host_limit_(void* arg1, void* arg2)
static int match_job_host_limit_(xce_sched_plugin_t* plugin,
xce_sched_ctx_t* ctx, xce_limit_t* limit)
{
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;
@ -235,21 +273,18 @@ static int match_job_host_limit_(void* arg1, void* arg2)
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)) {
if ((limit == NULL) || (ctx == NULL) || (limit == NULL)) {
return -EINVAL;
}
obj = args->ctx->schedobj;
plugin = args->plugin;
if ((obj == NULL) || (plugin == NULL)) {
obj = ctx->schedobj;
if (obj == NULL) {
return -EINVAL;
}
sched = plugin->sched;
ctx = args->ctx;
privdata = (limit_plugin_data_t*)plugin->data;
worker = ctx->worker;
@ -290,7 +325,10 @@ static int match_job_host_limit_(void* arg1, void* arg2)
return rc;
}
vres = kvmgr_lock(&(privdata->instances), vresname, OBJREF_NOP);
rc = key_find(privdata->instances, (void*)vresname, (void*)&vres);
if (rc < 0) {
return rc;
}
FREEUP(vresname);
if (vres == NULL) {
@ -300,7 +338,7 @@ static int match_job_host_limit_(void* arg1, void* arg2)
return rc;
}
rc = kvmgr_add(&(privdata->instances), vres->vres->name, vres);
rc = key_add(privdata->instances, vres->vres->name, vres);
if (rc < 0) {
return rc;
}
@ -363,10 +401,9 @@ static int limit_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;
match_limit_args_t args;
xlist_t lst;
int rc = 0;
limit_plugin_data_t* privdata = NULL;
iterator_t it;
if ((plugin == NULL) || (ctx == NULL)) {
return -EINVAL;
@ -379,13 +416,30 @@ static int limit_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
privdata = (limit_plugin_data_t*)plugin->data;
args.plugin = plugin;
args.ctx = ctx;
rc = iter_make(&it, privdata->limits);
if (rc < 0) {
return rc;
}
xlist_init(&lst, 2);
xlist_store(&lst, match_job_host_limit_);
rc = kvmgr_iter(&(privdata->limits), NULL, &lst, &args);
xlist_fini(&lst);
for (;;) {
xce_limit_t* limit = NULL;
rc = iter_next(&it, (void**)&limit);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
rc = match_job_host_limit_(plugin, ctx, limit);
if (rc < 0) {
break;
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
@ -416,8 +470,8 @@ static void limit_fini(xce_sched_plugin_t* plugin)
privdata = (limit_plugin_data_t*)plugin->data;
if (privdata != NULL) {
kvmgr_fini(&(privdata->limits));
kvmgr_fini(&(privdata->instances));
destroy(privdata->limits);
destroy(privdata->instances);
FREEUP(privdata);
}

261
src/scheduler/src/scheduler.c
View file

@ -317,6 +317,35 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
int rc = 0;
int ndecisions = 5; /* FIXME: make this configurable */
int njobs = 1; /* FIXME: make this configurable */
kvmgr_params_t jobs_params = { .key_type = KVMGR_KEY_INT,
.dynamic = 1,
.nslots = 5381, /* FIXME */
.delta = 1024,
.datakiller = kill_sched_job };
kvmgr_params_t queues_params
= { .key_type = KVMGR_KEY_STRING,
.dynamic = 0,
.nslots = 59,
.delta = 16,
.datakiller = (destroy_fn_t)kill_sched_queue };
kvmgr_params_t workers_params
= { .key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = 5381,
.delta = 64,
.datakiller = (destroy_fn_t)kill_sched_worker };
kvmgr_params_t users_params
= { .key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = 5381,
.delta = 64,
.datakiller = (destroy_fn_t)kill_sched_user };
kvmgr_params_t projects_params
= { .key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = 31,
.delta = 8,
.datakiller = (destroy_fn_t)kill_sched_project };
if ((sched == NULL) || (config == NULL) || (handler == NULL)) {
return -EINVAL;
@ -350,16 +379,11 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
sched_plugin_init(sched);
kvmgr_init(&(sched->jobs), int_hash_key, 1000, 1000,
(zarray_item_killer_t)kill_sched_job);
kvmgr_init(&(sched->queues), string_hash_key, 8, 8,
(zarray_item_killer_t)kill_sched_queue);
kvmgr_init(&(sched->workers), string_hash_key, 256, 256,
(zarray_item_killer_t)kill_sched_worker);
kvmgr_init(&(sched->users), string_hash_key, 256, 256,
(zarray_item_killer_t)kill_sched_user);
kvmgr_init(&(sched->projects), string_hash_key, 8, 8,
(zarray_item_killer_t)kill_sched_project);
sched->jobs = makeInstance(&xkvmgr_type, &jobs_params);
sched->queues = makeInstance(&xkvmgr_type, &queues_params);
sched->workers = makeInstance(&xkvmgr_type, &workers_params);
sched->users = makeInstance(&xkvmgr_type, &users_params);
sched->projects = makeInstance(&xkvmgr_type, &projects_params);
zarray_init(&(sched->sharedInst), 128,
(zarray_item_killer_t)kill_sched_resval);
@ -636,43 +660,66 @@ static int sched_reset_worker_(xce_scheduler_t* sched, xce_sched_worker_t* w)
static int sched_reset_workers(xce_scheduler_t* sched)
{
xlist_t lst;
iterator_t it;
int rc = 0;
xlist_init(&lst, 2);
xlist_store(&lst, sched_reset_worker_);
rc = kvmgr_iter(&(sched->workers), NULL, &lst, sched);
xlist_fini(&lst);
rc = iter_make(&it, sched->workers);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_worker_t* w = NULL;
rc = iter_next(&it, (void**)&w);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
sched_reset_worker_(sched, w);
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int charge_job_(xce_scheduler_t* sched, xce_sched_job_t* j)
{
if (j == NULL) {
return 0;
}
if (!JOB_IS_RUN(j->state)) {
return 0;
}
if (j->accounts == NULL) {
return 0;
}
return sched_charge_account(sched, j->accounts);
}
static int sched_account_jobs(xce_scheduler_t* sched)
{
xlist_t lst;
iterator_t it;
int rc = 0;
xlist_init(&lst, 2);
xlist_store(&lst, charge_job_);
rc = kvmgr_iter(&(sched->jobs), NULL, &lst, sched);
xlist_fini(&lst);
rc = iter_make(&it, sched->jobs);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_job_t* job = NULL;
rc = iter_next(&it, (void**)&job);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (!JOB_IS_RUN(job->state)) {
continue;
}
if (job->accounts != NULL) {
sched_charge_account(sched, job->accounts);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
@ -698,29 +745,45 @@ void sched_end_session(xce_scheduler_t* sched)
int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j)
{
return kvmgr_add(&(sched->jobs), (void*)j->jobid, j);
return key_add(sched->jobs, (void*)j->jobid, j);
}
xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
{
return kvmgr_lock(&(sched->jobs), (void*)jobid, OBJREF_NOP);
int rc = 0;
xce_sched_job_t* job = NULL;
rc = key_find(sched->jobs, (void*)jobid, (void**)&job);
if (rc <= 0) {
return NULL;
}
return job;
}
int sched_add_queue(xce_scheduler_t* sched, xce_sched_queue_t* q)
{
return kvmgr_add(&(sched->queues), q->name, q);
return key_add(sched->queues, (void*)q->name, q);
}
xce_sched_queue_t* sched_get_queue(xce_scheduler_t* sched, char* name)
{
return kvmgr_lock(&(sched->queues), (void*)name, OBJREF_NOP);
xce_sched_queue_t* q = NULL;
int rc = 0;
rc = key_find(sched->queues, (void*)name, (void**)&q);
if (rc <= 0) {
return NULL;
}
return q;
}
int sched_add_worker(xce_scheduler_t* sched, xce_sched_worker_t* w)
{
int rc;
rc = kvmgr_add(&(sched->workers), w->name, w);
rc = key_add(sched->workers, (void*)w->name, (void*)w);
if (rc >= 0) {
w->idx = rc;
}
@ -730,20 +793,35 @@ int sched_add_worker(xce_scheduler_t* sched, xce_sched_worker_t* w)
xce_sched_worker_t* sched_get_worker(xce_scheduler_t* sched, char* name)
{
return kvmgr_lock(&(sched->workers), (void*)name, OBJREF_NOP);
xce_sched_worker_t* w = NULL;
int rc = 0;
rc = key_find(sched->workers, (void*)name, (void**)&w);
if (rc <= 0) {
return NULL;
}
return w;
}
int sched_add_user(xce_scheduler_t* sched, xce_sched_user_t* u)
{
return kvmgr_add(&(sched->users), u->name, u);
int rc = 0;
rc = key_add(sched->users, (void*)u->name, (void*)u);
return rc;
}
xce_sched_user_t* sched_get_user(xce_scheduler_t* sched, char* name)
{
xce_sched_user_t* user = NULL;
user = (xce_sched_user_t*)kvmgr_lock(&(sched->users), (void*)name,
OBJREF_NOP);
int rc = 0;
rc = key_find(sched->users, (void*)name, (void**)&user);
if (rc < 0) {
return NULL;
}
if (user != NULL) {
return user;
}
@ -753,16 +831,27 @@ xce_sched_user_t* sched_get_user(xce_scheduler_t* sched, char* name)
int sched_add_project(xce_scheduler_t* sched, xce_sched_project_t* p)
{
return kvmgr_add(&(sched->projects), p->name, p);
int rc;
rc = key_add(sched->projects, (void*)p->name, (void*)p);
return rc;
}
xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name)
{
xce_sched_project_t* project = NULL;
int rc = 0;
project = (xce_sched_project_t*)kvmgr_lock(&(sched->projects), (void*)name,
OBJREF_NOP);
if (project != NULL) {
if ((name == NULL) || (strlen(name) == 0)) {
name = "default";
}
rc = key_find(sched->projects, (void*)name, (void **)&project);
if (rc < 0) {
return NULL;
}
if (rc > 0) {
return project;
}
@ -834,48 +923,50 @@ static xce_worker_group_t* find_match_worker_group(xlist_t* groups, void* id)
return wg;
}
static int _add_worker(void* arg1, void* arg2)
{
add_worker_args_t* args = (add_worker_args_t*)arg1;
xce_sched_worker_t* w = (xce_sched_worker_t*)arg2;
xce_worker_group_t* wg = NULL;
void* gid = NULL;
if ((arg1 == NULL) || (w == NULL)) {
return 0;
}
if (args->groupfn == NULL) {
args->groupfn = (sched_group_worker_fn_t)default_worker_groupfn_;
}
gid = args->groupfn(args->sched, w);
wg = find_match_worker_group(args->lst, gid);
if (wg == NULL) {
wg = (xce_worker_group_t*)calloc(1, sizeof(*wg));
wg->id = gid;
xlist_init(&(wg->members), 8);
xlist_store(args->lst, wg);
}
xlist_store(&(wg->members), w);
return 1;
}
int sched_collect_workers(xce_scheduler_t* sched, xlist_t* lst,
zarray_iter_skip_fn_t skipfn,
sched_group_worker_fn_t groupfn)
{
xlist_t l2;
iterator_t it;
int rc = 0;
add_worker_args_t args = { .lst = lst, .sched = sched, .groupfn = groupfn };
xlist_init(&l2, 2);
xlist_store(&l2, (void*)_add_worker);
if (groupfn == NULL) {
groupfn = default_worker_groupfn_;
}
rc = kvmgr_iter(&(sched->workers), skipfn, &l2, &args);
xlist_fini(&l2);
rc = iter_make(&it, sched->workers);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_worker_t* w = NULL;
void* gid = NULL;
xce_worker_group_t* wg = NULL;
rc = iter_next(&it, (void**)&w);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
gid = groupfn(sched, w);
wg = find_match_worker_group(lst, gid);
if (wg == NULL) {
wg = (xce_worker_group_t*)calloc(1, sizeof(*wg));
wg->id = gid;
xlist_init(&(wg->members), 8);
xlist_store(lst, wg);
}
xlist_store(&(wg->members), w);
}
iter_destroy(&it);
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
return rc;
}