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

improve

Browse source
This commit is contained in:
Rongsong Shen 2026-01-15 17:56:53 +08:00
parent ecf1916d78
commit 06209930f9
13 changed files with 573 additions and 241 deletions
Download patch file Download diff file Expand all files Collapse all files

39
src/include/xvec.h Normal file
View file

@ -0,0 +1,39 @@
#if !defined(_XVEC_H_)
#define _XVEC_H_ 1
#include <stddef.h>
#include <stdint.h>
#include <stdlib.h>
typedef enum {
XVEC_ADD_INT = 0,
XVEC_SUB_INT,
XVEC_ADD_FLOAT,
XVEC_SUB_FLOAT
} xvec_op_t;
typedef struct {
int count;
int elsize;
void* data;
} xvec_t;
typedef struct {
int count;
xvec_t idxs;
xvec_t vals;
} xvec_idval_t;
extern int xvec_init(xvec_t*, int, int);
extern void xvec_fini(xvec_t*);
extern int xvec_zero(xvec_t*);
extern int xvec_resize(xvec_t*, int);
extern int xvec_op(xvec_t*, xvec_t*, xvec_op_t);
extern int xvec_dup(xvec_t**, xvec_t*);
extern int xvec_put(xvec_t*, int, void*);
extern int xvec_get(xvec_t*, int, void*);
extern int xvec_idval_init(xvec_idval_t*, int);
extern void xvec_idval_fini(xvec_idval_t*);
extern int xvec_idval_store(xvec_idval_t*, int, float);
#endif

1
src/lib/meson.build
View file

@ -35,6 +35,7 @@ libxce_src = [ gen_srcs[0],
'misc.c',
'group.c',
'xcelimit.c',
'xvec.c',
]
incdir = include_directories('../include')

270
src/lib/xvec.c Normal file
View file

@ -0,0 +1,270 @@
#include <errno.h>
#include <stdlib.h>
#include "utils.h"
#include "xvec.h"
int xvec_init(xvec_t* v, int count, int elsize)
{
if ((v == NULL) || (count <= 0) || (elsize <= 0)) {
return -EINVAL;
}
v->data = (void*)calloc(count, elsize);
if (v->data == NULL) {
return -ENOMEM;
}
v->count = count;
v->elsize = elsize;
return v->count;
}
void xvec_fini(xvec_t* v)
{
if (v == NULL) {
return;
}
FREEUP(v->data);
memset(v, 0x0, sizeof(*v));
}
int xvec_zero(xvec_t* v)
{
if (v == NULL) {
return -EINVAL;
}
if (v->data == NULL) {
return -EINVAL;
}
memset(v->data, 0x0, v->count * v->elsize);
return 1;
}
int xvec_resize(xvec_t* v, int new_size)
{
void* p = NULL;
if (v == NULL) {
return -EINVAL;
}
if (new_size <= v->count) {
/* do thing for shrink now */
return 0;
}
p = (void*)realloc(v->data, new_size * v->elsize);
if (p == NULL) {
return -ENOMEM;
}
v->data = p;
/* zero fill the extended space */
memset(((char*)p) + v->count * v->elsize, 0x0,
(new_size - v->count) * v->elsize);
v->count = new_size;
return 0;
}
#define X_OP(op) X_OP_##op
#define X_OP_add +=
#define X_OP_sub -=
#define DEFINE_VEC_OP(op, typ) \
static inline int vec_##typ##_##op(typ* v1, typ* v2, int count) \
{ \
int i; \
for (i = 0; i < count; i++) { \
v1[i] X_OP(op) v2[i]; \
} \
return 0; \
}
DEFINE_VEC_OP(add, int)
DEFINE_VEC_OP(sub, int)
DEFINE_VEC_OP(add, float)
DEFINE_VEC_OP(sub, float)
int xvec_op(xvec_t* v1, xvec_t* v2, xvec_op_t op)
{
int rc = 0;
if ((v1 == NULL) || (v2 == NULL)) {
return -EINVAL;
}
if (v1->elsize != v2->elsize) {
return -EINVAL;
}
switch (op) {
case XVEC_ADD_INT:
case XVEC_SUB_INT:
if (v1->elsize != sizeof(int)) {
return -EINVAL;
}
break;
case XVEC_ADD_FLOAT:
case XVEC_SUB_FLOAT:
if (v1->elsize != sizeof(float)) {
return -EINVAL;
}
break;
}
if (v1->count < v2->count) {
rc = xvec_resize(v1, v2->count);
if (rc < 0) {
return rc;
}
}
switch (op) {
case XVEC_ADD_INT:
rc = vec_int_add((int*)v1->data, (int*)v2->data, v2->count);
break;
case XVEC_SUB_INT:
rc = vec_int_sub((int*)v1->data, (int*)v2->data, v2->count);
break;
case XVEC_ADD_FLOAT:
rc = vec_float_add((float*)v1->data, (float*)v2->data, v2->count);
break;
case XVEC_SUB_FLOAT:
rc = vec_float_sub((float*)v1->data, (float*)v2->data, v2->count);
break;
}
return rc;
}
int xvec_dup(xvec_t** dst, xvec_t* src)
{
xvec_t* v = NULL;
v = (xvec_t*)calloc(1, sizeof(*v));
*dst = v;
if (v == NULL) {
return -ENOMEM;
}
v->data = (void*)calloc(src->count, src->elsize);
if (v->data == NULL) {
return -ENOMEM;
}
memcpy(v->data, src->data, src->elsize * src->count);
v->elsize = src->elsize;
v->count = src->count;
return v->count;
}
int xvec_put(xvec_t* v, int idx, void* data)
{
int rc;
char* p;
if ((v == NULL) || (idx < 0) || (data == NULL)) {
return -EINVAL;
}
rc = xvec_resize(v, idx + 1);
if (rc < 0) {
return rc;
}
p = (char*)v->data;
memcpy(p + v->elsize * idx, data, v->elsize);
return 1;
}
int xvec_get(xvec_t* v, int idx, void* data)
{
char* p = NULL;
if ((v == NULL) || (idx < 0) || (data == NULL)) {
return -EINVAL;
}
if (idx >= v->count) {
memset(data, 0x0, v->elsize);
return 0;
}
p = (char*)v->data;
memcpy(data, p + v->elsize * idx, v->elsize);
return 1;
}
int xvec_idval_init(xvec_idval_t* v, int initsize)
{
int rc;
if ((v == NULL) || (initsize <= 0)) {
return -EINVAL;
}
memset(v, 0x0, sizeof(*v));
rc = xvec_init(&(v->idxs), initsize, sizeof(int));
if (rc < 0) {
return rc;
}
rc = xvec_init(&(v->vals), initsize, sizeof(float));
if (rc < 0) {
return rc;
}
v->count = 0;
return 0;
}
void xvec_idval_fini(xvec_idval_t* v)
{
if (v == NULL) {
return;
}
xvec_fini(&(v->idxs));
xvec_fini(&(v->vals));
memset(v, 0x0, sizeof(*v));
return;
}
int xvec_idval_store(xvec_idval_t* v, int idx, float val)
{
int rc;
int pos;
if ((v == NULL) || (idx < 0)) {
return -EINVAL;
}
pos = v->count;
rc = xvec_put(&(v->idxs), pos, &idx);
if (rc < 0) {
return rc;
}
rc = xvec_put(&(v->vals), pos, &val);
if (rc < 0) {
return rc;
}
v->count = pos + 1;
return v->count;
}

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

