diff --git a/src/controller/apiserver.c b/src/controller/apiserver.c index 4bc37747..73935500 100644 --- a/src/controller/apiserver.c +++ b/src/controller/apiserver.c @@ -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: diff --git a/src/controller/dataex.c b/src/controller/dataex.c index 3ef60172..9a9c6464 100644 --- a/src/controller/dataex.c +++ b/src/controller/dataex.c @@ -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); diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 0919a72a..d2490040 100644 --- a/src/controller/include/cluster.h +++ b/src/controller/include/cluster.h @@ -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*); diff --git a/src/controller/opget.c b/src/controller/opget.c index f9a94a23..7a3b1506 100644 --- a/src/controller/opget.c +++ b/src/controller/opget.c @@ -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); diff --git a/src/controller/reply.c b/src/controller/reply.c index b1789836..79324435 100644 --- a/src/controller/reply.c +++ b/src/controller/reply.c @@ -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); diff --git a/src/controller/sched.c b/src/controller/sched.c index f6cd22b5..3b741ac1 100644 --- a/src/controller/sched.c +++ b/src/controller/sched.c @@ -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); diff --git a/src/frontend/apiserver/backend/convert.go b/src/frontend/apiserver/backend/convert.go index 2bab7ea0..a8fe7a93 100644 --- a/src/frontend/apiserver/backend/convert.go +++ b/src/frontend/apiserver/backend/convert.go @@ -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 &. diff --git a/src/frontend/apiserver/backend/convert_pure.go b/src/frontend/apiserver/backend/convert_pure.go index 7d5348c4..48b0048c 100644 --- a/src/frontend/apiserver/backend/convert_pure.go +++ b/src/frontend/apiserver/backend/convert_pure.go @@ -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 diff --git a/src/frontend/apiserver/models/hosts.go b/src/frontend/apiserver/models/hosts.go index ab30636f..512a7903 100644 --- a/src/frontend/apiserver/models/hosts.go +++ b/src/frontend/apiserver/models/hosts.go @@ -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 } diff --git a/src/include/typemap.h b/src/include/typemap.h index cbb1827d..b5672137 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -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; diff --git a/src/include/utils.h b/src/include/utils.h index 8a87d4da..9b0306a3 100644 --- a/src/include/utils.h +++ b/src/include/utils.h @@ -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); diff --git a/src/include/xce/types.h b/src/include/xce/types.h index 50d441ba..d951cf16 100644 --- a/src/include/xce/types.h +++ b/src/include/xce/types.h @@ -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 { diff --git a/src/lib/utils.c b/src/lib/utils.c index d5537e45..8d809309 100644 --- a/src/lib/utils.c +++ b/src/lib/utils.c @@ -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); diff --git a/src/scheduler/src/schedobj.c b/src/scheduler/src/schedobj.c index 948be293..a35bdf35 100644 --- a/src/scheduler/src/schedobj.c +++ b/src/scheduler/src/schedobj.c @@ -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; diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index ac1e0604..83d14d3d 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -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; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 4171d744..fd4b80f7 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -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"); }