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

improve configuration handling

Browse source
This commit is contained in:
Rongsong Shen 2025-12-26 17:52:43 +08:00
parent a27c8bfe6e
commit 4769e6845f
8 changed files with 74 additions and 121 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/scheduler/include/plugin.h
View file

@ -11,8 +11,8 @@
typedef void* plugin_data_t; typedef void* plugin_data_t;
/* hooker of plugin, the first parameter should be xce_sched_plugin_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*, typedef int (*plugin_update_config_hook_t)(void*, xce_conf4sched_t*,
xce_sched_config_t*); xce_conf4sched_t*);
typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*, typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*,
xce_sched_queue_t*); xce_sched_queue_t*);
typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_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 int plugin_register(void*, void*);
extern void plugin_unregister(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_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_job_update(void*, xce_sched_job_t*, xce_sched_job_t*);
extern int plugin_worker_update(void*, xce_sched_worker_t*, extern int plugin_worker_update(void*, xce_sched_worker_t*,

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

@ -60,17 +60,11 @@ typedef struct {
xce_val_t* res_in_use; /* used resources in worker */ xce_val_t* res_in_use; /* used resources in worker */
} xce_sched_worker_t; } 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_job_t* make_sched_job(void*, xce_job_t*);
extern xce_sched_queue_t* make_sched_queue(void*, xce_queue_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_worker_t* make_sched_worker(void*, xce_worker_t*);
extern xce_sched_user_t* make_sched_user(void*, char*); extern xce_sched_user_t* make_sched_user(void*, char*);
extern xce_sched_project_t* make_sched_project(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_job(void*);
extern void kill_sched_queue(void*); extern void kill_sched_queue(void*);
extern void kill_sched_worker(void*); extern void kill_sched_worker(void*);

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

@ -49,31 +49,33 @@ typedef struct {
link))) link)))
typedef struct { typedef struct {
mylock_t lock; mylock_t lock;
int stop; int stop;
xcecli_ctx_t apictx; xcecli_ctx_t apictx;
xce_sched_config_t* config; xce_conf4sched_t* config;
kvmgr_t plugins; kvmgr_t plugins;
dataq_t* pend_updates[2]; dataq_t* pend_updates[2];
kvmgr_t jobs; resmgr_t resmgr;
kvmgr_t queues; kvmgr_t jobs;
kvmgr_t workers; kvmgr_t queues;
kvmgr_t users; kvmgr_t workers;
kvmgr_t projects; kvmgr_t users;
pthread_t jobpicker; kvmgr_t projects;
pthread_t allocator; pthread_t jobpicker;
cbuf_t wbuf; /* action to trigger scheduling */ pthread_t allocator;
cbuf_t jbuf; /* job picked for allocation */ cbuf_t wbuf; /* action to trigger scheduling */
cbuf_t decisions; /* scheduler decisions */ cbuf_t jbuf; /* job picked for allocation */
cbuf_t decisions; /* scheduler decisions */
} xce_scheduler_t; } xce_scheduler_t;
extern int sched_init(xce_scheduler_t*, xce_client_config_t*, extern int sched_init(xce_scheduler_t*, xce_client_config_t*,
xce_request_handler_t); xce_request_handler_t);
extern void sched_fini(xce_scheduler_t*); extern void sched_fini(xce_scheduler_t*);
extern int sched_update(xce_scheduler_t*); extern int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t*);
extern void sched_go(xce_scheduler_t*); extern int sched_update(xce_scheduler_t*);
extern void sched_begin_session(xce_scheduler_t*); extern void sched_go(xce_scheduler_t*);
extern void sched_end_session(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_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_queue_t* sched_get_queue(xce_scheduler_t*, char*);
extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*); extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*);

27
src/scheduler/src/handler.c
View file

