377 lines
11 KiB
Go
377 lines
11 KiB
Go
//go:build cgo
|
|
|
|
package backend
|
|
|
|
/*
|
|
#include <stdlib.h>
|
|
#include "client.h"
|
|
#include "val_helpers.h"
|
|
|
|
// Helper to index into xce_host_alloc_t** (CGO cannot dereference C pointer-to-pointer arrays).
|
|
static xce_host_alloc_t *job_alloc_host(xce_job_alloc_t *a, int i) { return a->allocs[i]; }
|
|
*/
|
|
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
|
|
}
|
|
if job.allocs != nil {
|
|
n := int(job.allocs.nhosts)
|
|
for i := 0; i < n; i++ {
|
|
h := C.job_alloc_host(job.allocs, C.int(i))
|
|
if h != nil && h.host != nil {
|
|
r.execHosts = append(r.execHosts, C.GoString(h.host))
|
|
}
|
|
}
|
|
}
|
|
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 {
|
|
workerName := goString(worker.name)
|
|
r := rawWorkerData{
|
|
name: workerName,
|
|
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)
|
|
}
|
|
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)
|
|
}
|
|
|
|
// 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: 0, // vcpus moved to myconf_t (--vcpus CLI flag); not in cluster_config_t
|
|
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))
|
|
}
|
|
}
|
|
|
|
// 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 &.
|
|
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 = float64(C.xce_val_fval(vp.val))
|
|
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
|
|
}
|