From 088afe4ea0dce4b358fe7d1273ad984f7951878d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 23 Dec 2025 18:12:13 +0800 Subject: [PATCH] scheduling job with fake resource --- src/include/preason.h | 1 + src/include/xcealloc.h | 15 +++ src/lib/meson.build | 3 +- src/lib/xcealloc.c | 119 ++++++++++++++++++ src/scheduler/include/plugin.h | 16 ++- src/scheduler/include/sched/types.h | 1 + src/scheduler/include/scheduler.h | 1 + src/scheduler/src/main.c | 21 ++-- src/scheduler/src/plugin.c | 41 ++++-- src/scheduler/src/plugins/default_allocator.c | 79 ++++++++++-- src/scheduler/src/plugins/fcfs.c | 6 +- src/scheduler/src/scheduler.c | 24 +++- 12 files changed, 281 insertions(+), 46 deletions(-) create mode 100644 src/include/xcealloc.h create mode 100644 src/lib/xcealloc.c diff --git a/src/include/preason.h b/src/include/preason.h index fc138d2..a073f6a 100644 --- a/src/include/preason.h +++ b/src/include/preason.h @@ -5,6 +5,7 @@ #include "xce/types.h" #define PREASON_BAD_RESREQ (1) +#define PREASON_NO_RESOURCE (2) extern void erase_job_preason(xce_job_preason_t*); extern int add_job_preason(xce_job_preason_t*, int, char*, char*); diff --git a/src/include/xcealloc.h b/src/include/xcealloc.h new file mode 100644 index 0000000..4520723 --- /dev/null +++ b/src/include/xcealloc.h @@ -0,0 +1,15 @@ +#if !defined(_XCEALLOC_H_) + +#define _XCEALLOC_H_ 1 + +#include + +#include "xce/types.h" + +extern xce_alloc_t* make_alloc(char*, char*, int, float, int); +extern xce_host_alloc_t* make_host_alloc(char*, int, xce_alloc_t**); +extern xce_job_alloc_t* make_job_alloc(uint64_t, int, xce_host_alloc_t**); +extern xce_alloc_t** make_allocs(int, ...); +extern xce_host_alloc_t** make_host_allocs(int, ...); + +#endif diff --git a/src/lib/meson.build b/src/lib/meson.build index 1a1d745..787e699 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -30,7 +30,8 @@ libxce_src = [ gen_srcs[0], 'idgen.c', 'delay.c', 'ordq.c', - 'preason.c' + 'preason.c', + 'xcealloc.c', ] incdir = include_directories('../include') diff --git a/src/lib/xcealloc.c b/src/lib/xcealloc.c new file mode 100644 index 0000000..d928b80 --- /dev/null +++ b/src/lib/xcealloc.c @@ -0,0 +1,119 @@ +#include +#include +#include +#include +#include + +#include "xce/types.h" +#include "xcealloc.h" + +xce_alloc_t* make_alloc(char* resname, char* reserve, int instid, float value, + int duration) +{ + xce_alloc_t* alloc = NULL; + + alloc = (xce_alloc_t*)calloc(1, sizeof(*alloc)); + if (alloc == NULL) { + return NULL; + } + + alloc->resname = strdup(resname); + if (reserve != NULL) { + alloc->reserve = strdup(reserve); + } else { + alloc->reserve = strdup(""); + } + + alloc->instid = instid; + alloc->value = value; + alloc->from = time(NULL); + alloc->duration = duration; + + return alloc; +} + +xce_alloc_t** make_allocs(int nallocs, ...) +{ + xce_alloc_t** allocs = NULL; + int i; + va_list ap; + + if (nallocs <= 0) { + return NULL; + } + + allocs = (xce_alloc_t**)calloc(nallocs, sizeof(*allocs)); + if (allocs == NULL) { + return NULL; + } + + va_start(ap, nallocs); + for (i = 0; i < nallocs; i++) { + xce_alloc_t* alloc = va_arg(ap, xce_alloc_t*); + + allocs[i] = alloc; + } + va_end(ap); + + return allocs; +} + +xce_host_alloc_t* make_host_alloc(char* host, int nalloc, xce_alloc_t** allocs) +{ + xce_host_alloc_t* halloc = NULL; + + halloc = (xce_host_alloc_t*)calloc(1, sizeof(*halloc)); + if (halloc == NULL) { + return NULL; + } + + halloc->nallocs = nalloc; + halloc->allocs = allocs; + halloc->host = strdup(host); + + return halloc; +} + +xce_host_alloc_t** make_host_allocs(int nhosts, ...) +{ + xce_host_alloc_t** hallocs = NULL; + int i; + va_list ap; + + if (nhosts == NULL) { + return NULL; + } + + hallocs = (xce_host_alloc_t**)calloc(1, sizeof(*hallocs)); + if (hallocs == NULL) { + return NULL; + } + + va_start(ap, nhosts); + for (i = 0; i < nhosts; i++) { + xce_host_alloc_t* halloc = va_arg(ap, xce_host_alloc_t*); + + hallocs[i] = halloc; + } + + va_end(ap); + + return hallocs; +} + +xce_job_alloc_t* make_job_alloc(uint64_t jobid, int nhost, + xce_host_alloc_t** hallocs) +{ + xce_job_alloc_t* jalloc = NULL; + + jalloc = (xce_job_alloc_t*)calloc(1, sizeof(*jalloc)); + if (jalloc == NULL) { + return NULL; + } + + jalloc->jobid = jobid; + jalloc->nhosts = nhost; + jalloc->allocs = hallocs; + + return jalloc; +} diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 575b489..ba6eb12 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -12,9 +12,12 @@ 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_conf4sched_t*); -typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*); -typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*); -typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_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*, + xce_sched_job_t*); +typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*, + xce_sched_worker_t*); typedef int (*plugin_commit_decision_hook_t)(void*, xce_decision_t*); /* initialize function of plugin */ @@ -56,9 +59,10 @@ typedef struct { extern int plugin_register(void*, void*); extern void plugin_unregister(void*); extern int plugin_config_update(void*, xce_conf4sched_t*); -extern int plugin_queue_update(void*, xce_sched_queue_t*); -extern int plugin_job_update(void*, xce_sched_job_t*); -extern int plugin_worker_update(void*, xce_sched_worker_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*, + xce_sched_worker_t*); extern int plugin_begin_session(void*); extern int plugin_end_session(void*); extern int plugin_commit_decision(void*, xce_decision_t*); diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 73334cb..5669de6 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -41,6 +41,7 @@ typedef struct { time_t queue_time; time_t start_time; unsigned int state; + unsigned int starting : 1; int32_t priority; xce_sched_queue_t* queue; xce_sched_user_t* user; diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 4cf7b9a..bb1badf 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -82,5 +82,6 @@ extern int sched_job_is_ready(xce_sched_job_t*); extern int sched_queue_is_ready(xce_sched_queue_t*); extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*); extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*); +extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*); #endif diff --git a/src/scheduler/src/main.c b/src/scheduler/src/main.c index 1d0b9bb..2e9bbfc 100644 --- a/src/scheduler/src/main.c +++ b/src/scheduler/src/main.c @@ -122,13 +122,12 @@ int main(int argc, char* argv[]) sched_go(&sched); while (1) { - xce_decision_t* decision = NULL; - int rc = 0; - xce_pendreason_t *preasons[1]; - xce_preason_set_t reasons = { - .nreasons = 1, - .reasons = preasons - }; + xce_decision_t* decision = NULL; + int rc = 0; + xce_pendreason_t* preasons[1]; + xce_preason_set_t reasons = { .nreasons = 1, .reasons = preasons }; + xce_job_alloc_t* allocs[1]; + xce_job_dispatch_t jdispatch = { .njobs = 1, .allocs = allocs }; cbuf_get(&(sched.decisions), &decision); @@ -144,11 +143,13 @@ int main(int argc, char* argv[]) break; case XCE_DECISION_DISPATCH: /* tell controller to dispatch job */ + allocs[0] = decision->u.alloc; + rc = sched_publish_dispatch(&sched, &jdispatch); break; case XCE_DECISION_PEND: /* set pend reason for jobs */ - preasons[ 0] = decision->u.reason; - rc = sched_publish_pend_reason(&sched, &reasons); + preasons[0] = decision->u.reason; + rc = sched_publish_pend_reason(&sched, &reasons); break; default: break; @@ -158,7 +159,7 @@ int main(int argc, char* argv[]) } sched_end_session(&sched); - + if (sched.stop) { break; } diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 5bad0d5..674327b 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -7,6 +7,11 @@ #include "scheduler.h" #include "xlist.h" +typedef struct { + void* old; + void* new; +} update_args_t; + int plugin_register(void* arg1, void* arg2) { xce_scheduler_t* sched = (xce_scheduler_t*)arg1; @@ -88,23 +93,26 @@ int plugin_config_update(void* arg, xce_conf4sched_t* config) static int plugin_job_update_(void* arg1, void* arg2) { xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + update_args_t* args = (update_args_t*)arg1; if ((arg1 == NULL) || (plugin == NULL)) { return 0; } if (plugin->hooks.job_hook != NULL) { - return plugin->hooks.job_hook(plugin, arg1); + return plugin->hooks.job_hook(plugin, (xce_sched_job_t*)args->old, + (xce_sched_job_t*)args->new); } return 0; } -int plugin_job_update(void* arg, xce_sched_job_t* job) +int plugin_job_update(void* arg, xce_sched_job_t* jold, xce_sched_job_t* job) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; xlist_t lst; int rc; + update_args_t args = { .old = jold, .new = job }; if (sched == NULL) { return -EINVAL; @@ -112,7 +120,7 @@ int plugin_job_update(void* arg, xce_sched_job_t* job) xlist_init(&lst, 3); xlist_store(&lst, plugin_job_update_); - rc = kvmgr_iter(&(sched->plugins), NULL, &lst, job); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args); xlist_fini(&lst); @@ -122,23 +130,27 @@ int plugin_job_update(void* arg, xce_sched_job_t* job) static int plugin_queue_update_(void* arg1, void* arg2) { xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + update_args_t* args = (update_args_t*)arg1; if ((arg1 == NULL) || (plugin == NULL)) { return 0; } if (plugin->hooks.queue_hook != NULL) { - return plugin->hooks.queue_hook(plugin, arg1); + return plugin->hooks.queue_hook(plugin, (xce_sched_queue_t*)args->old, + (xce_sched_queue_t*)args->new); } return 0; } -int plugin_queue_update(void* arg, xce_sched_queue_t* queue) +int plugin_queue_update(void* arg, xce_sched_queue_t* qold, + xce_sched_queue_t* queue) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; xlist_t lst; int rc; + update_args_t args = { .old = qold, .new = queue }; if (sched == NULL) { return -EINVAL; @@ -146,7 +158,7 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* queue) xlist_init(&lst, 3); xlist_store(&lst, plugin_queue_update_); - rc = kvmgr_iter(&(sched->plugins), NULL, &lst, queue); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args); xlist_fini(&lst); @@ -156,23 +168,27 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* queue) static int plugin_worker_update_(void* arg1, void* arg2) { xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + update_args_t* args = (update_args_t*)arg1; if ((arg1 == NULL) || (plugin == NULL)) { return 0; } if (plugin->hooks.worker_hook != NULL) { - return plugin->hooks.worker_hook(plugin, arg1); + return plugin->hooks.worker_hook(plugin, (xce_sched_worker_t*)args->old, + (xce_sched_worker_t*)args->new); } return 0; } -int plugin_worker_update(void* arg, xce_sched_worker_t* worker) +int plugin_worker_update(void* arg, xce_sched_worker_t* wold, + xce_sched_worker_t* worker) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; xlist_t lst; int rc; + update_args_t args = { .old = wold, .new = worker }; if (sched == NULL) { return -EINVAL; @@ -180,7 +196,7 @@ int plugin_worker_update(void* arg, xce_sched_worker_t* worker) xlist_init(&lst, 3); xlist_store(&lst, plugin_worker_update_); - rc = kvmgr_iter(&(sched->plugins), NULL, &lst, worker); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args); xlist_fini(&lst); @@ -253,8 +269,9 @@ int plugin_end_session(void* arg) return rc; } -static int plugin_commit_decision_(void* arg1, void* arg2) { - xce_sched_plugin_t *plugin = (xce_sched_plugin_t *)arg2; +static int plugin_commit_decision_(void* arg1, void* arg2) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; if (plugin == NULL) { return 0; @@ -263,7 +280,7 @@ static int plugin_commit_decision_(void* arg1, void* arg2) { if (plugin->hooks.commit_hook != NULL) { return plugin->hooks.commit_hook(plugin, arg1); } - + return 0; } diff --git a/src/scheduler/src/plugins/default_allocator.c b/src/scheduler/src/plugins/default_allocator.c index ea4f2dc..c5def9b 100644 --- a/src/scheduler/src/plugins/default_allocator.c +++ b/src/scheduler/src/plugins/default_allocator.c @@ -10,33 +10,93 @@ #include "scheduler.h" #include "utils.h" #include "xce/types.h" +#include "xcealloc.h" + +static uint32_t maxslots = 8; +static uint32_t slotsInUse = 0; static int default_allocator_init(xce_sched_plugin_t* plugin) { return 1; } -static void default_allocator_fini(xce_sched_plugin_t* plugin) { +static void default_allocator_fini(xce_sched_plugin_t* plugin) +{ if (plugin == NULL) { return; } FREEUP(plugin); } -static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j) +static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold, + xce_sched_job_t* j) { /* FIXME: we need to add logic to check whether the job should be handled by * me */ j->allocator = plugin; + switch (j->state) { + case JOB_STAT_PEND: + break; + case JOB_STAT_RUN: + break; + case JOB_STAT_DONE: + break; + default: + break; + } return 1; } static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid, xce_pendreason_t** preason) { - xce_pendreason_t* reason = make_pendreason(jobid); + xce_job_alloc_t* jalloc = NULL; - add_job_preason(reason->reasons, PREASON_BAD_RESREQ, NULL, NULL); + *preason = NULL; - *preason = reason; - return NULL; + if (slotsInUse < maxslots) { + xce_alloc_t* alloc = NULL; + xce_alloc_t** allocs = NULL; + xce_host_alloc_t* halloc = NULL; + xce_host_alloc_t** hallocs = NULL; + + alloc = make_alloc("slots", "", -1, 1.0, -1); + allocs = make_allocs(1, alloc); + halloc = make_host_alloc("pc13v", 1, allocs); + hallocs = make_host_allocs(1, halloc); + + jalloc = make_job_alloc(jobid, 1, hallocs); + } else { + xce_pendreason_t* reason = NULL; + + reason = make_pendreason(jobid); + add_job_preason(reason->reasons, PREASON_NO_RESOURCE, "pc13v", "slots"); + + *preason = reason; + } + + return jalloc; +} + +static int my_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision) +{ + int rc = 0; + uint64_t jobid = 0; + xce_sched_job_t* job = NULL; + + switch (decision->op) { + case XCE_DECISION_DISPATCH: + /* FIXME: */ + jobid = decision->u.alloc->jobid; + job = sched_get_job(plugin->sched, jobid); + if (job != NULL) { + slotsInUse++; + job->starting = 1; + } + + break; + default: + break; + } + + return rc; } int register_default_allocator(void* sched) @@ -44,7 +104,8 @@ int register_default_allocator(void* sched) static xce_sched_plugin_t default_allocator_plugin = { .name = "xce_sched_default_allocator", .hooks = { .config_hook = NULL, - .job_hook = (plugin_update_job_hook_t)my_update_job }, + .job_hook = (plugin_update_job_hook_t)my_update_job, + .commit_hook = (plugin_commit_decision_hook_t)my_commit }, .init_fn = (plugin_init_fn_t)default_allocator_init, .fini_fn = (plugin_fini_fn_t)default_allocator_fini, .begin_session_fn = NULL, @@ -54,9 +115,9 @@ int register_default_allocator(void* sched) .data = NULL }; - xce_sched_plugin_t *plugin = NULL; + xce_sched_plugin_t* plugin = NULL; - plugin= (xce_sched_plugin_t *)calloc(1, sizeof(*plugin)); + plugin = (xce_sched_plugin_t*)calloc(1, sizeof(*plugin)); if (plugin == NULL) { return -ENOMEM; } diff --git a/src/scheduler/src/plugins/fcfs.c b/src/scheduler/src/plugins/fcfs.c index 8cf314e..07ffd1d 100644 --- a/src/scheduler/src/plugins/fcfs.c +++ b/src/scheduler/src/plugins/fcfs.c @@ -124,7 +124,8 @@ static int job_match(void* arg1, va_list ap) return (j->job->jobid == jobid); } -static int fcfs_update_queue(xce_sched_plugin_t* plugin, xce_sched_queue_t* q) +static int fcfs_update_queue(xce_sched_plugin_t* plugin, + xce_sched_queue_t* qold, xce_sched_queue_t* q) { fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; ordq_item_t* it; @@ -164,7 +165,8 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name) return (fcfs_queue_t*)it->data; } -static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j) +static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold, + xce_sched_job_t* j) { fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; fcfs_queue_t* q = NULL; diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index ff46a6b..7f3df40 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -206,7 +206,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, void sched_fini(xce_scheduler_t* sched) { - void* res; + void* res; + uint64_t jobid; if (sched == NULL) { return; @@ -216,6 +217,8 @@ void sched_fini(xce_scheduler_t* sched) pthread_join(sched->jobpicker, &res); pthread_cancel(sched->allocator); + /* try to wake up allocator */ + cbuf_tryput(&(sched->jbuf), &jobid); pthread_join(sched->allocator, &res); xcecli_wakeup(sched->apictx, (int)-EINTR); @@ -241,9 +244,11 @@ static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config) plugin_config_update(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, q); + plugin_queue_update(sched, qold, q); } return 0; @@ -255,9 +260,10 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs, int i; for (i = 0; i < njobs; i++) { - xce_sched_job_t* j = make_sched_job(sched, jobs[i]); + xce_sched_job_t* jold = sched_get_job(sched, jobs[i]->jobid); + xce_sched_job_t* j = make_sched_job(sched, jobs[i]); - plugin_job_update(sched, j); + plugin_job_update(sched, jold, j); } return 0; @@ -269,9 +275,10 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers, int i; for (i = 0; i < nworkers; i++) { - xce_sched_worker_t* w = make_sched_worker(sched, workers[i]); + xce_sched_worker_t* wold = sched_get_worker(sched, workers[i]->name); + xce_sched_worker_t* w = make_sched_worker(sched, workers[i]); - plugin_worker_update(sched, w); + plugin_worker_update(sched, wold,w); } return 0; @@ -413,3 +420,8 @@ int sched_publish_pend_reason(xce_scheduler_t* sched, { return xcecli_set_pend_reason(sched->apictx, reasons); } + +int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* dispatch) +{ + return xcecli_jobdispatch(sched->apictx, dispatch); +}