@ -11,14 +11,10 @@
typedef void* plugin_data_t;
/* hooker of plugin, the first parameter should be xce_sched_plugin_t */
typedef int (*plugin_update_config_hook_t)(void*, xce_conf4sched_t*,
xce_conf4sched_t*);
typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*,
xce_sched_queue_t*);
typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*,
xce_sched_job_t*);
typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*,
xce_sched_worker_t*);
typedef int (*plugin_update_config_hook_t)(void*, xce_conf4sched_t*);
typedef int (*plugin_update_queue_hook_t)(void*, xce_sched_queue_t*);
typedef int (*plugin_update_job_hook_t)(void*, xce_sched_job_t*);
typedef int (*plugin_update_worker_hook_t)(void*, xce_sched_worker_t*);
typedef int (*plugin_job_state_change_hook_t)(void*, xce_sched_job_t*,
uint32_t);
typedef int (*plugin_commit_decision_hook_t)(void*, xce_decision_t*);
@ -40,7 +36,7 @@ typedef int (*plugin_begin_session_fn_t)(void*);
typedef int (*plugin_end_session_fn_t)(void*);
/* allocation function of job */
typedef int (*plugin_job_allocate_fn_t)(void*,xce_sched_ctx_t*);
typedef int (*plugin_job_allocate_fn_t)(void*, xce_sched_ctx_t*);
typedef struct {
plugin_update_config_hook_t config_hook;
@ -69,11 +65,10 @@ typedef struct {
extern int plugin_register(void*, void*);
extern void plugin_unregister(void*);
extern int plugin_config_update(void*, xce_conf4sched_t*, xce_conf4sched_t*);
extern int plugin_queue_update(void*, xce_sched_queue_t*, xce_sched_queue_t*);
extern int plugin_job_update(void*, xce_sched_job_t*, xce_sched_job_t*);
extern int plugin_worker_update(void*, xce_sched_worker_t*,
xce_sched_worker_t*);
extern int plugin_config_update(void*, xce_conf4sched_t*);
extern int plugin_queue_update(void*, xce_sched_queue_t*);
extern int plugin_job_update(void*, xce_sched_job_t*);
extern int plugin_worker_update(void*, xce_sched_worker_t*);
extern int plugin_job_state_change(void*, xce_sched_job_t*, uint32_t);
extern int plugin_begin_session(void*);
extern int plugin_end_session(void*);

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

@ -9,18 +9,15 @@
#include "typemap.h"
#include "xce/types.h"
#include "xlist.h"
#include "xvec.h"
#include "zarray.h"
/* FIXME: improve later */
typedef void* host_selection_t;
typedef struct {
int nres_shared;
int* idxs_shared;
float* vals_shared;
int nres_host;
int* idxs_host;
float* vals_host;
xvec_idval_t shared;
xvec_idval_t host;
} xce_sched_alloc_req_t;
typedef enum {
@ -100,6 +97,7 @@ typedef struct {
resreq_t* resreq;
xce_job_alloc_t* alloc;
xce_pendreason_t* reason;
xvec_idval_t unmeet;
xlist_t extra_res_requests;
xlist_t extra_host_res_requests;
xlist_t limits; /* non per host limits */

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

@ -15,6 +15,7 @@
#include "plugin.h"
#include "xce/types.h"
#include "xlist.h"
#include "xvec.h"
typedef void* (*sched_group_worker_fn_t)(void*, xce_sched_worker_t*);
@ -96,28 +97,28 @@ extern xce_sched_worker_t* sched_get_worker(xce_scheduler_t*, char*);
extern xce_sched_user_t* sched_get_user(xce_scheduler_t*, char*);
extern xce_sched_project_t* sched_get_project(xce_scheduler_t*, char*);
extern int sched_add_job(xce_scheduler_t*, xce_sched_job_t*);
extern int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*);
extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*);
extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*);
extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);
extern int sched_job_is_ready(xce_sched_job_t*);
extern int sched_queue_is_ready(xce_sched_queue_t*);
extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*);
extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);
extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*,
host_selection_t*);
extern int sched_check_shared_satisfy(xce_scheduler_t*, int, int*, float*, int*,
int*, float*);
extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*, int,
int*, float*, int*, int*, float*);
extern int sched_collect_workers(xce_scheduler_t*, xlist_t*,
zarray_iter_skip_fn_t,
sched_group_worker_fn_t);
extern int sched_charge_account(xce_scheduler_t*, xce_job_account_t*);
extern int sched_uncharge_account(xce_scheduler_t*, xce_job_account_t*);
extern int sched_commit_job_alloc(xce_scheduler_t*, xce_sched_job_t*);
extern uint64_t sched_get_id(xce_scheduler_t*);
extern int sched_add_queue(xce_scheduler_t*, xce_sched_queue_t*);
extern int sched_add_user(xce_scheduler_t*, xce_sched_user_t*);
extern int sched_add_worker(xce_scheduler_t*, xce_sched_worker_t*);
extern int sched_add_project(xce_scheduler_t*, xce_sched_project_t*);
extern int sched_job_is_ready(xce_sched_job_t*);
extern int sched_queue_is_ready(xce_sched_queue_t*);
extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*);
extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);
extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*,
host_selection_t*);
extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_idval_t*,
xvec_idval_t*);
extern int sched_check_host_satisfy(xce_scheduler_t*, xce_sched_worker_t*,
xvec_idval_t*, xvec_idval_t*);
extern int sched_collect_workers(xce_scheduler_t*, xlist_t*,
zarray_iter_skip_fn_t,
sched_group_worker_fn_t);
extern int sched_charge_account(xce_scheduler_t*, xce_job_account_t*);
extern int sched_uncharge_account(xce_scheduler_t*, xce_job_account_t*);
extern int sched_commit_job_alloc(xce_scheduler_t*, xce_sched_job_t*);
extern uint64_t sched_get_id(xce_scheduler_t*);
extern xce_alloc_t** sched_make_allocs(xce_scheduler_t*, char*, int, int*,
float*, int*);
extern int sched_ctx_init(xce_sched_ctx_t*, xce_sched_job_t*);

