lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

scheduling using real slots number

Browse source
This commit is contained in:
Rongsong Shen 2025-12-30 18:13:15 +08:00
parent 8db4458375
commit 780e9fb313
22 changed files with 980 additions and 105 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/include/kvmgr.h
View file

@ -37,6 +37,7 @@ extern void kvmgr_fini(kvmgr_t*);
extern void* kvmgr_lock(kvmgr_t*, void*, int);
extern void* kvmgr_lock_byid(kvmgr_t*, int, int);
extern int32_t kvmgr_add(kvmgr_t*, void*, void*);
extern int32_t kvmgr_put(kvmgr_t*, void*, int, void*);
extern void kvmgr_del(kvmgr_t*, void*);
extern int kvmgr_iter(kvmgr_t*, zarray_iter_skip_fn_t, xlist_t*, void*);
@ -44,6 +45,7 @@ extern int resmgr_init(resmgr_t*, int, int, zarray_item_killer_t);
extern void resmgr_fini(resmgr_t*);
extern int32_t resmgr_add_res(resmgr_t*, xce_resource_t*);
extern xce_resource_t* resmgr_lockres(resmgr_t*, char*, int);
extern xce_resource_t* resmgr_lock_byid(resmgr_t*, int, int);
extern void resmgr_del_res(resmgr_t*, xce_resource_t*);
extern int resmgr_iter(resmgr_t*, xlist_t*, void*);

1
src/include/preason.h
View file

@ -6,6 +6,7 @@
#define PREASON_BAD_RESREQ (1)
#define PREASON_NO_RESOURCE (2)
#define PREASON_MATCH_FAIL (3)
extern void erase_job_preason(xce_job_preason_t*);
extern int add_job_preason(xce_job_preason_t*, int, char*, char*);

27
src/include/xce/types.h
View file

@ -257,7 +257,7 @@ typedef struct {
typedef struct {
char* resname;
char* reserve;
uint32_t instid;
uint64_t instid;
double value;
uint64_t from;
int64_t duration;
@ -269,6 +269,14 @@ typedef struct {
double used;
} res_usage_t;
typedef struct {
uint64_t instid;
char* resname;
int residx;
double used;
double reserved;
} xce_res_account_t;
typedef struct {
xce_worker_t host;
char hostkey[XCE_MAXKEY_LEN];
@ -283,12 +291,28 @@ typedef struct {
xce_alloc_t** allocs;
} xce_host_alloc_t;
typedef struct {
char* host;
int naccount;
xce_res_account_t** accounts;
} xce_host_account_t;
typedef struct {
uint64_t jobid;
int nhosts;
xce_host_alloc_t** allocs;
int nres_shared;
xce_alloc_t** allocs_shared;
} xce_job_alloc_t;
typedef struct {
uint64_t jobid;
int nhosts;
xce_host_account_t** accounts;
int nres_shared;
xce_res_account_t** accounts_shared;
} xce_job_account_t;
typedef struct {
int njobs;
xce_job_alloc_t** allocs;
@ -440,6 +464,7 @@ typedef struct {
job_pidinfo_t* pidinfo;
unsigned int state;
xce_job_alloc_t* allocs;
xce_job_account_t* accounts;
int run_time;
xce_job_runusage_t* run_usage;
job_exitinfo_t* exit_info;

4
src/include/xcealloc.h
View file

@ -3,13 +3,15 @@
#define _XCEALLOC_H_ 1
#include <stdarg.h>
#include <stdint.h>
#include "xce/types.h"
extern xce_alloc_t* make_alloc(char*, char*, int, float, int);
extern xce_alloc_t* make_alloc(char*, char*, uint64_t, float, int);
extern xce_host_alloc_t* make_host_alloc(char*, int, xce_alloc_t**);
extern xce_job_alloc_t* make_job_alloc(uint64_t, int, xce_host_alloc_t**);
extern xce_alloc_t** make_allocs(int, ...);
extern xce_host_alloc_t** make_host_allocs(int, ...);
extern xce_job_account_t* make_job_account(xce_job_alloc_t*);
#endif

14
src/include/xlist.h
View file

@ -2,7 +2,10 @@
#define __XLIST_H__ 1
#include <stdarg.h>
typedef int (*xlist_iter_fn_t)(void*, void*);
typedef int (*xlist_find_fn_t)(void*, va_list);
typedef struct {
int count;
@ -13,9 +16,10 @@ typedef struct {
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_clear(xlist_t*);
extern int xlist_iter(xlist_t*, xlist_iter_fn_t, void*);
extern void xlist_fini(xlist_t*);
extern int xlist_init(xlist_t*, int);
extern int xlist_store(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, ...);
extern void xlist_fini(xlist_t*);
#endif

4
src/include/zarray.h
View file

@ -2,12 +2,15 @@
#define _ZARRAY_H_ 1
#include <stdarg.h>
#include "mylock.h"
#include "xlist.h"
typedef void (*zarray_item_killer_t)(void*);
typedef int (*zarray_iter_fn_t)(void*, void*);
typedef int (*zarray_iter_skip_fn_t)(void*);
typedef int (*zarray_find_cond_fn_t)(void*, va_list);
typedef struct {
mylock_t lock;
@ -29,6 +32,7 @@ extern int zarray_put(zarray_t*, int, void*);
extern int zarray_put_ex(zarray_t*, int, int*, void**);
extern int zarray_get_ex(zarray_t*, int, int*, void**);
extern void* zarray_get(zarray_t*, int);
extern void* zarray_find(zarray_t*, zarray_find_cond_fn_t, ...);
extern void zarray_fini(zarray_t*);
#endif

54
src/lib/kvmgr.c
View file

@ -53,7 +53,7 @@ static void* kvmgr_lock_(kvmgr_t* mgr, void* key, int op)
if (op == OBJREF_NOP) {
return ptr;
}
return obj_ref(ptr, op);
}
@ -108,7 +108,7 @@ void* kvmgr_lock_byid(kvmgr_t* mgr, int idx, int op)
if (op == OBJREF_NOP) {
return ptr;
}
return obj_ref(ptr, op);
}
@ -152,6 +152,47 @@ int32_t kvmgr_add(kvmgr_t* mgr, void* key, void* p)
return idx;
}
static int32_t kvmgr_put_(kvmgr_t* mgr, void* key, int idx, void* p)
{
hent_t* e = NULL;
int idx1 = -1;
e = ht_get(&(mgr->htab), key);
if (e != NULL) {
idx1 = e->data;
if (idx1 >= 0) {
zarray_del(&(mgr->data), idx1);
}
} else {
e = ht_add(&(mgr->htab), (void*)key);
if (e == NULL) {
return -ENOMEM;
}
}
if (idx1 != idx) {
zarray_del(&(mgr->data), idx);
}
e->data = idx;
return zarray_put(&(mgr->data), idx, p);
}
int32_t kvmgr_put(kvmgr_t* mgr, void* key, int idx, void* p)
{
int rc;
if ((mgr == NULL) || (key == NULL) || (p == NULL) || (idx < 0)) {
return -EINVAL;
}
mylock_lock(&(mgr->lock));
rc = kvmgr_put_(mgr, key, idx, p);
mylock_unlock(&(mgr->lock));
return rc;
}
static void kvmgr_del_(kvmgr_t* mgr, void* key)
{
hent_t* e = NULL;
@ -245,6 +286,15 @@ xce_resource_t* resmgr_lockres(resmgr_t* mgr, char* resname, int op)
return r;
}
xce_resource_t* resmgr_lock_byid(resmgr_t* mgr, int idx, int op)
{
xce_resource_t* r = NULL;
r = (xce_resource_t*)kvmgr_lock_byid(&(mgr->resdef), idx, op);
return r;
}
void resmgr_del_res(resmgr_t* mgr, xce_resource_t* r)
{
kvmgr_del(&(mgr->resdef), (void*)r->name);

3
src/lib/meson.build
View file

@ -68,7 +68,8 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_idgen.c',
'testsuite/test_delay.c',
'testsuite/test_ordq.c',
'testsuite/test_preason.c'
'testsuite/test_preason.c',
'testsuite/test_kvmgr.c',
]
libxce_test_deps = [libxce_deps, unity_dep, libxce_dep]

2
src/lib/testsuite/test.c
View file

@ -15,6 +15,7 @@ extern void test_idgen(void);
extern void test_preason(void);
extern void test_ordq(void);
extern void test_delay(void);
extern void test_kvmgr(void);
void setUp(void) { return; }
@ -34,6 +35,7 @@ int main(int argc, char* argv[])
test_preason();
test_ordq();
test_delay();
test_kvmgr();
return UNITY_END();
}

76
src/lib/testsuite/test_kvmgr.c Normal file
View file

@ -0,0 +1,76 @@
#include <stdio.h>
#include <stdlib.h>
#include "kvmgr.h"
#include <unity.h>
#include "objmgr.h"
typedef struct {
char* key;
int data;
} kv_data_t;
static void test_kvadd(void)
{
kvmgr_t mgr;
char *key = "key1";
kv_data_t data = { .key = key, .data = 100 };
kv_data_t* p = NULL;
kvmgr_init(&mgr, string_hash_key, 4, 4, NULL);
kvmgr_add(&mgr, key, &data);
p = kvmgr_lock(&mgr, key, OBJREF_NOP);
TEST_ASSERT_TRUE(p != NULL);
TEST_ASSERT_TRUE(p == &data);
kvmgr_fini(&mgr);
}
static void test_kvput(void)
{
kvmgr_t mgr;
char *key = "key1";
kv_data_t data = { .key = key, .data = 100 };
kv_data_t* p = NULL;
int idx = 10;
kvmgr_init(&mgr, string_hash_key, 4, 4, NULL);
kvmgr_put(&mgr, key, idx, &data);
p = kvmgr_lock(&mgr, key, OBJREF_NOP);
TEST_ASSERT_TRUE(p != NULL);
TEST_ASSERT_TRUE(p == &data);
p = kvmgr_lock_byid(&mgr, idx, OBJREF_NOP);
TEST_ASSERT_TRUE(p != NULL);
TEST_ASSERT_TRUE(p == &data);
kvmgr_fini(&mgr);
}
static void test_kvdel(void)
{
kvmgr_t mgr;
char *key = "key1";
kv_data_t data = { .key = key, .data = 100 };
kv_data_t* p = NULL;
kvmgr_init(&mgr, string_hash_key, 4, 4, NULL);
kvmgr_add(&mgr, key, &data);
p = kvmgr_lock(&mgr, key, OBJREF_NOP);
TEST_ASSERT_TRUE(p != NULL);
TEST_ASSERT_TRUE(p == &data);
kvmgr_del(&mgr, key);
p = kvmgr_lock(&mgr, key, OBJREF_NOP);
TEST_ASSERT_TRUE( p == NULL);
kvmgr_fini(&mgr);
}
void test_kvmgr(void)
{
printf("\n>> kvmgr test\n");
RUN_TEST(test_kvadd);
RUN_TEST(test_kvput);
RUN_TEST(test_kvdel);
}

97
src/lib/xcealloc.c
View file

@ -4,11 +4,12 @@
#include <stdlib.h>
#include <string.h>
#include "utils.h"
#include "xce/types.h"
#include "xcealloc.h"
xce_alloc_t* make_alloc(char* resname, char* reserve, int instid, float value,
int duration)
xce_alloc_t* make_alloc(char* resname, char* reserve, uint64_t instid,
float value, int duration)
{
xce_alloc_t* alloc = NULL;
@ -32,6 +33,24 @@ xce_alloc_t* make_alloc(char* resname, char* reserve, int instid, float value,
return alloc;
}
static xce_res_account_t* make_res_account(xce_alloc_t* alloc)
{
xce_res_account_t* account = NULL;
account = (xce_res_account_t*)calloc(1, sizeof(*account));
if (account == NULL) {
return NULL;
}
account->instid = alloc->instid;
account->resname = strdup(alloc->resname);
account->residx = -1;
account->used = 0.0;
account->reserved = alloc->value;
return account;
}
xce_alloc_t** make_allocs(int nallocs, ...)
{
xce_alloc_t** allocs = NULL;
@ -74,6 +93,33 @@ xce_host_alloc_t* make_host_alloc(char* host, int nalloc, xce_alloc_t** allocs)
return halloc;
}
static xce_host_account_t* make_host_account(xce_host_alloc_t* halloc)
{
xce_host_account_t* haccount = NULL;
int i;
haccount = (xce_host_account_t*)calloc(1, sizeof(*haccount));
if (haccount == NULL) {
return NULL;
}
haccount->accounts = (xce_res_account_t**)calloc(
halloc->nallocs, sizeof(xce_res_account_t*));
if (haccount->accounts == NULL) {
FREEUP(haccount);
return NULL;
}
for (i = 0; i < halloc->nallocs; i++) {
haccount->accounts[i] = make_res_account(halloc->allocs[i]);
}
haccount->naccount = halloc->nallocs;
haccount->host = strdup(halloc->host);
return haccount;
}
xce_host_alloc_t** make_host_allocs(int nhosts, ...)
{
xce_host_alloc_t** hallocs = NULL;
@ -111,9 +157,50 @@ xce_job_alloc_t* make_job_alloc(uint64_t jobid, int nhost,
return NULL;
}
jalloc->jobid = jobid;
jalloc->nhosts = nhost;
jalloc->allocs = hallocs;
jalloc->jobid = jobid;
jalloc->nhosts = nhost;
jalloc->allocs = hallocs;
jalloc->nres_shared = 0;
jalloc->allocs_shared = NULL;
return jalloc;
}
xce_job_account_t* make_job_account(xce_job_alloc_t* jalloc)
{
xce_job_account_t* account = NULL;
int i;
account = (xce_job_account_t*)calloc(1, sizeof(*account));
if (account == NULL) {
return NULL;
}
account->accounts
= (xce_host_account_t**)calloc(1, sizeof(xce_host_account_t*));
if (account->accounts == NULL) {
FREEUP(account);
return NULL;
}
for (i = 0; i < jalloc->nhosts; i++) {
account->accounts[i] = make_host_account(jalloc->allocs[i]);
}
account->nhosts = jalloc->nhosts;
account->accounts_shared = NULL;
if (jalloc->nres_shared) {
account->accounts_shared = (xce_res_account_t**)calloc(
jalloc->nres_shared, sizeof(xce_res_account_t*));
for (i = 0; i < jalloc->nres_shared; i++) {
account->accounts_shared[i]
= make_res_account(jalloc->allocs_shared[i]);
}
account->nres_shared = jalloc->nres_shared;
}
return account;
}

30
src/lib/xlist.c
View file

@ -86,6 +86,36 @@ int xlist_iter(xlist_t* l, xlist_iter_fn_t fn, void* data)
return 0;
}
void* xlist_find(xlist_t* l, xlist_find_fn_t fn, ...)
{
int i;
va_list ap;
void* it = NULL;
if ((l == NULL) || (fn == NULL)) {
errno = EINVAL;
return NULL;
}
va_start(ap, fn);
for (i = 0; i < l->count; i++) {
va_list ap1;
int rc;
va_copy(ap1, ap);
rc = fn((void*)l->data[i], ap1);
va_end(ap1);
if (rc) {
it = (void*)l->data[i];
break;
}
}
return it;
}
void xlist_fini(xlist_t* l)
{
if (l == NULL) {

64
src/lib/zarray.c
View file

@ -189,6 +189,40 @@ int zarray_iter(zarray_t* array, zarray_iter_fn_t f, void* arg)
return rc;
}
void* zarray_find(zarray_t* array, zarray_find_cond_fn_t fn, ...)
{
int i;
va_list ap;
void* data = NULL;
if ((array == NULL) || (fn == NULL)) {
errno = EINVAL;
return NULL;
}
va_start(ap, fn);
mylock_lock(&(array->lock));
for (i = 0; i < array->caps; i++) {
va_list ap1;
int rc;
va_copy(ap1, ap);
rc = fn(array->data[i], ap1);
va_end(ap1);
if (rc) {
data = array->data[i];
break;
}
}
mylock_unlock(&(array->lock));
va_end(ap);
return data;
}
int zarray_del(zarray_t* array, int idx)
{
void* ptr = NULL;
@ -197,10 +231,14 @@ int zarray_del(zarray_t* array, int idx)
return -EINVAL;
}
if ((idx < 0) && (idx >= array->caps)) {
if (idx < 0) {
return -EINVAL;
}
if (idx >= array->caps) {
return 0;
}
mylock_lock(&(array->lock));
ptr = array->data[idx];
array->data[idx] = NULL;
@ -273,32 +311,22 @@ int zarray_put(zarray_t* array, int idx, void* data)
int zarray_get_ex(zarray_t* array, int count, int* idxs, void** data)
{
int i, maxid;
int i;
if ((array == NULL) || (count <= 0) || (idxs == NULL) || (data == NULL)) {
return -EINVAL;
}
maxid = 0;
for (i = 0; i < count; i++) {
if (idxs[i] < 0) {
return -EINVAL;
}
if (idxs[i] > maxid) {
maxid = idxs[i];
}
}
if (maxid >= array->caps) {
return -EINVAL;
}
mylock_lock(&(array->lock));
for (i = 0; i < count; i++) {
int idx = idxs[i];
data[i] = array->data[idx];
if (idx >= array->caps) {
data[i] = NULL;
}
else {
data[i] = array->data[idx];
}
}
mylock_unlock(&(array->lock));

2
src/scheduler/include/plugin.h
View file

@ -35,7 +35,7 @@ typedef int (*plugin_begin_session_fn_t)(void*);
typedef int (*plugin_end_session_fn_t)(void*);
/* allocation function of job */
typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t,
typedef xce_job_alloc_t* (*plugin_job_allocate_fn_t)(void*, int64_t, resreq_t*,
xce_pendreason_t**);
typedef struct {

35
src/scheduler/include/sched/types.h
View file

@ -10,14 +10,36 @@
#include "xce/types.h"
#include "zarray.h"
/* FIXME: improve later */
typedef void* host_selection_t;
typedef enum {
RESREQ_OP_SIMPLE
= 0, /* simple just a list of resources and require values */
RESREQ_OP_AFFINITY, /* with affinity allocation */
RESREQ_OP_GPU, /* with GPU allocation */
RESREQ_OP_PARALL, /* parall job nhosts, ptile and other constraints */
RESREQ_OP_AND, /* success for all */
RESREQ_OP_OR /* success for any */
} resreq_op_t;
/* simple resource allocation just have numeric resource values */
typedef struct {
host_selection_t* select;
int nres;
uint32_t* resources;
float* amount;
} rank_req_t;
host_selection_t* selection;
int nres_shared;
int* idxs_shared;
float* vals_shared;
int nres_host;
int* idxs_host;
float* vals_host;
} resreq_simple_t;
typedef struct {
resreq_op_t op;
union {
resreq_simple_t simple;
} u;
} resreq_t;
typedef struct {
char* name;
@ -48,10 +70,11 @@ typedef struct {
xce_sched_queue_t* queue;
xce_sched_user_t* user;
xce_sched_project_t* project;
void* allocator;
resreq_t* resreq;
uint32_t mintasks;
uint32_t maxtasks;
xce_job_alloc_t* allocs;
xce_job_account_t* accounts;
} xce_sched_job_t;
typedef struct {

18
src/scheduler/include/scheduler.h
View file

@ -52,11 +52,17 @@ typedef struct {
mylock_t lock;
int stop;
xcecli_ctx_t apictx;
uint64_t nextid;
xce_conf4sched_t* config;
kvmgr_t plugins;
void* simple_allocator;
void* affinity_allocator;
void* gpu_allocator;
void* parallel_allocator;
dataq_t* pend_updates[2];
resmgr_t resmgr;
zarray_t sharedInst;
zarray_t sharedInUse;
kvmgr_t jobs;
kvmgr_t queues;
kvmgr_t workers;
@ -92,5 +98,17 @@ extern int sched_queue_is_ready(xce_sched_queue_t*);
extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*);
extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);
extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*,
host_selection_t*);
extern int sched_check_shared_satisfy(xce_scheduler_t*, int, int*, float*, int*,
int*, float*);
extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*, int,
int*, float*, int*, int*, float*);
extern int sched_collect_workers(xce_scheduler_t*, xlist_t*,
zarray_iter_skip_fn_t);
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*);
#endif

1
src/scheduler/meson.build
View file

@ -5,6 +5,7 @@ scheduler_srcs = [
'src/decision.c',
'src/schedobj.c',
'src/plugin.c',
'src/resource.c',
'src/plugins/fcfs.c',
'src/plugins/simple_allocator.c',
]

174
src/scheduler/src/plugins/simple_allocator.c
View file

@ -5,17 +5,23 @@
#include <string.h>
#include "decision.h"
#include "kvmgr.h"
#include "objmgr.h"
#include "plugin.h"
#include "preason.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xlist.h"
static uint32_t maxslots = 8;
static uint32_t slotsInUse = 0;
static int simple_allocator_init(xce_sched_plugin_t* plugin)
{
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
static int simple_allocator_init(xce_sched_plugin_t* plugin) { return 1; }
sched->simple_allocator = plugin;
return 1;
}
static void simple_allocator_fini(xce_sched_plugin_t* plugin)
{
@ -28,10 +34,7 @@ static void simple_allocator_fini(xce_sched_plugin_t* plugin)
static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold,
xce_sched_job_t* j)
{
/* FIXME: we need to add logic to check whether the job should be handled by
* me */
j->allocator = plugin;
#if 0
if (jold == NULL) {
if (JOB_IS_RUN(j->state)) {
slotsInUse++;
@ -41,36 +44,134 @@ static int my_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold,
slotsInUse--;
}
}
#endif
return 1;
}
static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid,
xce_pendreason_t** preason)
static int skip_bad_worker_(xce_sched_worker_t* w)
{
xce_job_alloc_t* jalloc = NULL;
if (w == NULL) {
return 1;
}
*preason = NULL;
/* add more check here */
return 0;
}
if (slotsInUse < maxslots) {
xce_alloc_t* alloc = NULL;
xce_alloc_t** allocs = NULL;
xce_host_alloc_t* halloc = NULL;
xce_host_alloc_t** hallocs = NULL;
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;
alloc = make_alloc("slots", "", -1, 1.0, -1);
allocs = make_allocs(1, alloc);
halloc = make_host_alloc("pc13v", 1, allocs);
hallocs = make_host_allocs(1, halloc);
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);
jalloc = make_job_alloc(jobid, 1, hallocs);
} else {
xce_pendreason_t* reason = NULL;
return 0;
}
reason = make_pendreason(jobid);
add_job_preason(reason->reasons, PREASON_NO_RESOURCE, "pc13v", "slots");
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)
{
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
xce_job_alloc_t* jalloc = NULL;
int n_unmeet;
int unmeet_idxs[resreq->u.simple.nres_shared + 1];
float unmeet_vals[resreq->u.simple.nres_shared + 1];
int rc;
xlist_t workers;
xce_pendreason_t* reason = NULL;
reason = make_pendreason(jobid);
if (reason == NULL) {
return NULL;
}
*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);
if (rc < 0) {
return NULL;
}
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(reason->reasons, PREASON_NO_RESOURCE, NULL,
r->name);
}
*preason = reason;
return NULL;
}
xlist_init(&workers, 128);
rc = sched_collect_workers(sched, &workers,
(zarray_iter_skip_fn_t)skip_bad_worker_);
if (rc < 0) {
return NULL;
}
xlist_find(&workers, (xlist_find_fn_t)worker_is_satisfy_, sched, jobid,
resreq, &jalloc, preason);
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,
/*FIXME: no duration now*/ NULL);
jalloc->nres_shared = resreq->u.simple.nres_shared;
} else {
jalloc->allocs_shared = NULL;
jalloc->nres_shared = 0;
}
}
return jalloc;
@ -78,34 +179,17 @@ static xce_job_alloc_t* my_alloc(xce_sched_plugin_t* plugin, int64_t jobid,
static int my_commit(xce_sched_plugin_t* plugin, xce_decision_t* decision)
{
int rc = 0;
uint64_t jobid = 0;
xce_sched_job_t* job = NULL;
switch (decision->op) {
case XCE_DECISION_DISPATCH:
/* FIXME: */
jobid = decision->u.alloc->jobid;
job = sched_get_job(plugin->sched, jobid);
if (job != NULL) {
slotsInUse++;
job->starting = 1;
}
break;
default:
break;
}
return rc;
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;
}

266
src/scheduler/src/resource.c Normal file
View file

@ -0,0 +1,266 @@
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include "kvmgr.h"
#include "sched/types.h"
#include "scheduler.h"
#include "utils.h"
#include "xce/types.h"
#include "zarray.h"
#include "objmgr.h"
static int get_resvals(zarray_t* array, int nres, int* idxs, float* vals)
{
xce_val_t* xvals[nres + 1];
int rc = 0;
int i;
if ((array == NULL) || (nres < 0) || (idxs == NULL) || (vals == NULL)) {
return -EINVAL;
}
if (nres == 0) {
return 0;
}
rc = zarray_get_ex(array, nres, idxs, (void**)xvals);
if (rc < 0) {
return rc;
}
for (i = 0; i < nres; i++) {
if (xvals[i] == NULL) {
vals[i] = 0.0;
} else {
vals[i] = xvals[i]->u.fvalue;
}
}
return nres;
}
static int put_resvals(zarray_t* array, int nres, int* idxs, float* vals)
{
xce_val_t* xvals[nres + 1];
int rc, i;
if ((array == NULL) || (nres < 0) || (idxs == NULL) || (vals == NULL)) {
return -EINVAL;
}
if (nres == 0) {
return 0;
}
rc = zarray_get_ex(array, nres, idxs, (void**)xvals);
if (rc < 0) {
return rc;
}
for (i = 0; i < nres; i++) {
if (idxs[i] <= 0) {
continue;
}
if (xvals[i] == NULL) {
xce_val_t* val = (xce_val_t*)calloc(1, sizeof(*val));
if (val == NULL) {
continue;
}
val->which = Value_u_fvalue;
val->u.fvalue = 0.0;
zarray_put(array, idxs[i], val);
xvals[i] = val;
}
}
for (i = 0; i < nres; i++) {
xvals[i]->u.fvalue = vals[i];
}
return nres;
}
static int unsatisfy(int nres, int* idxs, float* total, float* inuse,
float* wanted, int* unmeet_idxs, float* unmeet_vals)
{
int i, j;
int count;
count = 0;
for (i = 0; i < nres; i++) {
if (total[i] < inuse[i] + wanted[i]) {
count++;
}
}
if (count == 0) {
return 0;
}
j = 0;
for (i = 0; i < nres; i++) {
if (total[i] < inuse[i] + wanted[i]) {
unmeet_idxs[j] = idxs[i];
unmeet_vals[j] = inuse[i] + wanted[i] - total[i];
j++;
}
}
return count;
}
int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w,
host_selection_t* s)
{
/* FIXME: add real work later */
return 1;
}
int sched_check_shared_satisfy(xce_scheduler_t* sched, int nres, int* idxs,
float* wanted, int* n_unmeet, int* unmeet_idxs,
float* unmeet_vals)
{
float total[nres];
float inuse[nres];
int rc;
*n_unmeet = 0;
if (nres == 0) {
return 1;
}
rc = get_resvals(&(sched->sharedInst), nres, idxs, total);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(sched->sharedInUse), nres, idxs, inuse);
if (rc < 0) {
return rc;
}
rc = unsatisfy(nres, idxs, total, inuse, wanted, unmeet_idxs, unmeet_vals);
if (rc < 0) {
return rc;
}
if (rc > 0) {
*n_unmeet = rc;
return 0;
}
return 1;
}
int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w,
int nres, int* idxs, float* wanted, int* n_unmeet,
int* unmeet_idxs, float* unmeet_vals)
{
float total[nres];
float inuse[nres];
int rc;
rc = get_resvals(&(w->resvals), nres, idxs, total);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(w->resInUse), nres, idxs, inuse);
if (rc < 0) {
return rc;
}
rc = unsatisfy(nres, idxs, total, inuse, wanted, unmeet_idxs, unmeet_vals);
if (rc < 0) {
return rc;
}
if (rc > 0) {
*n_unmeet = rc;
return 0;
}
return 1;
}
static int sched_charge_res_accounts(xce_scheduler_t* sched, zarray_t* array,
int naccounts,
xce_res_account_t** accounts)
{
int i;
int idxs[naccounts + 1];
float vals[naccounts + 1];
if (naccounts < 0) {
return -EINVAL;
}
if (naccounts == 0) {
return 0;
}
for (i = 0; i < naccounts; i++) {
xce_res_account_t* acct = accounts[i];
if (acct->residx < 0) {
xce_resource_t* r
= resmgr_lockres(&(sched->resmgr), acct->resname, OBJREF_NOP);
if (r != NULL) {
acct->residx = r->idx;
}
}
}
for (i = 0; i < naccounts; i++) {
idxs[i] = accounts[i]->residx;
}
get_resvals(array, naccounts, idxs, vals);
for (i = 0; i < naccounts; i++) {
vals[i] += (accounts[i]->used + accounts[i]->reserved);
}
put_resvals(array, naccounts, idxs, vals);
return 0;
}
static int sched_charge_account(xce_scheduler_t* sched,
xce_job_account_t* account)
{
int i;
sched_charge_res_accounts(sched, &(sched->sharedInUse),
account->nres_shared, account->accounts_shared);
for (i = 0; i < account->nhosts; i++) {
xce_host_account_t* haccount = account->accounts[i];
xce_sched_worker_t* w = NULL;
if (haccount == NULL) {
continue;
}
w = sched_get_worker(sched, haccount->host);
if (w == NULL) {
continue;
}
sched_charge_res_accounts(sched, &(w->resInUse), haccount->naccount,
haccount->accounts);
}
return 0;
}
int sched_commit_job_alloc(xce_scheduler_t* sched, xce_sched_job_t* job)
{
if (job->accounts == NULL) {
return 0;
}
return sched_charge_account(sched, job->accounts);
}

