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

improve plugin

Browse source
This commit is contained in:
Rongsong Shen 2026-02-26 17:06:07 +08:00
parent 598d381379
commit 638173dec9
4 changed files with 269 additions and 219 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -60,7 +60,7 @@ typedef struct {
uint64_t nextid;
xce_conf4sched_t* config;
uint64_t configver;
kvmgr_t plugins;
var plugins;
void* seqjob_allocator;
void* parjob_allocator;
void* affinity_allocator;

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

@ -1,4 +1,5 @@
#include <errno.h>
#include <rtti.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
@ -27,7 +28,7 @@ int plugin_register(void* arg1, void* arg2)
return rc;
}
rc = kvmgr_add(&(sched->plugins), (void*)plugin->name, (void*)plugin);
rc = key_add(sched->plugins, (void*)plugin->name, (void*)plugin);
if (rc >= 0) {
plugin->plugin_id = rc;
}
@ -49,342 +50,349 @@ void plugin_unregister(void* arg)
return;
}
kvmgr_del(&(sched->plugins), plugin->name);
key_del(sched->plugins, plugin->name);
if (plugin->fini_fn != NULL) {
plugin->fini_fn(plugin);
}
}
static int plugin_config_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
xce_conf4sched_t* config = (xce_conf4sched_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.config_hook != NULL) {
return plugin->hooks.config_hook(plugin, config);
}
return 0;
}
int plugin_config_update(void* arg, xce_conf4sched_t* config)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
iterator_t it;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_config_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, config);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
xlist_fini(&lst);
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->hooks.config_hook != NULL) {
plugin->hooks.config_hook(plugin, config);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int plugin_job_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
xce_sched_job_t* job = (xce_sched_job_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.job_hook != NULL) {
return plugin->hooks.job_hook(plugin, job);
}
return 0;
}
int plugin_job_update(void* arg, xce_sched_job_t* job)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
iterator_t it;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_job_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, job);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
xlist_fini(&lst);
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->hooks.job_hook != NULL) {
plugin->hooks.job_hook(plugin, job);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int plugin_queue_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
xce_sched_queue_t* queue = (xce_sched_queue_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.queue_hook != NULL) {
return plugin->hooks.queue_hook(plugin, queue);
}
return 0;
}
int plugin_queue_update(void* arg, xce_sched_queue_t* queue)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
iterator_t it;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_queue_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, queue);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
xlist_fini(&lst);
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->hooks.queue_hook != NULL) {
plugin->hooks.queue_hook(plugin, queue);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int plugin_worker_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
xce_sched_worker_t* worker = (xce_sched_worker_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.worker_hook != NULL) {
return plugin->hooks.worker_hook(plugin, worker);
}
return 0;
}
int plugin_worker_update(void* arg, xce_sched_worker_t* worker)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
iterator_t it;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_worker_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, worker);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
xlist_fini(&lst);
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->hooks.worker_hook != NULL) {
plugin->hooks.worker_hook(plugin, worker);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int plugin_begin_session_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
if (plugin == NULL) {
return 0;
}
if (plugin->begin_session_fn != NULL) {
return plugin->begin_session_fn(plugin);
}
return 0;
}
int plugin_begin_session(void* arg)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
iterator_t it;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_begin_session_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, NULL);
xlist_fini(&lst);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->begin_session_fn != NULL) {
plugin->begin_session_fn(plugin);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int plugin_end_session_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
if (plugin == NULL) {
return 0;
}
if (plugin->end_session_fn != NULL) {
return plugin->end_session_fn(plugin);
}
return 0;
}
int plugin_end_session(void* arg)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
iterator_t it;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_end_session_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, NULL);
xlist_fini(&lst);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->begin_session_fn != NULL) {
plugin->end_session_fn(plugin);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int plugin_commit_decision_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
if (plugin == NULL) {
return 0;
}
if (plugin->hooks.commit_hook != NULL) {
return plugin->hooks.commit_hook(plugin, arg1);
}
return 0;
}
int plugin_commit_decision(void* arg1, xce_decision_t* decision)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg1;
xlist_t lst;
iterator_t it;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_commit_decision_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, decision);
xlist_fini(&lst);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->hooks.commit_hook != NULL) {
plugin->hooks.commit_hook(plugin, decision);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
typedef struct {
xce_sched_job_t* job;
uint32_t state;
} job_state_change_t;
static int plugin_job_state_change_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
job_state_change_t* args = (job_state_change_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.jstate_hook != NULL) {
return plugin->hooks.jstate_hook(plugin, args->job, args->state);
}
return 0;
}
int plugin_job_state_change(void* arg, xce_sched_job_t* job, uint32_t state)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
int rc;
job_state_change_t args = { .job = job, .state = state };
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
iterator_t it;
int rc;
if (sched == NULL) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_job_state_change_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
xlist_fini(&lst);
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->hooks.jstate_hook != NULL) {
plugin->hooks.jstate_hook(plugin, job, state);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int plugin_job_election_(xce_sched_job_t* job,
xce_sched_plugin_t* plugin)
{
if ((plugin == NULL) || (job == NULL)) {
return 0;
}
if (plugin->hooks.job_election_hook != NULL) {
return plugin->hooks.job_election_hook(plugin, job);
}
return 0;
}
int plugin_job_election(void* arg, xce_sched_job_t* job)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
iterator_t it;
int rc;
if ((sched == NULL) || (job == NULL)) {
return -EINVAL;
}
xlist_init(&lst, 3);
xlist_store(&lst, plugin_job_election_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, job);
xlist_fini(&lst);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->hooks.job_election_hook != NULL) {
plugin->hooks.job_election_hook(plugin, job);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
return rc;
}
static int plugin_prepare_allocation_(xce_sched_ctx_t* ctx,
xce_sched_plugin_t* plugin)
{
if ((plugin == NULL) || (ctx == NULL)) {
return 0;
}
if (plugin->hooks.prepare_allocation_hook != NULL) {
return plugin->hooks.prepare_allocation_hook(plugin, ctx);
}
return 0;
}
int plugin_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
iterator_t it;
int rc;
xce_sched_obj_t* obj = NULL;
@ -405,10 +413,29 @@ int plugin_prepare_allocation(void* arg, xce_sched_ctx_t* ctx)
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);
rc = iter_make(&it, sched->plugins);
if (rc < 0) {
return rc;
}
for (;;) {
xce_sched_plugin_t* plugin = NULL;
rc = iter_next(&it, (void**)&plugin);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (plugin->hooks.prepare_allocation_hook != NULL) {
plugin->hooks.prepare_allocation_hook(plugin, ctx);
}
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
if (rc < 0) {
return rc;

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

@ -386,8 +386,8 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
xce_resource_t* r = resmgr_lock_byid(
&(sched->resmgr), ((int*)ctx->unmeet.idxs.data)[i], OBJREF_NOP);
add_job_preason(schedobj->reason->reasons, PREASON_NO_RESOURCE,
w->name, r->name);
add_job_preason(ctx->reason->reasons, PREASON_NO_RESOURCE, w->name,
r->name);
}
return 0;
}

