From 4769e6845f6f5e811ce26de99b48a78eb0cbdf7d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 26 Dec 2025 17:52:43 +0800 Subject: [PATCH] improve configuration handling --- src/scheduler/include/plugin.h | 6 +-- src/scheduler/include/sched/types.h | 6 --- src/scheduler/include/scheduler.h | 48 ++++++++++++----------- src/scheduler/src/handler.c | 27 +++---------- src/scheduler/src/plugin.c | 4 +- src/scheduler/src/plugins/fcfs.c | 5 +-- src/scheduler/src/schedobj.c | 38 ------------------ src/scheduler/src/scheduler.c | 61 +++++++++++++++++------------ 8 files changed, 74 insertions(+), 121 deletions(-) diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 449b36e..b5e9c8f 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -11,8 +11,8 @@ typedef void* plugin_data_t; /* hooker of plugin, the first parameter should be xce_sched_plugin_t */ -typedef int (*plugin_update_config_hook_t)(void*, xce_sched_config_t*, - xce_sched_config_t*); +typedef int (*plugin_update_config_hook_t)(void*, xce_conf4sched_t*, + xce_conf4sched_t*); typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*, xce_sched_queue_t*); typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*, @@ -62,7 +62,7 @@ typedef struct { extern int plugin_register(void*, void*); extern void plugin_unregister(void*); -extern int plugin_config_update(void*, xce_sched_config_t*, xce_sched_config_t*); +extern int plugin_config_update(void*, xce_conf4sched_t*, xce_conf4sched_t*); extern int plugin_queue_update(void*, xce_sched_queue_t*, xce_sched_queue_t*); extern int plugin_job_update(void*, xce_sched_job_t*, xce_sched_job_t*); extern int plugin_worker_update(void*, xce_sched_worker_t*, diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 99af412..959e592 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -60,17 +60,11 @@ typedef struct { xce_val_t* res_in_use; /* used resources in worker */ } xce_sched_worker_t; -typedef struct { - char* clustername; - resmgr_t resmgr; -} xce_sched_config_t; - extern xce_sched_job_t* make_sched_job(void*, xce_job_t*); extern xce_sched_queue_t* make_sched_queue(void*, xce_queue_t*); extern xce_sched_worker_t* make_sched_worker(void*, xce_worker_t*); extern xce_sched_user_t* make_sched_user(void*, char*); extern xce_sched_project_t* make_sched_project(void*, char*); -extern xce_sched_config_t* make_sched_config(void*, xce_conf4sched_t*); extern void kill_sched_job(void*); extern void kill_sched_queue(void*); extern void kill_sched_worker(void*); diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index b8adcb8..f386594 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -49,31 +49,33 @@ typedef struct { link))) typedef struct { - mylock_t lock; - int stop; - xcecli_ctx_t apictx; - xce_sched_config_t* config; - kvmgr_t plugins; - dataq_t* pend_updates[2]; - kvmgr_t jobs; - kvmgr_t queues; - kvmgr_t workers; - kvmgr_t users; - kvmgr_t projects; - pthread_t jobpicker; - pthread_t allocator; - cbuf_t wbuf; /* action to trigger scheduling */ - cbuf_t jbuf; /* job picked for allocation */ - cbuf_t decisions; /* scheduler decisions */ + mylock_t lock; + int stop; + xcecli_ctx_t apictx; + xce_conf4sched_t* config; + kvmgr_t plugins; + dataq_t* pend_updates[2]; + resmgr_t resmgr; + kvmgr_t jobs; + kvmgr_t queues; + kvmgr_t workers; + kvmgr_t users; + kvmgr_t projects; + pthread_t jobpicker; + pthread_t allocator; + cbuf_t wbuf; /* action to trigger scheduling */ + cbuf_t jbuf; /* job picked for allocation */ + cbuf_t decisions; /* scheduler decisions */ } xce_scheduler_t; -extern int sched_init(xce_scheduler_t*, xce_client_config_t*, - xce_request_handler_t); -extern void sched_fini(xce_scheduler_t*); -extern int sched_update(xce_scheduler_t*); -extern void sched_go(xce_scheduler_t*); -extern void sched_begin_session(xce_scheduler_t*); -extern void sched_end_session(xce_scheduler_t*); +extern int sched_init(xce_scheduler_t*, xce_client_config_t*, + xce_request_handler_t); +extern void sched_fini(xce_scheduler_t*); +extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*); +extern int sched_update(xce_scheduler_t*); +extern void sched_go(xce_scheduler_t*); +extern void sched_begin_session(xce_scheduler_t*); +extern void sched_end_session(xce_scheduler_t*); extern xce_sched_job_t* sched_get_job(xce_scheduler_t*, uint64_t); extern xce_sched_queue_t* sched_get_queue(xce_scheduler_t*, char*); extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*); diff --git a/src/scheduler/src/handler.c b/src/scheduler/src/handler.c index 3e9f016..ecf79e4 100644 --- a/src/scheduler/src/handler.c +++ b/src/scheduler/src/handler.c @@ -18,26 +18,6 @@ static void enq_pend_update(xce_scheduler_t* sched, mylock_unlock(&(sched->lock)); } -static int sched_handle_config(xce_scheduler_t* sched, xce_conf4sched_t* config) -{ - xce_sched_update_request_t* req; - - TRACE("%s...", __func__); - TRACE("my cluster name %s", config->clustername); - - req = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); - if (req == NULL) { - return -ENOMEM; - } - - req->op = XCE_SCHED_OP_UPDATE_CONFIG; - req->u.config = config; - - enq_pend_update(sched, req); - - return 0; -} - static int sched_handle_jobpush(xce_scheduler_t* sched, xce_jobs_data_t* jobs) { TRACE("%s...", __func__); @@ -191,10 +171,13 @@ int scheduler_req_handler(xce_request_t* request, void* args) case OPCode_push: switch (request->body_which) { case Request_body_configpushmsg: - /* configuration arriving */ - rc = sched_handle_config( + /* configuration packet, it will only arrive when scheduler has + * register itself to controller, we can handle it synchronized + */ + rc = sched_init_config( sched, request->body.configpushmsg->data.conf4sched); if (rc >= 0) { + /* configuration will be saved in sched->config */ request->body.configpushmsg->data.conf4sched = NULL; } break; diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index f662937..0e4fe2a 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -72,8 +72,8 @@ static int plugin_config_update_(void* arg1, void* arg2) return 0; } -int plugin_config_update(void* arg, xce_sched_config_t* old, - xce_sched_config_t* config) +int plugin_config_update(void* arg, xce_conf4sched_t* old, + xce_conf4sched_t* config) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; xlist_t lst; diff --git a/src/scheduler/src/plugins/fcfs.c b/src/scheduler/src/plugins/fcfs.c index 026fbf8..63b62ae 100644 --- a/src/scheduler/src/plugins/fcfs.c +++ b/src/scheduler/src/plugins/fcfs.c @@ -86,9 +86,8 @@ static int fcfs_priv_data_init(fcfs_plugin_data_t* data) return 0; } -static int fcfs_update_config(xce_sched_plugin_t* plugin, - xce_sched_config_t* old, - xce_sched_config_t* config) +static int fcfs_update_config(xce_sched_plugin_t* plugin, xce_conf4sched_t* old, + xce_conf4sched_t* config) { fcfs_plugin_data_t* privdata = NULL; diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index 721b020..ae1c3a8 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -126,41 +126,3 @@ xce_sched_project_t* make_sched_project(void* arg, char* name) return project; } -xce_sched_config_t* make_sched_config(void* arg, xce_conf4sched_t* c) -{ - xce_scheduler_t* sched = (xce_scheduler_t*)arg; - xce_sched_config_t* config = NULL; - int i; - - if ((sched == NULL) || (c == NULL)) { - errno = EINVAL; - - return NULL; - } - - config = (xce_sched_config_t *)calloc(1, sizeof(*config)); - if (config == NULL) { - errno = ENOMEM; - return NULL; - } - - /* FIXME: */ - resmgr_init(&(config->resmgr), /* slots */128, /* delta */ 128, NULL); - config->clustername = strdup(c->clustername); - - for(i = 0; i < c->n_resources; i ++) { - xce_resource_t *r = NULL; - - MOVE_PTR(r, c->resources[i]); - resmgr_add_res(&(config->resmgr), r); - } - - return config; -} - -void kill_sched_config(void* arg) -{ - if (arg == NULL) { - return; - } -} diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 0c448d5..3a3f227 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -164,6 +164,12 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); dataq_init(sched->pend_updates[1], sched_pend_update_killer); + resmgr_init( + &(sched->resmgr), /* slots */ 128, /* delta */ 128, + /* should be NULL as resmgr will provide hash find by resource name, the + real resource stored in sched->config->resources[] */ + NULL); + kvmgr_init(&(sched->plugins), string_hash_key, 8, 8, (zarray_item_killer_t)plugin_unregister); kvmgr_init(&(sched->jobs), int_hash_key, 1000, 1000, @@ -239,40 +245,43 @@ void sched_fini(xce_scheduler_t* sched) return; } -static int sched_merge_config(xce_scheduler_t *sched, xce_sched_config_t *config) { - if (sched->config == NULL) { - sched->config = config; - return 0; - } - - /* FIXME: merge sched->config and config and free unused */ - return 1; -} -static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config) +static int sched_init_resources(xce_scheduler_t* sched) { - int i; - xce_sched_config_t *config2 = make_sched_config(sched, config); + int i; + xce_conf4sched_t* config = sched->config; - plugin_config_update(sched, sched->config, config2); + for (i = 0; i < config->n_resources; i++) { + xce_resource_t* r = config->resources[i]; - sched_merge_config(sched, config2); + resmgr_add_res(&(sched->resmgr), r); + } + return 0; +} + +static int sched_init_queues(xce_scheduler_t* sched) +{ + int i; + xce_conf4sched_t* config = sched->config; for (i = 0; i < config->n_queues; i++) { - xce_sched_queue_t* qold - = sched_get_queue(sched, config->queues[i]->queue); xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]); - plugin_queue_update(sched, qold, q); - - if (qold == NULL) { - sched_add_queue(sched, q); - } else { - /* FIXME: overwrite qold with q */ - } + plugin_queue_update(sched, NULL, q); + sched_add_queue(sched, q); } return 0; } +int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config) +{ + plugin_config_update(sched, NULL, config); + sched->config = config; + + sched_init_resources(sched); + sched_init_queues(sched); + + return 0; +} static int sched_update_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs) @@ -353,7 +362,9 @@ int sched_update(xce_scheduler_t* sched) req = ITPTR_TO_UPDATE_REQ(it); switch (req->op) { case XCE_SCHED_OP_UPDATE_CONFIG: - rc = sched_update_config(sched, req->u.config); + /* this will only be handled when scheduler has been started, so + * should never get here */ + rc = -EINVAL; break; case XCE_SCHED_OP_UPDATE_JOB: rc = sched_update_jobs(sched, req->u.jobupdate.njobs, @@ -455,8 +466,10 @@ xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name) return kvmgr_lock(&(sched->projects), (void*)name, OBJREF_NOP); } +/* FIXME this is just stub, we need to add more */ int sched_job_is_ready(xce_sched_job_t* j) { return 1; } +/* FIXME this is just stub, we need to add more */ int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; } int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision)