From 06209930f91a112ed9651a6c6d324d5c5a131520 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 15 Jan 2026 17:56:53 +0800 Subject: [PATCH] improve --- src/include/xvec.h | 39 +++ src/lib/meson.build | 1 + src/lib/xvec.c | 270 +++++++++++++++++ src/scheduler/include/plugin.h | 23 +- src/scheduler/include/sched/types.h | 10 +- src/scheduler/include/scheduler.h | 45 +-- src/scheduler/src/plugin.c | 47 +-- src/scheduler/src/plugins/fcfs.c | 10 +- src/scheduler/src/plugins/limit.c | 2 +- src/scheduler/src/plugins/simple_allocator.c | 27 +- src/scheduler/src/resource.c | 296 +++++++++++-------- src/scheduler/src/schedobj.c | 9 +- src/scheduler/src/scheduler.c | 35 +-- 13 files changed, 573 insertions(+), 241 deletions(-) create mode 100644 src/include/xvec.h create mode 100644 src/lib/xvec.c diff --git a/src/include/xvec.h b/src/include/xvec.h new file mode 100644 index 00000000..51ae10aa --- /dev/null +++ b/src/include/xvec.h @@ -0,0 +1,39 @@ +#if !defined(_XVEC_H_) +#define _XVEC_H_ 1 + +#include +#include +#include + +typedef enum { + XVEC_ADD_INT = 0, + XVEC_SUB_INT, + XVEC_ADD_FLOAT, + XVEC_SUB_FLOAT +} xvec_op_t; + +typedef struct { + int count; + int elsize; + void* data; +} xvec_t; + +typedef struct { + int count; + xvec_t idxs; + xvec_t vals; +} xvec_idval_t; + +extern int xvec_init(xvec_t*, int, int); +extern void xvec_fini(xvec_t*); +extern int xvec_zero(xvec_t*); +extern int xvec_resize(xvec_t*, int); +extern int xvec_op(xvec_t*, xvec_t*, xvec_op_t); +extern int xvec_dup(xvec_t**, xvec_t*); +extern int xvec_put(xvec_t*, int, void*); +extern int xvec_get(xvec_t*, int, void*); +extern int xvec_idval_init(xvec_idval_t*, int); +extern void xvec_idval_fini(xvec_idval_t*); +extern int xvec_idval_store(xvec_idval_t*, int, float); + +#endif diff --git a/src/lib/meson.build b/src/lib/meson.build index 1be21ce3..9b0c1106 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -35,6 +35,7 @@ libxce_src = [ gen_srcs[0], 'misc.c', 'group.c', 'xcelimit.c', + 'xvec.c', ] incdir = include_directories('../include') diff --git a/src/lib/xvec.c b/src/lib/xvec.c new file mode 100644 index 00000000..a671131b --- /dev/null +++ b/src/lib/xvec.c @@ -0,0 +1,270 @@ +#include +#include + +#include "utils.h" +#include "xvec.h" + +int xvec_init(xvec_t* v, int count, int elsize) +{ + if ((v == NULL) || (count <= 0) || (elsize <= 0)) { + return -EINVAL; + } + + v->data = (void*)calloc(count, elsize); + if (v->data == NULL) { + return -ENOMEM; + } + + v->count = count; + v->elsize = elsize; + + return v->count; +} + +void xvec_fini(xvec_t* v) +{ + if (v == NULL) { + return; + } + + FREEUP(v->data); + memset(v, 0x0, sizeof(*v)); +} + +int xvec_zero(xvec_t* v) +{ + if (v == NULL) { + return -EINVAL; + } + + if (v->data == NULL) { + return -EINVAL; + } + + memset(v->data, 0x0, v->count * v->elsize); + + return 1; +} + +int xvec_resize(xvec_t* v, int new_size) +{ + void* p = NULL; + + if (v == NULL) { + return -EINVAL; + } + + if (new_size <= v->count) { + /* do thing for shrink now */ + return 0; + } + + p = (void*)realloc(v->data, new_size * v->elsize); + if (p == NULL) { + return -ENOMEM; + } + + v->data = p; + /* zero fill the extended space */ + memset(((char*)p) + v->count * v->elsize, 0x0, + (new_size - v->count) * v->elsize); + v->count = new_size; + + return 0; +} + +#define X_OP(op) X_OP_##op +#define X_OP_add += +#define X_OP_sub -= + +#define DEFINE_VEC_OP(op, typ) \ + static inline int vec_##typ##_##op(typ* v1, typ* v2, int count) \ + { \ + int i; \ + for (i = 0; i < count; i++) { \ + v1[i] X_OP(op) v2[i]; \ + } \ + return 0; \ + } + +DEFINE_VEC_OP(add, int) +DEFINE_VEC_OP(sub, int) +DEFINE_VEC_OP(add, float) +DEFINE_VEC_OP(sub, float) + +int xvec_op(xvec_t* v1, xvec_t* v2, xvec_op_t op) +{ + int rc = 0; + + if ((v1 == NULL) || (v2 == NULL)) { + return -EINVAL; + } + + if (v1->elsize != v2->elsize) { + return -EINVAL; + } + + switch (op) { + case XVEC_ADD_INT: + case XVEC_SUB_INT: + if (v1->elsize != sizeof(int)) { + return -EINVAL; + } + break; + case XVEC_ADD_FLOAT: + case XVEC_SUB_FLOAT: + if (v1->elsize != sizeof(float)) { + return -EINVAL; + } + break; + } + + if (v1->count < v2->count) { + rc = xvec_resize(v1, v2->count); + + if (rc < 0) { + return rc; + } + } + + switch (op) { + case XVEC_ADD_INT: + rc = vec_int_add((int*)v1->data, (int*)v2->data, v2->count); + break; + case XVEC_SUB_INT: + rc = vec_int_sub((int*)v1->data, (int*)v2->data, v2->count); + break; + case XVEC_ADD_FLOAT: + rc = vec_float_add((float*)v1->data, (float*)v2->data, v2->count); + break; + case XVEC_SUB_FLOAT: + rc = vec_float_sub((float*)v1->data, (float*)v2->data, v2->count); + break; + } + + return rc; +} + +int xvec_dup(xvec_t** dst, xvec_t* src) +{ + xvec_t* v = NULL; + + v = (xvec_t*)calloc(1, sizeof(*v)); + *dst = v; + + if (v == NULL) { + return -ENOMEM; + } + + v->data = (void*)calloc(src->count, src->elsize); + if (v->data == NULL) { + return -ENOMEM; + } + + memcpy(v->data, src->data, src->elsize * src->count); + v->elsize = src->elsize; + v->count = src->count; + + return v->count; +} + +int xvec_put(xvec_t* v, int idx, void* data) +{ + int rc; + char* p; + + if ((v == NULL) || (idx < 0) || (data == NULL)) { + return -EINVAL; + } + + rc = xvec_resize(v, idx + 1); + if (rc < 0) { + return rc; + } + + p = (char*)v->data; + memcpy(p + v->elsize * idx, data, v->elsize); + + return 1; +} + +int xvec_get(xvec_t* v, int idx, void* data) +{ + char* p = NULL; + + if ((v == NULL) || (idx < 0) || (data == NULL)) { + return -EINVAL; + } + + if (idx >= v->count) { + memset(data, 0x0, v->elsize); + return 0; + } + + p = (char*)v->data; + memcpy(data, p + v->elsize * idx, v->elsize); + + return 1; +} + +int xvec_idval_init(xvec_idval_t* v, int initsize) +{ + int rc; + + if ((v == NULL) || (initsize <= 0)) { + return -EINVAL; + } + + memset(v, 0x0, sizeof(*v)); + rc = xvec_init(&(v->idxs), initsize, sizeof(int)); + if (rc < 0) { + return rc; + } + + rc = xvec_init(&(v->vals), initsize, sizeof(float)); + if (rc < 0) { + return rc; + } + + v->count = 0; + + return 0; +} + +void xvec_idval_fini(xvec_idval_t* v) +{ + if (v == NULL) { + return; + } + + xvec_fini(&(v->idxs)); + xvec_fini(&(v->vals)); + + memset(v, 0x0, sizeof(*v)); + return; +} + +int xvec_idval_store(xvec_idval_t* v, int idx, float val) +{ + int rc; + int pos; + + if ((v == NULL) || (idx < 0)) { + return -EINVAL; + } + + pos = v->count; + rc = xvec_put(&(v->idxs), pos, &idx); + if (rc < 0) { + return rc; + } + + rc = xvec_put(&(v->vals), pos, &val); + if (rc < 0) { + return rc; + } + + v->count = pos + 1; + + return v->count; +} diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 3485f287..e0d64494 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -11,14 +11,10 @@ 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*, - 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*, - xce_sched_job_t*); -typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*, - xce_sched_worker_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_job_state_change_hook_t)(void*, xce_sched_job_t*, uint32_t); typedef int (*plugin_commit_decision_hook_t)(void*, xce_decision_t*); @@ -40,7 +36,7 @@ typedef int (*plugin_begin_session_fn_t)(void*); typedef int (*plugin_end_session_fn_t)(void*); /* allocation function of job */ -typedef int (*plugin_job_allocate_fn_t)(void*,xce_sched_ctx_t*); +typedef int (*plugin_job_allocate_fn_t)(void*, xce_sched_ctx_t*); typedef struct { plugin_update_config_hook_t config_hook; @@ -69,11 +65,10 @@ typedef struct { extern int plugin_register(void*, void*); extern void plugin_unregister(void*); -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*, - xce_sched_worker_t*); +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_job_state_change(void*, xce_sched_job_t*, uint32_t); extern int plugin_begin_session(void*); extern int plugin_end_session(void*); diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 942b12bd..c0804e86 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -9,18 +9,15 @@ #include "typemap.h" #include "xce/types.h" #include "xlist.h" +#include "xvec.h" #include "zarray.h" /* FIXME: improve later */ typedef void* host_selection_t; typedef struct { - int nres_shared; - int* idxs_shared; - float* vals_shared; - int nres_host; - int* idxs_host; - float* vals_host; + xvec_idval_t shared; + xvec_idval_t host; } xce_sched_alloc_req_t; typedef enum { @@ -100,6 +97,7 @@ typedef struct { resreq_t* resreq; xce_job_alloc_t* alloc; xce_pendreason_t* reason; + xvec_idval_t unmeet; xlist_t extra_res_requests; xlist_t extra_host_res_requests; xlist_t limits; /* non per host limits */ diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index e6c1528b..f840c6c4 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -15,6 +15,7 @@ #include "plugin.h" #include "xce/types.h" #include "xlist.h" +#include "xvec.h" typedef void* (*sched_group_worker_fn_t)(void*, xce_sched_worker_t*); @@ -96,28 +97,28 @@ 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*); -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*); -extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*, - host_selection_t*); -extern int sched_check_shared_satisfy(xce_scheduler_t*, int, int*, float*, int*, - int*, float*); -extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*, int, - int*, float*, int*, int*, float*); -extern int sched_collect_workers(xce_scheduler_t*, xlist_t*, - zarray_iter_skip_fn_t, - sched_group_worker_fn_t); -extern int sched_charge_account(xce_scheduler_t*, xce_job_account_t*); -extern int sched_uncharge_account(xce_scheduler_t*, xce_job_account_t*); -extern int sched_commit_job_alloc(xce_scheduler_t*, xce_sched_job_t*); -extern uint64_t sched_get_id(xce_scheduler_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*); +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*); +extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*, + host_selection_t*); +extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_idval_t*, + xvec_idval_t*); +extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*, + xvec_idval_t*, xvec_idval_t*); +extern int sched_collect_workers(xce_scheduler_t*, xlist_t*, + zarray_iter_skip_fn_t, + sched_group_worker_fn_t); +extern int sched_charge_account(xce_scheduler_t*, xce_job_account_t*); +extern int sched_uncharge_account(xce_scheduler_t*, xce_job_account_t*); +extern int sched_commit_job_alloc(xce_scheduler_t*, xce_sched_job_t*); +extern uint64_t sched_get_id(xce_scheduler_t*); extern xce_alloc_t** sched_make_allocs(xce_scheduler_t*, char*, int, int*, float*, int*); extern int sched_ctx_init(xce_sched_ctx_t*, xce_sched_job_t*); diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 79dcd960..28785d4a 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -7,11 +7,6 @@ #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; @@ -62,26 +57,24 @@ void plugin_unregister(void* arg) static int plugin_config_update_(void* arg1, void* arg2) { xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2; - update_args_t* args = (update_args_t*)arg1; + xce_conf4sched_t* config = (xce_conf4sched_t*)arg1; if ((arg1 == NULL) || (plugin == NULL)) { return 0; } if (plugin->hooks.config_hook != NULL) { - return plugin->hooks.config_hook(plugin, args->old, args->new); + return plugin->hooks.config_hook(plugin, config); } return 0; } -int plugin_config_update(void* arg, xce_conf4sched_t* old, - xce_conf4sched_t* config) +int plugin_config_update(void* arg, xce_conf4sched_t* config) { xce_scheduler_t* sched = (xce_scheduler_t*)arg; xlist_t lst; int rc; - update_args_t args = { .old = old, .new = config }; if (sched == NULL) { return -EINVAL; @@ -89,7 +82,7 @@ int plugin_config_update(void* arg, xce_conf4sched_t* old, xlist_init(&lst, 3); xlist_store(&lst, plugin_config_update_); - rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, config); xlist_fini(&lst); @@ -99,26 +92,24 @@ int plugin_config_update(void* arg, xce_conf4sched_t* old, 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; + xce_sched_job_t* job = (xce_sched_job_t*)arg1; if ((arg1 == NULL) || (plugin == NULL)) { return 0; } if (plugin->hooks.job_hook != NULL) { - return plugin->hooks.job_hook(plugin, (xce_sched_job_t*)args->old, - (xce_sched_job_t*)args->new); + return plugin->hooks.job_hook(plugin, job); } return 0; } -int plugin_job_update(void* arg, xce_sched_job_t* jold, xce_sched_job_t* job) +int plugin_job_update(void* arg, 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; @@ -126,7 +117,7 @@ int plugin_job_update(void* arg, xce_sched_job_t* jold, xce_sched_job_t* job) xlist_init(&lst, 3); xlist_store(&lst, plugin_job_update_); - rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, job); xlist_fini(&lst); @@ -136,27 +127,24 @@ int plugin_job_update(void* arg, xce_sched_job_t* jold, 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; + xce_sched_queue_t* queue = (xce_sched_queue_t*)arg1; if ((arg1 == NULL) || (plugin == NULL)) { return 0; } if (plugin->hooks.queue_hook != NULL) { - return plugin->hooks.queue_hook(plugin, (xce_sched_queue_t*)args->old, - (xce_sched_queue_t*)args->new); + return plugin->hooks.queue_hook(plugin, queue); } return 0; } -int plugin_queue_update(void* arg, xce_sched_queue_t* qold, - xce_sched_queue_t* queue) +int plugin_queue_update(void* arg, 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; @@ -164,7 +152,7 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* qold, xlist_init(&lst, 3); xlist_store(&lst, plugin_queue_update_); - rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, queue); xlist_fini(&lst); @@ -174,27 +162,24 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* qold, 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; + xce_sched_worker_t* worker = (xce_sched_worker_t*)arg1; if ((arg1 == NULL) || (plugin == NULL)) { return 0; } if (plugin->hooks.worker_hook != NULL) { - return plugin->hooks.worker_hook(plugin, (xce_sched_worker_t*)args->old, - (xce_sched_worker_t*)args->new); + return plugin->hooks.worker_hook(plugin, worker); } return 0; } -int plugin_worker_update(void* arg, xce_sched_worker_t* wold, - xce_sched_worker_t* worker) +int plugin_worker_update(void* arg, 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; @@ -202,7 +187,7 @@ int plugin_worker_update(void* arg, xce_sched_worker_t* wold, xlist_init(&lst, 3); xlist_store(&lst, plugin_worker_update_); - rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, worker); xlist_fini(&lst); diff --git a/src/scheduler/src/plugins/fcfs.c b/src/scheduler/src/plugins/fcfs.c index 63b62ae9..8cf314e3 100644 --- a/src/scheduler/src/plugins/fcfs.c +++ b/src/scheduler/src/plugins/fcfs.c @@ -86,8 +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_conf4sched_t* old, - xce_conf4sched_t* config) +static int fcfs_update_config(xce_sched_plugin_t* plugin, + xce_conf4sched_t* config) { fcfs_plugin_data_t* privdata = NULL; @@ -124,8 +124,7 @@ 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* qold, xce_sched_queue_t* q) +static int fcfs_update_queue(xce_sched_plugin_t* plugin, xce_sched_queue_t* q) { fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data; ordq_item_t* it; @@ -165,8 +164,7 @@ 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* jold, - xce_sched_job_t* j) +static int fcfs_update_job(xce_sched_plugin_t* plugin, 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/plugins/limit.c b/src/scheduler/src/plugins/limit.c index 3aed17c2..91c5881f 100644 --- a/src/scheduler/src/plugins/limit.c +++ b/src/scheduler/src/plugins/limit.c @@ -27,7 +27,7 @@ static int limit_priv_data_init(limit_plugin_data_t* data) } static int limit_update_config(xce_sched_plugin_t* plugin, - xce_conf4sched_t* old, xce_conf4sched_t* config) + xce_conf4sched_t* config) { limit_plugin_data_t* privdata = NULL; xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; diff --git a/src/scheduler/src/plugins/simple_allocator.c b/src/scheduler/src/plugins/simple_allocator.c index f8664a1a..5fdb3b4a 100644 --- a/src/scheduler/src/plugins/simple_allocator.c +++ b/src/scheduler/src/plugins/simple_allocator.c @@ -46,9 +46,6 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; xce_sched_job_t* job = ctx->job; resreq_t* resreq = ctx->resreq; - int n_unmeet; - int unmeet_idxs[ctx->alloc_req.nres_shared + 1]; - float unmeet_vals[ctx->alloc_req.nres_shared + 1]; int rc; xlist_t workers; xce_pendreason_t* reason = NULL; @@ -59,9 +56,8 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) return -ENOMEM; } - rc = sched_check_shared_satisfy( - sched, ctx->alloc_req.nres_shared, ctx->alloc_req.idxs_shared, - ctx->alloc_req.vals_shared, &n_unmeet, unmeet_idxs, unmeet_vals); + rc = sched_check_shared_satisfy(sched, &(ctx->alloc_req.shared), + &(ctx->unmeet)); if (rc < 0) { return rc; } @@ -69,9 +65,9 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) if (rc == 0) { int i; - for (i = 0; i < n_unmeet; i++) { - xce_resource_t* r = resmgr_lock_byid(&(sched->resmgr), - unmeet_idxs[i], OBJREF_NOP); + for (i = 0; i < ctx->alloc_req.shared.count; i++) { + xce_resource_t* r = resmgr_lock_byid( + &(sched->resmgr), ((int*)ctx->unmeet.idxs.data)[i], OBJREF_NOP); add_job_preason(reason->reasons, PREASON_NO_RESOURCE, NULL, r->name); } @@ -90,12 +86,13 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) xlist_find(&workers, (xlist_find_fn_t)sched_try_alloc_on_hosts, sched, ctx); if (ctx->alloc != NULL) { - if (ctx->alloc_req.nres_shared > 0) { - ctx->alloc->allocs_shared = sched_make_allocs( - sched, "", ctx->alloc_req.nres_shared, - ctx->alloc_req.idxs_shared, ctx->alloc_req.vals_shared, - /*FIXME: no duration now*/ NULL); - ctx->alloc->nres_shared = ctx->alloc_req.nres_shared; + if (ctx->alloc_req.shared.count > 0) { + ctx->alloc->allocs_shared + = sched_make_allocs(sched, "", ctx->alloc_req.shared.count, + ctx->alloc_req.shared.idxs.data, + ctx->alloc_req.shared.vals.data, + /*FIXME: no duration now*/ NULL); + ctx->alloc->nres_shared = ctx->alloc_req.shared.count; } else { ctx->alloc->allocs_shared = NULL; ctx->alloc->nres_shared = 0; diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index a799fd8a..449f8c74 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -28,6 +28,8 @@ int get_resvals(zarray_t* array, int nres, int* idxs, float* vals) return 0; } + memset(vals, 0x0, sizeof(*vals) * nres); + rc = zarray_get_ex(array, nres, idxs, (void**)xvals); if (rc < 0) { return rc; @@ -35,10 +37,14 @@ int get_resvals(zarray_t* array, int nres, int* idxs, float* vals) for (i = 0; i < nres; i++) { if (xvals[i] == NULL) { - vals[i] = 0.0; - } else { - vals[i] = xvals[i]->u.fvalue; + continue; } + + if (xvals[i]->which != Value_u_fvalue) { + continue; + } + + vals[i] = xvals[i]->u.fvalue; } return nres; @@ -87,8 +93,8 @@ int put_resvals(zarray_t* array, int nres, int* idxs, float* vals) return nres; } -static int unsatisfy(int nres, int* idxs, float* total, float* inuse, - float* wanted, int* unmeet_idxs, float* unmeet_vals) +static int unsatisfy_(int nres, int* idxs, float* total, float* inuse, + float* wanted, int* unmeet_idxs, float* unmeet_vals) { int i, j; int count; @@ -123,73 +129,148 @@ int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w, return 1; } -int sched_check_shared_satisfy(xce_scheduler_t* sched, int nres, int* idxs, - float* wanted, int* n_unmeet, int* unmeet_idxs, - float* unmeet_vals) +static int sched_check_shared_satisfy_(xce_scheduler_t* sched, xvec_t* total, + xvec_t* inuse, xvec_idval_t* wanted, + xvec_idval_t* unmeet) { - float total[nres]; - float inuse[nres]; - int rc; + int nres = wanted->count; + int rc; - *n_unmeet = 0; - - if (nres == 0) { - return 1; - } - - rc = get_resvals(&(sched->sharedInst), nres, idxs, total); + rc = xvec_init(total, nres, sizeof(float)); if (rc < 0) { return rc; } - rc = get_resvals(&(sched->sharedInUse), nres, idxs, inuse); + rc = xvec_init(inuse, nres, sizeof(float)); if (rc < 0) { return rc; } - rc = unsatisfy(nres, idxs, total, inuse, wanted, unmeet_idxs, unmeet_vals); + rc = xvec_idval_init(unmeet, nres); + if (rc < 0) { + return rc; + } + + rc = get_resvals(&(sched->sharedInst), nres, (int*)wanted->idxs.data, + (float*)total->data); + if (rc < 0) { + return rc; + } + + rc = get_resvals(&(sched->sharedInUse), nres, (int*)wanted->idxs.data, + (float*)inuse->data); + if (rc < 0) { + return rc; + } + + rc = unsatisfy_(nres, (int*)wanted->idxs.data, (float*)total->data, + (float*)inuse->data, (float*)wanted->vals.data, + (int*)unmeet->idxs.data, (float*)unmeet->vals.data); if (rc < 0) { return rc; } if (rc > 0) { - *n_unmeet = rc; + unmeet->count = rc; + return 0; } return 1; } -int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w, - int nres, int* idxs, float* wanted, int* n_unmeet, - int* unmeet_idxs, float* unmeet_vals) +int sched_check_shared_satisfy(xce_scheduler_t* sched, xvec_idval_t* wanted, + xvec_idval_t* unmeet) { - float total[nres]; - float inuse[nres]; - int rc; + xvec_t total; + xvec_t inuse; + int rc; - rc = get_resvals(&(w->resvals), nres, idxs, total); + if (wanted->count == 0) { + return 1; + } + + rc = sched_check_shared_satisfy_(sched, &total, &inuse, wanted, unmeet); + if (rc < 0) { + xvec_idval_fini(unmeet); + } + + xvec_fini(&total); + xvec_fini(&inuse); + + return rc; +} + +static int sched_check_host_satisfy_(xce_scheduler_t* sched, + xce_sched_worker_t* w, xvec_t* total, + xvec_t* inuse, xvec_idval_t* wanted, + xvec_idval_t* unmeet) +{ + int nres = wanted->count; + int rc = 0; + + rc = xvec_init(total, nres, sizeof(float)); if (rc < 0) { return rc; } - rc = get_resvals(&(w->resInUse), nres, idxs, inuse); + rc = xvec_init(inuse, nres, sizeof(float)); if (rc < 0) { return rc; } - rc = unsatisfy(nres, idxs, total, inuse, wanted, unmeet_idxs, unmeet_vals); + rc = xvec_idval_init(unmeet, nres); + if (rc < 0) { + return rc; + } + + rc = get_resvals(&(w->resvals), nres, (int*)wanted->idxs.data, + (float*)total->data); + if (rc < 0) { + return rc; + } + + rc = get_resvals(&(w->resInUse), nres, (int*)wanted->idxs.data, + (float*)inuse->data); + if (rc < 0) { + return rc; + } + + rc = unsatisfy_(nres, (int*)wanted->idxs.data, (float*)total->data, + (float*)inuse->data, (float*)wanted->vals.data, + (int*)unmeet->idxs.data, (float*)unmeet->vals.data); if (rc < 0) { return rc; } if (rc > 0) { - *n_unmeet = rc; + unmeet->count = rc; return 0; } return 1; } +int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w, + xvec_idval_t* wanted, xvec_idval_t* unmeet) +{ + xvec_t total; + xvec_t inuse; + int rc; + + if (wanted->count == 0) { + return 1; + } + + rc = sched_check_host_satisfy_(sched, w, &total, &inuse, wanted, unmeet); + if (rc < 0) { + xvec_idval_fini(unmeet); + } + + xvec_fini(&total); + xvec_fini(&inuse); + + return rc; +} static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, int global, int naccounts, @@ -255,16 +336,11 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) { - xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*); - xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*); - xce_sched_job_t* job = ctx->job; - resreq_t* resreq = ctx->resreq; - int rc; - int unmeet_idxs[ctx->alloc_req.nres_host + ctx->limits.count - + ctx->limits_host.count + 1]; - float unmeet_vals[ctx->alloc_req.nres_host + ctx->limits.count - + ctx->limits_host.count + 1]; - int n_unmeet; + xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*); + xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*); + xce_sched_job_t* job = ctx->job; + resreq_t* resreq = ctx->resreq; + int rc; xce_host_alloc_t** hallocs = NULL; xce_host_alloc_t* halloc = NULL; xce_alloc_t** allocs = NULL; @@ -288,15 +364,19 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) sched_ctx_fill_host_requests(sched, ctx); - rc = sched_check_host_satisfy( - sched, w, ctx->alloc_req.nres_host, ctx->alloc_req.idxs_host, - ctx->alloc_req.vals_host, &n_unmeet, unmeet_idxs, unmeet_vals); - if (rc <= 0) { + rc = sched_check_host_satisfy(sched, w, &(ctx->alloc_req.host), + &(ctx->unmeet)); + if (rc < 0) { + /* FIXME: */ + return rc; + } + + if (rc == 0) { int i; - for (i = 0; i < n_unmeet; i++) { - xce_resource_t* r = resmgr_lock_byid(&(sched->resmgr), - unmeet_idxs[i], OBJREF_NOP); + for (i = 0; i < ctx->unmeet.count; i++) { + xce_resource_t* r = resmgr_lock_byid( + &(sched->resmgr), ((int*)ctx->unmeet.idxs.data)[i], OBJREF_NOP); add_job_preason(ctx->reason->reasons, PREASON_NO_RESOURCE, w->name, r->name); @@ -305,10 +385,10 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap) } allocs = sched_make_allocs( - sched, "", ctx->alloc_req.nres_host, ctx->alloc_req.idxs_host, - ctx->alloc_req.vals_host, /* FIXME: no duration now */ NULL); + sched, "", ctx->alloc_req.host.count, ctx->alloc_req.host.idxs.data, + ctx->alloc_req.host.vals.data, /* FIXME: no duration now */ NULL); - halloc = make_host_alloc(w->name, ctx->alloc_req.nres_host, allocs); + halloc = make_host_alloc(w->name, ctx->alloc_req.host.count, allocs); hallocs = make_host_allocs(1, halloc); ctx->alloc = make_job_alloc(job->jobid, 1, hallocs); @@ -370,100 +450,74 @@ int sched_commit_job_alloc(xce_scheduler_t* sched, xce_sched_job_t* job) return sched_charge_account(sched, job->accounts); } -int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) +static int fill_ctx_idval_vec_(xce_scheduler_t* sched, xvec_idval_t* v, + int nlists, xlist_t** lists) { - int i, count, j, k; - xlist_t* lists[] = { NULL, NULL, NULL }; + int count; + int i, rc; - count = ctx->resreq->u.simple.res_shared.count - + ctx->extra_res_requests.count; - ctx->alloc_req.idxs_shared = (int*)calloc(count + 1, sizeof(int)); - ctx->alloc_req.vals_shared = (float*)calloc(count + 1, sizeof(float)); - - if ((ctx->alloc_req.idxs_shared == NULL) - || (ctx->alloc_req.vals_shared == NULL)) { - FREEUP(ctx->alloc_req.idxs_shared); - FREEUP(ctx->alloc_req.vals_shared); - return -ENOMEM; + count = 0; + for (i = 0; i < nlists; i++) { + count += lists[i]->count; } - lists[0] = &(ctx->resreq->u.simple.res_shared); - lists[1] = &(ctx->extra_res_requests); + memset(v, 0x0, sizeof(*v)); + rc = xvec_idval_init(v, count); + if (rc < 0) { + return rc; + } - j = 0; + for (i = 0; i < nlists; i++) { + xlist_t* lst = lists[i]; + int j; - for (k = 0; lists[k] != NULL; k++) { - xlist_t* lst = lists[k]; - - for (i = 0; i < lst->count; i++) { - request_res_t* rq = xlist_get(lst, i); + for (j = 0; j < lst->count; j++) { + request_res_t* rq = (request_res_t*)xlist_get(lst, j); xce_resource_t* r = NULL; if (rq == NULL) { - continue; + return -EINVAL; } r = resmgr_lockres(&(sched->resmgr), rq->res, OBJREF_NOP); if (r == NULL) { - continue; + return -ENOENT; } - ctx->alloc_req.idxs_shared[j] = r->idx; - ctx->alloc_req.vals_shared[j] = rq->value; - j++; + rc = xvec_idval_store(v, r->idx, rq->value); + if (rc < 0) { + return rc; + } } } - ctx->alloc_req.nres_shared = j; + return v->count; +} - return 0; +int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) +{ + int rc = 0; + xlist_t* lists[] + = { &(ctx->resreq->u.simple.res_shared), &(ctx->extra_res_requests) }; + + rc = fill_ctx_idval_vec_(sched, &(ctx->alloc_req.shared), 2, &lists[0]); + if (rc < 0) { + xvec_idval_fini(&(ctx->alloc_req.shared)); + } + + return rc; } int sched_ctx_fill_host_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) { - int i, count, j, k; - xlist_t* lists[] = { NULL, NULL, NULL }; + int rc; + xlist_t* lists[] = { &(ctx->resreq->u.simple.res_host), + &(ctx->extra_host_res_requests) }; - count = ctx->resreq->u.simple.res_host.count - + ctx->extra_host_res_requests.count; - ctx->alloc_req.idxs_host = (int*)calloc(count + 1, sizeof(int)); - ctx->alloc_req.vals_host = (float*)calloc(count + 1, sizeof(float)); - - if ((ctx->alloc_req.idxs_host == NULL) - || (ctx->alloc_req.vals_host == NULL)) { - FREEUP(ctx->alloc_req.idxs_host); - FREEUP(ctx->alloc_req.vals_host); - return -ENOMEM; + rc = fill_ctx_idval_vec_(sched, &(ctx->alloc_req.host), 2, &lists[0]); + if (rc < 0) { + xvec_idval_fini(&(ctx->alloc_req.host)); } - lists[0] = &(ctx->resreq->u.simple.res_host); - lists[1] = &(ctx->extra_host_res_requests); - - j = 0; - - for (k = 0; lists[k] != NULL; k++) { - xlist_t* lst = lists[k]; - - for (i = 0; i < lst->count; i++) { - request_res_t* rq = xlist_get(lst, i); - xce_resource_t* r = NULL; - - if (rq == NULL) { - continue; - } - - r = resmgr_lockres(&(sched->resmgr), rq->res, OBJREF_NOP); - if (r == NULL) { - continue; - } - - ctx->alloc_req.idxs_host[j] = r->idx; - ctx->alloc_req.vals_host[j] = rq->value; - j++; - } - } - - ctx->alloc_req.nres_host = j; - - return 0; + return rc; } diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index a66d8555..c3dd21fb 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -11,6 +11,7 @@ #include "xcealloc.h" #include "xlist.h" #include "utils.h" +#include "xvec.h" void kill_sched_job(void* arg) { @@ -54,7 +55,7 @@ static resreq_t* make_sched_resreq(xce_scheduler_t* sched, xce_job_t* j) xlist_store(&(resreq->u.simple.res_host), (void*)req); } - req = make_resource_request("mem", 4096.0); + req = make_resource_request("mem", 1024.0); if (req != NULL) { xlist_store(&(resreq->u.simple.res_host), (void *)req); } @@ -261,10 +262,8 @@ void sched_ctx_fini(xce_sched_ctx_t* ctx) return; } - FREEUP(ctx->alloc_req.idxs_shared); - FREEUP(ctx->alloc_req.vals_shared); - FREEUP(ctx->alloc_req.idxs_host); - FREEUP(ctx->alloc_req.vals_host); + xvec_idval_fini(&(ctx->alloc_req.shared)); + xvec_idval_fini(&(ctx->alloc_req.host)); xlist_fini(&(ctx->extra_res_requests)); xlist_fini(&(ctx->extra_host_res_requests)); diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index f9f04d95..b06fc12c 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -350,7 +350,7 @@ static int sched_init_queues(xce_scheduler_t* sched) for (i = 0; i < config->n_queues; i++) { xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]); - plugin_queue_update(sched, NULL, q); + plugin_queue_update(sched, q); sched_add_queue(sched, q); } @@ -358,7 +358,7 @@ static int sched_init_queues(xce_scheduler_t* sched) } int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config) { - plugin_config_update(sched, NULL, config); + plugin_config_update(sched, config); sched->config = config; sched_init_resources(sched); @@ -374,14 +374,10 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs, int i; for (i = 0; i < njobs; i++) { - xce_sched_job_t* jold = sched_get_job(sched, jobs[i]->jobid); - xce_sched_job_t* j = make_sched_job(sched, jobs[i]); + xce_sched_job_t* j = make_sched_job(sched, jobs[i]); - plugin_job_update(sched, jold, j); - - if (jold == NULL) { - sched_add_job(sched, j); - } + plugin_job_update(sched, j); + sched_add_job(sched, j); } return 0; @@ -419,13 +415,10 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers, int i; for (i = 0; i < nworkers; i++) { - xce_sched_worker_t* wold = sched_get_worker(sched, workers[i]->name); - xce_sched_worker_t* w = make_sched_worker(sched, workers[i]); + xce_sched_worker_t* w = make_sched_worker(sched, workers[i]); - plugin_worker_update(sched, wold, w); - if (wold == NULL) { - sched_add_worker(sched, w); - } + plugin_worker_update(sched, w); + sched_add_worker(sched, w); } return 0; @@ -619,8 +612,9 @@ int sched_add_user(xce_scheduler_t* sched, xce_sched_user_t* u) xce_sched_user_t* sched_get_user(xce_scheduler_t* sched, char* name) { - xce_sched_user_t *user = NULL; - user = (xce_sched_user_t *) kvmgr_lock(&(sched->users), (void*)name, OBJREF_NOP); + xce_sched_user_t* user = NULL; + user = (xce_sched_user_t*)kvmgr_lock(&(sched->users), (void*)name, + OBJREF_NOP); if (user != NULL) { return user; @@ -636,9 +630,10 @@ int sched_add_project(xce_scheduler_t* sched, xce_sched_project_t* p) xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name) { - xce_sched_project_t *project = NULL; - - project = (xce_sched_project_t *) kvmgr_lock(&(sched->projects), (void*)name, OBJREF_NOP); + xce_sched_project_t* project = NULL; + + project = (xce_sched_project_t*)kvmgr_lock(&(sched->projects), (void*)name, + OBJREF_NOP); if (project != NULL) { return project; }