53
src/scheduler/src/schedobj.c
View file

@ -3,9 +3,12 @@
#include <string.h>
#include "controller.capnp.h"
#include "kvmgr.h"
#include "objmgr.h"
#include "sched/types.h"
#include "scheduler.h"
#include "xce/types.h"
#include "xcealloc.h"
void kill_sched_job(void* arg)
{
@ -14,6 +17,32 @@ void kill_sched_job(void* arg)
}
}
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 = (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;
return resreq;
}
xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
{
xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job));
@ -34,6 +63,13 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
job->mintasks = j->submit->mintasks;
job->maxtasks = j->submit->maxtasks;
job->priority = 0; /* FIXME: support job level priority */
job->resreq = make_sched_resreq(sched, j);
job->allocs = NULL;
job->accounts = NULL;
if (j->allocs != NULL) {
MOVE_PTR(job->allocs, j->allocs);
job->accounts = make_job_account(job->allocs);
}
return job;
}
@ -89,6 +125,7 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xce_sched_worker_t* worker = NULL;
int i;
if ((sched == NULL) || (w == NULL)) {
return NULL;
@ -102,9 +139,19 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
worker->name = strdup(w->name);
worker->status = w->status;
zarray_init(&(worker->resvals), 128, (zarray_item_killer_t)kill_sched_resval);
zarray_init(&(worker->resInUse), 128, (zarray_item_killer_t)kill_sched_resval);
/* FIXME: */
zarray_init(&(worker->resvals), 128,
(zarray_item_killer_t)kill_sched_resval);
zarray_init(&(worker->resInUse), 128,
(zarray_item_killer_t)kill_sched_resval);
/* fill resource values */
for (i = 0; i < w->nresvals_to_sched; i++) {
int idx = w->resvals_to_sched[i]->idx;
xce_val_t* val = NULL;
MOVE_PTR(val, w->resvals_to_sched[i]->val);
zarray_put(&(worker->resvals), idx, val);
}
return worker;
}