29
src/scheduler/src/scheduler.c
View file

@ -23,8 +23,15 @@ static void* sched_job_picker(void* arg)
xce_sched_plugin_t* job_picker = NULL;
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
char* job_pick_plugin = "xce_sched_fcfs";
int rc = 0;
rc = key_find(sched->plugins, job_pick_plugin, (void**)&job_picker);
if (rc < 0) {
errno = -rc;
return NULL;
}
job_picker = kvmgr_lock(&(sched->plugins), job_pick_plugin, OBJREF_NOP);
if ((job_picker == NULL) || (job_picker->job_elect_fn == NULL)) {
return NULL;
}
@ -288,6 +295,22 @@ static int sched_register_plugins(xce_scheduler_t* sched)
return 0;
}
static int sched_plugin_init(xce_scheduler_t* sched)
{
kvmgr_params_t params = { .key_type = KVMGR_KEY_STRING,
.dynamic = 1,
.nslots = 16,
.delta = 8,
.datakiller = plugin_unregister };
sched->plugins = makeInstance(&xkvmgr_type, &params);
if (sched->plugins == NULL) {
return -errno;
}
return 0;
}
int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
xce_request_handler_t handler)
{
@ -325,8 +348,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
groupmgr_init(&(sched->pgmgr), 16, 16);
groupmgr_init(&(sched->hgmgr), 64, 64);
kvmgr_init(&(sched->plugins), string_hash_key, 8, 8,
(zarray_item_killer_t)plugin_unregister);
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,