From cd0bca31385fdd20c436d15f983e998c6b607072 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 22 Dec 2025 18:24:43 +0800 Subject: [PATCH] save works --- src/scheduler/include/decision.h | 5 +++ src/scheduler/include/plugin.h | 14 ++++--- src/scheduler/include/sched/types.h | 1 + src/scheduler/include/scheduler.h | 1 + src/scheduler/meson.build | 1 + src/scheduler/src/decision.c | 30 ++++++++++++++ src/scheduler/src/plugin_default_allocator.c | 41 ++++++++++++++++++++ src/scheduler/src/scheduler.c | 40 ++++++++++++++++++- 8 files changed, 127 insertions(+), 6 deletions(-) create mode 100644 src/scheduler/src/plugin_default_allocator.c diff --git a/src/scheduler/include/decision.h b/src/scheduler/include/decision.h index cd06e74..b247fa4 100644 --- a/src/scheduler/include/decision.h +++ b/src/scheduler/include/decision.h @@ -3,6 +3,7 @@ #define _DECISION_H_ 1 #include "xce/types.h" +#include typedef enum { XCE_DECISION_NONE = 0, @@ -13,9 +14,13 @@ typedef enum { typedef struct { xce_decision_op_t op; + uint64_t jobid; union { xce_job_alloc_t* alloc; } u; } xce_decision_t; +extern xce_decision_t* make_pend_decision(uint64_t); +extern xce_decision_t* make_dispatch_decision(xce_job_alloc_t*); + #endif diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index cea6c5c..7ae3b97 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -5,6 +5,7 @@ #include #include +#include "decision.h" #include "sched/types.h" typedef void* plugin_data_t; @@ -14,6 +15,7 @@ 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_commit_decision_hook_t)(void*, xce_decision_t*); /* initialize function of plugin */ typedef int (*plugin_init_fn_t)(void*); @@ -27,13 +29,14 @@ typedef int (*plugin_begin_session_fn_t)(void*); typedef int (*plugin_end_session_fn_t)(void*); /* allocation function of job */ -typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t*); +typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t); typedef struct { - plugin_update_config_hook_t config_hook; - plugin_update_queue_hook_t queue_hook; - plugin_update_job_hook_t job_hook; - plugin_update_worker_hook_t worker_hook; + plugin_update_config_hook_t config_hook; + plugin_update_queue_hook_t queue_hook; + plugin_update_job_hook_t job_hook; + plugin_update_worker_hook_t worker_hook; + plugin_commit_decision_hook_t commit_hook; } plugin_hooks_t; typedef struct { @@ -59,5 +62,6 @@ extern int plugin_begin_session(void*); extern int plugin_end_session(void*); extern int register_fcfs(void*); +extern int register_default_allocator(void*); #endif diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 8bd282d..73334cb 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -45,6 +45,7 @@ typedef struct { xce_sched_queue_t* queue; xce_sched_user_t* user; xce_sched_project_t* project; + void* allocator; uint32_t mintasks; uint32_t maxtasks; xce_job_alloc_t* allocs; diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index ea89ad2..7bacc87 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -80,5 +80,6 @@ extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*); extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*); 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*); #endif diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index b471bdf..0b84644 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -6,6 +6,7 @@ scheduler_srcs = [ 'src/schedobj.c', 'src/plugin.c', 'src/plugin_fcfs.c', + 'src/plugin_default_allocator.c', ] incdir = include_directories('src', diff --git a/src/scheduler/src/decision.c b/src/scheduler/src/decision.c index 5af397f..46deaba 100644 --- a/src/scheduler/src/decision.c +++ b/src/scheduler/src/decision.c @@ -2,3 +2,33 @@ #include #include "decision.h" + +xce_decision_t* make_pend_decision(uint64_t jobid) +{ + xce_decision_t* decision; + + decision = (xce_decision_t*)calloc(1, sizeof(*decision)); + if (decision == NULL) { + return NULL; + } + + decision->op = XCE_DECISION_PEND; + decision->jobid = jobid; + + return decision; +} + +xce_decision_t* make_dispatch_decision(xce_job_alloc_t* alloc) +{ + xce_decision_t* decision; + decision = (xce_decision_t*)calloc(1, sizeof(*decision)); + if (decision == NULL) { + return NULL; + } + + decision->op = XCE_DECISION_DISPATCH; + decision->u.alloc = alloc; + decision->jobid = alloc->jobid; + + return decision; +} diff --git a/src/scheduler/src/plugin_default_allocator.c b/src/scheduler/src/plugin_default_allocator.c new file mode 100644 index 0000000..495635d --- /dev/null +++ b/src/scheduler/src/plugin_default_allocator.c @@ -0,0 +1,41 @@ +#include +#include +#include +#include +#include + +#include "plugin.h" +#include "scheduler.h" +#include "utils.h" + +static int default_allocator_init(void* arg) { return 1; } + +static void default_allocator_fini(void* arg) { return; } + +static int my_update_job(xce_sched_plugin_t *plugin, xce_sched_job_t *j) { + /* FIXME: we need to add logic to check whether the job should be handled by me */ + j->allocator = plugin; + return 1; +} + +static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid) +{ + return NULL; +} + +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 }, + .init_fn = (plugin_init_fn_t)default_allocator_init, + .fini_fn = (plugin_fini_fn_t)default_allocator_fini, + .begin_session_fn = NULL, + .end_session_fn = NULL, + .job_elect_fn = NULL, + .job_alloc_fn = (plugin_job_allocate_fn_t)my_alloc, + .data = NULL +}; + +int register_default_allocator(void* sched) +{ + return plugin_register(sched, &default_allocator_plugin); +} diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 33aa3f6..6c27ec6 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -7,6 +7,7 @@ #include "client.h" #include "controller.capnp.h" +#include "decision.h" #include "kvmgr.h" #include "objmgr.h" #include "plugin.h" @@ -29,7 +30,37 @@ static void* sched_job_picker(void* arg) static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid) { - return NULL; + xce_sched_job_t* job = sched_get_job(sched, jobid); + xce_sched_plugin_t* plugin = NULL; + xce_job_alloc_t* alloc = NULL; + + if (job == NULL) { + ERROR("%s job %lu can not be found", __func__, jobid); + return NULL; + } + + if (job->allocator == NULL) { + ERROR("%s job %lu has invalid allocator", __func__, jobid); + return NULL; + } + + plugin = (xce_sched_plugin_t*)job->allocator; + if (plugin->job_alloc_fn == NULL) { + ERROR("%s plugin %s has no allocation function", __func__, + plugin->name); + return NULL; + } + + alloc = plugin->job_alloc_fn(job->allocator, jobid); + if (alloc == NULL) { + return make_pend_decision(jobid); + } + + return make_dispatch_decision(alloc); +} + +static int commit_decision(xce_scheduler_t *sched, xce_decision_t *decision) { + return sched_commit_decision(sched, decision); } static void* sched_allocator(void* arg) @@ -54,6 +85,8 @@ static void* sched_allocator(void* arg) /* calculate decision for jobid */ decision = calc_decision(sched, jobid); + commit_decision(sched, decision); + cbuf_put(&(sched->decisions), &decision); } @@ -98,6 +131,7 @@ static void sched_pend_update_killer(data_item_t* it) static int sched_register_plugins(xce_scheduler_t* sched) { register_fcfs(sched); + register_default_allocator(sched); return 0; } @@ -366,3 +400,7 @@ xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name) int sched_job_is_ready(xce_sched_job_t* j) { 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) { + return 0; +}