From d7afd72e5b4f7c5e81794616a87fb8f5e4dcc003 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 1 May 2026 21:22:45 +0800 Subject: [PATCH] fix front end issues --- CLAUDE.md | 10 ++ src/frontend/apiserver/backend/clib.go | 21 ++++ src/frontend/apiserver/backend/convert.go | 12 +++ .../apiserver/backend/convert_pure.go | 18 ++++ src/frontend/apiserver/backend/handler.go | 27 +++-- .../apiserver/backend/handler_test.go | 63 ++++++++---- .../apiserver/backend/integration_test.go | 60 ++++++----- .../apiserver/backend/notification.go | 3 + src/frontend/apiserver/controllers/queues.go | 4 +- src/frontend/apiserver/models/hosts.go | 3 - src/frontend/apiserver/models/hosts_test.go | 4 +- src/frontend/apiserver/models/jobs.go | 17 +++- src/frontend/apiserver/models/jobs_test.go | 99 +++++++++++++++++++ src/include/resreq.h | 7 ++ src/lib/helper.c | 2 + src/lib/resreq.c | 15 +++ src/lib/testsuite/test_select.c | 29 ++++++ src/scheduler/src/scheduler.c | 12 ++- 18 files changed, 345 insertions(+), 61 deletions(-) create mode 100644 src/frontend/apiserver/models/jobs_test.go diff --git a/CLAUDE.md b/CLAUDE.md index 1f541b43..1f183f46 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -105,3 +105,13 @@ Fall back to Grep/Glob/Read **only** when the graph doesn't cover what you need. 2. Use `detect_changes` for code review. 3. Use `get_affected_flows` to understand impact. 4. Use `query_graph` pattern="tests_for" to check coverage. + +## graphify + +This project has a graphify knowledge graph at graphify-out/. + +Rules: +- Before answering architecture or codebase questions, read graphify-out/GRAPH_REPORT.md for god nodes and community structure +- If graphify-out/wiki/index.md exists, navigate it instead of reading raw files +- For cross-module "how does X relate to Y" questions, prefer `graphify query ""`, `graphify path "" ""`, or `graphify explain ""` over grep — these traverse the graph's EXTRACTED + INFERRED edges instead of scanning files +- After modifying code files in this session, run `graphify update .` to keep the graph current (AST-only, no API cost) diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index 4a97e7fa..4fc63897 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -266,6 +266,27 @@ func handleNotifyData(nd *C.xce_notify_data_t) C.int { return 0 } switch nd.event { + case C.NotifyData_data_updateJob: + jp := (*C.xce_job_t)(C.xce_notify_data_ptr(nd)) + if jp == nil { + return 0 + } + gHandler.OnJobUpdated(cJobToModel(jp)) + + case C.NotifyData_data_jobFinishInfo: + fi := (*C.xce_jobfinish_t)(C.xce_notify_data_ptr(nd)) + if fi == nil { + return 0 + } + gHandler.OnJobFinished(uint64(fi.jobid), int(fi.exit_code)) + + case C.NotifyData_data_jobFailInfo: + ff := (*C.job_failinfo_t)(C.xce_notify_data_ptr(nd)) + if ff == nil { + return 0 + } + gHandler.OnJobFinished(uint64(ff.jobid), int(ff.code)) + case C.NotifyData_data_schedUpdateResvals: rvl := (*C.xce_resval_list_t)(C.xce_notify_data_ptr(nd)) if rvl == nil { diff --git a/src/frontend/apiserver/backend/convert.go b/src/frontend/apiserver/backend/convert.go index 2771c3b4..2bab7ea0 100644 --- a/src/frontend/apiserver/backend/convert.go +++ b/src/frontend/apiserver/backend/convert.go @@ -6,6 +6,9 @@ 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]; } */ import "C" @@ -63,6 +66,15 @@ func cJobToModel(job *C.xce_job_t) *models.JobInfo { } 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) } diff --git a/src/frontend/apiserver/backend/convert_pure.go b/src/frontend/apiserver/backend/convert_pure.go index ce9eb116..535c553b 100644 --- a/src/frontend/apiserver/backend/convert_pure.go +++ b/src/frontend/apiserver/backend/convert_pure.go @@ -2,6 +2,7 @@ package backend import ( "apiserver/models" + "encoding/json" "time" ) @@ -110,6 +111,7 @@ type rawJobData struct { startTime int64 finishTime int64 submit *rawSubmitData + execHosts []string } // rawWorkerData mirrors the C xce_worker_t fields used by the Go model. @@ -189,6 +191,22 @@ func rawJobToModel(r rawJobData) *models.JobInfo { if r.finishTime > 0 { info.EndTime = time.Unix(r.finishTime, 0) } + if len(r.execHosts) > 0 { + type hostSlot struct { + HostName string `json:"host_name"` + NumSlots int `json:"num_slots"` + } + type execHostsJSON struct { + HostSlots []hostSlot `json:"host_slots"` + NumHosts int `json:"num_hosts"` + } + slots := make([]hostSlot, len(r.execHosts)) + for i, h := range r.execHosts { + slots[i] = hostSlot{HostName: h, NumSlots: 1} + } + data, _ := json.Marshal(execHostsJSON{HostSlots: slots, NumHosts: len(slots)}) + info.ExecHosts = models.Jsonb{String: string(data)} + } return info } diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index 7f41ce85..8efba352 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -20,6 +20,19 @@ func (h *DefaultHandler) OnJobDeleted(jobid uint64) { store.Jobs.Delete(jobid) } +func (h *DefaultHandler) OnJobFinished(jobid uint64, exitCode int) { + existing, ok := store.Jobs.Get(jobid) + if !ok { + return + } + if exitCode == 0 { + existing.JobStatus = models.JOBSTAT["DONE"] + } else { + existing.JobStatus = models.JOBSTAT["EXIT"] + } + store.Jobs.Put(jobid, existing) +} + func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) { store.Queues.Put(queue.QueueName, *queue) } @@ -38,14 +51,14 @@ func (h *DefaultHandler) OnHostUpdated(incoming *models.HostInfo) { store.Hosts.Put(merged.HostName, merged) } -// OnHostConfigUpdated applies config-sourced fields onto the existing host entry. -// All other fields (worker-sourced) are preserved from the prior entry. -// NumCores, NumThreads, Topology, SchedStop have no current write path in -// rawWorkerData and remain zero until a future push source populates them -// (analogous to the DisptWindow deferral). +// OnHostConfigUpdated applies config-sourced fields onto an existing worker-registered host. +// Hosts not yet registered by a worker are skipped — config entries like "default" (which sets +// cluster-wide defaults) or hosts whose daemons haven't started must not appear in the host list. func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) { - existing, _ := store.Hosts.Get(incoming.HostName) - existing.HostName = incoming.HostName + existing, ok := store.Hosts.Get(incoming.HostName) + if !ok { + return + } existing.HostModel = incoming.HostModel existing.HostType = incoming.HostType existing.MaxSlot = incoming.MaxSlot diff --git a/src/frontend/apiserver/backend/handler_test.go b/src/frontend/apiserver/backend/handler_test.go index 53f68aea..dfc609c0 100644 --- a/src/frontend/apiserver/backend/handler_test.go +++ b/src/frontend/apiserver/backend/handler_test.go @@ -11,31 +11,39 @@ func resetStores() { store.InitStores() } -func TestOnHostConfigUpdated_NewHost(t *testing.T) { +func TestOnHostConfigUpdated_UnregisteredHostIgnored(t *testing.T) { + // Config-only entries (e.g. "default", hosts whose daemons haven't started) + // must not be inserted into the Hosts store. Only OnHostUpdated (worker push) creates entries. resetStores() h := DefaultHandler{} - host := &models.HostInfo{ + h.OnHostConfigUpdated(&models.HostInfo{ HostName: "node01", HostModel: "x86_64", HostType: "LINUX64", MaxSlot: 8, - } + }) - h.OnHostConfigUpdated(host) + _, ok := store.Hosts.Get("node01") + if ok { + t.Error("OnHostConfigUpdated must not create a new host entry; only workers can register hosts") + } +} - got, ok := store.Hosts.Get("node01") - if !ok { - t.Fatal("host node01 should exist in store after OnHostConfigUpdated") - } - if got.HostModel != "x86_64" { - t.Errorf("HostModel: got %q, want %q", got.HostModel, "x86_64") - } - if got.HostType != "LINUX64" { - t.Errorf("HostType: got %q, want %q", got.HostType, "LINUX64") - } - if got.MaxSlot != 8 { - t.Errorf("MaxSlot: got %d, want 8", got.MaxSlot) +func TestOnHostConfigUpdated_DefaultPseudoHostIgnored(t *testing.T) { + // The "default" host is a cluster-wide config sentinel, never a real worker. + resetStores() + h := DefaultHandler{} + + h.OnHostConfigUpdated(&models.HostInfo{ + HostName: "default", + HostType: "x86_64-linux-gnu", + MaxSlot: -1, + }) + + _, ok := store.Hosts.Get("default") + if ok { + t.Error("OnHostConfigUpdated must not create an entry for the 'default' pseudo-host") } } @@ -249,20 +257,27 @@ func TestOnHostUpdated_NewHost(t *testing.T) { } } -func TestDispatchConfigPush_CallsOnHostConfigUpdatedForEachHost(t *testing.T) { +func TestDispatchConfigPush_MergesConfigOntoRegisteredHosts(t *testing.T) { + // Config push should merge onto hosts already registered by workers, and + // silently skip any host not yet in the store (e.g. "default", offline workers). resetStores() h := &DefaultHandler{} + // Simulate worker registration arriving before config push. + store.Hosts.Put("node01", models.HostInfo{HostName: "node01", NumCPUs: 32}) + store.Hosts.Put("node02", models.HostInfo{HostName: "node02", NumCPUs: 64}) + hosts := []models.HostInfo{ {HostName: "node01", HostModel: "x86_64", HostType: "LINUX64", MaxSlot: 4}, {HostName: "node02", HostModel: "arm64", HostType: "LINUX64", MaxSlot: 8}, + {HostName: "default", HostModel: "x86_64", HostType: "LINUX64", MaxSlot: -1}, } dispatchConfigPush(h, nil, nil, hosts, nil) got1, ok1 := store.Hosts.Get("node01") if !ok1 { - t.Fatal("node01 should be in store after dispatchConfigPush") + t.Fatal("node01 should remain in store after dispatchConfigPush") } if got1.HostModel != "x86_64" { t.Errorf("node01 HostModel: got %q, want x86_64", got1.HostModel) @@ -270,10 +285,13 @@ func TestDispatchConfigPush_CallsOnHostConfigUpdatedForEachHost(t *testing.T) { if got1.MaxSlot != 4 { t.Errorf("node01 MaxSlot: got %d, want 4", got1.MaxSlot) } + if got1.NumCPUs != 32 { + t.Errorf("node01 NumCPUs: got %d, want 32 (worker field must be preserved)", got1.NumCPUs) + } got2, ok2 := store.Hosts.Get("node02") if !ok2 { - t.Fatal("node02 should be in store after dispatchConfigPush") + t.Fatal("node02 should remain in store after dispatchConfigPush") } if got2.HostModel != "arm64" { t.Errorf("node02 HostModel: got %q, want arm64", got2.HostModel) @@ -281,6 +299,13 @@ func TestDispatchConfigPush_CallsOnHostConfigUpdatedForEachHost(t *testing.T) { if got2.MaxSlot != 8 { t.Errorf("node02 MaxSlot: got %d, want 8", got2.MaxSlot) } + + if _, ok := store.Hosts.Get("default"); ok { + t.Error("'default' pseudo-host must not be inserted by config push") + } + if n := store.Hosts.Count(); n != 2 { + t.Errorf("Hosts store should have exactly 2 entries, got %d", n) + } } func TestDispatchConfigPush_CallsOnClusterUpdated(t *testing.T) { diff --git a/src/frontend/apiserver/backend/integration_test.go b/src/frontend/apiserver/backend/integration_test.go index c768664f..b2b8b381 100644 --- a/src/frontend/apiserver/backend/integration_test.go +++ b/src/frontend/apiserver/backend/integration_test.go @@ -20,11 +20,14 @@ func resetIntegrationStores() { // ========== §1. P0-1: Cluster Config Push ========== -// --- 1.1 P0: Config push populates all stores in a single call --- +// --- 1.1 P0: Config push populates queues/resources/clusters; hosts require prior worker registration --- func TestIntegration_ConfigPushPopulatesAllStores(t *testing.T) { resetIntegrationStores() h := &DefaultHandler{} + // Worker registers node01 before config push arrives. + store.Hosts.Put("node01", models.HostInfo{HostName: "node01", NumCPUs: 8}) + queues := []models.QueueInfo{{QueueName: "normal", Priority: 30}} resources := []models.ResourceDefinitionInfo{{ResourceName: "mem", Description: "memory", ValueType: 1}} hosts := []models.HostInfo{{HostName: "node01", HostModel: "x86_64"}} @@ -40,10 +43,13 @@ func TestIntegration_ConfigPushPopulatesAllStores(t *testing.T) { } got, ok := store.Hosts.Get("node01") if !ok { - t.Fatal("Hosts store should contain 'node01'") + t.Fatal("Hosts store should contain 'node01' (pre-registered by worker)") } if got.HostModel != "x86_64" { - t.Errorf("node01 HostModel = %q, want x86_64", got.HostModel) + t.Errorf("node01 HostModel = %q, want x86_64 (config merged)", got.HostModel) + } + if got.NumCPUs != 8 { + t.Errorf("node01 NumCPUs = %d, want 8 (worker field preserved)", got.NumCPUs) } gotC, ok := store.Clusters.Get("c1") if !ok { @@ -394,32 +400,36 @@ func TestIntegration_CrossFeature_HeartbeatAndClusterData(t *testing.T) { } } -// --- 5.2 P1: Config push + host visibility in same host --- -func TestIntegration_CrossFeature_ConfigPushAndHostVisible(t *testing.T) { +// --- 5.2 P1: Config push alone must NOT create host entries --- +func TestIntegration_CrossFeature_ConfigPushDoesNotCreateHost(t *testing.T) { + // Hosts like "default" (cluster-wide defaults) and offline workers must not appear + // in the host store. Only OnHostUpdated (worker registration) creates entries. resetIntegrationStores() h := &DefaultHandler{} dispatchConfigPush(h, nil, nil, []models.HostInfo{ {HostName: "node01", HostModel: "x86_64"}, + {HostName: "default", HostType: "x86_64-linux-gnu", MaxSlot: -1}, }, nil) - got, ok := store.Hosts.Get("node01") - if !ok { - t.Fatal("node01 should exist after config push") + if store.Hosts.Has("node01") { + t.Error("node01 must not be created by config push alone — only worker push registers hosts") } - if got.HostModel != "x86_64" { - t.Errorf("HostModel = %q, want x86_64", got.HostModel) + if store.Hosts.Has("default") { + t.Error("'default' pseudo-host must not be created by config push") } - if !store.Hosts.Has("node01") { - t.Error("store.Hosts.Has('node01') should be true — config-push data is visible in the host store") + if n := store.Hosts.Count(); n != 0 { + t.Errorf("Hosts store must be empty after config-only push, got %d entries", n) } } -// --- 5.3 P1: Resource push after host config push — same host --- -func TestIntegration_CrossFeature_ResourceAfterHostConfig(t *testing.T) { +// --- 5.3 P1: Resource push and host registration are independent --- +func TestIntegration_CrossFeature_ResourceAfterWorkerRegistration(t *testing.T) { + // Worker registers first, then resource push and config push arrive in any order. resetIntegrationStores() h := &DefaultHandler{} + h.OnHostUpdated(&models.HostInfo{HostName: "node01", HostStatus: 0, NumCPUs: 8}) h.OnHostConfigUpdated(&models.HostInfo{HostName: "node01", HostModel: "x86_64"}) h.OnResourceLocationUpdated(&models.ResourceDistributionInfo{ ResourceName: "mem", Location: "node01", NumericValue: 65536.0, @@ -428,10 +438,10 @@ func TestIntegration_CrossFeature_ResourceAfterHostConfig(t *testing.T) { gotHost, ok := store.Hosts.Get("node01") if !ok { - t.Fatal("node01 should exist in Hosts store") + t.Fatal("node01 should exist in Hosts store after worker registration") } if gotHost.HostModel != "x86_64" { - t.Errorf("HostModel = %q, want x86_64 (should be preserved)", gotHost.HostModel) + t.Errorf("HostModel = %q, want x86_64 (config merged)", gotHost.HostModel) } gotRes, ok := store.ResourceLocations.Get("mem:node01") @@ -442,36 +452,38 @@ func TestIntegration_CrossFeature_ResourceAfterHostConfig(t *testing.T) { t.Errorf("NumericValue = %f, want 65536.0", gotRes.NumericValue) } - // Host config should not have been overwritten by resource push + // Resource push must not have overwritten host config gotHost2, _ := store.Hosts.Get("node01") if gotHost2.HostModel != "x86_64" { t.Errorf("HostModel after resource push = %q, want x86_64 (unchanged)", gotHost2.HostModel) } } -// --- 5.4 P1: Worker push after config push preserves config fields --- -func TestIntegration_CrossFeature_WorkerPreservesConfigFields(t *testing.T) { +// --- 5.4 P1: Config push onto registered host preserves worker fields --- +func TestIntegration_CrossFeature_ConfigPreservesWorkerFields(t *testing.T) { + // Canonical order: worker registers first, then config push merges config fields. + // Worker-sourced fields (HostStatus, CPUFactor, NumCPUs) must be preserved. resetIntegrationStores() h := &DefaultHandler{} - h.OnHostConfigUpdated(&models.HostInfo{HostName: "node01", HostModel: "x86_64", MaxSlot: 8}) h.OnHostUpdated(&models.HostInfo{HostName: "node01", HostStatus: 3, CPUFactor: 2.5, NumCPUs: 16}) + h.OnHostConfigUpdated(&models.HostInfo{HostName: "node01", HostModel: "x86_64", MaxSlot: 8}) got, ok := store.Hosts.Get("node01") if !ok { t.Fatal("node01 should exist after both pushes") } if got.HostModel != "x86_64" { - t.Errorf("HostModel = %q, want x86_64 (config preserved)", got.HostModel) + t.Errorf("HostModel = %q, want x86_64 (config applied)", got.HostModel) } if got.MaxSlot != 8 { - t.Errorf("MaxSlot = %d, want 8 (config preserved)", got.MaxSlot) + t.Errorf("MaxSlot = %d, want 8 (config applied)", got.MaxSlot) } if got.HostStatus != 3 { - t.Errorf("HostStatus = %d, want 3 (worker updated)", got.HostStatus) + t.Errorf("HostStatus = %d, want 3 (worker field preserved)", got.HostStatus) } if got.CPUFactor != 2.5 { - t.Errorf("CPUFactor = %f, want 2.5 (worker updated)", got.CPUFactor) + t.Errorf("CPUFactor = %f, want 2.5 (worker field preserved)", got.CPUFactor) } } diff --git a/src/frontend/apiserver/backend/notification.go b/src/frontend/apiserver/backend/notification.go index 34f36051..f26bc6ff 100644 --- a/src/frontend/apiserver/backend/notification.go +++ b/src/frontend/apiserver/backend/notification.go @@ -9,6 +9,9 @@ type NotificationHandler interface { OnJobCreated(job *models.JobInfo) OnJobUpdated(job *models.JobInfo) OnJobDeleted(jobid uint64) + // OnJobFinished is called when the controller reports job completion or + // dispatch failure. exitCode==0 → DONE, non-zero → EXIT. + OnJobFinished(jobid uint64, exitCode int) OnQueueUpdated(queue *models.QueueInfo) OnHostUpdated(host *models.HostInfo) OnHostConfigUpdated(host *models.HostInfo) diff --git a/src/frontend/apiserver/controllers/queues.go b/src/frontend/apiserver/controllers/queues.go index 0541d1b2..8cbad41b 100644 --- a/src/frontend/apiserver/controllers/queues.go +++ b/src/frontend/apiserver/controllers/queues.go @@ -600,9 +600,9 @@ func formatQueueStatus(state int64) string { status = "open" } - if state&int64(models.QUEUESTAT["INACTIVE_ADM"]) != 0 { // INACT_WINDOW + if state&int64(models.QUEUESTAT["INACTIVE_ADM"]) != 0 { // INACT_ADMIN status += ":inactive_adm" - } else if state&int64(models.QUEUESTAT["INACTIVE_WIN"]) != 0 { // INACT_ADMIN + } else if state&int64(models.QUEUESTAT["INACTIVE_WIN"]) != 0 { // INACT_WINDOW status += ":inactive_win" } else { status += ":active" diff --git a/src/frontend/apiserver/models/hosts.go b/src/frontend/apiserver/models/hosts.go index 0e449126..ab30636f 100644 --- a/src/frontend/apiserver/models/hosts.go +++ b/src/frontend/apiserver/models/hosts.go @@ -18,9 +18,6 @@ var HOSTSTAT = map[string]int{ } func HostStateStr(state int) string { - if state == 0 { - return "UNKNOWN" - } if (state & HOSTSTAT["UNAVAIL"]) != 0 { return "UNAVAIL" } else if (state & HOSTSTAT["CLOSED"]) != 0 { diff --git a/src/frontend/apiserver/models/hosts_test.go b/src/frontend/apiserver/models/hosts_test.go index bf2ef11d..99ce5dab 100644 --- a/src/frontend/apiserver/models/hosts_test.go +++ b/src/frontend/apiserver/models/hosts_test.go @@ -9,9 +9,9 @@ func TestHostStateStr(t *testing.T) { want string }{ { - name: "zero value returns UNKNOWN when no worker has reported", + name: "zero value (no bad flags set) returns OK — C status=0 means healthy", state: 0, - want: "UNKNOWN", + want: "OK", }, { name: "UNAVAIL flag", diff --git a/src/frontend/apiserver/models/jobs.go b/src/frontend/apiserver/models/jobs.go index 3b44d8aa..772cf33e 100644 --- a/src/frontend/apiserver/models/jobs.go +++ b/src/frontend/apiserver/models/jobs.go @@ -70,8 +70,9 @@ var JOBSTAT = map[string]int{ "SSUSP": 0x40, "PDONE": 0x80, "PERR": 0x100, + "WAIT": 0x200, "RUNKWN": 0x8000, - "UNKNOWN": 0x1000, + "UNKNOWN": 0x10000, // C JOB_STAT_UNKWN: worker lost contact with controller } var LAUNCHTYPE = map[string]int{ @@ -90,12 +91,16 @@ func JobStateStr(state int) string { return "USUSP" } else if (state & JOBSTAT["SSUSP"]) != 0 { return "SSUSP" + } else if (state & JOBSTAT["RUNKWN"]) != 0 { + return "RUNKWN" } return "RUN" } else if (state & JOBSTAT["EXIT"]) != 0 { return "EXIT" } else if (state & JOBSTAT["DONE"]) != 0 { return "DONE" + } else if (state & JOBSTAT["UNKNOWN"]) != 0 { + return "UNKWN" } return "ERR" } @@ -245,6 +250,16 @@ func (jsb *Jsonb) IsZero() bool { return jsb.String == "" } +func (jsb Jsonb) MarshalJSON() ([]byte, error) { + if jsb.String == "" { + return []byte(`null`), nil + } + if json.Valid([]byte(jsb.String)) { + return []byte(jsb.String), nil + } + return json.Marshal(jsb.String) +} + func (tm *CustomTime) Tm() time.Time { return tm.Time } diff --git a/src/frontend/apiserver/models/jobs_test.go b/src/frontend/apiserver/models/jobs_test.go new file mode 100644 index 00000000..d4a7b794 --- /dev/null +++ b/src/frontend/apiserver/models/jobs_test.go @@ -0,0 +1,99 @@ +package models + +import "testing" + +func TestJobStateStr(t *testing.T) { + tests := []struct { + name string + state int + want string + }{ + { + name: "PEND", + state: JOBSTAT["PEND"], + want: "PEND", + }, + { + name: "PSUSP (PEND+PSUSP)", + state: JOBSTAT["PEND"] | JOBSTAT["PSUSP"], + want: "PSUSP", + }, + { + name: "RUN", + state: JOBSTAT["RUN"], + want: "RUN", + }, + { + name: "USUSP (RUN+USUSP)", + state: JOBSTAT["RUN"] | JOBSTAT["USUSP"], + want: "USUSP", + }, + { + name: "SSUSP (RUN+SSUSP)", + state: JOBSTAT["RUN"] | JOBSTAT["SSUSP"], + want: "SSUSP", + }, + { + name: "RUNKWN (RUN+RUNKWN — lost contact with remote cluster)", + state: JOBSTAT["RUN"] | JOBSTAT["RUNKWN"], + want: "RUNKWN", + }, + { + name: "EXIT", + state: JOBSTAT["EXIT"], + want: "EXIT", + }, + { + name: "DONE", + state: JOBSTAT["DONE"], + want: "DONE", + }, + { + name: "UNKNOWN (0x10000 — worker lost contact with controller)", + state: JOBSTAT["UNKNOWN"], + want: "UNKWN", + }, + { + name: "zero state returns ERR", + state: 0, + want: "ERR", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := JobStateStr(tt.state) + if got != tt.want { + t.Errorf("JobStateStr(0x%X) = %q, want %q", tt.state, got, tt.want) + } + }) + } +} + +func TestJobStatMapValues(t *testing.T) { + // Verify bitmask values match C JOB_STAT_* defines in src/include/xce/types.h. + expected := map[string]int{ + "PEND": 0x01, + "RUN": 0x02, + "EXIT": 0x04, + "DONE": 0x08, + "PSUSP": 0x10, + "USUSP": 0x20, + "SSUSP": 0x40, + "PDONE": 0x80, + "PERR": 0x100, + "WAIT": 0x200, + "RUNKWN": 0x8000, + "UNKNOWN": 0x10000, + } + for key, want := range expected { + got, ok := JOBSTAT[key] + if !ok { + t.Errorf("JOBSTAT[%q] missing", key) + continue + } + if got != want { + t.Errorf("JOBSTAT[%q] = 0x%X, want 0x%X", key, got, want) + } + } +} diff --git a/src/include/resreq.h b/src/include/resreq.h index 39c77870..b40a6801 100644 --- a/src/include/resreq.h +++ b/src/include/resreq.h @@ -91,6 +91,13 @@ typedef struct xce_sched_job_ xce_sched_job_t; extern xce_select_node_t* makeSelectNode(sel_node_type_t type, uintptr_t op1, uintptr_t op2); +/* Resreq subsystem lifecycle — call once at process startup/shutdown. + * resreq_init sets the optimization level and initializes the intern table. + * resreq_fini tears down the intern table. + * optimize_level: 0 = disabled (API server); >0 = CSE+memoization (scheduler). */ +extern void resreq_init(int optimize_level); +extern void resreq_fini(void); + /* Optimization level — set once at process startup. * 0 = disabled (API server); >0 = CSE+memoization (scheduler). */ extern void sel_optimize_set(int level); diff --git a/src/lib/helper.c b/src/lib/helper.c index ecd904de..0dc33e54 100644 --- a/src/lib/helper.c +++ b/src/lib/helper.c @@ -269,6 +269,8 @@ void* xce_notify_data_ptr(xce_notify_data_t* d) return (void*)d->data.pdone_job; case NotifyData_data_schedJobRusages: return d->data.sched_job_rusages; + case NotifyData_data_updateJob: + return d->data.update_job; default: break; } diff --git a/src/lib/resreq.c b/src/lib/resreq.c index 8b6bfc2d..d1b5cf7d 100644 --- a/src/lib/resreq.c +++ b/src/lib/resreq.c @@ -17,6 +17,21 @@ #include "xlist.h" #include "xvec.h" +/* ── resreq subsystem lifecycle ───────────────────────────────────────────── */ + +void resreq_init(int optimize_level) +{ + sel_optimize_set(optimize_level); + sel_node_table_init(); +} + +void resreq_fini(void) +{ + sel_node_table_fini(); +} + +/* ── type helpers ─────────────────────────────────────────────────────────── */ + res_type_t valtype_to_res_type(int v) { switch (v) { diff --git a/src/lib/testsuite/test_select.c b/src/lib/testsuite/test_select.c index 47250eb8..8241617a 100644 --- a/src/lib/testsuite/test_select.c +++ b/src/lib/testsuite/test_select.c @@ -80,6 +80,34 @@ test_select_parse_valid_expr(void) sel_optimize_test_reset(); } +static void +test_select_parse_valid_any(void) +{ + /* parse_resreq_primary with a well-formed select expression must + * succeed (rc >= 0) and produce a non-NULL root node. */ + char * errmsg = NULL; + xce_primary_resreq_t resreq; + + /* resmgr fixture is created for teardown symmetry; Phase 2 threads it + * into parse_resreq_primary — for Phase 1 we use the 3-arg signature. */ + var resmgr = make_test_resmgr(); + + sel_optimize_set(0); + + int rc = parse_resreq_primary(&resreq, "type==any", resmgr, &errmsg); + + /* Phase 4: expect success */ + TEST_ASSERT_TRUE(rc >= 0); + TEST_ASSERT_NOT_NULL(resreq.select); + TEST_ASSERT_NOT_NULL(resreq.select->root); + + kill_primary_resreq(&resreq); + FREEUP(errmsg); + free_test_resmgr(resmgr); + sel_optimize_test_reset(); +} + + static void test_select_parse_unknown_resource(void) { @@ -2311,6 +2339,7 @@ test_select(void) RUN_TEST(test_optimize_reset_restores_unset); RUN_TEST(test_select_parse_valid_expr); + RUN_TEST(test_select_parse_valid_any); RUN_TEST(test_select_parse_unknown_resource); RUN_TEST(test_select_parse_type_mismatch); diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index 6293405b..a6fba60d 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -254,8 +254,12 @@ static void* sched_allocator(void* arg) /* job is pend */ xce_pendreason_t* reason = job->schedobj->reason; - reason->jobid = job->jobid; - decision = make_pend_decision(reason); + WARN("job %ld is pending but pend reason is NULL", jobid); + + if (reason != NULL) { + reason->jobid = job->jobid; + decision = make_pend_decision(reason); + } } else { MOVE_PTR(job->allocs, job->schedobj->alloc); job->allocs->jobid = job->jobid; @@ -375,7 +379,7 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, memset(sched, 0x0, sizeof(*sched)); - sel_optimize_set(1); /* enable CSE intern table for scheduler process */ + resreq_init(1); /* enable CSE intern table for scheduler process */ sched->nextid = 1; sched->configver = 0; @@ -476,6 +480,8 @@ void sched_fini(xce_scheduler_t* sched) destroy(sched->pgmgr); destroy(sched->hgmgr); + resreq_fini(); + mylock_fini(&(sched->lock)); return;