wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

controller will pass resource <name, value> pair than <idx, value> pair

Browse source
This commit is contained in:
Rongsong Shen 2026-05-13 07:18:03 +00:00
parent 9f6741c2c4
commit d47b3375e7
16 changed files with 183 additions and 86 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/controller/apiserver.c
View file

@ -48,7 +48,7 @@ static int xce_apiserver_push_workers(cluster_t* c, endpoint_t* ep)
for (i = 0; i < workers.n_total; i++) {
xce_worker_t* w = workers.workers[i];
free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched);
free_name_val_list(&(w->resvals_to_sched), w->nresvals_to_sched);
}
unlock_collect_workers(&wlist);
@ -364,9 +364,9 @@ static int apiserver_worker_change(int type, int situation, void* data,
situation);
xlist_init(&resvalist, 256);
collect_resval_with_idx(&(w->host.resvals), &resvalist);
collect_resval(c->resmgr, &(w->host.resvals), &resvalist);
w->host.nresvals_to_sched = resvalist.count;
w->host.resvals_to_sched = (xce_val_with_idx_t**)resvalist.data;
w->host.resvals_to_sched = (xce_name_val_t**)resvalist.data;
switch (situation) {
case HOOK_AFTER_EVENT:

92
src/controller/dataex.c
View file

@ -19,57 +19,66 @@ int check_controller(void* arg)
return 1;
}
static int collect_resval_with_idx_(zarray_t* array, xlist_t* l)
int collect_resval(var resmgr, zarray_t* array, xlist_t* l)
{
int i;
iterator_t it;
int rc;
for (i = 0; i < array->caps; i++) {
xce_val_with_idx_t* val;
xce_val_t* src = array->data[i];
lock(resmgr, RTTI_LOCK_READ);
rc = iter_make(&it, resmgr);
if (rc < 0) {
unlock(resmgr);
return rc;
}
if (array->data[i] == NULL) {
for (;;) {
xce_resource_t* r = NULL;
xce_val_t* src = NULL;
xce_name_val_t* nv = NULL;
rc = iter_next(&it, (void**)&r);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (r == NULL) {
continue;
}
val = (xce_val_with_idx_t*)calloc(1, sizeof(*val));
if (val == NULL) {
return -ENOMEM;
src = (xce_val_t*)zarray_get(array, r->idx);
if (src == NULL) {
continue;
}
val->idx = i;
val->val = (xce_val_t*)calloc(1, sizeof(xce_val_t));
if (val->val == NULL) {
return -ENOMEM;
nv = (xce_name_val_t*)calloc(1, sizeof(*nv));
if (nv == NULL) {
rc = -ENOMEM;
break;
}
nv->name = strdup(r->name);
nv->val = (xce_val_t*)calloc(1, sizeof(xce_val_t));
if (nv->val == NULL) {
FREEUP(nv->name);
FREEUP(nv);
rc = -ENOMEM;
break;
}
if (src->which == Value_u_svalue) {
val->val->which = Value_u_svalue;
if (src->u.svalue != NULL) {
val->val->u.svalue = strdup(src->u.svalue);
} else {
val->val->u.svalue = strdup("-");
}
nv->val->which = Value_u_svalue;
nv->val->u.svalue = strdup(src->u.svalue ? src->u.svalue : "-");
} else {
memcpy(val->val, array->data[i], sizeof(xce_val_t));
memcpy(nv->val, src, sizeof(xce_val_t));
}
xlist_store(l, val);
xlist_store(l, nv);
}
return l->count;
}
iter_destroy(&it);
unlock(resmgr);
int collect_resval_with_idx(zarray_t* array, xlist_t* l)
{
int rc = 0;
mylock_lock(&(array->lock));
rc = collect_resval_with_idx_(array, l);
mylock_unlock(&(array->lock));
return rc;
return (rc == RTTI_ITER_NO_DATA || rc == 0) ? l->count : rc;
}
static void lock_objs(xlist_t* l, int op)
@ -138,7 +147,7 @@ static void lock_collect_jobs(xlist_t* l, int op)
}
}
static void collect_resvals_workers(xlist_t* l)
static void collect_resvals_workers(var resmgr, xlist_t* l)
{
int i;
@ -153,11 +162,11 @@ static void collect_resvals_workers(xlist_t* l)
xlist_init(&lst, 128);
collect_resval_with_idx(&(w->resvals), &lst);
collect_resval(resmgr, &(w->resvals), &lst);
w->nresvals_to_sched = lst.count;
MOVE_PTR(p, lst.data);
w->resvals_to_sched = (xce_val_with_idx_t**)p;
w->resvals_to_sched = (xce_name_val_t**)p;
xlist_fini(&lst);
}
@ -352,17 +361,14 @@ void collect_workers(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
lock_collect_workers(l, RTTI_LOCK_WRITE);
collect_resvals_workers(l);
collect_resvals_workers(c->resmgr, l);
return;
}
void collect_shared_resvals(cluster_t* c, collect_check_fn_t fn, xlist_t* l)
{
collect_resval_with_idx(&(c->sharedinst), l);
return;
collect_resval(c->resmgr, &(c->sharedinst), l);
}
int notify(cluster_t* c, endpoint_t* ep, xce_notify_data_t* data)
@ -401,7 +407,7 @@ void unlock_collect_workers(xlist_t* l)
continue;
}
free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched);
free_name_val_list(&(w->resvals_to_sched), w->nresvals_to_sched);
w1 = CONTAINER_OF(w, worker_t, host);
unlock(w1);

2
src/controller/include/cluster.h
View file

@ -118,7 +118,7 @@ extern int on_task_finish(cluster_t*, xce_task_finish_t*);
extern int xce_controller_res_init(cluster_t*);
extern int map_resource_to_host(char*, cluster_t*);
extern int release_resource_from_worker(char*, cluster_t*);
extern int collect_resval_with_idx(zarray_t*, xlist_t*);
extern int collect_resval(var, zarray_t*, xlist_t*);
extern int do_set_preason(cluster_t*, endpoint_t*, xce_request_t*);
extern int do_clr_preason(cluster_t*, endpoint_t*, xce_request_t*);
extern int handle_apiserver_leave(cluster_t*, endpoint_t*);

2
src/controller/opget.c
View file

@ -125,7 +125,7 @@ static int do_reply_workerlist(cluster_t* c, endpoint_t* ep,
for (i = 0; i < workerdata.n_total; i++) {
xce_worker_t* w = workerdata.workers[i];
free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched);
free_name_val_list(&(w->resvals_to_sched), w->nresvals_to_sched);
}
unlock_collect_workers(&wlist);

2
src/controller/reply.c
View file

@ -63,7 +63,7 @@ int handle_clusterconf(cluster_t* c, endpoint_t* ep, xce_request_t* request)
for (i = 0; i < clusterconf.wlist.count; i++) {
xce_worker_t* w = clusterconf.wlist.workers[i];
free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched);
free_name_val_list(&(w->resvals_to_sched), w->nresvals_to_sched);
}
unlock_collect_queues(&qlist);

16
src/controller/sched.c
View file

@ -117,7 +117,7 @@ static int xcesched_push_configs(cluster_t* c)
configdata.data.conf4sched->resources = (xce_resource_t**)rlist.data;
collect_shared_resvals(c, NULL, &xreslist);
configdata.data.conf4sched->nresvals = xreslist.count;
configdata.data.conf4sched->resvals = (xce_val_with_idx_t**)xreslist.data;
configdata.data.conf4sched->resvals = (xce_name_val_t**)xreslist.data;
configdata.data.conf4sched->n_ugroups = c->myconf->cluster_config->nugroups;
configdata.data.conf4sched->ugroups = c->myconf->cluster_config->ugroups;
@ -142,8 +142,8 @@ static int xcesched_push_configs(cluster_t* c)
rc = xce_pushconfig_req(c->sched.ep, c->myhost, &configdata);
free_resval_with_idx_list(&(configdata.data.conf4sched->resvals),
configdata.data.conf4sched->nresvals);
free_name_val_list(&(configdata.data.conf4sched->resvals),
configdata.data.conf4sched->nresvals);
FREEUP(configdata.data.conf4sched);
@ -223,7 +223,7 @@ static int xcesched_push_workers(cluster_t* c)
for (i = 0; i < workers.n_total; i++) {
xce_worker_t* w = workers.workers[i];
free_resval_with_idx_list(&(w->resvals_to_sched), w->nresvals_to_sched);
free_name_val_list(&(w->resvals_to_sched), w->nresvals_to_sched);
}
unlock_collect_workers(&wlist);
@ -439,9 +439,9 @@ static int sched_worker_change(int type, int situation, void* data,
situation);
xlist_init(&resvalist, 256);
collect_resval_with_idx(&(w->host.resvals), &resvalist);
collect_resval(c->resmgr, &(w->host.resvals), &resvalist);
w->host.nresvals_to_sched = resvalist.count;
w->host.resvals_to_sched = (xce_val_with_idx_t**)resvalist.data;
w->host.resvals_to_sched = (xce_name_val_t**)resvalist.data;
mylock_lock(&(c->sched.lock));
if (c->sched.ep != NULL) {
@ -460,8 +460,8 @@ static int sched_worker_change(int type, int situation, void* data,
#if 0
/*FIXME*/
free_resval_with_idx_list(&(w->host.resvals_to_sched),
w->host.nresvals_to_sched);
free_name_val_list(&(w->host.resvals_to_sched),
w->host.nresvals_to_sched);
#endif
xlist_fini(&resvalist);

62
src/frontend/apiserver/backend/convert.go
View file

@ -80,8 +80,9 @@ func cJobToModel(job *C.xce_job_t) *models.JobInfo {
// cWorkerToModel extracts C xce_worker_t fields into a rawWorkerData and maps it to HostInfo.
func cWorkerToModel(worker *C.xce_worker_t) *models.HostInfo {
workerName := goString(worker.name)
r := rawWorkerData{
name: goString(worker.name),
name: workerName,
status: int(worker.status),
}
if worker.attrs != nil {
@ -96,6 +97,16 @@ func cWorkerToModel(worker *C.xce_worker_t) *models.HostInfo {
r.maxTmp = int(a.maxtmp)
r.nDisks = int(a.ndisks)
}
n := int(worker.nresvals_to_sched)
if n > 0 {
entries := unsafe.Slice(worker.resvals_to_sched, n)
r.resvals = make([]models.ResourceDistributionInfo, 0, n)
for _, nv := range entries {
if rd := cNameValToWorkerResval(workerName, nv); rd != nil {
r.resvals = append(r.resvals, *rd)
}
}
}
return rawWorkerToModel(r)
}
@ -267,6 +278,55 @@ func freeJobSubmit(js *C.xce_job_submit_t, sd *C.job_submit_data_t, si *C.job_st
}
}
// cNameValToWorkerResval converts a C xce_name_val_t pointer to a ResourceDistributionInfo.
// workerName is set as Location. Returns nil if nv or nv.val is nil.
func cNameValToWorkerResval(workerName string, nv *C.xce_name_val_t) *models.ResourceDistributionInfo {
if nv == nil {
return nil
}
rd := &models.ResourceDistributionInfo{
ResourceName: C.GoString(nv.name),
Location: workerName,
IsDynamic: true,
}
if nv.val == nil {
logger.Warn("worker resval with nil value, skipping",
zap.String("resource", rd.ResourceName),
zap.String("host", workerName))
return nil
}
switch C.xce_val_which(nv.val) {
case C.Value_u_ivalue32:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_i32(nv.val))
case C.Value_u_ivalue64:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_i64(nv.val))
case C.Value_u_uvalue32:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_u32(nv.val))
case C.Value_u_uvalue64:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_u64(nv.val))
case C.Value_u_fvalue:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_fval(nv.val))
case C.Value_u_svalue:
rd.ValueType = int(models.STRING)
rd.StringValue = C.GoString(C.xce_val_str(nv.val))
case C.Value_u_bvalue:
rd.ValueType = int(models.BOOLEAN)
rd.BooleanValue = C.xce_val_bool(nv.val) != 0
default:
logger.Warn("unknown xce_val_t discriminant in worker resval, skipping",
zap.Int("which", int(C.xce_val_which(nv.val))),
zap.String("resource", rd.ResourceName),
zap.String("host", workerName))
return nil
}
return rd
}
// cResvalToDistribution converts a C xce_resval_t pointer to a Go ResourceDistributionInfo.
// Returns nil if vp.val is NULL (nil guard prevents CGO panic).
// vp.val is a *C.xce_val_t pointer field — pass directly, no &.

6
src/frontend/apiserver/backend/convert_pure.go
View file

@ -126,6 +126,7 @@ type rawWorkerData struct {
maxSwap int
maxTmp int
nDisks int
resvals []models.ResourceDistributionInfo
}
// rawQueueData mirrors the C xce_queue_t fields used by the Go model.
@ -204,8 +205,9 @@ func rawJobToModel(r rawJobData) *models.JobInfo {
// rawWorkerToModel maps a rawWorkerData to a models.HostInfo.
func rawWorkerToModel(r rawWorkerData) *models.HostInfo {
info := &models.HostInfo{
HostName: r.name,
HostStatus: r.status,
HostName: r.name,
HostStatus: r.status,
ResourceValues: r.resvals,
}
if r.hasAttrs {
info.HostType = r.hostType

5
src/frontend/apiserver/models/hosts.go
View file

@ -86,8 +86,9 @@ type HostInfo struct {
NumUSusp int `json:"num_ususp"`
//NumRsv int `json:"num_rsv"`
NumDone int `json:"num_done"`
Topology string `json:"topology"` // JSONB
SchedStop string `json:"sched_stop"` // JSONB
Topology string `json:"topology"` // JSONB
SchedStop string `json:"sched_stop"` // JSONB
ResourceValues []ResourceDistributionInfo `json:"resource_values"` // per-worker resource values
//RealLoad string `json:"real_load"` // JSONB
}

4
src/include/typemap.h
View file

@ -46,8 +46,8 @@ typedef struct {
xce_group_config_t** ugroups;
int n_hgroups;
xce_group_config_t** hgroups;
int nresvals;
xce_val_with_idx_t** resvals;
int nresvals;
xce_name_val_t** resvals;
int nlimits;
xce_limit_config_t** limits;
int nhproviders;

2
src/include/utils.h
View file

@ -127,7 +127,7 @@ extern xce_timewindow_config_t* dup_timewindow_config(xce_timewindow_config_t*);
extern void free_resource_thresholds(xce_res_threshold_t**, int);
extern xce_res_threshold_t** dup_resource_thresholds(xce_res_threshold_t**,
int);
extern void free_resval_with_idx_list(xce_val_with_idx_t***, int);
extern void free_name_val_list(xce_name_val_t***, int);
extern void free_strlist(xce_strlist_t*);
extern int fmt_jobid_ex(char*, size_t, uint64_t, char*, char*);
extern int fmt_jobid(char*, size_t, uint64_t);

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

@ -213,8 +213,8 @@ typedef struct {
int nthresholds;
xce_res_threshold_t** thresholds;
/* following two vars used to push resource values to sched */
int nresvals_to_sched;
xce_val_with_idx_t** resvals_to_sched;
int nresvals_to_sched;
xce_name_val_t** resvals_to_sched;
} xce_worker_t;
typedef struct {

22
src/lib/utils.c
View file

@ -385,10 +385,10 @@ int copy_strlist(xce_strlist_t* dst, xce_strlist_t src)
}
return 0;
}
void free_resval_with_idx_list(xce_val_with_idx_t*** vals, int count)
void free_name_val_list(xce_name_val_t*** vals, int count)
{
xce_val_with_idx_t** resvals = *vals;
int i;
xce_name_val_t** resvals = *vals;
int i;
if (resvals == NULL) {
return;
@ -399,18 +399,22 @@ void free_resval_with_idx_list(xce_val_with_idx_t*** vals, int count)
}
for (i = 0; i < count; i++) {
xce_val_with_idx_t* val = resvals[i];
xce_name_val_t* nv = resvals[i];
if (val == NULL) {
if (nv == NULL) {
continue;
}
if (val->val->which == Value_u_svalue) {
FREEUP(val->val->u.svalue);
FREEUP(nv->name);
if (nv->val != NULL) {
if (nv->val->which == Value_u_svalue) {
FREEUP(nv->val->u.svalue);
}
FREEUP(nv->val);
}
FREEUP(val->val);
FREEUP(val);
FREEUP(nv);
}
FREEUP(resvals);

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

@ -272,11 +272,23 @@ xce_sched_worker_t* make_sched_worker(void* arg, xce_worker_t* w)
/* 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;
xce_name_val_t* nv = w->resvals_to_sched[i];
xce_resource_t* r = NULL;
xce_val_t* val = NULL;
MOVE_PTR(val, w->resvals_to_sched[i]->val);
zarray_put(&(worker->resvals), idx, val);
if (nv == NULL || nv->name == NULL) {
continue;
}
if (key_find(sched->resmgr, (void*)nv->name, (void**)&r) < 0
|| r == NULL) {
WARN("%s unknown resource '%s' in worker resvals, skipping",
__func__, nv->name);
continue;
}
MOVE_PTR(val, nv->val);
zarray_put(&(worker->resvals), r->idx, val);
}
return worker;

20
src/scheduler/src/scheduler.c
View file

@ -576,11 +576,23 @@ 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;
xce_name_val_t* nv = sched->config->resvals[i];
xce_resource_t* r = NULL;
xce_val_t* val = NULL;
MOVE_PTR(val, sched->config->resvals[i]->val);
zarray_put(&(sched->sharedInst), idx, val);
if (nv == NULL || nv->name == NULL) {
continue;
}
if (key_find(sched->resmgr, (void*)nv->name, (void**)&r) < 0
|| r == NULL) {
WARN("%s unknown resource '%s' in resvals, skipping",
__func__, nv->name);
continue;
}
MOVE_PTR(val, nv->val);
zarray_put(&(sched->sharedInst), r->idx, val);
}
return 0;

4
src/schema/controller.capnp
View file

@ -126,7 +126,7 @@ struct Worker $C.mapname("xce_worker_t"){
status @2: UInt32;
dispatchWindow @3: TimeWindow $C.mapname("dispatch_window_config");
thresholds @4: List(Threshold) $C.mapname("thresholds") $C.maplistcount("nthresholds");
resvals @5: List(ValueWithIdx) $C.mapname("resvals_to_sched") $C.maplistcount("nresvals_to_sched");
resvals @5: List(NameValue) $C.mapname("resvals_to_sched") $C.maplistcount("nresvals_to_sched");
deadSince @6: UInt64 = 0 $C.mapname("dead_since"); # epoch seconds; set when worker becomes UNAVAILABLE
}
@ -227,7 +227,7 @@ struct Conf4Sched $C.mapname("xce_conf4sched_t"){
queues @3: List(QueueConf) $C.maplistcount("n_queues");
ugroups @4: List(GroupConfig) $C.maplistcount("n_ugroups");
hgroups @5: List(GroupConfig) $C.maplistcount("n_hgroups");
resvals @6: List(ValueWithIdx) $C.mapname("resvals") $C.maplistcount("nresvals");
resvals @6: List(NameValue) $C.mapname("resvals") $C.maplistcount("nresvals");
hproviders @7: List(ScalerConfig) $C.mapname("hproviders") $C.maplistcount("nhproviders");
limits @8: List(LimitConfig) $C.mapname("limits") $C.maplistcount("nlimits");
}