151
src/scheduler/src/scheduler.c
View file

@ -11,8 +11,10 @@
#include "kvmgr.h"
#include "objmgr.h"
#include "plugin.h"
#include "preason.h"
#include "scheduler.h"
#include "utils.h"
#include "xcealloc.h"
static void* sched_job_picker(void* arg)
{
@ -40,19 +42,32 @@ static xce_decision_t* calc_decision(xce_scheduler_t* sched, uint64_t jobid)
return NULL;
}
if (job->allocator == NULL) {
ERROR("%s job %lu has invalid allocator", __func__, jobid);
return NULL;
if (job->resreq == NULL) {
xce_pendreason_t* reason = NULL;
ERROR("%s job %lu has invalid resource requirement", __func__, jobid);
reason = make_pendreason(jobid);
add_job_preason(reason->reasons, PREASON_BAD_RESREQ, NULL, NULL);
return make_pend_decision(reason);
}
plugin = (xce_sched_plugin_t*)job->allocator;
if (plugin->job_alloc_fn == NULL) {
ERROR("%s plugin %s has no allocation function", __func__,
plugin->name);
return NULL;
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);
break;
default:
break;
}
alloc = plugin->job_alloc_fn(job->allocator, jobid, &reason);
if (alloc == NULL) {
return make_pend_decision(reason);
}
@ -86,10 +101,10 @@ static void* sched_allocator(void* arg)
/* calculate decision for jobid */
decision = calc_decision(sched, jobid);
commit_decision(sched, decision);
cbuf_put(&(sched->decisions), &decision);
if (decision != NULL) {
commit_decision(sched, decision);
cbuf_put(&(sched->decisions), &decision);
}
}
return NULL;
@ -153,6 +168,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
memset(sched, 0x0, sizeof(*sched));
sched->nextid = 1;
mylock_init(&(sched->lock));
cbuf_init(&(sched->wbuf), 1, sizeof(int));
cbuf_init(&(sched->jbuf), njobs, sizeof(uint64_t));
@ -185,6 +202,7 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
zarray_init(&(sched->sharedInst), 128,
(zarray_item_killer_t)kill_sched_resval);
zarray_init(&(sched->sharedInUse), 128, NULL);
sched_register_plugins(sched);
@ -261,6 +279,21 @@ static int sched_init_resources(xce_scheduler_t* sched)
return 0;
}
static int sched_init_sharedInst(xce_scheduler_t* sched)
{
int i;
for (i = 0; i < sched->config->nresvals; i++) {
int idx = sched->config->resvals[i]->idx;
xce_val_t* val = NULL;
MOVE_PTR(val, sched->config->resvals[i]->val);
zarray_put(&(sched->sharedInst), idx, val);
}
return 0;
}
static int sched_init_queues(xce_scheduler_t* sched)
{
int i;
@ -281,6 +314,7 @@ int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
sched->config = config;
sched_init_resources(sched);
sched_init_sharedInst(sched);
sched_init_queues(sched);
return 0;
@ -477,7 +511,23 @@ int sched_queue_is_ready(xce_sched_queue_t* q) { return 1; }
int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision)
{
return plugin_commit_decision(sched, decision);
int rc = 0;
xce_sched_job_t* job = NULL;
rc = plugin_commit_decision(sched, decision);
if (rc < 0) {
return rc;
}
if (decision->op == XCE_DECISION_DISPATCH) {
job = sched_get_job(sched, decision->u.alloc->jobid);
job->allocs = decision->u.alloc;
job->accounts = make_job_account(job->allocs);
return sched_commit_job_alloc(sched, job);
}
return 0;
}
int sched_publish_pend_reason(xce_scheduler_t* sched,
@ -490,3 +540,76 @@ int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* dispatch)
{
return xcecli_jobdispatch(sched->apictx, dispatch);
}
static int _add_worker(void* arg1, void* arg2)
{
xlist_t* l = (xlist_t*)arg1;
xce_sched_worker_t* w = (xce_sched_worker_t*)arg2;
if ((l == NULL) || (w == NULL)) {
return 0;
}
xlist_store(l, w);
return 1;
}
int sched_collect_workers(xce_scheduler_t* sched, xlist_t* lst,
zarray_iter_skip_fn_t skipfn)
{
xlist_t l2;
int rc = 0;
xlist_init(&l2, 2);
xlist_store(&l2, (void*)_add_worker);
rc = kvmgr_iter(&(sched->workers), skipfn, &l2, lst);
xlist_fini(&l2);
return rc;
}
uint64_t sched_get_id(xce_scheduler_t* sched)
{
uint64_t id;
mylock_lock(&(sched->lock));
id = sched->nextid++;
mylock_unlock(&(sched->lock));
return id;
}
xce_alloc_t** sched_make_allocs(xce_scheduler_t* sched, char* reserve, int nres,
int* idxs, float* vals, int* duration)
{
xce_alloc_t** allocs = NULL;
int i;
if ((sched == NULL) || (nres <= 0) || (idxs == NULL) || (vals == NULL)) {
errno = EINVAL;
return NULL;
}
allocs = (xce_alloc_t**)calloc(nres, sizeof(*allocs));
if (allocs == NULL) {
return NULL;
}
for (i = 0; i < nres; i++) {
xce_resource_t* r = (xce_resource_t*)resmgr_lock_byid(
&(sched->resmgr), idxs[i], OBJREF_NOP);
int d = (duration != NULL) ? duration[i] : -1;
allocs[i] = NULL;
if (r == NULL) {
continue;
}
allocs[i]
= make_alloc(r->name, reserve, sched_get_id(sched), vals[i], d);
}
return allocs;
}

7
src/schema/controller.capnp
View file

@ -359,7 +359,7 @@ struct Update $C.mapname("xce_update_data_t") {
struct ResAlloc $C.mapname("xce_alloc_t"){
name @0: Text $C.mapname("resname");
resvId @1: Text $C.mapname("reserve");
instid @2: UInt32;
instid @2: UInt64;
value @3: Float64;
from @4: UInt64; # alloc start time
duration @5: Int64;
@ -371,8 +371,9 @@ struct HostAlloc $C.mapname("xce_host_alloc_t"){
}
struct JobAlloc $C.mapname("xce_job_alloc_t"){
jobid @0: UInt64;
allocs @1: List(HostAlloc) $C.mapname("allocs") $C.maplistcount("nhosts");
jobid @0: UInt64;
allocs @1: List(HostAlloc) $C.mapname("allocs") $C.maplistcount("nhosts");
allocsShared @2: List(ResAlloc) $C.mapname("allocs_shared") $C.maplistcount("nres_shared");
}
struct JobDispatch $C.mapname("xce_job_dispatch_t"){