//go:build cgo package backend /* #include #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]; } // Helpers to index into the reason pointer arrays of xce_job_preason_t. static xce_host_reason_t *job_host_reason(xce_job_preason_t *p, int i) { return p->host_based_reasons[i]; } static xce_res_reason_t *job_res_reason(xce_job_preason_t *p, int i) { return p->res_based_reasons[i]; } // Helper to index into xce_alloc_t** (same CGO limitation as above) -- one // host's per-resource allocation list, scanned for the "slots" entry. static xce_alloc_t *host_alloc_res(xce_host_alloc_t *h, int i) { return h->allocs[i]; } // Helpers to index into job_pidinfo_t's pid/pgid arrays (same CGO limitation). static unsigned int pidinfo_pid(job_pidinfo_t *p, int i) { return p->pids[i]; } static unsigned int pidinfo_pgid(job_pidinfo_t *p, int i) { return p->pgids[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) } // cStringOrNil returns a C copy of s, or nil for an empty string so an unset // field stays NULL in the C struct. The caller frees the result. func cStringOrNil(s string) *C.char { if s == "" { return nil } return C.CString(s) } // cJobArrayToModels decodes a compact Data.jobArray push descriptor // (*C.xce_job_submit_t: base jobid, array_size, one shared job_submit_data_t // spec — the same JobSubmit shape apiserver_push_jobarray() sends, // job-array-single-wal-record task 5.2) and expands it locally into N // models.JobInfo entries via rawJobArrayToModels, mirroring cJobToModel's // single-job field extraction below but for the array-shaped descriptor // (task 5.4, Req 6.2, 6.5). Decodes the shared submit spec exactly once // (same field-by-field extraction cJobToModel uses for job.submit) since // every element shares it by construction — never once per element. func cJobArrayToModels(jobArray *C.xce_job_submit_t) []*models.JobInfo { if jobArray == nil || jobArray.submit == nil { return nil } s := jobArray.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), fromHost: goString(s.from_host), } 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), exdata: goString(si.exdata), } } return rawJobArrayToModels(uint64(jobArray.jobid), uint64(jobArray.array_size), rd) } // cPidInfoToRaw decodes a C job_pidinfo_t — the worker's start report // (pids/pgids plus exdata, whose {"url": ...} JSON is the interactive // session URL for ttyd/XJOB jobs) — into the rawPidInfo the pure layer // consumes. Shared by cJobToModel (the full xce_job_t conversion, where the // capnp Job wire carries pidinfo at Job.pidinfo@9) and the standalone // jobPidInfo notify path (NotifyData_data_jobPidInfo), which delivers the // same struct the moment WAL_JOB_START commits. func cPidInfoToRaw(pi *C.job_pidinfo_t) *rawPidInfo { if pi == nil { return nil } p := &rawPidInfo{exdata: goString(pi.exdata)} for i := 0; i < int(pi.npids); i++ { p.pids = append(p.pids, int(C.pidinfo_pid(pi, C.int(i)))) } for i := 0; i < int(pi.npgids); i++ { p.pgids = append(p.pgids, int(C.pidinfo_pgid(pi, C.int(i)))) } return p } // 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), fromHost: goString(s.from_host), } 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), exdata: goString(si.exdata), } } r.submit = rd } if u := job.run_usage; u != nil { r.runUsage = &rawRunUsage{ maxMem: float64(u.max_mem), avgMem: float64(u.avg_mem), usedMem: float64(u.used_mem), cpuTime: int(u.cpu_time), peakCPUTime: int(u.peak_cpu_time), peakCPUDuration: int(u.peak_cpu_duration), memEfficiency: float64(u.mem_efficiency), avgCPUEfficiency: float64(u.avg_cpu_efficiency), peakCPUEfficiency: float64(u.peak_cpu_efficiency), peakCPU: float64(u.peak_cpu), } } 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 { continue } // bug-1743-followup: this host's own "slots" resource // allocation (xce_host_alloc_t.allocs[]), mirroring the C // scheduler's own get_nslots_in_alloc()/repl_worker_alloc_ // nslots() resname scan -- defaults to 1 when absent, same as // those. nslots := 1 nallocs := int(h.nallocs) for j := 0; j < nallocs; j++ { res := C.host_alloc_res(h, C.int(j)) if res != nil && res.resname != nil && C.GoString(res.resname) == "slots" { nslots = int(res.value) break } } r.execHosts = append(r.execHosts, rawHostSlot{host: C.GoString(h.host), slots: nslots}) } } // PEND reason: the scheduler pushes these via setpreason; the controller // stores them on job->pend_reasons. Maps to models.JobInfo.PendingReason. // A detail-only record (main_reason 0) gets its overall-reason NAME from // the first detail row at the pure layer (rawJobToModel); the C record is // carried through unmutated (the scheduler-framework trace/verify stack // counts main_reason as its own occurrence and its committed .replay // baselines encode main_reason=0 detail-only PENDs). if job.pend_reasons != nil { pr := job.pend_reasons // main_reason 0 is detail-only (the allocator's ordinary shortfall // shape): leave the NAME empty so rawJobToModel's promotion fills it // from the first detail row. Pre-filling pendReasonName(0) here // ("unknown(0)") made the promotion unreachable on the live C path — // the advisor P0 verification caught jobs whose res_reasons named the // real cause while main_reason_name read unknown(0). mainName := "" if pr.main_reason != 0 { mainName = pendReasonName(int(pr.main_reason)) } r.pendReason = &rawPendReason{ MainReason: int(pr.main_reason), MainReasonName: mainName, } for i := 0; i < int(pr.nhost_reasons); i++ { hr := C.job_host_reason(pr, C.int(i)) if hr != nil && hr.host != nil { r.pendReason.HostReasons = append(r.pendReason.HostReasons, rawHostReason{Host: goString(hr.host), Reason: int(hr.reason), ReasonName: pendReasonName(int(hr.reason))}) } } for i := 0; i < int(pr.nres_reasons); i++ { rr := C.job_res_reason(pr, C.int(i)) if rr != nil && rr.resource != nil { r.pendReason.ResReasons = append(r.pendReason.ResReasons, rawResReason{Resource: goString(rr.resource), Host: goString(rr.host), Reason: int(rr.reason), ReasonName: pendReasonName(int(rr.reason))}) } } } // Pidinfo: the worker's start report — pids/pgids plus exdata, whose // {"url": ...} JSON is the interactive session URL (ttyd/XJOB). The // capnp Job wire carries pidinfo (Job.pidinfo@9 / JobPIDInfo.exdata@3) // but the conversion dropped it, so interactive_job_url never reached // the API and xcecli's browser flow could never fire. r.pidInfo = cPidInfoToRaw(job.pidinfo) 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), lastReport: int64(res.last_report), }) } // 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), }) } // cStrList converts an xce_strlist_t to a Go slice. func cStrList(l *C.xce_strlist_t) []string { n := int(l.count) if n == 0 || l.data == nil { return nil } out := make([]string, 0, n) for _, p := range unsafe.Slice(l.data, n) { if p != nil { out = append(out, C.GoString(p)) } } return out } // cClusterConfUserGroups extracts the user groups from a cluster_config_t // (bug-1934: the self_ug job scope resolves them). func cClusterConfUserGroups(conf *C.cluster_config_t) []models.UserGroupInfo { n := int(conf.nugroups) if n == 0 || conf.ugroups == nil { return nil } out := make([]models.UserGroupInfo, 0, n) for _, gp := range unsafe.Slice(conf.ugroups, n) { if gp == nil || gp.name == nil { continue } out = append(out, models.UserGroupInfo{ Name: C.GoString(gp.name), Members: cStrList(&gp.members), ExMembers: cStrList(&gp.ex_members), }) } return out } // 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) { rsi := submitStartInfo(job) si := (*C.job_startinfo_t)(C.calloc(1, C.sizeof_job_startinfo_t)) si.cmdline = C.CString(rsi.cmdline) // bug-1500/bug-1780 (same fix both sides): carry the exec user on the // start_info. Without it the worker receives a NULL/empty execuser and // runs every job as "nobody". si.execuser = cStringOrNil(rsi.execUser) si.launch_type = cStringOrNil(rsi.launchType) si.job_pre_exec = cStringOrNil(rsi.preExec) si.job_post_exec = cStringOrNil(rsi.postExec) si.exechome = cStringOrNil(rsi.execHome) si.execcwd = cStringOrNil(rsi.execCWD) si.exdata = cStringOrNil(rsi.exdata) 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) } if job.FromHost != "" { sd.from_host = C.CString(job.FromHost) } 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.execuser)) 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.exechome)) C.free(unsafe.Pointer(si.execcwd)) C.free(unsafe.Pointer(si.exdata)) 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.from_host)) 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 }