47
src/scheduler/src/plugin.c
View file

@ -7,11 +7,6 @@
#include "scheduler.h"
#include "xlist.h"
typedef struct {
void* old;
void* new;
} update_args_t;
int plugin_register(void* arg1, void* arg2)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg1;
@ -62,26 +57,24 @@ void plugin_unregister(void* arg)
static int plugin_config_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
update_args_t* args = (update_args_t*)arg1;
xce_conf4sched_t* config = (xce_conf4sched_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.config_hook != NULL) {
return plugin->hooks.config_hook(plugin, args->old, args->new);
return plugin->hooks.config_hook(plugin, config);
}
return 0;
}
int plugin_config_update(void* arg, xce_conf4sched_t* old,
xce_conf4sched_t* config)
int plugin_config_update(void* arg, xce_conf4sched_t* config)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
int rc;
update_args_t args = { .old = old, .new = config };
if (sched == NULL) {
return -EINVAL;
@ -89,7 +82,7 @@ int plugin_config_update(void* arg, xce_conf4sched_t* old,
xlist_init(&lst, 3);
xlist_store(&lst, plugin_config_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, config);
xlist_fini(&lst);
@ -99,26 +92,24 @@ int plugin_config_update(void* arg, xce_conf4sched_t* old,
static int plugin_job_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
update_args_t* args = (update_args_t*)arg1;
xce_sched_job_t* job = (xce_sched_job_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.job_hook != NULL) {
return plugin->hooks.job_hook(plugin, (xce_sched_job_t*)args->old,
(xce_sched_job_t*)args->new);
return plugin->hooks.job_hook(plugin, job);
}
return 0;
}
int plugin_job_update(void* arg, xce_sched_job_t* jold, xce_sched_job_t* job)
int plugin_job_update(void* arg, xce_sched_job_t* job)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
int rc;
update_args_t args = { .old = jold, .new = job };
if (sched == NULL) {
return -EINVAL;
@ -126,7 +117,7 @@ int plugin_job_update(void* arg, xce_sched_job_t* jold, xce_sched_job_t* job)
xlist_init(&lst, 3);
xlist_store(&lst, plugin_job_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, job);
xlist_fini(&lst);
@ -136,27 +127,24 @@ int plugin_job_update(void* arg, xce_sched_job_t* jold, xce_sched_job_t* job)
static int plugin_queue_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
update_args_t* args = (update_args_t*)arg1;
xce_sched_queue_t* queue = (xce_sched_queue_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.queue_hook != NULL) {
return plugin->hooks.queue_hook(plugin, (xce_sched_queue_t*)args->old,
(xce_sched_queue_t*)args->new);
return plugin->hooks.queue_hook(plugin, queue);
}
return 0;
}
int plugin_queue_update(void* arg, xce_sched_queue_t* qold,
xce_sched_queue_t* queue)
int plugin_queue_update(void* arg, xce_sched_queue_t* queue)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
int rc;
update_args_t args = { .old = qold, .new = queue };
if (sched == NULL) {
return -EINVAL;
@ -164,7 +152,7 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* qold,
xlist_init(&lst, 3);
xlist_store(&lst, plugin_queue_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, queue);
xlist_fini(&lst);
@ -174,27 +162,24 @@ int plugin_queue_update(void* arg, xce_sched_queue_t* qold,
static int plugin_worker_update_(void* arg1, void* arg2)
{
xce_sched_plugin_t* plugin = (xce_sched_plugin_t*)arg2;
update_args_t* args = (update_args_t*)arg1;
xce_sched_worker_t* worker = (xce_sched_worker_t*)arg1;
if ((arg1 == NULL) || (plugin == NULL)) {
return 0;
}
if (plugin->hooks.worker_hook != NULL) {
return plugin->hooks.worker_hook(plugin, (xce_sched_worker_t*)args->old,
(xce_sched_worker_t*)args->new);
return plugin->hooks.worker_hook(plugin, worker);
}
return 0;
}
int plugin_worker_update(void* arg, xce_sched_worker_t* wold,
xce_sched_worker_t* worker)
int plugin_worker_update(void* arg, xce_sched_worker_t* worker)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
xlist_t lst;
int rc;
update_args_t args = { .old = wold, .new = worker };
if (sched == NULL) {
return -EINVAL;
@ -202,7 +187,7 @@ int plugin_worker_update(void* arg, xce_sched_worker_t* wold,
xlist_init(&lst, 3);
xlist_store(&lst, plugin_worker_update_);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, &args);
rc = kvmgr_iter(&(sched->plugins), NULL, &lst, worker);
xlist_fini(&lst);

10
src/scheduler/src/plugins/fcfs.c
View file

@ -86,8 +86,8 @@ static int fcfs_priv_data_init(fcfs_plugin_data_t* data)
return 0;
}
static int fcfs_update_config(xce_sched_plugin_t* plugin, xce_conf4sched_t* old,
xce_conf4sched_t* config)
static int fcfs_update_config(xce_sched_plugin_t* plugin,
xce_conf4sched_t* config)
{
fcfs_plugin_data_t* privdata = NULL;
@ -124,8 +124,7 @@ static int job_match(void* arg1, va_list ap)
return (j->job->jobid == jobid);
}
static int fcfs_update_queue(xce_sched_plugin_t* plugin,
xce_sched_queue_t* qold, xce_sched_queue_t* q)
static int fcfs_update_queue(xce_sched_plugin_t* plugin, xce_sched_queue_t* q)
{
fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
ordq_item_t* it;
@ -165,8 +164,7 @@ static fcfs_queue_t* fcfs_get_queue(fcfs_plugin_data_t* data, char* name)
return (fcfs_queue_t*)it->data;
}
static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* jold,
xce_sched_job_t* j)
static int fcfs_update_job(xce_sched_plugin_t* plugin, xce_sched_job_t* j)
{
fcfs_plugin_data_t* privdata = (fcfs_plugin_data_t*)plugin->data;
fcfs_queue_t* q = NULL;

2
src/scheduler/src/plugins/limit.c
View file

@ -27,7 +27,7 @@ static int limit_priv_data_init(limit_plugin_data_t* data)
}
static int limit_update_config(xce_sched_plugin_t* plugin,
xce_conf4sched_t* old, xce_conf4sched_t* config)
xce_conf4sched_t* config)
{
limit_plugin_data_t* privdata = NULL;
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;

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

@ -46,9 +46,6 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
xce_scheduler_t* sched = (xce_scheduler_t*)plugin->sched;
xce_sched_job_t* job = ctx->job;
resreq_t* resreq = ctx->resreq;
int n_unmeet;
int unmeet_idxs[ctx->alloc_req.nres_shared + 1];
float unmeet_vals[ctx->alloc_req.nres_shared + 1];
int rc;
xlist_t workers;
xce_pendreason_t* reason = NULL;
@ -59,9 +56,8 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
return -ENOMEM;
}
rc = sched_check_shared_satisfy(
sched, ctx->alloc_req.nres_shared, ctx->alloc_req.idxs_shared,
ctx->alloc_req.vals_shared, &n_unmeet, unmeet_idxs, unmeet_vals);
rc = sched_check_shared_satisfy(sched, &(ctx->alloc_req.shared),
&(ctx->unmeet));
if (rc < 0) {
return rc;
}
@ -69,9 +65,9 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
if (rc == 0) {
int i;
for (i = 0; i < n_unmeet; i++) {
xce_resource_t* r = resmgr_lock_byid(&(sched->resmgr),
unmeet_idxs[i], OBJREF_NOP);
for (i = 0; i < ctx->alloc_req.shared.count; i++) {
xce_resource_t* r = resmgr_lock_byid(
&(sched->resmgr), ((int*)ctx->unmeet.idxs.data)[i], OBJREF_NOP);
add_job_preason(reason->reasons, PREASON_NO_RESOURCE, NULL,
r->name);
}
@ -90,12 +86,13 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
xlist_find(&workers, (xlist_find_fn_t)sched_try_alloc_on_hosts, sched, ctx);
if (ctx->alloc != NULL) {
if (ctx->alloc_req.nres_shared > 0) {
ctx->alloc->allocs_shared = sched_make_allocs(
sched, "", ctx->alloc_req.nres_shared,
ctx->alloc_req.idxs_shared, ctx->alloc_req.vals_shared,
/*FIXME: no duration now*/ NULL);
ctx->alloc->nres_shared = ctx->alloc_req.nres_shared;
if (ctx->alloc_req.shared.count > 0) {
ctx->alloc->allocs_shared
= sched_make_allocs(sched, "", ctx->alloc_req.shared.count,
ctx->alloc_req.shared.idxs.data,
ctx->alloc_req.shared.vals.data,
/*FIXME: no duration now*/ NULL);
ctx->alloc->nres_shared = ctx->alloc_req.shared.count;
} else {
ctx->alloc->allocs_shared = NULL;
ctx->alloc->nres_shared = 0;

296
src/scheduler/src/resource.c
View file

@ -28,6 +28,8 @@ int get_resvals(zarray_t* array, int nres, int* idxs, float* vals)
return 0;
}
memset(vals, 0x0, sizeof(*vals) * nres);
rc = zarray_get_ex(array, nres, idxs, (void**)xvals);
if (rc < 0) {
return rc;
@ -35,10 +37,14 @@ int get_resvals(zarray_t* array, int nres, int* idxs, float* vals)
for (i = 0; i < nres; i++) {
if (xvals[i] == NULL) {
vals[i] = 0.0;
} else {
vals[i] = xvals[i]->u.fvalue;
continue;
}
if (xvals[i]->which != Value_u_fvalue) {
continue;
}
vals[i] = xvals[i]->u.fvalue;
}
return nres;
@ -87,8 +93,8 @@ int put_resvals(zarray_t* array, int nres, int* idxs, float* vals)
return nres;
}
static int unsatisfy(int nres, int* idxs, float* total, float* inuse,
float* wanted, int* unmeet_idxs, float* unmeet_vals)
static int unsatisfy_(int nres, int* idxs, float* total, float* inuse,
float* wanted, int* unmeet_idxs, float* unmeet_vals)
{
int i, j;
int count;
@ -123,73 +129,148 @@ int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w,
return 1;
}
int sched_check_shared_satisfy(xce_scheduler_t* sched, int nres, int* idxs,
float* wanted, int* n_unmeet, int* unmeet_idxs,
float* unmeet_vals)
static int sched_check_shared_satisfy_(xce_scheduler_t* sched, xvec_t* total,
xvec_t* inuse, xvec_idval_t* wanted,
xvec_idval_t* unmeet)
{
float total[nres];
float inuse[nres];
int rc;
int nres = wanted->count;
int rc;
*n_unmeet = 0;
if (nres == 0) {
return 1;
}
rc = get_resvals(&(sched->sharedInst), nres, idxs, total);
rc = xvec_init(total, nres, sizeof(float));
if (rc < 0) {
return rc;
}
rc = get_resvals(&(sched->sharedInUse), nres, idxs, inuse);
rc = xvec_init(inuse, nres, sizeof(float));
if (rc < 0) {
return rc;
}
rc = unsatisfy(nres, idxs, total, inuse, wanted, unmeet_idxs, unmeet_vals);
rc = xvec_idval_init(unmeet, nres);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(sched->sharedInst), nres, (int*)wanted->idxs.data,
(float*)total->data);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(sched->sharedInUse), nres, (int*)wanted->idxs.data,
(float*)inuse->data);
if (rc < 0) {
return rc;
}
rc = unsatisfy_(nres, (int*)wanted->idxs.data, (float*)total->data,
(float*)inuse->data, (float*)wanted->vals.data,
(int*)unmeet->idxs.data, (float*)unmeet->vals.data);
if (rc < 0) {
return rc;
}
if (rc > 0) {
*n_unmeet = rc;
unmeet->count = rc;
return 0;
}
return 1;
}
int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w,
int nres, int* idxs, float* wanted, int* n_unmeet,
int* unmeet_idxs, float* unmeet_vals)
int sched_check_shared_satisfy(xce_scheduler_t* sched, xvec_idval_t* wanted,
xvec_idval_t* unmeet)
{
float total[nres];
float inuse[nres];
int rc;
xvec_t total;
xvec_t inuse;
int rc;
rc = get_resvals(&(w->resvals), nres, idxs, total);
if (wanted->count == 0) {
return 1;
}
rc = sched_check_shared_satisfy_(sched, &total, &inuse, wanted, unmeet);
if (rc < 0) {
xvec_idval_fini(unmeet);
}
xvec_fini(&total);
xvec_fini(&inuse);
return rc;
}
static int sched_check_host_satisfy_(xce_scheduler_t* sched,
xce_sched_worker_t* w, xvec_t* total,
xvec_t* inuse, xvec_idval_t* wanted,
xvec_idval_t* unmeet)
{
int nres = wanted->count;
int rc = 0;
rc = xvec_init(total, nres, sizeof(float));
if (rc < 0) {
return rc;
}
rc = get_resvals(&(w->resInUse), nres, idxs, inuse);
rc = xvec_init(inuse, nres, sizeof(float));
if (rc < 0) {
return rc;
}
rc = unsatisfy(nres, idxs, total, inuse, wanted, unmeet_idxs, unmeet_vals);
rc = xvec_idval_init(unmeet, nres);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(w->resvals), nres, (int*)wanted->idxs.data,
(float*)total->data);
if (rc < 0) {
return rc;
}
rc = get_resvals(&(w->resInUse), nres, (int*)wanted->idxs.data,
(float*)inuse->data);
if (rc < 0) {
return rc;
}
rc = unsatisfy_(nres, (int*)wanted->idxs.data, (float*)total->data,
(float*)inuse->data, (float*)wanted->vals.data,
(int*)unmeet->idxs.data, (float*)unmeet->vals.data);
if (rc < 0) {
return rc;
}
if (rc > 0) {
*n_unmeet = rc;
unmeet->count = rc;
return 0;
}
return 1;
}
int sched_check_host_satisfy(xce_scheduler_t* sched, xce_sched_worker_t* w,
xvec_idval_t* wanted, xvec_idval_t* unmeet)
{
xvec_t total;
xvec_t inuse;
int rc;
if (wanted->count == 0) {
return 1;
}
rc = sched_check_host_satisfy_(sched, w, &total, &inuse, wanted, unmeet);
if (rc < 0) {
xvec_idval_fini(unmeet);
}
xvec_fini(&total);
xvec_fini(&inuse);
return rc;
}
static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array,
int global, int naccounts,
@ -255,16 +336,11 @@ static int sched_update_res_accounts(xce_scheduler_t* sched, zarray_t* array,
static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
{
xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*);
xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*);
xce_sched_job_t* job = ctx->job;
resreq_t* resreq = ctx->resreq;
int rc;
int unmeet_idxs[ctx->alloc_req.nres_host + ctx->limits.count
+ ctx->limits_host.count + 1];
float unmeet_vals[ctx->alloc_req.nres_host + ctx->limits.count
+ ctx->limits_host.count + 1];
int n_unmeet;
xce_scheduler_t* sched = va_arg(ap, xce_scheduler_t*);
xce_sched_ctx_t* ctx = va_arg(ap, xce_sched_ctx_t*);
xce_sched_job_t* job = ctx->job;
resreq_t* resreq = ctx->resreq;
int rc;
xce_host_alloc_t** hallocs = NULL;
xce_host_alloc_t* halloc = NULL;
xce_alloc_t** allocs = NULL;
@ -288,15 +364,19 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
sched_ctx_fill_host_requests(sched, ctx);
rc = sched_check_host_satisfy(
sched, w, ctx->alloc_req.nres_host, ctx->alloc_req.idxs_host,
ctx->alloc_req.vals_host, &n_unmeet, unmeet_idxs, unmeet_vals);
if (rc <= 0) {
rc = sched_check_host_satisfy(sched, w, &(ctx->alloc_req.host),
&(ctx->unmeet));
if (rc < 0) {
/* FIXME: */
return rc;
}
if (rc == 0) {
int i;
for (i = 0; i < n_unmeet; i++) {
xce_resource_t* r = resmgr_lock_byid(&(sched->resmgr),
unmeet_idxs[i], OBJREF_NOP);
for (i = 0; i < ctx->unmeet.count; i++) {
xce_resource_t* r = resmgr_lock_byid(
&(sched->resmgr), ((int*)ctx->unmeet.idxs.data)[i], OBJREF_NOP);
add_job_preason(ctx->reason->reasons, PREASON_NO_RESOURCE, w->name,
r->name);
@ -305,10 +385,10 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
}
allocs = sched_make_allocs(
sched, "", ctx->alloc_req.nres_host, ctx->alloc_req.idxs_host,
ctx->alloc_req.vals_host, /* FIXME: no duration now */ NULL);
sched, "", ctx->alloc_req.host.count, ctx->alloc_req.host.idxs.data,
ctx->alloc_req.host.vals.data, /* FIXME: no duration now */ NULL);
halloc = make_host_alloc(w->name, ctx->alloc_req.nres_host, allocs);
halloc = make_host_alloc(w->name, ctx->alloc_req.host.count, allocs);
hallocs = make_host_allocs(1, halloc);
ctx->alloc = make_job_alloc(job->jobid, 1, hallocs);
@ -370,100 +450,74 @@ int sched_commit_job_alloc(xce_scheduler_t* sched, xce_sched_job_t* job)
return sched_charge_account(sched, job->accounts);
}
int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
static int fill_ctx_idval_vec_(xce_scheduler_t* sched, xvec_idval_t* v,
int nlists, xlist_t** lists)
{
int i, count, j, k;
xlist_t* lists[] = { NULL, NULL, NULL };
int count;
int i, rc;
count = ctx->resreq->u.simple.res_shared.count
+ ctx->extra_res_requests.count;
ctx->alloc_req.idxs_shared = (int*)calloc(count + 1, sizeof(int));
ctx->alloc_req.vals_shared = (float*)calloc(count + 1, sizeof(float));
if ((ctx->alloc_req.idxs_shared == NULL)
|| (ctx->alloc_req.vals_shared == NULL)) {
FREEUP(ctx->alloc_req.idxs_shared);
FREEUP(ctx->alloc_req.vals_shared);
return -ENOMEM;
count = 0;
for (i = 0; i < nlists; i++) {
count += lists[i]->count;
}
lists[0] = &(ctx->resreq->u.simple.res_shared);
lists[1] = &(ctx->extra_res_requests);
memset(v, 0x0, sizeof(*v));
rc = xvec_idval_init(v, count);
if (rc < 0) {
return rc;
}
j = 0;
for (i = 0; i < nlists; i++) {
xlist_t* lst = lists[i];
int j;
for (k = 0; lists[k] != NULL; k++) {
xlist_t* lst = lists[k];
for (i = 0; i < lst->count; i++) {
request_res_t* rq = xlist_get(lst, i);
for (j = 0; j < lst->count; j++) {
request_res_t* rq = (request_res_t*)xlist_get(lst, j);
xce_resource_t* r = NULL;
if (rq == NULL) {
continue;
return -EINVAL;
}
r = resmgr_lockres(&(sched->resmgr), rq->res, OBJREF_NOP);
if (r == NULL) {
continue;
return -ENOENT;
}
ctx->alloc_req.idxs_shared[j] = r->idx;
ctx->alloc_req.vals_shared[j] = rq->value;
j++;
rc = xvec_idval_store(v, r->idx, rq->value);
if (rc < 0) {
return rc;
}
}
}
ctx->alloc_req.nres_shared = j;
return v->count;
}
return 0;
int sched_ctx_fill_shared_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
{
int rc = 0;
xlist_t* lists[]
= { &(ctx->resreq->u.simple.res_shared), &(ctx->extra_res_requests) };
rc = fill_ctx_idval_vec_(sched, &(ctx->alloc_req.shared), 2, &lists[0]);
if (rc < 0) {
xvec_idval_fini(&(ctx->alloc_req.shared));
}
return rc;
}
int sched_ctx_fill_host_requests(xce_scheduler_t* sched, xce_sched_ctx_t* ctx)
{
int i, count, j, k;
xlist_t* lists[] = { NULL, NULL, NULL };
int rc;
xlist_t* lists[] = { &(ctx->resreq->u.simple.res_host),
&(ctx->extra_host_res_requests) };
count = ctx->resreq->u.simple.res_host.count
+ ctx->extra_host_res_requests.count;
ctx->alloc_req.idxs_host = (int*)calloc(count + 1, sizeof(int));
ctx->alloc_req.vals_host = (float*)calloc(count + 1, sizeof(float));
if ((ctx->alloc_req.idxs_host == NULL)
|| (ctx->alloc_req.vals_host == NULL)) {
FREEUP(ctx->alloc_req.idxs_host);
FREEUP(ctx->alloc_req.vals_host);
return -ENOMEM;
rc = fill_ctx_idval_vec_(sched, &(ctx->alloc_req.host), 2, &lists[0]);
if (rc < 0) {
xvec_idval_fini(&(ctx->alloc_req.host));
}
lists[0] = &(ctx->resreq->u.simple.res_host);
lists[1] = &(ctx->extra_host_res_requests);
j = 0;
for (k = 0; lists[k] != NULL; k++) {
xlist_t* lst = lists[k];
for (i = 0; i < lst->count; i++) {
request_res_t* rq = xlist_get(lst, i);
xce_resource_t* r = NULL;
if (rq == NULL) {
continue;
}
r = resmgr_lockres(&(sched->resmgr), rq->res, OBJREF_NOP);
if (r == NULL) {
continue;
}
ctx->alloc_req.idxs_host[j] = r->idx;
ctx->alloc_req.vals_host[j] = rq->value;
j++;
}
}
ctx->alloc_req.nres_host = j;
return 0;
return rc;
}

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

@ -11,6 +11,7 @@
#include "xcealloc.h"
#include "xlist.h"
#include "utils.h"
#include "xvec.h"
void kill_sched_job(void* arg)
{
@ -54,7 +55,7 @@ static resreq_t* make_sched_resreq(xce_scheduler_t* sched, xce_job_t* j)
xlist_store(&(resreq->u.simple.res_host), (void*)req);
}
req = make_resource_request("mem", 4096.0);
req = make_resource_request("mem", 1024.0);
if (req != NULL) {
xlist_store(&(resreq->u.simple.res_host), (void *)req);
}
@ -261,10 +262,8 @@ void sched_ctx_fini(xce_sched_ctx_t* ctx)
return;
}
FREEUP(ctx->alloc_req.idxs_shared);
FREEUP(ctx->alloc_req.vals_shared);
FREEUP(ctx->alloc_req.idxs_host);
FREEUP(ctx->alloc_req.vals_host);
xvec_idval_fini(&(ctx->alloc_req.shared));
xvec_idval_fini(&(ctx->alloc_req.host));
xlist_fini(&(ctx->extra_res_requests));
xlist_fini(&(ctx->extra_host_res_requests));

35
src/scheduler/src/scheduler.c
View file

@ -350,7 +350,7 @@ static int sched_init_queues(xce_scheduler_t* sched)
for (i = 0; i < config->n_queues; i++) {
xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]);
plugin_queue_update(sched, NULL, q);
plugin_queue_update(sched, q);
sched_add_queue(sched, q);
}
@ -358,7 +358,7 @@ static int sched_init_queues(xce_scheduler_t* sched)
}
int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
{
plugin_config_update(sched, NULL, config);
plugin_config_update(sched, config);
sched->config = config;
sched_init_resources(sched);
@ -374,14 +374,10 @@ static int sched_update_jobs(xce_scheduler_t* sched, int njobs,
int i;
for (i = 0; i < njobs; i++) {
xce_sched_job_t* jold = sched_get_job(sched, jobs[i]->jobid);
xce_sched_job_t* j = make_sched_job(sched, jobs[i]);
xce_sched_job_t* j = make_sched_job(sched, jobs[i]);
plugin_job_update(sched, jold, j);
if (jold == NULL) {
sched_add_job(sched, j);
}
plugin_job_update(sched, j);
sched_add_job(sched, j);
}
return 0;
@ -419,13 +415,10 @@ static int sched_update_workers(xce_scheduler_t* sched, int nworkers,
int i;
for (i = 0; i < nworkers; i++) {
xce_sched_worker_t* wold = sched_get_worker(sched, workers[i]->name);
xce_sched_worker_t* w = make_sched_worker(sched, workers[i]);
xce_sched_worker_t* w = make_sched_worker(sched, workers[i]);
plugin_worker_update(sched, wold, w);
if (wold == NULL) {
sched_add_worker(sched, w);
}
plugin_worker_update(sched, w);
sched_add_worker(sched, w);
}
return 0;
@ -619,8 +612,9 @@ int sched_add_user(xce_scheduler_t* sched, xce_sched_user_t* u)
xce_sched_user_t* sched_get_user(xce_scheduler_t* sched, char* name)
{
xce_sched_user_t *user = NULL;
user = (xce_sched_user_t *) kvmgr_lock(&(sched->users), (void*)name, OBJREF_NOP);
xce_sched_user_t* user = NULL;
user = (xce_sched_user_t*)kvmgr_lock(&(sched->users), (void*)name,
OBJREF_NOP);
if (user != NULL) {
return user;
@ -636,9 +630,10 @@ int sched_add_project(xce_scheduler_t* sched, xce_sched_project_t* p)
xce_sched_project_t* sched_get_project(xce_scheduler_t* sched, char* name)
{
xce_sched_project_t *project = NULL;
project = (xce_sched_project_t *) kvmgr_lock(&(sched->projects), (void*)name, OBJREF_NOP);
xce_sched_project_t* project = NULL;
project = (xce_sched_project_t*)kvmgr_lock(&(sched->projects), (void*)name,
OBJREF_NOP);
if (project != NULL) {
return project;
}