From b1faeb94b6adca87cc5709952712f57eab58578e Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 27 Feb 2026 15:15:16 +0800 Subject: [PATCH] improve scheduler --- src/lib/kvmgr.c | 2 + src/scheduler/include/scheduler.h | 10 +- src/scheduler/src/plugins/limit.c | 174 ++++++++++++------- src/scheduler/src/scheduler.c | 273 ++++++++++++++++++++---------- 4 files changed, 303 insertions(+), 156 deletions(-) diff --git a/src/lib/kvmgr.c b/src/lib/kvmgr.c index 816980b..76d2f59 100644 --- a/src/lib/kvmgr.c +++ b/src/lib/kvmgr.c @@ -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; } } diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index adfe6d0..09ea571 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -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 */ diff --git a/src/scheduler/src/plugins/limit.c b/src/scheduler/src/plugins/limit.c index 32309e8..394a7c6 100644 --- a/src/scheduler/src/plugins/limit.c +++ b/src/scheduler/src/plugins/limit.c @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -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,39 +218,54 @@ 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 }; - - 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; + if (obj->configver >= sched->configver) { + return 0; } + reset_sched_obj(obj); + + ctx.sched = plugin->sched; + ctx.worker = NULL; + ctx.schedobj = job->schedobj; + + rc = iter_make(&it, privdata->limits); + if (rc < 0) { + return rc; + } + + 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; @@ -273,7 +308,7 @@ static int match_job_host_limit_(void* arg1, void* arg2) } xlist_init(hlimits, 8); - + zarray_put(&(obj->hlimits), worker->idx, hlimits); } @@ -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,14 +416,31 @@ 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); } diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 6006fb0..b658894 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -314,9 +314,38 @@ static int sched_plugin_init(xce_scheduler_t* sched) int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, xce_request_handler_t handler) { - int rc = 0; - int ndecisions = 5; /* FIXME: make this configurable */ - int njobs = 1; /* FIXME: make this configurable */ + 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; - int rc = 0; + 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; - int rc = 0; + 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; - int rc = 0; - add_worker_args_t args = { .lst = lst, .sched = sched, .groupfn = groupfn }; + iterator_t it; + int rc = 0; - 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; }