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

support limit

Browse source
This commit is contained in:
Rongsong Shen 2026-01-12 15:20:51 +08:00
parent bef32ace38
commit ecf1916d78
33 changed files with 3036 additions and 243 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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

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

@ -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*);

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

@ -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

1
src/scheduler/meson.build
View file

@ -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',

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

@ -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;
}

443
src/scheduler/src/plugins/limit.c Normal file
View file

@ -0,0 +1,443 @@
#include <errno.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#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);
}

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

@ -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,

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

@ -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;
}

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

@ -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));
}

173
src/scheduler/src/scheduler.c
View file

@ -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;