From abea3375ca023df81e8d7253589d51f9e0ac7e87 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 22 Dec 2025 10:53:29 +0800 Subject: [PATCH] save --- src/lib/kvmgr.c | 8 + src/scheduler/include/plugin.h | 56 ++++++ src/scheduler/include/sched/types.h | 69 ++++++++ src/scheduler/include/scheduler.h | 68 +++---- src/scheduler/meson.build | 3 + src/scheduler/src/plugin.c | 188 ++++++++++++++++++++ src/scheduler/src/plugin_fcfs.c | 266 ++++++++++++++++++++++++++++ src/scheduler/src/schedobj.c | 131 ++++++++++++++ src/scheduler/src/scheduler.c | 122 ++++++++++++- 9 files changed, 866 insertions(+), 45 deletions(-) create mode 100644 src/scheduler/include/plugin.h create mode 100644 src/scheduler/include/sched/types.h create mode 100644 src/scheduler/src/plugin.c create mode 100644 src/scheduler/src/plugin_fcfs.c create mode 100644 src/scheduler/src/schedobj.c diff --git a/src/lib/kvmgr.c b/src/lib/kvmgr.c index 70444c7..cce8a96 100644 --- a/src/lib/kvmgr.c +++ b/src/lib/kvmgr.c @@ -50,6 +50,10 @@ static void* kvmgr_lock_(kvmgr_t* mgr, void* key, int op) return NULL; } + if (op == OBJREF_NOP) { + return ptr; + } + return obj_ref(ptr, op); } @@ -101,6 +105,10 @@ void* kvmgr_lock_byid(kvmgr_t* mgr, int idx, int op) return NULL; } + if (op == OBJREF_NOP) { + return ptr; + } + return obj_ref(ptr, op); } diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h new file mode 100644 index 0000000..60a0699 --- /dev/null +++ b/src/scheduler/include/plugin.h @@ -0,0 +1,56 @@ +#if !defined(_PLUGIN_H_) + +#define _PLUGIN_H_ 1 + +#include +#include + +#include "sched/types.h" + +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*); + +/* initialize function of plugin */ +typedef int (*plugin_init_fn_t)(void*); + +typedef void (*plugin_fini_fn_t)(void*); + +/* plugin function to elect job to be scheduler */ +typedef int64_t (*plugin_job_election_fn_t)(void*); + +/* allocation function of job */ +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_hooks_t; + +typedef struct { + char* name; + void* sched; + plugin_init_fn_t init_fn; + plugin_fini_fn_t fini_fn; + plugin_job_election_fn_t job_elect_fn; + plugin_job_allocate_fn_t job_alloc_fn; + plugin_hooks_t hooks; + plugin_data_t data; +} xce_sched_plugin_t; + +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 register_fcfs(void*); + +#endif diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h new file mode 100644 index 0000000..8bd282d --- /dev/null +++ b/src/scheduler/include/sched/types.h @@ -0,0 +1,69 @@ +#if !defined(_SCHED_TYPES_H_) + +#define _SCHED_TYPES_H_ 1 + +#include +#include + +#include "typemap.h" +#include "xce/types.h" + +typedef void* host_selection_t; + +typedef struct { + host_selection_t* select; + int nres; + uint32_t* resources; + float* amount; +} rank_req_t; + +typedef struct { + char* name; +} xce_sched_user_t; + +typedef struct { + char* name; +} xce_sched_project_t; + +typedef struct { + char* name; + uint32_t attrs; + uint32_t status; + int32_t priority; + int32_t accept_interval; + int32_t sched_delay; + char* resreq; +} xce_sched_queue_t; + +typedef struct { + uint64_t jobid; + time_t sub_time; + time_t queue_time; + time_t start_time; + unsigned int state; + int32_t priority; + xce_sched_queue_t* queue; + xce_sched_user_t* user; + xce_sched_project_t* project; + uint32_t mintasks; + uint32_t maxtasks; + xce_job_alloc_t* allocs; +} xce_sched_job_t; + +typedef struct { + char* name; + xce_val_t* res_avail; /* resources in worker */ + xce_val_t* res_in_use; /* used resources in worker */ +} xce_sched_worker_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 void kill_sched_job(void*); +extern void kill_sched_queue(void*); +extern void kill_sched_worker(void*); +extern void kill_sched_user(void*); +extern void kill_sched_project(void*); +#endif diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index 93a821e..acd3d5d 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -9,7 +9,9 @@ #include "client.h" #include "dataq.h" #include "decision.h" +#include "kvmgr.h" #include "mylock.h" +#include "plugin.h" #include "xce/types.h" #include "xlist.h" @@ -41,50 +43,17 @@ typedef struct { - offsetof(xce_sched_update_request_t, \ link))) -typedef void* host_selection_t; - -typedef struct { - host_selection_t* select; - int nres; - uint32_t* resources; - float* amount; -} rank_req_t; - -typedef struct { - uint64_t jobid; - time_t sub_time; - time_t queue_time; - time_t start_time; - unsigned int state; - char* queue; - char* user; - char* project; - uint32_t mintasks; - uint32_t maxtasks; - xce_job_alloc_t* allocs; -} xce_sched_job_t; - -typedef struct { - char* name; - uint32_t attrs; - uint32_t status; - int32_t priority; - int32_t accept_interval; - int32_t sched_delay; - char* resreq; -} xce_sched_queue_t; - -typedef struct { - char* name; - xce_val_t* res_avail; /* resources in worker */ - xce_val_t* res_in_use; /* used resources in worker */ -} xce_sched_worker_t; - typedef struct { mylock_t lock; int stop; xcecli_ctx_t apictx; + 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 */ @@ -92,10 +61,21 @@ typedef struct { 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 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 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*); +extern xce_sched_user_t* sched_get_user(xce_scheduler_t*, char*); +extern xce_sched_project_t* sched_get_project(xce_scheduler_t*, char*); +extern int sched_add_job(xce_scheduler_t*, xce_sched_job_t*); +extern int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*); +extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*); +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*); #endif diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 33ecc93..b471bdf 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -3,6 +3,9 @@ scheduler_srcs = [ 'src/handler.c', 'src/scheduler.c', 'src/decision.c', + 'src/schedobj.c', + 'src/plugin.c', + 'src/plugin_fcfs.c', ] incdir = include_directories('src', diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c new file mode 100644 index 0000000..e75aedd --- /dev/null +++ b/src/scheduler/src/plugin.c @@ -0,0 +1,188 @@ +#include +#include +#include +#include + +#include "plugin.h" +#include "scheduler.h" +#include "xlist.h" + +int plugin_register(void* arg1, void* arg2) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg1; + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + int rc = 0; + + if ((sched == NULL) && (plugin == NULL)) { + return -EINVAL; + } + + plugin->sched = sched; + if (plugin->init_fn != NULL) { + rc = plugin->init_fn(plugin); + } + + if (rc < 0) { + return rc; + } + + rc = kvmgr_add(&(sched->plugins), (void*)plugin->name, (void*)plugin); + + return rc; +} + +void plugin_unregister(void* arg) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + xce_scheduler_t* sched = NULL; + + if (plugin == NULL) { + return; + } + + sched = (xce_scheduler_t*)plugin->sched; + if (sched == NULL) { + return; + } + + kvmgr_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; + + if ((arg1 == NULL) || (plugin == NULL)) { + return 0; + } + + if (plugin->hooks.config_hook != NULL) { + return plugin->hooks.config_hook(plugin, arg1); + } + + return 0; +} + +int plugin_config_update(void* arg, xce_conf4sched_t* config) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xlist_t lst; + 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); + + xlist_fini(&lst); + + return rc; +} + +static int plugin_job_update_(void* arg1, void* arg2) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + + if ((arg1 == NULL) || (plugin == NULL)) { + return 0; + } + + if (plugin->hooks.job_hook != NULL) { + return plugin->hooks.job_hook(plugin, arg1); + } + + return 0; +} + +int plugin_job_update(void* arg, xce_sched_job_t* job) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xlist_t lst; + 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); + + xlist_fini(&lst); + + return rc; +} + +static int plugin_queue_update_(void* arg1, void* arg2) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + + if ((arg1 == NULL) || (plugin == NULL)) { + return 0; + } + + if (plugin->hooks.queue_hook != NULL) { + return plugin->hooks.queue_hook(plugin, arg1); + } + + return 0; +} + +int plugin_queue_update(void* arg, xce_sched_queue_t* queue) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xlist_t lst; + 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); + + xlist_fini(&lst); + + return rc; +} + +static int plugin_worker_update_(void* arg1, void* arg2) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; + + if ((arg1 == NULL) || (plugin == NULL)) { + return 0; + } + + if (plugin->hooks.worker_hook != NULL) { + return plugin->hooks.worker_hook(plugin, arg1); + } + + return 0; +} + +int plugin_worker_update(void* arg, xce_sched_worker_t* worker) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xlist_t lst; + 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); + + xlist_fini(&lst); + + return rc; +} diff --git a/src/scheduler/src/plugin_fcfs.c b/src/scheduler/src/plugin_fcfs.c new file mode 100644 index 0000000..3dde74c --- /dev/null +++ b/src/scheduler/src/plugin_fcfs.c @@ -0,0 +1,266 @@ +#include +#include +#include +#include + +#include "ordq.h" +#include "plugin.h" +#include "sched/types.h" +#include "scheduler.h" +#include "utils.h" + +typedef struct { + xce_sched_queue_t* queue; + ordq_t pendJobs; +} fcfs_queue_t; + +typedef struct { + xce_sched_job_t* job; +} fcfs_job_t; + +typedef struct { + ordq_t queues; +} fcfs_plugin_data_t; + +static void fcfs_job_killer(void* arg) +{ + if (arg == NULL) { + return; + } + + free(arg); +} + +static void fcfs_queue_killer(void* arg) +{ + fcfs_queue_t* q = (fcfs_queue_t*)arg; + + if (q == NULL) { + return; + } + + ordq_fini(&(q->pendJobs)); + q->queue = NULL; + + FREEUP(q); +} + +static int fcfs_queue_cmp(void* ctx, void* arg1, void* arg2) +{ + fcfs_queue_t* q1 = (fcfs_queue_t*)arg1; + fcfs_queue_t* q2 = (fcfs_queue_t*)arg2; + + if (q1->queue->priority > q2->queue->priority) { + return -1; + } else if (q1->queue->priority < q2->queue->priority) { + return 1; + } + + return 0; +} + +static int fcfs_job_cmp(void* ctx, void* arg1, void* arg2) +{ + fcfs_job_t* j1 = (fcfs_job_t*)arg1; + fcfs_job_t* j2 = (fcfs_job_t*)arg2; + + if (j1->job->priority > j2->job->priority) { + return -1; + } else if (j1->job->priority < j2->job->priority) { + return 1; + } else if (j1->job->queue_time < j2->job->queue_time) { + return -1; + } else if (j1->job->queue_time > j2->job->queue_time) { + return 1; + } + + return 0; +} + +static int fcfs_priv_data_init(fcfs_plugin_data_t* data) +{ + ordq_init(&(data->queues), (ordq_data_killer_t)fcfs_queue_killer, + (ordq_cmp_fn_t)fcfs_queue_cmp); + return 0; +} + +static int fcfs_update_config(void* arg, xce_conf4sched_t* config) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + fcfs_plugin_data_t* privdata = NULL; + + TRACE("%s ...", __func__); + + privdata = (fcfs_plugin_data_t*)plugin->data; + if (privdata == NULL) { + privdata = (fcfs_plugin_data_t*)calloc(1, sizeof(*privdata)); + if (privdata == NULL) { + return -ENOMEM; + } + + fcfs_priv_data_init(privdata); + plugin->data = privdata; + } + + return 0; +} + +static int skip_not_matched_queue(void* arg1, void* arg2) +{ + char* name = (char*)arg1; + fcfs_queue_t* q = (fcfs_queue_t*)arg2; + + if ((name == NULL) || (q == NULL)) { + return 1; + } + + return (strcmp(name, q->queue->name) != 0); +} + +static int fcfs_update_queue(void* arg, xce_sched_queue_t* q) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; + ordq_item_t* it; + fcfs_queue_t* queue = NULL; + + TRACE("%s queue %s...", __func__, q->name); + + privdata->queues.ctx = q->name; + it = ordq_get(&(privdata->queues), (ordq_skip_fn_t)skip_not_matched_queue); + if (it == NULL) { + queue = (fcfs_queue_t*)calloc(1, sizeof(*queue)); + if (queue == NULL) { + return -ENOMEM; + } + + queue->queue = q; + ordq_init(&(queue->pendJobs), (ordq_data_killer_t)fcfs_job_killer, + (ordq_cmp_fn_t)fcfs_job_cmp); + } else { + queue = (fcfs_queue_t*)it->data; + free(it); + } + + ordq_put(&(privdata->queues), (void*)queue); + + return 0; +} + +static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name) +{ + ordq_item_t* it = NULL; + + data->queues.ctx = name; + it = ordq_peek(&(data->queues), (ordq_skip_fn_t)skip_not_matched_queue); + if (it == NULL) { + return NULL; + } + + return (fcfs_queue_t*)it->data; +} + +static int fcfs_update_job(void* arg, xce_sched_job_t* j) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; + fcfs_queue_t* q = NULL; + fcfs_job_t* job = NULL; + + TRACE("%s...", __func__); + if (privdata == NULL) { + return -EINVAL; + } + + if ((j->state & JOB_STAT_PEND) == 0) { + /* skip non pend job */ + return 0; + } + + q = fcfs_get_queue(privdata, j->queue->name); + if (q == NULL) { + return -EINVAL; + } + + job = (fcfs_job_t*)calloc(1, sizeof(*job)); + if (job == NULL) { + return -ENOMEM; + } + + job->job = j; + + ordq_put(&(q->pendJobs), job); + + return 0; +} + +static int skip_no_pend_job_queue(void* arg1, void* arg2) +{ + fcfs_queue_t* q = (fcfs_queue_t*)arg2; + + if (q == NULL) { + return 1; + } + + return (ordq_peek(&(q->pendJobs), NULL) == NULL); +} + +static int skip_not_ready_job(void* arg1, void* arg2) +{ + fcfs_job_t* job = (fcfs_job_t*)arg2; + + return !sched_job_is_ready(job->job); +} + +static int64_t fcfs_elect_job(void* arg) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; + ordq_item_t* it = NULL; + fcfs_queue_t* q = NULL; + fcfs_job_t* job = NULL; + uint64_t jobid; + + TRACE("%s ...", __func__); + + it = ordq_peek(&(privdata->queues), (ordq_skip_fn_t)skip_no_pend_job_queue); + if (it == NULL) { + TRACE("%s no job to be elected", __func__); + + return 0; + } + + q = (fcfs_queue_t*)it->data; + it = ordq_get(&(q->pendJobs), skip_not_ready_job); + if (it == NULL) { + TRACE("%s no job to be elected", __func__); + + return 0; + } + + job = (fcfs_job_t*)it->data; + jobid = job->job->jobid; + free(job); + free(it); + + if (1) { + char idstring[128]; + + fmt_jobid(idstring, sizeof(idstring) - 1, jobid); + TRACE("%s job %s has been elected", __func__, idstring); + } + return jobid; +} + +static xce_sched_plugin_t fcfs_plugin = { + .name = "xce_sched_fcfs", + .hooks = { .config_hook = fcfs_update_config, + .queue_hook = fcfs_update_queue, + .job_hook = fcfs_update_job, + .worker_hook = NULL }, + .job_elect_fn = fcfs_elect_job, + .job_alloc_fn = NULL, + .data = NULL +}; + +int register_fcfs(void* sched) { return plugin_register(sched, &fcfs_plugin); } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c new file mode 100644 index 0000000..fc996d4 --- /dev/null +++ b/src/scheduler/src/schedobj.c @@ -0,0 +1,131 @@ +#include +#include + +#include "sched/types.h" +#include "scheduler.h" +#include "xce/types.h" + +void kill_sched_job(void* arg) +{ + if (arg == NULL) { + return; + } +} + +xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j) +{ + xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + + if ((sched == NULL) || (job == NULL)) { + return NULL; + } + + job->jobid = j->jobid; + job->sub_time = j->sub_time; + job->queue_time = j->queue_time; + job->start_time = j->start_time; + job->state = j->state; + job->user = sched_get_user(sched, j->submit->user); + job->queue = sched_get_queue(sched, j->submit->queue); + job->project = sched_get_project(sched, j->submit->project_name); + job->mintasks = j->submit->mintasks; + job->maxtasks = j->submit->maxtasks; + job->priority = 0; /* FIXME: support job level priority */ + + sched_add_job(sched, job); + + return job; +} + +void kill_sched_queue(void* arg) +{ + if (arg == NULL) { + return; + } +} + +xce_sched_queue_t* make_sched_queue(void* arg, xce_queue_t* q) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xce_sched_queue_t* queue = NULL; + + if ((sched == NULL) || (q == NULL)) { + return NULL; + } + + queue = (xce_sched_queue_t*)calloc(1, sizeof(*queue)); + if (queue == NULL) { + return NULL; + } + + queue->name = strdup(q->queue); + queue->attrs = q->attrs; + queue->status = q->status; + queue->priority = q->priority; + queue->accept_interval = q->accept_interval; + queue->sched_delay = q->sched_delay; + + sched_add_queue(sched, queue); + + return queue; +} + +void kill_sched_worker(void* arg) +{ + if (arg == NULL) { + return; + } +} + +xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xce_sched_worker_t* worker = NULL; + + if ((sched == NULL) || (w == NULL)) { + return NULL; + } + + worker = (xce_sched_worker_t*)calloc(1, sizeof(*worker)); + if (worker == NULL) { + return NULL; + } + + worker->name = strdup(w->name); + /* FIXME: */ + + sched_add_worker(sched, worker); + + return worker; +} + +void kill_sched_user(void* arg) +{ + if (arg == NULL) { + return; + } +} + +xce_sched_user_t* make_sched_user(void* arg, char* name) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xce_sched_user_t* user = NULL; + + return user; +} + +void kill_sched_project(void* arg) +{ + if (arg == NULL) { + return; + } +} + +xce_sched_project_t* make_sched_project(void* arg, char* name) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xce_sched_project_t* project = NULL; + + return project; +} diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 93ff8e7..7796e41 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -7,12 +7,32 @@ #include "client.h" #include "controller.capnp.h" +#include "kvmgr.h" +#include "objmgr.h" +#include "plugin.h" #include "scheduler.h" #include "utils.h" static uint64_t pick_job(xce_scheduler_t* sched) { - uint64_t jobid = 0; + static xce_sched_plugin_t* job_picker = NULL; + uint64_t jobid = 0; + + if (job_picker == NULL) { + char* job_picker_plugin = "xce_sched_fcfs"; + + /* set job_pick_plugin according configuration */ + job_picker + = kvmgr_lock(&(sched->plugins), job_picker_plugin, OBJREF_NOP); + if ((job_picker == NULL) || (job_picker->job_elect_fn == NULL)) { + /* something wrong */ + return 0; + } + } + + if (job_picker->job_elect_fn != NULL) { + jobid = job_picker->job_elect_fn(job_picker); + } return jobid; } @@ -111,6 +131,13 @@ static void sched_pend_update_killer(data_item_t* it) free(req); } +static int sched_register_plugins(xce_scheduler_t* sched) +{ + register_fcfs(sched); + + return 0; +} + int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, xce_request_handler_t handler) { @@ -135,6 +162,21 @@ 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); + kvmgr_init(&(sched->plugins), string_hash_key, 8, 8, + (zarray_item_killer_t)plugin_unregister); + 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, + (zarray_item_killer_t)kill_sched_queue); + kvmgr_init(&(sched->workers), string_hash_key, 256, 256, + (zarray_item_killer_t)kill_sched_worker); + kvmgr_init(&(sched->users), string_hash_key, 256, 256, + (zarray_item_killer_t)kill_sched_user); + kvmgr_init(&(sched->projects), string_hash_key, 8, 8, + (zarray_item_killer_t)kill_sched_project); + + sched_register_plugins(sched); + rc = xcecli_ctx_init(&(sched->apictx), config); if (rc < 0) { return rc; @@ -194,18 +236,44 @@ void sched_fini(xce_scheduler_t* sched) static int sched_update_config(xce_scheduler_t* sched, xce_conf4sched_t* config) { + int i; + + plugin_config_update(sched, config); + + for (i = 0; i < config->n_queues; i++) { + xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]); + + plugin_queue_update(sched, q); + } + return 0; } static int sched_update_jobs(xce_scheduler_t* sched, int njobs, xce_job_t** jobs) { + int i; + + for (i = 0; i < njobs; i++) { + xce_sched_job_t* j = make_sched_job(sched, jobs[i]); + + plugin_job_update(sched, j); + } + return 0; } static int sched_update_workers(xce_scheduler_t* sched, int nworkers, xce_worker_t** workers) { + int i; + + for (i = 0; i < nworkers; i++) { + xce_sched_worker_t* w = make_sched_worker(sched, workers[i]); + + plugin_worker_update(sched, w); + } + return 0; } @@ -266,3 +334,55 @@ void sched_go(xce_scheduler_t* sched) return; } + +int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j) +{ + return kvmgr_add(&(sched->jobs), (void*)j->jobid, j); +} + +xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid) +{ + return kvmgr_lock(&(sched->jobs), (void*)jobid, OBJREF_NOP); +} + +int sched_add_queue(xce_scheduler_t* sched, xce_sched_queue_t* q) +{ + return kvmgr_add(&(sched->queues), q->name, q); +} + +xce_sched_queue_t* sched_get_queue(xce_scheduler_t* sched, char* name) +{ + return kvmgr_lock(&(sched->queues), (void*)name, OBJREF_NOP); +} + +int sched_add_worker(xce_scheduler_t* sched, xce_sched_worker_t* w) +{ + return kvmgr_add(&(sched->workers), w->name, w); +} + +xce_sched_worker_t* sched_get_worker(xce_scheduler_t* sched, char* name) +{ + return kvmgr_lock(&(sched->workers), (void*)name, OBJREF_NOP); +} + +int sched_add_user(xce_scheduler_t* sched, xce_sched_user_t* u) +{ + return kvmgr_add(&(sched->users), u->name, u); +} + +xce_sched_user_t* sched_get_user(xce_scheduler_t* sched, char* name) +{ + return kvmgr_lock(&(sched->users), (void*)name, OBJREF_NOP); +} + +int sched_add_project(xce_scheduler_t* sched, xce_sched_project_t* p) +{ + return kvmgr_add(&(sched->projects), p->name, p); +} + +xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name) +{ + return kvmgr_lock(&(sched->projects), (void*)name, OBJREF_NOP); +} + +int sched_job_is_ready(xce_sched_job_t* j) { return 1; }