@ -18,26 +18,6 @@ static void enq_pend_update(xce_scheduler_t* sched,
mylock_unlock(&(sched->lock)); 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) static int sched_handle_jobpush(xce_scheduler_t* sched, xce_jobs_data_t* jobs)
{ {
TRACE("%s...", __func__); TRACE("%s...", __func__);
@ -191,10 +171,13 @@ int scheduler_req_handler(xce_request_t* request, void* args)
case OPCode_push: case OPCode_push:
switch (request->body_which) { switch (request->body_which) {
case Request_body_configpushmsg: case Request_body_configpushmsg:
/* configuration arriving */ /* configuration packet, it will only arrive when scheduler has
rc = sched_handle_config( * register itself to controller, we can handle it synchronized
*/
rc = sched_init_config(
sched, request->body.configpushmsg->data.conf4sched); sched, request->body.configpushmsg->data.conf4sched);
if (rc >= 0) { if (rc >= 0) {
/* configuration will be saved in sched->config */
request->body.configpushmsg->data.conf4sched = NULL; request->body.configpushmsg->data.conf4sched = NULL;
} }
break; break;

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

@ -72,8 +72,8 @@ static int plugin_config_update_(void* arg1, void* arg2)
return 0; return 0;
} }
int plugin_config_update(void* arg, xce_sched_config_t* old, int plugin_config_update(void* arg, xce_conf4sched_t* old,
xce_sched_config_t* config) xce_conf4sched_t* config)
{ {
xce_scheduler_t* sched = (xce_scheduler_t*)arg; xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst; xlist_t lst;

5
src/scheduler/src/plugins/fcfs.c
View file

@ -86,9 +86,8 @@ static int fcfs_priv_data_init(fcfs_plugin_data_t* data)
return 0; return 0;
} }
static int fcfs_update_config(xce_sched_plugin_t* plugin, static int fcfs_update_config(xce_sched_plugin_t* plugin, xce_conf4sched_t* old,
xce_sched_config_t* old, xce_conf4sched_t* config)
xce_sched_config_t* config)
{ {
fcfs_plugin_data_t* privdata = NULL; fcfs_plugin_data_t* privdata = NULL;

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

@ -126,41 +126,3 @@ xce_sched_project_t* make_sched_project(void* arg, char* name)
return project; 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;
}
}

61
src/scheduler/src/scheduler.c
View file

@ -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)); sched->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
dataq_init(sched->pend_updates[1], sched_pend_update_killer); 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, kvmgr_init(&(sched->plugins), string_hash_key, 8, 8,
(zarray_item_killer_t)plugin_unregister); (zarray_item_killer_t)plugin_unregister);
kvmgr_init(&(sched->jobs), int_hash_key, 1000, 1000, kvmgr_init(&(sched->jobs), int_hash_key, 1000, 1000,
@ -239,40 +245,43 @@ void sched_fini(xce_scheduler_t* sched)
return; return;
} }
static int sched_merge_config(xce_scheduler_t *sched, xce_sched_config_t *config) { static int sched_init_resources(xce_scheduler_t* sched)
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)
{ {
int i; int i;
xce_sched_config_t *config2 = make_sched_config(sched, config); 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++) { 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]); xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]);
plugin_queue_update(sched, qold, q); plugin_queue_update(sched, NULL, q);
sched_add_queue(sched, q);
if (qold == NULL) {
sched_add_queue(sched, q);
} else {
/* FIXME: overwrite qold with q */
}
} }
return 0; 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, static int sched_update_jobs(xce_scheduler_t* sched, int njobs,
xce_job_t** jobs) xce_job_t** jobs)
@ -353,7 +362,9 @@ int sched_update(xce_scheduler_t* sched)
req = ITPTR_TO_UPDATE_REQ(it); req = ITPTR_TO_UPDATE_REQ(it);
switch (req->op) { switch (req->op) {
case XCE_SCHED_OP_UPDATE_CONFIG: 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; break;
case XCE_SCHED_OP_UPDATE_JOB: case XCE_SCHED_OP_UPDATE_JOB:
rc = sched_update_jobs(sched, req->u.jobupdate.njobs, 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); 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; } 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_queue_is_ready(xce_sched_queue_t* q) { return 1; }
int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision)