myxce/src/frontend/apiserver/backend/convert.go
Rongsong Shen bc3103a6d5 feat: P1-1 handle resource value push via OPCode_notify
Add OPCode_notify handler in go_push_handler top-level switch
(not inside handlePush) to receive resource value updates from
the controller via xce_notify_data_t/schedUpdateResvals.

- Add 8 CGO union accessors for xce_val_t (which, i32, i64,
  u32, u64, fval, str, bool) since CGO cannot access union fields
- Add handleNotifyData() dispatching NotifyData_data_schedUpdateResvals
  events, iterating xce_resval_list_t and calling cResvalToDistribution
- Add cResvalToDistribution in convert.go with 7 separate switch
  cases for Value_u_* union discriminants (no combining i32/i64)
- Add rawResvalData + rawResvalToDistribution pure-Go version
  in convert_pure.go for testability without CGO
- Add 10 TDD tests covering all 7 value types, nil val guard,
  unknown discriminant, and boolean false
- All 42 backend tests passing
2026-04-17 13:54:03 +00:00

304 lines
9 KiB
Go

//go:build cgo
package backend
/*
#include <stdlib.h>
#include "client.h"
*/
import "C"
import (
"apiserver/models"
logger "apiserver/utils"
"unsafe"
"go.uber.org/zap"
)
// goString safely converts a C string to a Go string, handling nil.
func goString(s *C.char) string {
if s == nil {
return ""
}
return C.GoString(s)
}
// cJobToModel extracts C xce_job_t fields into a rawJobData and maps it to JobInfo.
func cJobToModel(job *C.xce_job_t) *models.JobInfo {
r := rawJobData{
jobID: int64(job.jobid),
state: int(job.state),
runTime: int(job.run_time),
subTime: int64(job.sub_time),
queueTime: int64(job.queue_time),
startTime: int64(job.start_time),
finishTime: int64(job.finish_time),
}
if job.submit != nil {
s := job.submit
rd := &rawSubmitData{
queue: goString(s.queue),
user: goString(s.user),
jobName: goString(s.job_name),
projectName: goString(s.project_name),
userGroup: goString(s.user_group),
jobDesc: goString(s.job_description),
resReq: goString(s.resreq),
minTasks: int(s.mintasks),
maxTasks: int(s.maxtasks),
}
if s.start_info != nil {
si := s.start_info
rd.startInfo = &rawStartInfo{
cmdline: goString(si.cmdline),
launchType: goString(si.launch_type),
preExec: goString(si.job_pre_exec),
postExec: goString(si.job_post_exec),
execHome: goString(si.exechome),
execCWD: goString(si.execcwd),
execUser: goString(si.execuser),
}
}
r.submit = rd
}
return rawJobToModel(r)
}
// cWorkerToModel extracts C xce_worker_t fields into a rawWorkerData and maps it to HostInfo.
func cWorkerToModel(worker *C.xce_worker_t) *models.HostInfo {
r := rawWorkerData{
name: goString(worker.name),
status: int(worker.status),
}
if worker.attrs != nil {
a := worker.attrs
r.hasAttrs = true
r.hostType = goString(a.hosttype)
r.hostModel = goString(a.hostmodel)
r.cpuFactor = float64(a.cpufactor)
r.maxCPUs = int(a.maxcpus)
r.maxMem = int(a.maxmem)
r.maxSwap = int(a.maxswap)
r.maxTmp = int(a.maxtmp)
r.nDisks = int(a.ndisks)
}
return rawWorkerToModel(r)
}
// cQueueToModel extracts C xce_queue_t fields and maps them to QueueInfo.
func cQueueToModel(queue *C.xce_queue_t) *models.QueueInfo {
return rawQueueToModel(rawQueueData{
name: goString(queue.queue),
status: int(queue.status),
priority: int(queue.priority),
})
}
// cResourceToModel extracts C xce_resource_t fields and maps them to ResourceDefinitionInfo.
func cResourceToModel(res *C.xce_resource_t) *models.ResourceDefinitionInfo {
return rawResourceToModel(rawResourceData{
name: goString(res.name),
desc: goString(res.desc),
valType: int(res.valtype),
ordType: int(res.ordtype),
flags: int(res.flags),
interval: int(res.interval),
})
}
// cClusterConfQueues extracts queue models from a cluster_config_t.
func cClusterConfQueues(conf *C.cluster_config_t) []models.QueueInfo {
n := int(conf.nqueues)
if n == 0 || conf.queues == nil {
return nil
}
qps := unsafe.Slice(conf.queues, n)
out := make([]models.QueueInfo, 0, n)
for _, qp := range qps {
if qp != nil {
out = append(out, *cQueueToModel(qp))
}
}
return out
}
// cClusterConfResources extracts resource definition models from a cluster_config_t.
func cClusterConfResources(conf *C.cluster_config_t) []models.ResourceDefinitionInfo {
n := int(conf.nres)
if n == 0 || conf.resources == nil {
return nil
}
rps := unsafe.Slice(conf.resources, n)
out := make([]models.ResourceDefinitionInfo, 0, n)
for _, rp := range rps {
if rp != nil {
out = append(out, *cResourceToModel(rp))
}
}
return out
}
// cHostConfigToModel extracts C xce_hostconfig_t fields and maps them to HostInfo.
func cHostConfigToModel(hc *C.xce_hostconfig_t) *models.HostInfo {
return rawHostConfigToModel(rawHostConfigData{
name: goString(hc.name),
hostmodel: goString(hc.hostmodel),
hosttype: goString(hc.hosttype),
mxj: int(hc.mxj),
})
}
// cClusterConfHostConfigs extracts host config models from a cluster_config_t.
func cClusterConfHostConfigs(conf *C.cluster_config_t) []models.HostInfo {
n := int(conf.nhconfigs)
if n == 0 || conf.hconfigs == nil {
return nil
}
hps := unsafe.Slice(conf.hconfigs, n)
out := make([]models.HostInfo, 0, n)
for _, hp := range hps {
if hp != nil {
out = append(out, *cHostConfigToModel(hp))
}
}
return out
}
// cClusterConfToClusterInfo extracts C cluster_config_t cluster-level fields and maps them to ClusterInfo.
// Returns nil when clustername is empty so dispatchConfigPush skips the OnClusterUpdated call.
// C struct field names are snake_case (clustername, heartbeat_interval, etc.) — not Cap'n Proto names.
func cClusterConfToClusterInfo(conf *C.cluster_config_t) *models.ClusterInfo {
name := goString(conf.clustername)
if name == "" {
return nil
}
return rawClusterToModel(rawClusterData{
name: name,
heartbeatInterval: int(conf.heartbeat_interval),
vcpus: int(conf.vcpus),
inactiveHeartbeatCount: int(conf.inactive_heartbeat_count),
})
}
// modelJobToSubmit converts a Go JobInfo to C xce_job_submit_t heap structs.
// The caller must free all returned C memory by calling freeJobSubmit.
func modelJobToSubmit(job *models.JobInfo) (*C.xce_job_submit_t, *C.job_submit_data_t, *C.job_startinfo_t) {
si := (*C.job_startinfo_t)(C.calloc(1, C.sizeof_job_startinfo_t))
si.cmdline = C.CString(job.Command)
if job.LaunchType != "" {
si.launch_type = C.CString(job.LaunchType)
}
if job.PreExecCmd != "" {
si.job_pre_exec = C.CString(job.PreExecCmd)
}
if job.PostExecCmd != "" {
si.job_post_exec = C.CString(job.PostExecCmd)
}
sd := (*C.job_submit_data_t)(C.calloc(1, C.sizeof_job_submit_data_t))
sd.queue = C.CString(job.QueueName)
sd.user = C.CString(job.SubmitUser)
sd.job_name = C.CString(job.JobName)
if job.ProjectName != "" {
sd.project_name = C.CString(job.ProjectName)
}
if job.UserGroup != "" {
sd.user_group = C.CString(job.UserGroup)
}
if job.JobDescription != "" {
sd.job_description = C.CString(job.JobDescription)
}
if job.ResReq != "" {
sd.resreq = C.CString(job.ResReq)
}
sd.mintasks = C.uint(job.NumProcessors)
if sd.mintasks == 0 {
sd.mintasks = 1
}
sd.maxtasks = C.uint(job.MaxNumProcessors)
if sd.maxtasks < sd.mintasks {
sd.maxtasks = sd.mintasks
}
sd.start_info = si
js := (*C.xce_job_submit_t)(C.calloc(1, C.sizeof_xce_job_submit_t))
js.jobid = C.uint64_t(job.Unique_JobID)
js.array_size = 1
js.submit = sd
return js, sd, si
}
// freeJobSubmit releases C memory allocated by modelJobToSubmit.
func freeJobSubmit(js *C.xce_job_submit_t, sd *C.job_submit_data_t, si *C.job_startinfo_t) {
if si != nil {
C.free(unsafe.Pointer(si.cmdline))
C.free(unsafe.Pointer(si.launch_type))
C.free(unsafe.Pointer(si.job_pre_exec))
C.free(unsafe.Pointer(si.job_post_exec))
C.free(unsafe.Pointer(si))
}
if sd != nil {
C.free(unsafe.Pointer(sd.queue))
C.free(unsafe.Pointer(sd.user))
C.free(unsafe.Pointer(sd.job_name))
C.free(unsafe.Pointer(sd.project_name))
C.free(unsafe.Pointer(sd.user_group))
C.free(unsafe.Pointer(sd.job_description))
C.free(unsafe.Pointer(sd.resreq))
C.free(unsafe.Pointer(sd))
}
if js != nil {
C.free(unsafe.Pointer(js))
}
}
// 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 &.
func cResvalToDistribution(vp *C.xce_resval_t) *models.ResourceDistributionInfo {
rd := &models.ResourceDistributionInfo{
ResourceName: C.GoString(vp.name),
Location: C.GoString(vp.host),
IsDynamic: true, // resource values pushed from workers are always dynamic
}
// Guard: vp.val may be NULL if the controller sends a resval with no value set.
// Dereferencing NULL in xce_val_which() would segfault in C.
if vp.val == nil {
logger.Warn("resval with nil value, skipping",
zap.String("resource", rd.ResourceName),
zap.String("host", rd.Location))
return nil
}
switch C.xce_val_which(vp.val) {
case C.Value_u_ivalue32:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_i32(vp.val))
case C.Value_u_ivalue64:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_i64(vp.val))
case C.Value_u_uvalue32:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_u32(vp.val))
case C.Value_u_uvalue64:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = float64(C.xce_val_u64(vp.val))
case C.Value_u_fvalue:
rd.ValueType = int(models.NUMERIC)
rd.NumericValue = C.xce_val_fval(vp.val) // C.double → float64, no explicit cast needed
case C.Value_u_svalue:
rd.ValueType = int(models.STRING)
rd.StringValue = C.GoString(C.xce_val_str(vp.val))
case C.Value_u_bvalue:
rd.ValueType = int(models.BOOLEAN)
rd.BooleanValue = C.xce_val_bool(vp.val) != 0
default:
logger.Warn("unknown xce_val_t discriminant, skipping",
zap.Int("which", int(C.xce_val_which(vp.val))),
zap.String("resource", rd.ResourceName),
zap.String("host", rd.Location))
return nil
}
return rd
}