From ecf1916d78a7169fecae2e5fdc444d9cb3007740 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 12 Jan 2026 15:20:51 +0800 Subject: [PATCH] support limit --- src/controller/sched.c | 52 +- src/controller/testsuite/test.c | 2 +- src/controller/xceobjs.c | 6 +- src/include/clusterconfig.h | 14 +- src/include/config.h | 2 +- src/include/group.h | 26 + src/include/typemap.h | 8 +- src/include/xce/types.h | 99 +++- src/include/xcelimit.h | 29 ++ src/include/xlist.h | 2 + src/lib/group.c | 379 +++++++++++++++ src/lib/kvmgr.c | 1 + src/lib/meson.build | 5 + src/lib/preason.c | 7 +- src/lib/testsuite/test.c | 93 +++- src/lib/testsuite/test_group.c | 271 +++++++++++ src/lib/testsuite/test_limit.c | 324 +++++++++++++ src/lib/testsuite/test_preason.c | 22 + src/lib/testsuite/test_xlist.c | 139 ++++++ src/lib/xceapi.c | 5 +- src/lib/xcelimit.c | 484 +++++++++++++++++++ src/lib/xlist.c | 46 +- src/scheduler/include/plugin.h | 26 +- src/scheduler/include/sched/types.h | 44 +- src/scheduler/include/scheduler.h | 18 +- src/scheduler/meson.build | 1 + src/scheduler/src/plugin.c | 67 +++ src/scheduler/src/plugins/limit.c | 443 +++++++++++++++++ src/scheduler/src/plugins/simple_allocator.c | 145 ++---- src/scheduler/src/resource.c | 196 +++++++- src/scheduler/src/schedobj.c | 114 ++++- src/scheduler/src/scheduler.c | 173 +++++-- src/schema/controller.capnp | 36 +- 33 files changed, 3036 insertions(+), 243 deletions(-) create mode 100644 src/include/group.h create mode 100644 src/include/xcelimit.h create mode 100644 src/lib/group.c create mode 100644 src/lib/testsuite/test_group.c create mode 100644 src/lib/testsuite/test_limit.c create mode 100644 src/lib/testsuite/test_xlist.c create mode 100644 src/lib/xcelimit.c create mode 100644 src/scheduler/src/plugins/limit.c diff --git a/src/controller/sched.c b/src/controller/sched.c index e6eeaa7..c999901 100644 --- a/src/controller/sched.c +++ b/src/controller/sched.c @@ -8,6 +8,7 @@ #include "worker.h" #include "xce/cluster_def.h" #include "xce/types.h" +#include "xcelimit.h" extern void collect_shared_resvals(cluster_t*, collect_check_fn_t, int*, xce_val_with_idx_t***); @@ -52,8 +53,44 @@ static int xcesched_push_workerlist(cluster_t* c, xlist_t* l) static int xcesched_push_configs(cluster_t* c) { - xce_configdata_t configdata; - int rc = 0; + xce_configdata_t configdata; + int rc = 0; + xce_limit_config_t limit1_config = { + .name = "limit1", + .per_user = 0, + .per_queue = 0, + .per_project = 0, + .per_host = 0, + .users + = &(xce_group_config_t) { .members + = { .count = 1, .data = (char*[]) { "all" } }, + .ex_members + = { .count = 1, + .data = (char*[]) { "rshen" } } }, + .queues = NULL, + .projects = NULL, + .hosts = NULL, + .val + = &(xce_limit_val_t) { .resname = "slots", .percent = 0, .val = 3.0 } + }; + xce_limit_config_t limit2_config = { + .name = "limits2", + .per_user = 1, + .per_queue = 1, + .per_project = 1, + .per_host = 1, + .users = &(xce_group_config_t) { .members + = { .count = 1, + .data = (char*[]) { "rshen" } } }, + .queues + = &(xce_group_config_t) { .members = { .count = 1, + .data = (char*[]) { "xce" } } }, + .projects = NULL, + .hosts = NULL, + .val = &(xce_limit_val_t) { .resname = "mem", .percent = 1, .val = 0.2 } + }; + xce_limit_config_t* limits[] = { &limit1_config, &limit2_config }; + int nlimits = 2; memset(&configdata, 0x0, sizeof(configdata)); @@ -76,6 +113,17 @@ static int xcesched_push_configs(cluster_t* c) configdata.data.conf4sched->ugroups = c->myconf->cluster_config->ugroups; configdata.data.conf4sched->n_hgroups = c->myconf->cluster_config->nhgroups; configdata.data.conf4sched->hgroups = c->myconf->cluster_config->hgroups; +#if 0 /* FIXME: we use builtin limit currently */ + configdata.data.conf4sched->nlimits = c->myconf->cluster_config->nlimits; + configdata.data.conf4sched->limits = c->myconf->cluster_config->limits; +#else + configdata.data.conf4sched->nlimits = nlimits; + configdata.data.conf4sched->limits = &(limits[0]); +#endif + configdata.data.conf4sched->nhproviders + = c->myconf->cluster_config->nhproviders; + configdata.data.conf4sched->hproviders + = c->myconf->cluster_config->hproviders; /* add resources */ configdata.data_which = ConfigData_data_schedconfig; diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index a369cec..081d8a9 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -1701,7 +1701,7 @@ static int print_workerlist(xce_workerlist_t wlist) return 0; } -static int print_grouplist(xce_grouplist_t glist) { return 0; } +static int print_grouplist(xce_group_configlist_t glist) { return 0; } static int print_configlist(xce_configlist_t clist) { return 0; } diff --git a/src/controller/xceobjs.c b/src/controller/xceobjs.c index 867117b..5a35c49 100644 --- a/src/controller/xceobjs.c +++ b/src/controller/xceobjs.c @@ -784,7 +784,8 @@ static void* xqueue_create(objmgr_t* mgr, objtype_t* typeval, va_list ap) } if (q->queue.users != NULL) { - q1->queue.users = (xce_group_t*)calloc(1, sizeof(xce_group_t)); + q1->queue.users + = (xce_group_config_t*)calloc(1, sizeof(xce_group_config_t)); if (q1->queue.users == NULL) { queue_obj_killer(q1); SET_ERROR(ENOMEM); @@ -826,7 +827,8 @@ static void* xqueue_create(objmgr_t* mgr, objtype_t* typeval, va_list ap) } } if (q->queue.hosts != NULL) { - q1->queue.hosts = (xce_group_t*)calloc(1, sizeof(xce_group_t)); + q1->queue.hosts + = (xce_group_config_t*)calloc(1, sizeof(xce_group_config_t)); if (q1->queue.hosts == NULL) { queue_obj_killer(q1); SET_ERROR(ENOMEM); diff --git a/src/include/clusterconfig.h b/src/include/clusterconfig.h index 13af38d..a9d9a61 100644 --- a/src/include/clusterconfig.h +++ b/src/include/clusterconfig.h @@ -28,14 +28,6 @@ typedef struct { int port; } master_candidate_t; -typedef struct { - char* name; - char* prov_type; - char* conf_path; - char* script_path; - char* script_option; -} scaler_config_t; - typedef struct { char* clustername; int port; @@ -51,9 +43,9 @@ typedef struct { int nqueues; xce_queue_t** queues; int nugroups; - xce_group_t** ugroups; + xce_group_config_t** ugroups; int nhgroups; - xce_group_t** hgroups; + xce_group_config_t** hgroups; int nhconfigs; xce_hostconfig_t** hconfigs; jobmgr_config_t* jobmgr; @@ -63,6 +55,8 @@ typedef struct { xstrings_config_t* xstrings; int nhproviders; scaler_config_t** hproviders; + int nlimits; + xce_limit_config_t** limits; } cluster_config_t; #endif diff --git a/src/include/config.h b/src/include/config.h index b3026eb..a88523e 100644 --- a/src/include/config.h +++ b/src/include/config.h @@ -9,7 +9,7 @@ typedef uint64_t jobid_t; #define BUG_EMAIL "rshen@shenrs.eu" #define DEFAULT_WORKDIR "/tmp/xce" -#define XCE_MAXNAME_LEN (64) +#define XCE_MAXNAME_LEN (256) #define XCE_MAXHOSTNAME_LEN (256) /* hostname suggest to 256 */ #define XCE_MAXDESC_LEN (4096) /* enlarge to 4096 */ #define XCE_MAXKEY_LEN (256) diff --git a/src/include/group.h b/src/include/group.h new file mode 100644 index 0000000..63c9e84 --- /dev/null +++ b/src/include/group.h @@ -0,0 +1,26 @@ +#if !defined(_GROUP_H_) + +#define _GROUP_H_ 1 + +#include "kvmgr.h" +#include "xce/types.h" +#include "xlist.h" +#include "zarray.h" + +typedef struct { + kvmgr_t groups; +} groupmgr_t; + +extern int groupmgr_init(groupmgr_t*, int, int); +extern void groupmgr_fini(groupmgr_t*); +extern int groupmgr_add(groupmgr_t*, xce_group_t*); +extern xce_group_t* groupmgr_lock(groupmgr_t*, char*, int); +extern void groupmgr_del(groupmgr_t*, xce_group_t*); +extern int groupmgr_iter(groupmgr_t*, xlist_t*, void*); +extern xce_group_t* build_group(groupmgr_t*, xce_group_config_t*, char*); +extern int is_in_group(xce_group_t*, char*); +extern xce_group_t* get_group(groupmgr_t*, char*); +extern int fmt_group(char*, int, xce_group_t*); +extern int is_group_all(xce_group_t*); + +#endif diff --git a/src/include/typemap.h b/src/include/typemap.h index c9e77bd..47593fb 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -42,11 +42,15 @@ typedef struct { int n_queues; xce_queue_t** queues; int n_ugroups; - xce_group_t** ugroups; + xce_group_config_t** ugroups; int n_hgroups; - xce_group_t** hgroups; + xce_group_config_t** hgroups; int nresvals; xce_val_with_idx_t** resvals; + int nlimits; + xce_limit_config_t** limits; + int nhproviders; + scaler_config_t** hproviders; } xce_conf4sched_t; typedef struct { diff --git a/src/include/xce/types.h b/src/include/xce/types.h index 9dd1958..8611a9e 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -81,6 +81,7 @@ typedef struct { #define XCE_RESF_RELEASE (0x20) #define XCE_RESF_DEFAULT_HOST (0x40) #define XCE_RESF_CONSUMABLE (0x80) +#define XCE_RESF_LIMIT (0x100) typedef struct { int count; @@ -111,12 +112,12 @@ typedef struct { char* name; xce_strlist_t members; xce_strlist_t ex_members; -} xce_group_t; +} xce_group_config_t; typedef struct { - int count; - xce_group_t** groups; -} xce_grouplist_t; + int count; + xce_group_config_t** groups; +} xce_group_configlist_t; typedef struct { xce_val_t* value; @@ -501,8 +502,8 @@ typedef struct { int32_t accept_interval; int32_t sched_delay; char* resreq; - xce_group_t* users; - xce_group_t* hosts; + xce_group_config_t* users; + xce_group_config_t* hosts; int32_t nthresholds; xce_res_threshold_t** thresholds; xce_timewindow_t* runwindow; @@ -525,13 +526,13 @@ typedef struct { } xce_queuelist_t; typedef struct { - char* name; - xce_reslist_t reslist; - xce_queuelist_t qlist; - xce_workerlist_t wlist; - xce_grouplist_t uglist; - xce_grouplist_t hglist; - xce_configlist_t cfglist; + char* name; + xce_reslist_t reslist; + xce_queuelist_t qlist; + xce_workerlist_t wlist; + xce_group_configlist_t uglist; + xce_group_configlist_t hglist; + xce_configlist_t cfglist; } xce_clusterconf_t; typedef enum { @@ -695,4 +696,76 @@ typedef struct { xce_pendreason_t** reasons; } xce_preason_set_t; +typedef struct { + char* name; + char* prov_type; + char* conf_path; + char* script_path; + char* script_option; +} scaler_config_t; + +typedef struct xce_group_ xce_group_t; +typedef struct xce_g_member_ xce_g_member_t; + +struct xce_g_member_ { + unsigned int is_group : 1; + union { + char* name; + xce_group_t* g; + } u; +}; + +struct xce_group_ { + char* name; + unsigned int dirty : 1; + xlist_t affecteds; /* groups affected by this group */ + xlist_t members; + xlist_t ex_members; +}; + +typedef enum { + CONSUMER_USER = 0, + CONSUMER_QUEUE, + CONSUMER_PROJECT, + CONSUMER_HOST, + CONSUMER_MAX /* keep this the last */ +} xce_consumer_type_t; + +typedef struct { + xce_consumer_type_t ctype; + unsigned int per_consumer : 1; + xce_group_t* group; +} xce_consumer_t; + +typedef struct { + char* resname; + unsigned int percent : 1; + float val; +} xce_limit_val_t; + +typedef struct { + char* name; + unsigned int per_user : 1; + unsigned int per_queue : 1; + unsigned int per_project : 1; + unsigned int per_host : 1; + xce_group_config_t* users; + xce_group_config_t* queues; + xce_group_config_t* projects; + xce_group_config_t* hosts; + xce_limit_val_t* val; +} xce_limit_config_t; + +typedef struct { + char* name; + xce_consumer_t* consumers[CONSUMER_MAX]; + xce_limit_val_t val; +} xce_limit_t; + +typedef struct { + xce_resource_t* vres; + xce_resource_t* res; + xce_limit_t* limit; +} xce_vres_t; + #endif diff --git a/src/include/xcelimit.h b/src/include/xcelimit.h new file mode 100644 index 0000000..67e4021 --- /dev/null +++ b/src/include/xcelimit.h @@ -0,0 +1,29 @@ +#if !defined(_XCELIMIT_H_) + +#define _XCELIMIT_H_ 1 + +#include +#include +#include + +#include "group.h" +#include "kvmgr.h" +#include "xce/types.h" + +extern int build_consumer(xce_consumer_t**, groupmgr_t*, xce_consumer_type_t, + int, xce_group_config_t*, char*); +extern void kill_consumer(xce_consumer_t*); +extern int build_limit(xce_limit_t**, groupmgr_t*, groupmgr_t*, groupmgr_t*, + groupmgr_t*, xce_limit_config_t*); +extern int match_limit(xce_limit_t*, int, ...); +extern int build_vres_name(char**, xce_limit_t*, char*, char*, char*); +extern int is_host_based_limit(resmgr_t*, xce_limit_t*); +extern int fmt_consumer(char*, int, xce_consumer_t*); +extern int fmt_limit(char*, int, xce_limit_t*); +extern void kill_limit(xce_limit_t*); +extern int instantiate_limit(xce_vres_t**, resmgr_t*, xce_limit_t*, char*, + char*, char*); +extern void kill_vres(xce_vres_t*); +extern int add_vres(xce_resource_t**, resmgr_t*, char*, int); + +#endif diff --git a/src/include/xlist.h b/src/include/xlist.h index 4728719..d5b5644 100644 --- a/src/include/xlist.h +++ b/src/include/xlist.h @@ -18,6 +18,8 @@ typedef int (*xlist_action_fn_t)(xlist_t*, void*); extern int xlist_init(xlist_t*, int); extern int xlist_store(xlist_t*, void*); +extern void* xlist_get(xlist_t*, int); +extern int xlist_del(xlist_t*, void*); extern void xlist_clear(xlist_t*); extern int xlist_iter(xlist_t*, xlist_iter_fn_t, void*); extern void* xlist_find(xlist_t*, xlist_find_fn_t, ...); diff --git a/src/lib/group.c b/src/lib/group.c new file mode 100644 index 0000000..206503a --- /dev/null +++ b/src/lib/group.c @@ -0,0 +1,379 @@ +#include +#include +#include + +#include "group.h" +#include "kvmgr.h" +#include "objmgr.h" +#include "utils.h" +#include "xce/types.h" + +static void group_killer(xce_group_t* g) +{ + int i; + + if (g == NULL) { + return; + } + + for (i = 0; i < g->members.count; i++) { + xce_g_member_t* m = (xce_g_member_t*)xlist_get(&(g->members), i); + + if (!m->is_group) { + FREEUP(m->u.name); + } + FREEUP(m); + } + + for (i = 0; i < g->ex_members.count; i++) { + xce_g_member_t* m = (xce_g_member_t*)xlist_get(&(g->ex_members), i); + + if (!m->is_group) { + FREEUP(m->u.name); + } + FREEUP(m); + } + + for (i = 0; i < g->affecteds.count; i++) { + xce_group_t* g1 = (xce_group_t*)xlist_get(&(g->affecteds), i); + + xlist_del(&(g1->members), g); + xlist_del(&(g1->ex_members), g); + } + + xlist_fini(&(g->members)); + xlist_fini(&(g->ex_members)); + xlist_fini(&(g->affecteds)); + + FREEUP(g->name); + FREEUP(g); +} + +int groupmgr_init(groupmgr_t* mgr, int nslots, int delta) +{ + int rc = 0; + xce_group_config_t all = { .name = "all" }; + xce_group_config_t empty = { .name = "empty" }; + xce_group_t* g = NULL; + + memset(mgr, 0x0, sizeof(*mgr)); + + rc = kvmgr_init(&(mgr->groups), string_hash_key, nslots, delta, + (zarray_item_killer_t)group_killer); + + if (rc < 0) { + return rc; + } + + /* create builtin all group */ + g = build_group(mgr, &all, NULL); + if (g == NULL) { + return -ENOMEM; + } + + /* create builtin empty group */ + g = build_group(mgr, &empty, NULL); + if (g == NULL) { + return -ENOMEM; + } + + return 1; +} + +void groupmgr_fini(groupmgr_t* mgr) { kvmgr_fini(&(mgr->groups)); } + +int groupmgr_add(groupmgr_t* mgr, xce_group_t* g) +{ + return kvmgr_add(&(mgr->groups), (void*)g->name, g); +} + +xce_group_t* groupmgr_lock(groupmgr_t* mgr, char* name, int op) +{ + return (xce_group_t*)kvmgr_lock(&(mgr->groups), (void*)name, op); +} + +void groupmgr_del(groupmgr_t* mgr, xce_group_t* g) +{ + kvmgr_del(&(mgr->groups), (void*)g->name); +} + +int groupmgr_iter(groupmgr_t* mgr, xlist_t* l, void* arg) +{ + return kvmgr_iter(&(mgr->groups), NULL, l, arg); +} + +xce_group_t* get_group(groupmgr_t* mgr, char* name) +{ + xce_group_t* g = NULL; + + if ((mgr == NULL) || (name == NULL)) { + return NULL; + } + + g = groupmgr_lock(mgr, name, OBJREF_NOP); + return g; +} + +xce_g_member_t* make_g_member(groupmgr_t* mgr, char* name) +{ + xce_g_member_t* m = NULL; + xce_group_t* g = NULL; + + if ((mgr == NULL) || (name == NULL)) { + errno = EINVAL; + return NULL; + } + + m = (xce_g_member_t*)calloc(1, sizeof(*m)); + if (m == NULL) { + return NULL; + } + + g = get_group(mgr, name); + if (g != NULL) { + m->is_group = 1; + m->u.g = g; + } else { + m->is_group = 0; + m->u.name = strdup(name); + } + + return m; +} + +static int update_group_dirty(xce_group_t* g) +{ + int i; + int count = 0; + + for (i = 0; i < g->affecteds.count; i++) { + xce_group_t* g1 = (xce_group_t*)xlist_get(&(g->affecteds), i); + + if (g1 == NULL) { + continue; + } + + g1->dirty = 1; + count++; + + count += update_group_dirty(g1); + } + + return count; +} + +int fmt_group(char* buf, int len, xce_group_t* g) +{ + int i, pos; + + if (strcmp(g->name, "all") == 0) { + return snprintf(buf, len, "all"); + } + + if (strcmp(g->name, "empty") == 0) { + return snprintf(buf, len, "[]"); + } + + snprintf(buf, len, "["); + pos = strlen(buf); + + for (i = 0; i < g->members.count; i++) { + xce_g_member_t* m = (xce_g_member_t*)xlist_get(&(g->members), i); + + if (m == NULL) { + continue; + } + + snprintf(buf + pos, len - pos, "%s ", + (m->is_group) ? m->u.g->name : m->u.name); + pos = strlen(buf); + } + + for (i = 0; i < g->ex_members.count; i++) { + xce_g_member_t* m = (xce_g_member_t*)xlist_get(&(g->ex_members), i); + + if (m == NULL) { + continue; + } + + snprintf(buf + pos, len - pos, "~%s ", + (m->is_group) ? m->u.g->name : m->u.name); + pos = strlen(buf); + } + + while ((pos > 0) && (buf[pos - 1] == ' ')) { + pos--; + } + + snprintf(buf + pos, len - pos, "]"); + + return strlen(buf); +} + +xce_group_t* build_group(groupmgr_t* mgr, xce_group_config_t* g_config, + char* default_group_name) +{ + xce_group_t* g = NULL; + char* g_name; + int i; + + if ((mgr == NULL) || (g_config == NULL)) { + errno = EINVAL; + return NULL; + } + + g_name = g_config->name; + if ((g_name == NULL) || (strlen(g_name) == 0)) { + g_name = default_group_name; + } + + if (g_name == NULL) { + return NULL; + } + + g = get_group(mgr, g_name); + if (g != NULL) { + for (i = 0; i < g->members.count; i++) { + xce_g_member_t* m = (xce_g_member_t*)xlist_get(&(g->members), i); + if (m == NULL) { + continue; + } + if (m->is_group == 0) { + FREEUP(m->u.name); + } + + FREEUP(m); + } + g->members.count = 0; + + for (i = 0; i < g->ex_members.count; i++) { + xce_g_member_t* m = (xce_g_member_t*)xlist_get(&(g->ex_members), i); + if (m == NULL) { + continue; + } + if (m->is_group == 0) { + FREEUP(m->u.name); + } + + FREEUP(m); + } + g->ex_members.count = 0; + g->dirty = 1; + } else { + g = (xce_group_t*)calloc(1, sizeof(*g)); + if (g == NULL) { + return NULL; + } + + xlist_init(&(g->affecteds), 4); + xlist_init(&(g->members), 4); + xlist_init(&(g->ex_members), 4); + g->name = strdup(g_name); + g->dirty = 0; + } + + for (i = 0; i < g_config->members.count; i++) { + char* g_name = g_config->members.data[i]; + xce_g_member_t* m = make_g_member(mgr, g_name); + + if (m == NULL) { + continue; + } + + xlist_store(&(g->members), m); + if (m->is_group) { + xlist_store(&(m->u.g->affecteds), g); + } + } + + for (i = 0; i < g_config->ex_members.count; i++) { + char* g_name = g_config->ex_members.data[i]; + xce_g_member_t* m = make_g_member(mgr, g_name); + + if (m == NULL) { + continue; + } + + xlist_store(&(g->ex_members), m); + if (m->is_group) { + xlist_store(&(m->u.g->affecteds), g); + } + } + + groupmgr_add(mgr, g); + + if (g->dirty) { + update_group_dirty(g); + } + + return g; +} + +static int check_in_member(xce_g_member_t* m, va_list ap) +{ + char* name = va_arg(ap, char*); + + if ((m == NULL) || (name == NULL)) { + return 0; + } + + if (m->is_group) { + if (is_in_group(m->u.g, name)) { + return 1; + } + + return 0; + } + + if (strcmp(m->u.name, name) == 0) { + return 1; + } + + return 0; +} + +int is_in_group(xce_group_t* g, char* name) +{ + if ((g == NULL) || (name == NULL)) { + return -EINVAL; + } + + /* everything should be in all group */ + if (strcmp(g->name, "all") == 0) { + return 1; + } + + /* nothing is in empty group */ + if (strcmp(g->name, "empty") == 0) { + return 0; + } + + if (strcmp(g->name, name) == 0) { + return 1; + } + + if (xlist_find(&(g->ex_members), (xlist_find_fn_t)check_in_member, name) + != NULL) { + return 0; + } + + if (xlist_find(&(g->members), (xlist_find_fn_t)check_in_member, name) + != NULL) { + return 1; + } + + return 0; +} + +int is_group_all(xce_group_t* g) +{ + if (g == NULL) { + return 0; + } + + if (g->name == NULL) { + return 0; + } + + return (strcmp(g->name, "all") == 0); +} diff --git a/src/lib/kvmgr.c b/src/lib/kvmgr.c index c73911d..b44fadc 100644 --- a/src/lib/kvmgr.c +++ b/src/lib/kvmgr.c @@ -392,3 +392,4 @@ int jobmgr_iter(jobmgr_t* mgr, xlist_t* l, void* arg) { return kvmgr_iter(&(mgr->jobs), NULL, l, arg); } + diff --git a/src/lib/meson.build b/src/lib/meson.build index 9c0fa44..1be21ce 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -33,6 +33,8 @@ libxce_src = [ gen_srcs[0], 'preason.c', 'xcealloc.c', 'misc.c', + 'group.c', + 'xcelimit.c', ] incdir = include_directories('../include') @@ -71,6 +73,9 @@ libxce_test_src = ['testsuite/test.c', 'testsuite/test_ordq.c', 'testsuite/test_preason.c', 'testsuite/test_kvmgr.c', + 'testsuite/test_group.c', + 'testsuite/test_xlist.c', + 'testsuite/test_limit.c', ] libxce_test_deps = [libxce_deps, unity_dep, libxce_dep] diff --git a/src/lib/preason.c b/src/lib/preason.c index 11411af..6baefd2 100644 --- a/src/lib/preason.c +++ b/src/lib/preason.c @@ -94,10 +94,15 @@ static int add_res_preason(xce_job_preason_t* reasons, int code, char* host, int count; xce_res_reason_t* p1; - if ((reasons == NULL) || (host == NULL) || (res == NULL)) { + if ((reasons == NULL) || (res == NULL)) { return -EINVAL; } + /* FIXME: global resource reason */ + if (host == NULL) { + host = ""; + } + h = strdup(host); r = strdup(res); p1 = (xce_res_reason_t*)calloc(1, sizeof(*p1)); diff --git a/src/lib/testsuite/test.c b/src/lib/testsuite/test.c index 3ca78f6..b191006 100644 --- a/src/lib/testsuite/test.c +++ b/src/lib/testsuite/test.c @@ -1,3 +1,4 @@ +#include #include #include #include @@ -16,26 +17,94 @@ extern void test_preason(void); extern void test_ordq(void); extern void test_delay(void); extern void test_kvmgr(void); +extern void test_group(void); +extern void test_xlist(void); +extern void test_limit(void); void setUp(void) { return; } void tearDown(void) { return; } +static char doc[] = "xcelib_test"; +static char args_doc[] = ""; + +#define MAX_TESTS (256) + +struct arguments { + int ntests; + char* tests[MAX_TESTS]; +}; + +static struct argp_option options[] = { + { "run", 't', "STR", 0, "run test" }, + { 0 } +}; + +static error_t parse_opt(int key, char* arg, struct argp_state* state) +{ + struct arguments* arguments = state->input; + + switch (key) { + case 't': + if (arguments->ntests >= MAX_TESTS) { + fprintf(stderr, "Too many tests\n"); + exit(-1); + } + arguments->tests[arguments->ntests] = arg; + arguments->ntests++; + break; + default: + return ARGP_ERR_UNKNOWN; + } + + return 0; +} + +static struct argp argp = { options, parse_opt, args_doc, doc }; +struct arguments arguments; + +static int check_test(struct arguments* args, char* x) +{ + int i; + + if (args->ntests == 0) { + return 1; + } + for (i = 0; i < args->ntests; i++) { + if (strcmp(x, args->tests[i]) == 0) { + return 1; + } + } + return 0; +} + +#define TEST(x) \ + do { \ + if (check_test(&arguments, #x)) { \ + x(); \ + } \ + } while (0) + int main(int argc, char* argv[]) { + argp_parse(&argp, argc, argv, 0, 0, &arguments); + UNITY_BEGIN(); - test_config(); - test_idgen(); - test_ht_string(); - test_ht_int(); - test_zarray(); - test_xstrings(); - test_error(); - test_utils(); - test_preason(); - test_ordq(); - test_delay(); - test_kvmgr(); + TEST(test_config); + TEST(test_idgen); + TEST(test_ht_string); + TEST(test_ht_int); + TEST(test_zarray); + TEST(test_xstrings); + TEST(test_error); + TEST(test_utils); + TEST(test_preason); + TEST(test_ordq); + TEST(test_delay); + TEST(test_kvmgr); + TEST(test_group); + TEST(test_xlist); + TEST(test_limit); return UNITY_END(); } diff --git a/src/lib/testsuite/test_group.c b/src/lib/testsuite/test_group.c new file mode 100644 index 0000000..e935ad2 --- /dev/null +++ b/src/lib/testsuite/test_group.c @@ -0,0 +1,271 @@ +#include +#include +#include +#include + +#include "group.h" +#include "utils.h" + +static void test_all(void) +{ + groupmgr_t mgr; + xce_group_t* all = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + all = get_group(&mgr, "all"); + + TEST_ASSERT_TRUE(all != NULL); + + rc = is_in_group(all, "rshen"); + + TEST_ASSERT_TRUE(rc > 0); + + groupmgr_fini(&mgr); +} + +static void test_empty(void) +{ + groupmgr_t mgr; + xce_group_t* empty = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + empty = get_group(&mgr, "empty"); + + TEST_ASSERT_TRUE(empty != NULL); + + rc = is_in_group(empty, "rshen"); + + TEST_ASSERT_TRUE(rc == 0); + + groupmgr_fini(&mgr); +} + +static void test_all_1(void) +{ + groupmgr_t mgr; + xce_group_config_t gdata = { + .name = "g1", + .members = { .count = 1, .data = (char*[]) { "all" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + g = build_group(&mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc > 0); + + groupmgr_fini(&mgr); +} + +static void test_all_2(void) +{ + groupmgr_t mgr; + xce_group_config_t gdata = { + .name = "g1", + .members = { .count = 2, .data = (char*[]) { "g2", "all" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + g = build_group(&mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc > 0); + + groupmgr_fini(&mgr); +} + +static void test_empty_1(void) +{ + groupmgr_t mgr; + xce_group_config_t gdata = { + .name = "g1", + .ex_members = { .count = 1, .data = (char*[]) { "all" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + g = build_group(&mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc == 0); + + groupmgr_fini(&mgr); +} + +static void test_empty_2(void) +{ + groupmgr_t mgr; + xce_group_config_t gdata = { + .name = "g1", + .members = { .count = 1, .data = (char*[]) { "empty" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + g = build_group(&mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc == 0); + + groupmgr_fini(&mgr); +} + +static void test_include(void) +{ + groupmgr_t mgr; + xce_group_config_t gdata = { + .name = "g1", + .members = { .count = 1, .data = (char*[]) { "rshen" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + g = build_group(&mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc > 0); + + groupmgr_fini(&mgr); +} + +static void test_exclude(void) +{ + groupmgr_t mgr; + xce_group_config_t gdata = { + .name = "g1", + .members = { .count = 1, .data = (char*[]) { "all" } }, + .ex_members = { .count = 1, .data = (char*[]) { "rshen" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + g = build_group(&mgr, &gdata, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc == 0); + + groupmgr_fini(&mgr); +} + +static void test_nest(void) +{ + groupmgr_t mgr; + xce_group_config_t gdata1 = { + .name = "g1", + .members = { .count = 1, .data = (char*[]) { "rshen" } } + }; + xce_group_config_t gdata2 = { + .name = "g2", + .members = { .count = 1, .data = (char*[]) { "g1" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + g = build_group(&mgr, &gdata1, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + g = build_group(&mgr, &gdata2, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc != 0); + + groupmgr_fini(&mgr); +} + +static void test_nest2(void) +{ + groupmgr_t mgr; + xce_group_config_t gdata1 = { + .name = "g1", + .members = { .count = 1, .data = (char*[]) { "rshen" } } + }; + xce_group_config_t gdata2 = { + .name = "g2", + .members = { .count = 1, .data = (char*[]) { "all" } }, + .ex_members = { .count = 1, .data = (char*[]) { "g1" } } + }; + + xce_group_t* g = NULL; + int rc = 0; + + groupmgr_init(&mgr, 16, 16); + + g = build_group(&mgr, &gdata1, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + g = build_group(&mgr, &gdata2, NULL); + + TEST_ASSERT_TRUE(g != NULL); + + rc = is_in_group(g, "rshen"); + + TEST_ASSERT_TRUE(rc == 0); + + groupmgr_fini(&mgr); +} + +void test_group(void) +{ + printf("\n>> group test\n\n"); + + RUN_TEST(test_all); + RUN_TEST(test_empty); + RUN_TEST(test_all_1); + RUN_TEST(test_all_2); + RUN_TEST(test_empty_1); + RUN_TEST(test_empty_2); + RUN_TEST(test_include); + RUN_TEST(test_exclude); + RUN_TEST(test_nest); + RUN_TEST(test_nest2); +} diff --git a/src/lib/testsuite/test_limit.c b/src/lib/testsuite/test_limit.c new file mode 100644 index 0000000..e0983b4 --- /dev/null +++ b/src/lib/testsuite/test_limit.c @@ -0,0 +1,324 @@ +#include +#include +#include +#include + +#include "group.h" +#include "xce/types.h" +#include "xcelimit.h" + +static void test_build_consumer1(void) +{ + xce_consumer_t* c = NULL; + groupmgr_t mgr; + xce_group_config_t gdata = { + .members = { .count = 2, .data = (char*[]) { "rshen", "jsxing" } } + }; + int rc = 0; + char buf[1024]; + + groupmgr_init(&mgr, 4, 4); + rc = build_consumer(&c, &mgr, CONSUMER_USER, 1, &gdata, "user_group1"); + + TEST_ASSERT_TRUE(rc == 1); + + rc = fmt_consumer(buf, sizeof(buf) - 1, c); + TEST_ASSERT_TRUE(rc > 0); + + printf("\n >>> consumer: %s\n", buf); + + kill_consumer(c); + + groupmgr_fini(&mgr); + return; +} + +static void test_build_consumer2(void) +{ + xce_consumer_t* c = NULL; + groupmgr_t mgr; + xce_group_config_t gdata = { + .members = { .count = 1, .data = (char*[]) { "all" } }, + .ex_members = { .count = 2, .data = (char*[]) { "rshen", "jsxing" } } + }; + int rc = 0; + char buf[1024]; + + groupmgr_init(&mgr, 4, 4); + rc = build_consumer(&c, &mgr, CONSUMER_USER, 1, &gdata, "user_group1"); + + TEST_ASSERT_TRUE(rc == 1); + + rc = fmt_consumer(buf, sizeof(buf) - 1, c); + + TEST_ASSERT_TRUE(rc > 0); + + printf("\n >>> consumer: %s\n", buf); + + kill_consumer(c); + + groupmgr_fini(&mgr); + return; +} + +static void test_build_limit_ok1(void) +{ + groupmgr_t umgr, qmgr, hmgr, pmgr; + xce_limit_t* limit; + int rc; + xce_limit_config_t limit_config = { + .name = "limit1", + .per_user = 0, + .per_queue = 0, + .per_project = 0, + .per_host = 0, + .users + = &(xce_group_config_t) { .members = { .count = 1, + .data = (char*[]) { "u1" } } }, + .queues = NULL, + .hosts = NULL, + .projects = NULL, + .val + = &(xce_limit_val_t) { .resname = "mem", .percent = 1, .val = 80.0 } + }; + char buf[1024]; + + groupmgr_init(&umgr, 4, 4); + groupmgr_init(&qmgr, 4, 4); + groupmgr_init(&hmgr, 4, 4); + groupmgr_init(&pmgr, 4, 4); + + rc = build_limit(&limit, &umgr, &qmgr, &hmgr, &pmgr, &limit_config); + TEST_ASSERT_TRUE(rc == 1); + + memset(buf, 0x0, sizeof(buf)); + + rc = fmt_limit(buf, sizeof(buf) - 1, limit); + TEST_ASSERT_TRUE(rc > 0); + + printf("%s\n", buf); + + kill_limit(limit); + + groupmgr_fini(&pmgr); + groupmgr_fini(&hmgr); + groupmgr_fini(&qmgr); + groupmgr_fini(&umgr); +} + +static void test_match_limit_ok1(void) +{ + groupmgr_t umgr, qmgr, hmgr, pmgr; + xce_limit_t* limit; + int rc; + xce_limit_config_t limit_config = { + .name = "limit1", + .per_user = 0, + .per_queue = 0, + .per_project = 0, + .per_host = 0, + .users + = &(xce_group_config_t) { .members = { .count = 1, + .data = (char*[]) { "all" } } }, + .queues + = &(xce_group_config_t) { .members = { .count = 1, + .data = (char*[]) { "all" } } }, + .hosts = NULL, + .projects + = &(xce_group_config_t) { .members = { .count = 1, + .data = (char*[]) { "all" } } }, + .val + = &(xce_limit_val_t) { .resname = "mem", .percent = 1, .val = 80.0 } + }; + char buf[1024]; + + groupmgr_init(&umgr, 4, 4); + groupmgr_init(&qmgr, 4, 4); + groupmgr_init(&hmgr, 4, 4); + groupmgr_init(&pmgr, 4, 4); + + rc = build_limit(&limit, &umgr, &qmgr, &hmgr, &pmgr, &limit_config); + TEST_ASSERT_TRUE(rc == 1); + + memset(buf, 0x0, sizeof(buf)); + + rc = fmt_limit(buf, sizeof(buf) - 1, limit); + TEST_ASSERT_TRUE(rc > 0); + + printf("%s\n", buf); + + rc = match_limit(limit, 1, "rshen", "xce", "project1"); + TEST_ASSERT_TRUE(rc > 0); + + kill_limit(limit); + + groupmgr_fini(&pmgr); + groupmgr_fini(&hmgr); + groupmgr_fini(&qmgr); + groupmgr_fini(&umgr); +} + +static void test_match_limit_ok2(void) +{ + groupmgr_t umgr, qmgr, hmgr, pmgr; + xce_limit_t* limit; + int rc; + xce_limit_config_t limit_config = { + .name = "limit1", + .per_user = 0, + .per_queue = 0, + .per_project = 0, + .per_host = 0, + .users = NULL, + .queues = NULL, + .hosts = NULL, + .projects = NULL, + .val + = &(xce_limit_val_t) { .resname = "mem", .percent = 1, .val = 80.0 } + }; + char buf[1024]; + + groupmgr_init(&umgr, 4, 4); + groupmgr_init(&qmgr, 4, 4); + groupmgr_init(&hmgr, 4, 4); + groupmgr_init(&pmgr, 4, 4); + + rc = build_limit(&limit, &umgr, &qmgr, &hmgr, &pmgr, &limit_config); + TEST_ASSERT_TRUE(rc == 1); + + memset(buf, 0x0, sizeof(buf)); + + rc = fmt_limit(buf, sizeof(buf) - 1, limit); + TEST_ASSERT_TRUE(rc > 0); + + printf("%s\n", buf); + + rc = match_limit(limit, 1, "rshen", "xce", "project1"); + TEST_ASSERT_TRUE(rc > 0); + + kill_limit(limit); + + groupmgr_fini(&pmgr); + groupmgr_fini(&hmgr); + groupmgr_fini(&qmgr); + groupmgr_fini(&umgr); +} + +static void test_match_limit1(void) +{ + groupmgr_t umgr, qmgr, hmgr, pmgr; + xce_limit_t* limit; + int rc; + xce_limit_config_t limit_config = { + .name = "limit1", + .per_user = 0, + .per_queue = 0, + .per_project = 0, + .per_host = 0, + .users + = &(xce_group_config_t) { .members + = { .count = 1, .data = (char*[]) { "all" } }, + .ex_members + = { .count = 1, + .data = (char*[]) { "rshen" } } }, + .queues + = &(xce_group_config_t) { .members = { .count = 1, + .data = (char*[]) { "all" } } }, + .hosts = NULL, + .projects + = &(xce_group_config_t) { .members = { .count = 1, + .data = (char*[]) { "all" } } }, + .val + = &(xce_limit_val_t) { .resname = "mem", .percent = 1, .val = 80.0 } + }; + char buf[1024]; + + groupmgr_init(&umgr, 4, 4); + groupmgr_init(&qmgr, 4, 4); + groupmgr_init(&hmgr, 4, 4); + groupmgr_init(&pmgr, 4, 4); + + rc = build_limit(&limit, &umgr, &qmgr, &hmgr, &pmgr, &limit_config); + TEST_ASSERT_TRUE(rc == 1); + + memset(buf, 0x0, sizeof(buf)); + + rc = fmt_limit(buf, sizeof(buf) - 1, limit); + TEST_ASSERT_TRUE(rc > 0); + + printf("%s\n", buf); + + rc = match_limit(limit, 1, "rshen", "xce", "project1"); + TEST_ASSERT_TRUE(rc == 0); + + kill_limit(limit); + + groupmgr_fini(&pmgr); + groupmgr_fini(&hmgr); + groupmgr_fini(&qmgr); + groupmgr_fini(&umgr); +} + +static void test_match_limit2(void) +{ + groupmgr_t umgr, qmgr, hmgr, pmgr; + xce_limit_t* limit; + int rc; + xce_limit_config_t limit_config = { + .name = "limit1", + .per_user = 0, + .per_queue = 0, + .per_project = 0, + .per_host = 0, + .users + = &(xce_group_config_t) { .members + = { .count = 1, .data = (char*[]) { "all" } }, + .ex_members + = { .count = 1, + .data = (char*[]) { "rshen" } } }, + .queues = NULL, + .hosts = NULL, + .projects = NULL, + .val + = &(xce_limit_val_t) { .resname = "slots", .percent = 0, .val = 3.0 } + }; + char buf[1024]; + + groupmgr_init(&umgr, 4, 4); + groupmgr_init(&qmgr, 4, 4); + groupmgr_init(&hmgr, 4, 4); + groupmgr_init(&pmgr, 4, 4); + + rc = build_limit(&limit, &umgr, &qmgr, &hmgr, &pmgr, &limit_config); + TEST_ASSERT_TRUE(rc == 1); + + memset(buf, 0x0, sizeof(buf)); + + rc = fmt_limit(buf, sizeof(buf) - 1, limit); + TEST_ASSERT_TRUE(rc > 0); + + printf("%s\n", buf); + + rc = match_limit(limit, 1, "rshen", "normal", "default"); + TEST_ASSERT_TRUE(rc == 0); + + kill_limit(limit); + + groupmgr_fini(&pmgr); + groupmgr_fini(&hmgr); + groupmgr_fini(&qmgr); + groupmgr_fini(&umgr); +} + +void test_limit(void) +{ + printf("\n>> limit test\n\n"); + RUN_TEST(test_build_consumer1); + RUN_TEST(test_build_consumer2); + RUN_TEST(test_build_limit_ok1); + RUN_TEST(test_match_limit_ok1); + RUN_TEST(test_match_limit_ok2); + RUN_TEST(test_match_limit1); + RUN_TEST(test_match_limit2); +} + diff --git a/src/lib/testsuite/test_preason.c b/src/lib/testsuite/test_preason.c index 0a8ed29..99a7fa6 100644 --- a/src/lib/testsuite/test_preason.c +++ b/src/lib/testsuite/test_preason.c @@ -58,6 +58,27 @@ static void test_add_res_reason(void) erase_job_preason(&reason); } +static void test_add_shared_res_reason(void) +{ + xce_job_preason_t reason; + + memset(&reason, 0x0, sizeof(reason)); + + add_job_preason(&reason, 1, NULL, "res1"); + TEST_ASSERT_TRUE(reason.main_reason == 0); + TEST_ASSERT_TRUE((reason.nres_reasons == 1) + && (reason.res_based_reasons != NULL) + && (reason.res_based_reasons[0] != NULL)); + TEST_ASSERT_TRUE( + (reason.res_based_reasons[0]->reason == 1) + && (strcmp(reason.res_based_reasons[0]->host, "") == 0) + && (strcmp(reason.res_based_reasons[0]->resource, "res1") == 0)); + TEST_ASSERT_TRUE((reason.nhost_reasons == 0) + && (reason.host_based_reasons == NULL)); + erase_job_preason(&reason); +} + + static void test_add_res_host_reason(void) { xce_job_preason_t reason; @@ -161,6 +182,7 @@ void test_preason(void) RUN_TEST(test_main_reason); RUN_TEST(test_add_host_reason); RUN_TEST(test_add_res_reason); + RUN_TEST(test_add_shared_res_reason); RUN_TEST(test_add_res_host_reason); RUN_TEST(test_add_main_reset_other); RUN_TEST(test_add_more_host_reason); diff --git a/src/lib/testsuite/test_xlist.c b/src/lib/testsuite/test_xlist.c new file mode 100644 index 0000000..4b192d5 --- /dev/null +++ b/src/lib/testsuite/test_xlist.c @@ -0,0 +1,139 @@ +#include +#include +#include +#include +#include + +#include "utils.h" +#include "xlist.h" + +static void test_xlist1(void) +{ + xlist_t lst; + char* hello = "Hello"; + void* p = NULL; + int idx; + + xlist_init(&lst, 2); + idx = xlist_store(&lst, (void*)hello); + + TEST_ASSERT_TRUE(idx >= 0); + p = xlist_get(&lst, idx); + + TEST_ASSERT_TRUE(p == hello); + xlist_fini(&lst); +} + +static int find_check_(void* data, va_list ap) +{ + char* p = va_arg(ap, char*); + + if (data == p) { + return 1; + } + + if (p == NULL) { + return 0; + } + + if (data == NULL) { + return 0; + } + + if (strcmp((char*)data, p) == 0) { + return 1; + } + + return 0; +} + +static void test_xlist_find(void) +{ + xlist_t lst; + char* hello = "Hello"; + void* p = NULL; + int idx; + + xlist_init(&lst, 2); + idx = xlist_store(&lst, (void*)hello); + + TEST_ASSERT_TRUE(idx >= 0); + + p = xlist_find(&lst, (xlist_find_fn_t)find_check_, "Hello"); + + TEST_ASSERT_TRUE(p != NULL); + + TEST_ASSERT_TRUE(strcmp((char*)p, "Hello") == 0); + + xlist_fini(&lst); +} + +static void test_xlist_del(void) +{ + xlist_t lst; + char* hello = "Hello"; + void* p = NULL; + int idx; + int i; + + xlist_init(&lst, 2); + for (i = 0; i < 5; i++) { + idx = xlist_store(&lst, (void*)hello); + TEST_ASSERT_TRUE(idx >= 0); + } + + p = xlist_find(&lst, (xlist_find_fn_t)find_check_, "Hello"); + + TEST_ASSERT_TRUE(p != NULL); + + TEST_ASSERT_TRUE(strcmp((char*)p, "Hello") == 0); + + xlist_del(&lst, (void*)hello); + + p = xlist_find(&lst, (xlist_find_fn_t)find_check_, "Hello"); + + TEST_ASSERT_TRUE(p == NULL); + TEST_ASSERT_TRUE(lst.count == 0); + + xlist_fini(&lst); +} + +static int do_iter_(void* arg1, void* arg2) +{ + int* count = (int*)arg2; + int i = (int)(long)arg1; + + (*count) += i; + + return 0; +} + +static void test_xlist_iter(void) +{ + xlist_t lst; + int idx; + int i; + int count = 0; + + xlist_init(&lst, 2); + for (i = 0; i < 5; i++) { + idx = xlist_store(&lst, (void*)(long)i); + TEST_ASSERT_TRUE(idx >= 0); + } + + xlist_iter(&lst, (xlist_iter_fn_t)do_iter_, &count); + + TEST_ASSERT_TRUE(count == 10); + + xlist_fini(&lst); +} + +void test_xlist(void) +{ + printf("\n>> xlist test\n\n"); + + RUN_TEST(test_xlist1); + RUN_TEST(test_xlist_find); + RUN_TEST(test_xlist_del); + RUN_TEST(test_xlist_iter); +} diff --git a/src/lib/xceapi.c b/src/lib/xceapi.c index a913f99..2d2eac6 100644 --- a/src/lib/xceapi.c +++ b/src/lib/xceapi.c @@ -368,7 +368,8 @@ int xce_free_queues(int nqueues, xce_queue_t* queues) return 0; } -int xce_grouplist(xce_ctx_t* ctx, int group_type, xce_group_t*** groups) +int xce_group_configlist(xce_ctx_t* ctx, int group_type, + xce_group_config_t*** groups) { #if 0 int table; @@ -388,7 +389,7 @@ int xce_grouplist(xce_ctx_t* ctx, int group_type, xce_group_t*** groups) #endif } -int xce_free_groups(int ngroups, xce_group_t* groups) +int xce_free_groups(int ngroups, xce_group_config_t* groups) { /* TODO */ return 0; diff --git a/src/lib/xcelimit.c b/src/lib/xcelimit.c new file mode 100644 index 0000000..2b0c5d8 --- /dev/null +++ b/src/lib/xcelimit.c @@ -0,0 +1,484 @@ +#include +#include +#include +#include +#include +#include + +#include "group.h" +#include "kvmgr.h" +#include "objmgr.h" +#include "utils.h" +#include "xce/types.h" +#include "xcelimit.h" + +int fmt_consumer(char* buf, int len, xce_consumer_t* c) +{ + struct { + xce_consumer_type_t type; + char* name; + char* per_name; + } consumer_mapping[] = { + { CONSUMER_USER, "USER", "PER_USER" }, + { CONSUMER_HOST, "HOST", "PER_HOST" }, + { CONSUMER_PROJECT, "PROJECT", "PER_PROJECT" }, + { CONSUMER_QUEUE, "QUEUE", "PER_QUEUE" }, + { -1, NULL, NULL } + }; + int i; + + if ((buf == NULL) || (c == NULL) || (len <= 0)) { + return -EINVAL; + } + for (i = 0; consumer_mapping[i].type >= 0; i++) { + if (c->ctype == consumer_mapping[i].type) { + char* t; + int l; + int rc; + + if (c->per_consumer) { + t = consumer_mapping[i].per_name; + } else { + t = consumer_mapping[i].name; + } + + snprintf(buf, len, "%s = ", t); + l = strlen(buf); + + rc = fmt_group(buf + l, len - l, c->group); + if (rc < 0) { + return rc; + } + + return strlen(buf); + } + } + + return -EINVAL; +} + +int build_consumer(xce_consumer_t** pconsumer, groupmgr_t* mgr, + xce_consumer_type_t type, int per_consumer, + xce_group_config_t* g_config, char* default_name_prefix) +{ + xce_consumer_t* c = NULL; + xce_group_t* g = NULL; + + *pconsumer = NULL; + c = (xce_consumer_t*)calloc(1, sizeof(*c)); + if (c == NULL) { + return -ENOMEM; + } + + *pconsumer = c; + + c->ctype = type; + c->per_consumer = (per_consumer != 0) ? 1 : 0; + c->group = NULL; + + if (g_config != NULL) { + char* default_g_name = NULL; + char consumer_group_name[2 * XCE_MAXNAME_LEN + 1]; + + if ((g_config->name == NULL) || (strlen(g_config->name) == 0)) { + char* suffix = NULL; + + switch (type) { + case CONSUMER_USER: + suffix = "ug"; + break; + case CONSUMER_PROJECT: + suffix = "pg"; + break; + case CONSUMER_HOST: + suffix = "hg"; + break; + case CONSUMER_QUEUE: + suffix = "qg"; + break; + default: + suffix = "unknown"; + break; + } + + sprintf(consumer_group_name, "%s-%s-group", default_name_prefix, + suffix); + default_g_name = consumer_group_name; + } + else { + default_g_name = g_config->name; + } + + g = build_group(mgr, g_config, default_g_name); + if (g == NULL) { + return -ENOMEM; + } + } else { + g = get_group(mgr, "all"); + } + + c->group = g; + + return 1; +} + +void kill_consumer(xce_consumer_t* c) +{ + if (c == NULL) { + return; + } + + /* leave c->group managed by group manager */ + FREEUP(c); +} + +int fmt_limit(char* buf, int len, xce_limit_t* limit) +{ + int pos = 0; + int i, rc; + + rc = snprintf(buf + pos, len - pos, "limit: %s\n", limit->name); + if (rc < 0) { + return rc; + } + + pos += rc; + + for (i = 0; i < CONSUMER_MAX; i++) { + xce_consumer_t* c = limit->consumers[i]; + + if (c != NULL) { + rc = fmt_consumer(buf + pos, len - pos, c); + if (rc < 0) { + return rc; + } + + pos += rc; + rc = snprintf(buf + pos, len - pos, "\n"); + if (rc < 0) { + return rc; + } + pos += rc; + } + } + + rc = snprintf(buf + pos, len - pos, "\n%s = %f%s\n", limit->val.resname, + limit->val.val, (limit->val.percent != 0) ? "%" : ""); + if (rc < 0) { + return rc; + } + + return strlen(buf); +} + +void kill_limit(xce_limit_t* limit) +{ + int i; + + if (limit == NULL) { + return; + } + + FREEUP(limit->name); + + for (i = 0; i < CONSUMER_MAX; i++) { + xce_consumer_t* c = limit->consumers[i]; + + if (c != NULL) { + kill_consumer(c); + } + } + + FREEUP(limit->val.resname); + FREEUP(limit); +} + +int build_limit(xce_limit_t** plimit, groupmgr_t* ugmgr, groupmgr_t* qgmgr, + groupmgr_t* hgmgr, groupmgr_t* pgmgr, + xce_limit_config_t* config) +{ + xce_limit_t* limit = NULL; + int rc = 0; + + *plimit = NULL; + limit = (xce_limit_t*)calloc(1, sizeof(*limit)); + if (limit == NULL) { + return -ENOMEM; + } + + *plimit = limit; + + limit->name = strdup(config->name); + + rc = build_consumer(&(limit->consumers[CONSUMER_USER]), ugmgr, + CONSUMER_USER, config->per_user, config->users, + config->name); + if (rc < 0) { + return rc; + } + + rc = build_consumer(&(limit->consumers[CONSUMER_HOST]), hgmgr, + CONSUMER_HOST, config->per_host, config->hosts, + config->name); + if (rc < 0) { + return rc; + } + + rc = build_consumer(&(limit->consumers[CONSUMER_QUEUE]), qgmgr, + CONSUMER_QUEUE, config->per_queue, config->queues, + config->name); + if (rc < 0) { + return rc; + } + + rc = build_consumer(&(limit->consumers[CONSUMER_PROJECT]), pgmgr, + CONSUMER_PROJECT, config->per_project, config->projects, + config->name); + if (rc < 0) { + return rc; + } + + limit->val.resname = strdup(config->val->resname); + limit->val.percent = config->val->percent; + limit->val.val = config->val->val; + + return 1; +} + +int match_limit(xce_limit_t* limit, int global, ...) +{ + va_list ap; + char* user = NULL; + char* queue = NULL; + char* project = NULL; + char* host = NULL; + xce_consumer_t* c = NULL; + + va_start(ap, global); + user = va_arg(ap, char*); + queue = va_arg(ap, char*); + project = va_arg(ap, char*); + + if ((limit == NULL) || (user == NULL) || (queue == NULL) + || (project == NULL)) { + return -EINVAL; + } + + if (!global) { + host = va_arg(ap, char*); + if (host == NULL) { + return -EINVAL; + } + } + + c = limit->consumers[CONSUMER_USER]; + if (c != NULL) { + if (is_in_group(c->group, user) <= 0) { + return 0; + } + } + + c = limit->consumers[CONSUMER_QUEUE]; + if (c != NULL) { + if (is_in_group(c->group, queue) <= 0) { + return 0; + } + } + + c = limit->consumers[CONSUMER_PROJECT]; + if (c != NULL) { + if (is_in_group(c->group, project) <= 0) { + return 0; + } + } + + if (!global) { + c = limit->consumers[CONSUMER_HOST]; + if (c != NULL) { + if (is_in_group(c->group, host) <= 0) { + return 0; + } + } + } + + return 1; +} + +int build_vres_name(char** vname, xce_limit_t* limit, char* user, char* queue, + char* project) +{ + char resname[2 * XCE_MAXNAME_LEN + 1]; + xce_consumer_t* c = NULL; + char* p = NULL; + int rc = 0; + + *vname = NULL; + if ((vname == NULL) || (limit == NULL) || (user == NULL) || (queue == NULL) + || (project == NULL)) { + return -EINVAL; + } + + memset(resname, 0x0, sizeof(resname)); + + p = &resname[0]; + + rc = sprintf(p, "limit=%s", limit->name); + if (rc < 0) { + return rc; + } + + p += rc; + + c = limit->consumers[CONSUMER_USER]; + + if ((c != NULL) && (c->per_consumer == 1)) { + rc = sprintf(p, "/u=%s", user); + if (rc < 0) { + return rc; + } + + p += rc; + } + + c = limit->consumers[CONSUMER_QUEUE]; + if ((c != NULL) && (c->per_consumer == 1)) { + rc = sprintf(p, "/q=%s", queue); + if (rc < 0) { + return rc; + } + + p += rc; + } + + c = limit->consumers[CONSUMER_PROJECT]; + if ((c != NULL) && (c->per_consumer == 1)) { + rc = sprintf(p, "/p=%s", project); + if (rc < 0) { + return rc; + } + + p += rc; + } + + *vname = strdup(resname); + + if ((*vname) != NULL) { + return 1; + } + + return -ENOMEM; +} + +int is_host_based_limit(resmgr_t* mgr, xce_limit_t* limit) +{ + xce_consumer_t* c = NULL; + xce_resource_t* r = NULL; + + if (limit == NULL) { + return -EINVAL; + } + + r = resmgr_lockres(mgr, limit->val.resname, OBJREF_NOP); + if ((r->flags & (XCE_RESF_GLOBAL | XCE_RESF_SHARED)) == 0) { + if (limit->val.percent != 0) { + return 1; + } + } + + c = limit->consumers[CONSUMER_HOST]; + if (c == NULL) { + return 0; + } + + return c->per_consumer; +} + +int add_vres(xce_resource_t** pvres, resmgr_t* mgr, char* name, int global) +{ + xce_resource_t* vres = NULL; + int rc; + + vres = (xce_resource_t*)calloc(1, sizeof(*vres)); + *pvres = vres; + + if (vres == NULL) { + return -ENOMEM; + } + + vres->name = strdup(name); + vres->idx = -1; + vres->desc = NULL; + vres->valtype = XCE_VAL_NUMERIC; + vres->ordtype = XCE_ORD_DECR; + vres->flags = XCE_RESF_LIMIT | XCE_RESF_CONSUMABLE | XCE_RESF_RELEASE; + if (global) { + vres->flags |= (XCE_RESF_SHARED | XCE_RESF_GLOBAL); + } else { + vres->flags |= XCE_RESF_DEFAULT_HOST; + } + + rc = resmgr_add_res(mgr, vres); + if (rc < 0) { + return rc; + } + + vres->idx = rc; + return rc; +} + +int instantiate_limit(xce_vres_t** pvres, resmgr_t* mgr, xce_limit_t* limit, + char* user, char* queue, char* project) +{ + xce_vres_t* vres = NULL; + int rc = 0; + char* vresname = NULL; + int shared = 1; + + if ((pvres == NULL) || (mgr == NULL) || (limit == NULL) || (user == NULL) + || (queue == NULL) || (project == NULL)) { + return -EINVAL; + } + + vres = (xce_vres_t*)calloc(1, sizeof(*vres)); + *pvres = vres; + + if (vres == NULL) { + return -ENOMEM; + } + + vres->limit = limit; + + if (is_host_based_limit(mgr, limit)) { + shared = 0; + } + + rc = build_vres_name(&(vresname), limit, user, queue, project); + if (rc < 0) { + return rc; + } + + vres->vres = resmgr_lockres(mgr, vresname, OBJREF_NOP); + if (vres->vres == NULL) { + rc = add_vres(&(vres->vres), mgr, vresname, shared); + if (rc < 0) { + FREEUP(vresname); + return rc; + } + } + + FREEUP(vresname); + + vres->res = resmgr_lockres(mgr, limit->val.resname, OBJREF_NOP); + if (vres->res == NULL) { + return -ENOENT; + } + + return rc; +} + +void kill_vres(xce_vres_t* vres) +{ + if (vres == NULL) { + return; + } +} diff --git a/src/lib/xlist.c b/src/lib/xlist.c index 0fd5923..ab48adb 100644 --- a/src/lib/xlist.c +++ b/src/lib/xlist.c @@ -46,6 +46,7 @@ static int xlist_ensure(xlist_t* l) int xlist_store(xlist_t* l, void* data) { int rc; + int idx; if (l == NULL) { return -EINVAL; @@ -56,10 +57,26 @@ int xlist_store(xlist_t* l, void* data) return rc; } - l->data[l->count] = data; + idx = l->count; + l->data[idx] = data; l->count++; - return 1; + return idx; +} + +void* xlist_get(xlist_t* l, int idx) +{ + if ((l == NULL) || (idx < 0)) { + errno = EINVAL; + return NULL; + } + + if (idx >= l->caps) { + errno = 0; + return NULL; + } + + return (void*)l->data[idx]; } void xlist_clear(xlist_t* l) @@ -86,6 +103,31 @@ int xlist_iter(xlist_t* l, xlist_iter_fn_t fn, void* data) return 0; } +int xlist_del(xlist_t* l, void* data) +{ + int i, j; + + if (l == NULL) { + return -EINVAL; + } + + j = -1; + i = 0; + while (i < l->count) { + if (l->data[i] == data) { + for (j = i + 1; j < l->count; j++) { + l->data[j - 1] = l->data[j]; + } + + l->count--; + } else { + i++; + } + } + + return 1; +} + void* xlist_find(xlist_t* l, xlist_find_fn_t fn, ...) { int i; diff --git a/src/scheduler/include/plugin.h b/src/scheduler/include/plugin.h index 5e82e34..3485f28 100644 --- a/src/scheduler/include/plugin.h +++ b/src/scheduler/include/plugin.h @@ -22,6 +22,8 @@ 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*); +typedef int (*plugin_job_election_hook_t)(void*, xce_sched_ctx_t*); +typedef int (*plugin_prepare_allocation_hook_t)(void*, xce_sched_ctx_t*); /* initialize function of plugin */ typedef int (*plugin_init_fn_t)(void*); @@ -31,25 +33,30 @@ typedef void (*plugin_fini_fn_t)(void*); /* plugin function to elect job to be scheduler */ typedef void* (*plugin_job_election_fn_t)(void*); +/* plugin callback when a job has been elected */ + 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, resreq_t*, - xce_pendreason_t**); +typedef int (*plugin_job_allocate_fn_t)(void*,xce_sched_ctx_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_job_state_change_hook_t jstate_hook; - plugin_update_worker_hook_t worker_hook; - plugin_commit_decision_hook_t commit_hook; + plugin_update_config_hook_t config_hook; + plugin_update_queue_hook_t queue_hook; + plugin_update_job_hook_t job_hook; + plugin_job_state_change_hook_t jstate_hook; + plugin_update_worker_hook_t worker_hook; + plugin_commit_decision_hook_t commit_hook; + plugin_job_election_hook_t job_election_hook; + plugin_prepare_allocation_hook_t prepare_allocation_hook; } plugin_hooks_t; typedef struct { char* name; void* sched; + uint32_t plugin_id; plugin_init_fn_t init_fn; plugin_fini_fn_t fini_fn; plugin_begin_session_fn_t begin_session_fn; @@ -71,7 +78,10 @@ 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*); extern int plugin_commit_decision(void*, xce_decision_t*); +extern int plugin_prepare_allocation(void*, xce_sched_ctx_t*); +extern int plugin_job_election(void*, xce_sched_ctx_t*); extern int register_fcfs(void*); +extern int register_limit(void*); extern int register_simple_allocator(void*); #endif diff --git a/src/scheduler/include/sched/types.h b/src/scheduler/include/sched/types.h index 114fdab..942b12b 100644 --- a/src/scheduler/include/sched/types.h +++ b/src/scheduler/include/sched/types.h @@ -8,11 +8,21 @@ #include "kvmgr.h" #include "typemap.h" #include "xce/types.h" +#include "xlist.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; +} xce_sched_alloc_req_t; + typedef enum { RESREQ_OP_SIMPLE = 0, /* simple just a list of resources and require values */ @@ -25,13 +35,14 @@ typedef enum { /* simple resource allocation just have numeric resource values */ typedef struct { - host_selection_t* selection; - int nres_shared; - int* idxs_shared; - float* vals_shared; - int nres_host; - int* idxs_host; - float* vals_host; + char* res; + float value; +} request_res_t; + +typedef struct { + host_selection_t selection; + xlist_t res_shared; + xlist_t res_host; } resreq_simple_t; typedef struct { @@ -65,7 +76,6 @@ 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; @@ -84,6 +94,24 @@ typedef struct { zarray_t resInUse; } xce_sched_worker_t; +typedef struct { + xce_sched_job_t* job; + xce_sched_worker_t* worker; + resreq_t* resreq; + xce_job_alloc_t* alloc; + xce_pendreason_t* reason; + xlist_t extra_res_requests; + xlist_t extra_host_res_requests; + xlist_t limits; /* non per host limits */ + xlist_t limits_host; /* PER HOST limits */ + xce_sched_alloc_req_t alloc_req; +} xce_sched_ctx_t; + +typedef struct { + void* id; + xlist_t members; +} xce_worker_group_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*); diff --git a/src/scheduler/include/scheduler.h b/src/scheduler/include/scheduler.h index cd7a654..e6c1528 100644 --- a/src/scheduler/include/scheduler.h +++ b/src/scheduler/include/scheduler.h @@ -9,12 +9,15 @@ #include "client.h" #include "dataq.h" #include "decision.h" +#include "group.h" #include "kvmgr.h" #include "mylock.h" #include "plugin.h" #include "xce/types.h" #include "xlist.h" +typedef void* (*sched_group_worker_fn_t)(void*, xce_sched_worker_t*); + typedef enum { XCE_SCHED_OP_NONE = 0, XCE_SCHED_OP_UPDATE_CONFIG, @@ -61,6 +64,10 @@ typedef struct { void* parallel_allocator; dataq_t* pend_updates[2]; resmgr_t resmgr; + groupmgr_t ugmgr; + groupmgr_t qgmgr; + groupmgr_t pgmgr; + groupmgr_t hgmgr; zarray_t sharedInst; zarray_t sharedInUse; kvmgr_t jobs; @@ -105,12 +112,21 @@ extern int sched_check_shared_satisfy(xce_scheduler_t*, int, int*, float*, int*, 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); + 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*); +extern void sched_ctx_fini(xce_sched_ctx_t*); +extern int sched_try_alloc_on_hosts(xce_worker_group_t*, va_list); +extern int sched_ctx_fill_shared_requests(xce_scheduler_t*, xce_sched_ctx_t*); +extern int sched_ctx_fill_host_requests(xce_scheduler_t*, xce_sched_ctx_t*); +extern int get_resvals(zarray_t*, int, int*, float*); +extern int put_resvals(zarray_t*, int, int*, float*); +extern request_res_t* make_resource_request(char*, float); #endif diff --git a/src/scheduler/meson.build b/src/scheduler/meson.build index 236cdcf..44feb8b 100644 --- a/src/scheduler/meson.build +++ b/src/scheduler/meson.build @@ -8,6 +8,7 @@ scheduler_srcs = [ 'src/resource.c', 'src/plugins/fcfs.c', 'src/plugins/simple_allocator.c', + 'src/plugins/limit.c', ] incdir = include_directories('src', diff --git a/src/scheduler/src/plugin.c b/src/scheduler/src/plugin.c index 0e4fe2a..79dcd96 100644 --- a/src/scheduler/src/plugin.c +++ b/src/scheduler/src/plugin.c @@ -32,6 +32,9 @@ int plugin_register(void* arg1, void* arg2) } rc = kvmgr_add(&(sched->plugins), (void*)plugin->name, (void*)plugin); + if (rc >= 0) { + plugin->plugin_id = rc; + } return rc; } @@ -345,3 +348,67 @@ int plugin_job_state_change(void* arg, xce_sched_job_t* job, uint32_t state) return rc; } + +static int plugin_job_election_(xce_sched_ctx_t* ctx, + xce_sched_plugin_t* plugin) +{ + if ((plugin == NULL) || (ctx == NULL)) { + return 0; + } + + if (plugin->hooks.job_election_hook != NULL) { + return plugin->hooks.job_election_hook(plugin, ctx); + } + + return 0; +} + +int plugin_job_election(void* arg, xce_sched_ctx_t* ctx) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xlist_t lst; + int rc; + + if ((sched == NULL) || (ctx == NULL)) { + return -EINVAL; + } + + xlist_init(&lst, 3); + xlist_store(&lst, plugin_job_election_); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, ctx); + xlist_fini(&lst); + + return rc; +} + +static int plugin_prepare_allocation_(xce_sched_ctx_t* ctx, + xce_sched_plugin_t* plugin) +{ + if ((plugin == NULL) || (ctx == NULL)) { + return 0; + } + + if (plugin->hooks.job_election_hook != NULL) { + return plugin->hooks.prepare_allocation_hook(plugin, ctx); + } + + return 0; +} + +int plugin_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) +{ + xce_scheduler_t* sched = (xce_scheduler_t*)arg; + xlist_t lst; + int rc; + + if ((sched == NULL) || (ctx == NULL)) { + return -EINVAL; + } + + xlist_init(&lst, 3); + xlist_store(&lst, plugin_prepare_allocation_); + rc = kvmgr_iter(&(sched->plugins), NULL, &lst, ctx); + xlist_fini(&lst); + + return rc; +} diff --git a/src/scheduler/src/plugins/limit.c b/src/scheduler/src/plugins/limit.c new file mode 100644 index 0000000..3aed17c --- /dev/null +++ b/src/scheduler/src/plugins/limit.c @@ -0,0 +1,443 @@ +#include +#include +#include +#include + +#include "kvmgr.h" +#include "objmgr.h" +#include "plugin.h" +#include "sched/types.h" +#include "scheduler.h" +#include "utils.h" +#include "xcelimit.h" +#include "xlist.h" + +typedef struct { + kvmgr_t limits; + kvmgr_t instances; +} limit_plugin_data_t; + +static int limit_priv_data_init(limit_plugin_data_t* data) +{ + kvmgr_init(&(data->limits), string_hash_key, 64, 64, + (zarray_item_killer_t)kill_limit); + kvmgr_init(&(data->instances), string_hash_key, 64, 64, + (zarray_item_killer_t)kill_vres); + return 0; +} + +static int limit_update_config(xce_sched_plugin_t* plugin, + xce_conf4sched_t* old, xce_conf4sched_t* config) +{ + limit_plugin_data_t* privdata = NULL; + xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; + int i; + xce_limit_config_t** limits = config->limits; + int nlimits = config->nlimits; + + TRACE("%s ...", __func__); + privdata = (limit_plugin_data_t*)plugin->data; + + for (i = 0; i < nlimits; i++) { + int rc; + xce_limit_t* limit; + + rc = build_limit(&limit, &(sched->ugmgr), &(sched->qgmgr), + &(sched->pgmgr), &(sched->hgmgr), limits[i]); + if (rc < 0) { + return rc; + } + + rc = kvmgr_add(&(privdata->limits), limit->name, limit); + if (rc < 0) { + kill_limit(limit); + + return rc; + } + } + + return 0; +} + +typedef struct { + xce_sched_plugin_t* plugin; + xce_sched_ctx_t* ctx; +} match_job_limit_args_t; + +static int update_limit_res_value(zarray_t* array, xce_vres_t* vres) +{ + float val = 0.0; + int rc = 0; + + if ((array == NULL) || (vres == NULL)) { + return -EINVAL; + } + + if (vres->limit->val.percent == 0) { + val = vres->limit->val.val; + } else { + rc = get_resvals(array, 1, &(vres->res->idx), &val); + if (rc < 0) { + return rc; + } + + val = val * vres->limit->val.val; + } + + rc = put_resvals(array, 1, &(vres->vres->idx), &val); + + return rc; +} + +static float get_vres_value(xce_sched_ctx_t* ctx, xce_vres_t* vres) +{ + int i; + resreq_t* resreq = ctx->resreq; + + if (resreq->op != RESREQ_OP_SIMPLE) { + return 0.0; + } + + for (i = 0; i < resreq->u.simple.res_shared.count; i++) { + request_res_t* rq + = (request_res_t*)xlist_get(&(resreq->u.simple.res_shared), i); + if (strcmp(vres->res->name, rq->res) == 0) { + return rq->value; + } + } + + for (i = 0; i < resreq->u.simple.res_host.count; i++) { + request_res_t* rq + = (request_res_t*)xlist_get(&(resreq->u.simple.res_host), i); + + if (strcmp(vres->res->name, rq->res) == 0) { + return rq->value; + } + } + + return 0.0; +} + +static int match_job_global_limit_(void* arg1, void* arg2) +{ + xce_limit_t* limit = (xce_limit_t*)arg2; + match_job_limit_args_t* args = (match_job_limit_args_t*)arg1; + xce_sched_ctx_t* ctx = NULL; + xce_sched_plugin_t* plugin = NULL; + xce_sched_job_t* job = NULL; + int rc = 0; + char* vresname = NULL; + xce_vres_t* vres = NULL; + limit_plugin_data_t* privdata = NULL; + xce_scheduler_t* sched = NULL; + + if ((limit == NULL) || (args == NULL)) { + return -EINVAL; + } + + ctx = args->ctx; + plugin = args->plugin; + if ((ctx == NULL) || (plugin == NULL)) { + return -EINVAL; + } + + job = ctx->job; + sched = plugin->sched; + privdata = (limit_plugin_data_t*)plugin->data; + if ((job == NULL) || (privdata == NULL) || (sched == NULL)) { + return -EINVAL; + } + + if (is_host_based_limit(&(sched->resmgr), limit) > 0) { + return 0; + } + + rc = match_limit(limit, 1, job->user->name, job->queue->name, + job->project->name); + TRACE("%s match_limit limit %s user %s queue %s project %s rc %d", __func__, + limit->name, job->user->name, job->queue->name, job->project->name, + rc); + if (rc <= 0) { + return rc; + } + + rc = build_vres_name(&vresname, limit, job->user->name, job->queue->name, + job->project->name); + if (rc < 0) { + return rc; + } + + vres = kvmgr_lock(&(privdata->instances), vresname, OBJREF_NOP); + FREEUP(vresname); + + if (vres == NULL) { + rc = instantiate_limit(&vres, &(sched->resmgr), limit, job->user->name, + job->queue->name, job->project->name); + if (rc < 0) { + return rc; + } + + rc = kvmgr_add(&(privdata->instances), vres->vres->name, vres); + if (rc < 0) { + return rc; + } + } + + rc = update_limit_res_value(&(sched->sharedInst), vres); + if (rc >= 0) { + rc = xlist_store(&(ctx->limits), vres); + } + return rc; +} + +static int limit_job_election(void* arg, xce_sched_ctx_t* ctx) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + limit_plugin_data_t* privdata = (limit_plugin_data_t*)plugin->data; + xce_sched_job_t* job = (xce_sched_job_t*)ctx->job; + int rc = 0; + xlist_t lst; + match_job_limit_args_t args; + + TRACE("%s job %lu plugin %s", __func__, job->jobid, plugin->name); + + args.plugin = plugin; + args.ctx = ctx; + + xlist_init(&lst, 3); + xlist_store(&lst, match_job_global_limit_); + rc = kvmgr_iter(&(privdata->limits), NULL, &lst, &args); + xlist_fini(&lst); + + return rc; +} + +static int match_job_host_limit_(void* arg1, void* arg2) +{ + xce_limit_t* limit = (xce_limit_t*)arg2; + match_job_limit_args_t* args = (match_job_limit_args_t*)arg1; + xce_sched_ctx_t* ctx = NULL; + xce_sched_plugin_t* plugin = NULL; + xce_sched_job_t* job = NULL; + int rc = 0; + char* vresname = NULL; + xce_vres_t* vres = NULL; + limit_plugin_data_t* privdata = NULL; + xce_scheduler_t* sched = NULL; + xce_sched_worker_t* worker = NULL; + + if ((limit == NULL) || (args == NULL)) { + return -EINVAL; + } + + ctx = args->ctx; + plugin = args->plugin; + if ((ctx == NULL) || (plugin == NULL)) { + return -EINVAL; + } + + job = ctx->job; + sched = plugin->sched; + privdata = (limit_plugin_data_t*)plugin->data; + worker = ctx->worker; + + if ((job == NULL) || (sched == NULL) || (privdata == NULL)) { + return -EINVAL; + } + + if (worker == NULL) { + return 0; + } + + if (is_host_based_limit(&(sched->resmgr), limit) <= 0) { + return 0; + } + + rc = match_limit(limit, 0, job->user->name, job->queue->name, + job->project->name, worker->name); + TRACE("%s match_limit limit %s user %s queue %s project %s rc %d", __func__, + limit->name, job->user->name, job->queue->name, job->project->name, + rc); + if (rc <= 0) { + return rc; + } + + rc = build_vres_name(&vresname, limit, job->user->name, job->queue->name, + job->project->name); + if (rc <= 0) { + return rc; + } + + vres = kvmgr_lock(&(privdata->instances), vresname, OBJREF_NOP); + FREEUP(vresname); + + if (vres == NULL) { + rc = instantiate_limit(&vres, &(sched->resmgr), limit, job->user->name, + job->queue->name, job->project->name); + if (rc < 0) { + return rc; + } + + rc = kvmgr_add(&(privdata->instances), vres->vres->name, vres); + if (rc < 0) { + return rc; + } + } + + rc = update_limit_res_value(&(worker->resvals), vres); + if (rc >= 0) { + rc = xlist_store(&(ctx->limits_host), vres); + } + + return rc; +} + +static int build_limit_vres_requests(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx, xlist_t* lst, + xlist_t* requests) +{ + int i; + + if ((sched == NULL) || (ctx == NULL) || (lst == NULL) + || (requests == NULL)) { + return -EINVAL; + } + + for (i = 0; i < requests->count; i++) { + request_res_t* rq = (request_res_t*)xlist_get(requests, i); + if (rq == NULL) { + continue; + } + + FREEUP(rq->res); + FREEUP(rq); + } + + xlist_clear(requests); + + for (i = 0; i < lst->count; i++) { + xce_vres_t* vres = (xce_vres_t*)xlist_get(lst, i); + request_res_t* rq = NULL; + + if (vres == NULL) { + continue; + } + + rq = (request_res_t*)calloc(1, sizeof(*rq)); + if (rq == NULL) { + continue; + } + + rq->res = strdup(vres->vres->name); + rq->value = get_vres_value(ctx, vres); + + xlist_store(requests, rq); + } + return 0; +} +static int limit_prepare_allocation(void* arg, xce_sched_ctx_t* ctx) +{ + xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg; + xce_sched_job_t* job = NULL; + xce_sched_worker_t* worker = NULL; + match_job_limit_args_t args; + xlist_t lst; + int rc = 0; + limit_plugin_data_t* privdata = NULL; + + if ((plugin == NULL) || (ctx == NULL)) { + return -EINVAL; + } + + worker = ctx->worker; + if (worker == NULL) { + /* we need to fill global limit requests here */ + return build_limit_vres_requests(plugin->sched, ctx, &(ctx->limits), + &(ctx->extra_res_requests)); + } + + job = ctx->job; + privdata = (limit_plugin_data_t*)plugin->data; + + TRACE("%s job %lu plugin %s", __func__, job->jobid, plugin->name); + + args.plugin = plugin; + args.ctx = ctx; + + xlist_fini(&(ctx->limits_host)); + xlist_init(&(ctx->limits_host), 32); + + xlist_init(&lst, 2); + xlist_store(&lst, match_job_host_limit_); + rc = kvmgr_iter(&(privdata->limits), NULL, &lst, &args); + xlist_fini(&lst); + + if (rc < 0) { + return rc; + } + /* fill host based limits here */ + rc = build_limit_vres_requests(plugin->sched, ctx, &(ctx->limits_host), + &(ctx->extra_host_res_requests)); + + return rc; +} + +static int limit_init(xce_sched_plugin_t* plugin) +{ + limit_plugin_data_t* privdata = NULL; + + privdata = (limit_plugin_data_t*)calloc(1, sizeof(*privdata)); + if (privdata == NULL) { + return -ENOMEM; + } + + limit_priv_data_init(privdata); + + plugin->data = privdata; + + return 0; +} + +static void limit_fini(xce_sched_plugin_t* plugin) +{ + limit_plugin_data_t* privdata = NULL; + + if (plugin == NULL) { + return; + } + + privdata = (limit_plugin_data_t*)plugin->data; + if (privdata != NULL) { + kvmgr_fini(&(privdata->limits)); + kvmgr_fini(&(privdata->instances)); + + FREEUP(privdata); + } + + FREEUP(plugin); +} + +int register_limit(void* sched) +{ + static xce_sched_plugin_t limit_plugin = { + .name = "xce_sched_limit", + .hooks + = { .job_election_hook = (plugin_job_election_hook_t)limit_job_election, + .config_hook = (plugin_update_config_hook_t)limit_update_config, + .prepare_allocation_hook + = (plugin_prepare_allocation_hook_t)limit_prepare_allocation }, + .init_fn = (plugin_init_fn_t)limit_init, + .fini_fn = (plugin_fini_fn_t)limit_fini, + .data = NULL + }; + + xce_sched_plugin_t* plugin + = (xce_sched_plugin_t*)calloc(1, sizeof(*plugin)); + if (plugin == NULL) { + return -ENOMEM; + } + + memcpy(plugin, &limit_plugin, sizeof(*plugin)); + + return plugin_register(sched, plugin); +} diff --git a/src/scheduler/src/plugins/simple_allocator.c b/src/scheduler/src/plugins/simple_allocator.c index f631c99..f8664a1 100644 --- a/src/scheduler/src/plugins/simple_allocator.c +++ b/src/scheduler/src/plugins/simple_allocator.c @@ -31,24 +31,6 @@ static void simple_allocator_fini(xce_sched_plugin_t* plugin) FREEUP(plugin); } -static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold, - xce_sched_job_t* j) -{ -#if 0 - if (jold == NULL) { - if (JOB_IS_RUN(j->state)) { - slotsInUse++; - } - } else { - if (JOB_IS_RUN(jold->state) && JOB_IS_FINISH(j->state)) { - slotsInUse--; - } - } -#endif - - return 1; -} - static int skip_bad_worker_(xce_sched_worker_t* w) { if (w == NULL) { @@ -59,81 +41,29 @@ static int skip_bad_worker_(xce_sched_worker_t* w) return 0; } -static int worker_is_satisfy_(xce_sched_worker_t* w, va_list ap) -{ - xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*); - uint64_t jobid = va_arg(ap, uint64_t); - resreq_t* resreq = va_arg(ap, resreq_t*); - xce_job_alloc_t** jalloc = va_arg(ap, xce_job_alloc_t**); - xce_pendreason_t** preason = va_arg(ap, xce_pendreason_t**); - int rc; - int unmeet_idxs[resreq->u.simple.nres_host + 1]; - float unmeet_vals[resreq->u.simple.nres_host + 1]; - int n_unmeet; - xce_host_alloc_t** hallocs = NULL; - xce_host_alloc_t* halloc = NULL; - xce_alloc_t** allocs = NULL; - - rc = sched_meet_host_selection(sched, w, resreq->u.simple.selection); - if (rc == 0) { - add_job_preason((*preason)->reasons, PREASON_MATCH_FAIL, w->name, NULL); - - return 0; - } - - rc = sched_check_host_satisfy( - sched, w, resreq->u.simple.nres_host, resreq->u.simple.idxs_host, - resreq->u.simple.vals_host, &n_unmeet, unmeet_idxs, unmeet_vals); - 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); - - add_job_preason((*preason)->reasons, PREASON_NO_RESOURCE, w->name, - r->name); - } - return 0; - } - - allocs = sched_make_allocs( - sched, "", resreq->u.simple.nres_host, resreq->u.simple.idxs_host, - resreq->u.simple.vals_host, /* FIXME: no duration now */ NULL); - - halloc = make_host_alloc(w->name, resreq->u.simple.nres_host, allocs); - hallocs = make_host_allocs(1, halloc); - - *jalloc = make_job_alloc(jobid, 1, hallocs); - - /* host is OK */ - return 1; -} - -static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid, - resreq_t* resreq, xce_pendreason_t** preason) +static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx) { xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched; - xce_job_alloc_t* jalloc = NULL; + xce_sched_job_t* job = ctx->job; + resreq_t* resreq = ctx->resreq; int n_unmeet; - int unmeet_idxs[resreq->u.simple.nres_shared + 1]; - float unmeet_vals[resreq->u.simple.nres_shared + 1]; + 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; - reason = make_pendreason(jobid); + reason = make_pendreason(job->jobid); + ctx->reason = reason; if (reason == NULL) { - return NULL; + return -ENOMEM; } - *preason = reason; - rc = sched_check_shared_satisfy( - sched, resreq->u.simple.nres_shared, resreq->u.simple.idxs_shared, - resreq->u.simple.vals_shared, &n_unmeet, unmeet_idxs, unmeet_vals); + sched, ctx->alloc_req.nres_shared, ctx->alloc_req.idxs_shared, + ctx->alloc_req.vals_shared, &n_unmeet, unmeet_idxs, unmeet_vals); if (rc < 0) { - return NULL; + return rc; } if (rc == 0) { @@ -146,63 +76,42 @@ static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid, r->name); } - *preason = reason; - return NULL; + return 0; } xlist_init(&workers, 128); rc = sched_collect_workers(sched, &workers, - (zarray_iter_skip_fn_t)skip_bad_worker_); + (zarray_iter_skip_fn_t)skip_bad_worker_, NULL); if (rc < 0) { - return NULL; + return rc; } - xlist_find(&workers, (xlist_find_fn_t)worker_is_satisfy_, sched, jobid, - resreq, &jalloc, preason); + xlist_find(&workers, (xlist_find_fn_t)sched_try_alloc_on_hosts, sched, ctx); - if (jalloc != NULL) { - if (resreq->u.simple.nres_shared > 0) { - jalloc->allocs_shared = sched_make_allocs( - sched, "", resreq->u.simple.nres_shared, - resreq->u.simple.idxs_shared, resreq->u.simple.vals_shared, + 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); - jalloc->nres_shared = resreq->u.simple.nres_shared; + ctx->alloc->nres_shared = ctx->alloc_req.nres_shared; } else { - jalloc->allocs_shared = NULL; - jalloc->nres_shared = 0; + ctx->alloc->allocs_shared = NULL; + ctx->alloc->nres_shared = 0; } + + return 1; } - return jalloc; -} - -static int my_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision) -{ - return 0; -} - -static int my_jstate_changed(xce_sched_plugin_t* plugin, xce_sched_job_t* job, - uint32_t state) -{ -#if 0 - if (JOB_IS_RUN(job->state) && JOB_IS_FINISH(state)) { - slotsInUse--; - } -#endif - return 0; } int register_simple_allocator(void* sched) { static xce_sched_plugin_t simple_allocator_plugin = { - .name = "xce_sched_simple_allocator", - .hooks - = { .config_hook = NULL, - .job_hook = (plugin_update_job_hook_t)my_update_job, - .jstate_hook = (plugin_job_state_change_hook_t)my_jstate_changed, - .commit_hook = (plugin_commit_decision_hook_t)my_commit }, + .name = "xce_sched_simple_allocator", + .hooks = { .config_hook = NULL, .commit_hook = NULL }, .init_fn = (plugin_init_fn_t)simple_allocator_init, .fini_fn = (plugin_fini_fn_t)simple_allocator_fini, .begin_session_fn = NULL, diff --git a/src/scheduler/src/resource.c b/src/scheduler/src/resource.c index 15d3102..a799fd8 100644 --- a/src/scheduler/src/resource.c +++ b/src/scheduler/src/resource.c @@ -4,13 +4,17 @@ #include "kvmgr.h" #include "objmgr.h" +#include "plugin.h" +#include "preason.h" #include "sched/types.h" #include "scheduler.h" #include "utils.h" #include "xce/types.h" +#include "xcealloc.h" +#include "xcelimit.h" #include "zarray.h" -static int get_resvals(zarray_t* array, int nres, int* idxs, float* vals) +int get_resvals(zarray_t* array, int nres, int* idxs, float* vals) { xce_val_t* xvals[nres + 1]; int rc = 0; @@ -40,7 +44,7 @@ static int get_resvals(zarray_t* array, int nres, int* idxs, float* vals) return nres; } -static int put_resvals(zarray_t* array, int nres, int* idxs, float* vals) +int put_resvals(zarray_t* array, int nres, int* idxs, float* vals) { xce_val_t* xvals[nres + 1]; int rc, i; @@ -188,7 +192,7 @@ int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w, } static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, - int naccounts, + int global, int naccounts, xce_res_account_t** accounts, int charge) { int i; @@ -210,6 +214,19 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, xce_resource_t* r = resmgr_lockres(&(sched->resmgr), acct->resname, OBJREF_NOP); + /* It's possible that r return NULL if acct->resname is a + * virtual resource of limit */ + if (r == NULL) { + int rc; + + rc = add_vres(&r, &(sched->resmgr), acct->resname, global); + if (rc < 0) { + return rc; + } + + r = resmgr_lockres(&(sched->resmgr), acct->resname, OBJREF_NOP); + } + if (r != NULL) { acct->residx = r->idx; } @@ -236,12 +253,83 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array, return 0; } +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_host_alloc_t** hallocs = NULL; + xce_host_alloc_t* halloc = NULL; + xce_alloc_t** allocs = NULL; + + rc = sched_meet_host_selection(sched, w, resreq->u.simple.selection); + if (rc == 0) { + add_job_preason(ctx->reason->reasons, PREASON_MATCH_FAIL, w->name, + NULL); + + return 0; + } + + ctx->worker = w; + + rc = plugin_prepare_allocation(sched, ctx); + if (rc < 0) { + ctx->worker = NULL; + + return rc; + } + + 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) { + int i; + + for (i = 0; i < n_unmeet; i++) { + xce_resource_t* r = resmgr_lock_byid(&(sched->resmgr), + unmeet_idxs[i], OBJREF_NOP); + + add_job_preason(ctx->reason->reasons, PREASON_NO_RESOURCE, w->name, + r->name); + } + return 0; + } + + 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); + + halloc = make_host_alloc(w->name, ctx->alloc_req.nres_host, allocs); + hallocs = make_host_allocs(1, halloc); + + ctx->alloc = make_job_alloc(job->jobid, 1, hallocs); + + /* host is OK */ + return 1; +} + +int sched_try_alloc_on_hosts(xce_worker_group_t* g, va_list ap) +{ + /* FIXME: !!! */ + xce_sched_worker_t* w = xlist_get(&(g->members), 0); + return sched_try_alloc_on_host_(w, ap); +} + static int sched_update_account(xce_scheduler_t* sched, xce_job_account_t* account, int charge) { int i; - sched_update_res_accounts(sched, &(sched->sharedInUse), + sched_update_res_accounts(sched, &(sched->sharedInUse), 1, account->nres_shared, account->accounts_shared, charge); for (i = 0; i < account->nhosts; i++) { @@ -257,7 +345,7 @@ static int sched_update_account(xce_scheduler_t* sched, continue; } - sched_update_res_accounts(sched, &(w->resInUse), haccount->naccount, + sched_update_res_accounts(sched, &(w->resInUse), 0, haccount->naccount, haccount->accounts, charge); } return 0; @@ -281,3 +369,101 @@ 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) +{ + int i, count, j, k; + xlist_t* lists[] = { NULL, NULL, NULL }; + + 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; + } + + lists[0] = &(ctx->resreq->u.simple.res_shared); + lists[1] = &(ctx->extra_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_shared[j] = r->idx; + ctx->alloc_req.vals_shared[j] = rq->value; + j++; + } + } + + ctx->alloc_req.nres_shared = j; + + return 0; +} + +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 }; + + 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; + } + + 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; +} diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index effd786..a66d855 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -9,6 +9,8 @@ #include "scheduler.h" #include "xce/types.h" #include "xcealloc.h" +#include "xlist.h" +#include "utils.h" void kill_sched_job(void* arg) { @@ -17,28 +19,45 @@ void kill_sched_job(void* arg) } } +request_res_t *make_resource_request(char *res, float value) { + request_res_t *req = NULL; + + req = (request_res_t *)calloc(1, sizeof(*req)); + if (req == NULL) { + return NULL; + } + + req->res = strdup(res); + req->value = value; + + return req; +} + static resreq_t* make_sched_resreq(xce_scheduler_t* sched, xce_job_t* j) { - resreq_t* resreq = NULL; - xce_resource_t* slots - = resmgr_lockres(&(sched->resmgr), (void*)"slots", OBJREF_NOP); + resreq_t* resreq = NULL; + request_res_t* req = NULL; resreq = (resreq_t*)calloc(1, sizeof(*resreq)); if (resreq == NULL) { return NULL; } - /* FIXME: need parse real resreq from job */ - resreq->op = RESREQ_OP_SIMPLE; - resreq->u.simple.selection = NULL; - resreq->u.simple.nres_shared = 0; - resreq->u.simple.idxs_shared = NULL; - resreq->u.simple.vals_shared = NULL; - resreq->u.simple.nres_host = 1; - resreq->u.simple.idxs_host = (int*)calloc(1, sizeof(int)); - resreq->u.simple.vals_host = (float*)calloc(1, sizeof(float)); - resreq->u.simple.idxs_host[0] = slots->idx; - resreq->u.simple.vals_host[0] = 1.0; + resreq->op = RESREQ_OP_SIMPLE; + resreq->u.simple.selection = NULL; + + xlist_init(&(resreq->u.simple.res_shared), 4); + xlist_init(&(resreq->u.simple.res_host), 4); + + req = make_resource_request("slots", 1.0); + if (req != NULL) { + xlist_store(&(resreq->u.simple.res_host), (void*)req); + } + + req = make_resource_request("mem", 4096.0); + if (req != NULL) { + xlist_store(&(resreq->u.simple.res_host), (void *)req); + } return resreq; } @@ -158,9 +177,14 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w) void kill_sched_user(void* arg) { + xce_sched_user_t *user = (xce_sched_user_t *)arg; + if (arg == NULL) { return; } + + FREEUP(user->name); + FREEUP(user); } xce_sched_user_t* make_sched_user(void* arg, char* name) @@ -168,14 +192,28 @@ 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; + user = (xce_sched_user_t *)calloc(1, sizeof(*user)); + if (user == NULL) { + return NULL; + } + + user->name = strdup(name); + + sched_add_user(sched, user); + return user; } void kill_sched_project(void* arg) { + xce_sched_project_t *project = (xce_sched_project_t *)arg; + if (arg == NULL) { return; } + + FREEUP(project->name); + FREEUP(project); } xce_sched_project_t* make_sched_project(void* arg, char* name) @@ -183,5 +221,53 @@ 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; + project = (xce_sched_project_t *)calloc(1, sizeof(*project)); + if (project == NULL) { + return NULL; + } + + if ((name == NULL) || (strlen(name) == 0)) { + project->name = strdup("default"); + } + else{ + project->name = strdup(name); + } + + sched_add_project(sched, project); + return project; } + +int sched_ctx_init(xce_sched_ctx_t* ctx, xce_sched_job_t* job) +{ + if ((ctx == NULL) || (job == NULL)) { + return -EINVAL; + } + + memset(ctx, 0x0, sizeof(*ctx)); + ctx->job = job; + + xlist_init(&(ctx->limits), 4); + xlist_init(&(ctx->limits_host), 4); + xlist_init(&(ctx->extra_res_requests), 4); + xlist_init(&(ctx->extra_host_res_requests), 4); + + return 0; +} + +void sched_ctx_fini(xce_sched_ctx_t* ctx) +{ + if (ctx == NULL) { + 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); + + xlist_fini(&(ctx->extra_res_requests)); + xlist_fini(&(ctx->extra_host_res_requests)); + xlist_fini(&(ctx->limits)); + xlist_fini(&(ctx->limits_host)); +} diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 4584a19..f9f04d9 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -31,49 +31,71 @@ static void* sched_job_picker(void* arg) return job_picker->job_elect_fn(job_picker); } -static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid) +static int sched_make_simple_allocation(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) { - xce_sched_job_t* job = sched_get_job(sched, jobid); xce_sched_plugin_t* plugin = NULL; - xce_job_alloc_t* alloc = NULL; - xce_pendreason_t* reason = NULL; - if (job == NULL) { - ERROR("%s job %lu can not be found", __func__, jobid); - return NULL; + plugin = sched->simple_allocator; + if (plugin->job_alloc_fn == NULL) { + ERROR("%s plugin %s has no allocation function", __func__, + plugin->name); + return -EINVAL; } + plugin_prepare_allocation(sched, ctx); + + sched_ctx_fill_shared_requests(sched, ctx); + + return plugin->job_alloc_fn(plugin, ctx); +} + +static int sched_make_allocation(xce_scheduler_t* sched, xce_sched_ctx_t* ctx) +{ + int rc = 0; + xce_sched_job_t* job = ctx->job; + if (job->resreq == NULL) { xce_pendreason_t* reason = NULL; - ERROR("%s job %lu has invalid resource requirement", __func__, jobid); - reason = make_pendreason(jobid); + ERROR("%s job %lu has invalid resource requirement", __func__, + job->jobid); + reason = make_pendreason(job->jobid); add_job_preason(reason->reasons, PREASON_BAD_RESREQ, NULL, NULL); - return make_pend_decision(reason); + ctx->reason = reason; + + return 0; } switch (job->resreq->op) { case RESREQ_OP_SIMPLE: - plugin = sched->simple_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(plugin, jobid, job->resreq, &reason); + ctx->resreq = job->resreq; + rc = sched_make_simple_allocation(sched, ctx); break; default: + rc = 0; break; } + return rc; +} - if (alloc == NULL) { - return make_pend_decision(reason); +static xce_decision_t* calc_decision(xce_scheduler_t* sched, + xce_sched_ctx_t* ctx) +{ + int rc = 0; + + rc = sched_make_allocation(sched, ctx); + if (rc < 0) { + return NULL; } - return make_dispatch_decision(alloc); + if (rc == 0) { + return make_pend_decision(ctx->reason); + } + + return make_dispatch_decision(ctx->alloc); } static int commit_decision(xce_scheduler_t* sched, xce_decision_t* decision) @@ -87,8 +109,10 @@ static void* sched_allocator(void* arg) xce_use(sched); while (1) { - uint64_t jobid; - xce_decision_t* decision = NULL; + uint64_t jobid; + xce_decision_t* decision = NULL; + xce_sched_job_t* job = NULL; + xce_sched_ctx_t schedctx; pthread_testcancel(); @@ -100,12 +124,24 @@ static void* sched_allocator(void* arg) continue; } + job = sched_get_job(sched, jobid); + if (job == NULL) { + TRACE("%s job %lu can not be found", __func__, jobid); + continue; + } + + sched_ctx_init(&schedctx, job); + + plugin_job_election(sched, &schedctx); + /* calculate decision for jobid */ - decision = calc_decision(sched, jobid); + decision = calc_decision(sched, &schedctx); if (decision != NULL) { commit_decision(sched, decision); cbuf_put(&(sched->decisions), &decision); } + + sched_ctx_fini(&schedctx); } return NULL; @@ -151,6 +187,7 @@ static void sched_pend_update_killer(data_item_t* it) static int sched_register_plugins(xce_scheduler_t* sched) { register_fcfs(sched); + register_limit(sched); register_simple_allocator(sched); return 0; @@ -187,6 +224,10 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, /* should be NULL as resmgr will provide hash find by resource name, the real resource stored in sched->config->resources[] */ NULL); + groupmgr_init(&(sched->ugmgr), 64, 64); + groupmgr_init(&(sched->qgmgr), 16, 16); + groupmgr_init(&(sched->pgmgr), 16, 16); + groupmgr_init(&(sched->hgmgr), 64, 64); kvmgr_init(&(sched->plugins), string_hash_key, 8, 8, (zarray_item_killer_t)plugin_unregister); @@ -262,6 +303,12 @@ void sched_fini(xce_scheduler_t* sched) dataq_fini(sched->pend_updates[0]); dataq_fini(sched->pend_updates[1]); + resmgr_fini(&(sched->resmgr)); + groupmgr_fini(&(sched->ugmgr)); + groupmgr_fini(&(sched->qgmgr)); + groupmgr_fini(&(sched->pgmgr)); + groupmgr_fini(&(sched->hgmgr)); + mylock_fini(&(sched->lock)); return; @@ -572,7 +619,14 @@ 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) { - return 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; + } + + return make_sched_user(sched, name); } int sched_add_project(xce_scheduler_t* sched, xce_sched_project_t* p) @@ -582,7 +636,14 @@ 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) { - return 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; + } + + return make_sched_project(sched, name); } /* FIXME this is just stub, we need to add more */ @@ -623,30 +684,74 @@ int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* dispatch) return xcecli_jobdispatch(sched->apictx, dispatch); } +typedef struct { + xlist_t* lst; + xce_scheduler_t* sched; + sched_group_worker_fn_t groupfn; +} add_worker_args_t; + +static void* default_worker_groupfn_(void* arg, xce_sched_worker_t* w) +{ + return (void*)w->name; +} + +static int wgroup_matched(xce_worker_group_t* wg, va_list ap) +{ + void* id = va_arg(ap, void*); + + return (wg->id == id); +} + +static xce_worker_group_t* find_match_worker_group(xlist_t* groups, void* id) +{ + xce_worker_group_t* wg = NULL; + + wg = xlist_find(groups, (xlist_find_fn_t)wgroup_matched, id); + + return wg; +} + static int _add_worker(void* arg1, void* arg2) { - xlist_t* l = (xlist_t*)arg1; - xce_sched_worker_t* w = (xce_sched_worker_t*)arg2; + add_worker_args_t* args = (add_worker_args_t*)arg1; + xce_sched_worker_t* w = (xce_sched_worker_t*)arg2; + xce_worker_group_t* wg = NULL; + void* gid = NULL; - if ((l == NULL) || (w == NULL)) { + if ((arg1 == NULL) || (w == NULL)) { return 0; } - xlist_store(l, w); + if (args->groupfn == NULL) { + args->groupfn = (sched_group_worker_fn_t)default_worker_groupfn_; + } + + gid = args->groupfn(args->sched, w); + wg = find_match_worker_group(args->lst, gid); + if (wg == NULL) { + wg = (xce_worker_group_t*)calloc(1, sizeof(*wg)); + wg->id = gid; + xlist_init(&(wg->members), 8); + xlist_store(args->lst, wg); + } + + xlist_store(&(wg->members), w); return 1; } int sched_collect_workers(xce_scheduler_t* sched, xlist_t* lst, - zarray_iter_skip_fn_t skipfn) + zarray_iter_skip_fn_t skipfn, + sched_group_worker_fn_t groupfn) { - xlist_t l2; - int rc = 0; + xlist_t l2; + int rc = 0; + add_worker_args_t args = { .lst = lst, .sched = sched, .groupfn = groupfn }; xlist_init(&l2, 2); xlist_store(&l2, (void*)_add_worker); - rc = kvmgr_iter(&(sched->workers), skipfn, &l2, lst); + rc = kvmgr_iter(&(sched->workers), skipfn, &l2, &args); xlist_fini(&l2); return rc; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index a0dfea6..2896197 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -74,7 +74,7 @@ struct TimeWindow $C.mapname("xce_timewindow_config_t") { segs @0: List(TimeWindowSeg) $C.mapname("segs") $C.maplistcount("nsegs"); } -struct Group $C.mapname("xce_group_t") { +struct GroupConfig $C.mapname("xce_group_config_t") { name @0: Text $C.mapname("name"); members @1: List(Text) $C.mapname("members.data") $C.maplistcount("members.count"); exmembers @2: List(Text) $C.mapname("ex_members.data") $C.maplistcount("ex_members.count"); @@ -149,6 +149,12 @@ struct Value $C.mapname("xce_val_t"){ } } +struct LimitValue $C.mapname("xce_limit_val_t") { + resname @0: Text $C.mapname("resname"); + percentage @1: Bool $C.mapname("percent"); + value @2: Float64 $C.mapname("val"); +} + struct ValueWithIdx $C.mapname("xce_val_with_idx_t") { idx @0: UInt32; val @1: Value $C.mapname("val"); @@ -191,8 +197,8 @@ struct QueueConf $C.mapname("xce_queue_t") { acceptIntervals @4: Int32 $C.mapname("accept_interval"); schedDelay @5: Int32 $C.mapname("sched_delay"); resreq @6: Text; - users @7: Group $C.mapname("users"); - hosts @8: Group $C.mapname("hosts"); + users @7: GroupConfig $C.mapname("users"); + hosts @8: GroupConfig $C.mapname("hosts"); thresholds @9: List(Threshold) $C.maplistcount("nthresholds"); preExec @10: Text $C.mapname("pre_exec"); postExec @11: Text $C.mapname("post_exec"); @@ -205,9 +211,11 @@ struct Conf4Sched $C.mapname("xce_conf4sched_t"){ configvars @1: List(ConfigVar) $C.mapname("confvars") $C.maplistcount("n_confvars"); resources @2: List(Resource) $C.maplistcount("n_resources"); queues @3: List(QueueConf) $C.maplistcount("n_queues"); - ugroups @4: List(Group) $C.maplistcount("n_ugroups"); - hgroups @5: List(Group) $C.maplistcount("n_hgroups"); + ugroups @4: List(GroupConfig) $C.maplistcount("n_ugroups"); + hgroups @5: List(GroupConfig) $C.maplistcount("n_hgroups"); resvals @6: List(ValueWithIdx) $C.mapname("resvals") $C.maplistcount("nresvals"); + hproviders @7: List(ScalerConfig) $C.mapname("hproviders") $C.maplistcount("nhproviders"); + limits @8: List(LimitConfig) $C.mapname("limits") $C.maplistcount("nlimits"); } struct Conf4Worker $C.mapname("xce_conf4worker_t") { @@ -609,6 +617,19 @@ struct ScalerConfig $C.mapname("scaler_config_t") { scriptOption @4: Text $C.mapname("script_option"); } +struct LimitConfig $C.mapname("xce_limit_config_t") { + name @0: Text $C.mapname("name"); + users @1: GroupConfig $C.mapname("users"); + queues @2: GroupConfig $C.mapname("queues"); + projects @3: GroupConfig $C.mapname("projects"); + hosts @4: GroupConfig $C.mapname("hosts"); + val @5: LimitValue $C.mapname("val"); + perUser @6: Bool $C.mapname("per_user"); + perQueue @7: Bool $C.mapname("per_queue"); + perProject @8: Bool $C.mapname("per_project"); + perHost @9: Bool $C.mapname("per_host"); +} + struct ClusterConfig $C.mapname("cluster_config_t") { clusterName @0: Text $C.mapname("clustername"); port @1: UInt32 $C.mapname("port"); @@ -620,8 +641,8 @@ struct ClusterConfig $C.mapname("cluster_config_t") { vcpus @7: UInt32 $C.mapname("vcpus"); resources @8: List(Resource) $C.mapname("resources") $C.maplistcount("nres"); queues @9: List(QueueConf) $C.mapname("queues") $C.maplistcount("nqueues"); - ugroups @10: List(Group) $C.mapname("ugroups") $C.maplistcount("nugroups"); - hgroups @11: List(Group) $C.mapname("hgroups") $C.maplistcount("nhgroups"); + ugroups @10: List(GroupConfig) $C.mapname("ugroups") $C.maplistcount("nugroups"); + hgroups @11: List(GroupConfig) $C.mapname("hgroups") $C.maplistcount("nhgroups"); hostConfigs @12: List(HostConfig) $C.mapname("hconfigs") $C.maplistcount("nhconfigs"); jobmgr @13: HashZArrayConfig; workermgr @14: HashZArrayConfig; @@ -629,4 +650,5 @@ struct ClusterConfig $C.mapname("cluster_config_t") { resmgr @16: ResmgrConfig; xstrings @17: HashZArrayConfig; hproviders @18: List(ScalerConfig) $C.mapname("hproviders") $C.maplistcount("nhproviders"); + limits @19: List(LimitConfig) $C.mapname("limits") $C.maplistcount("nlimits"); }