Merge branch 'merge_api_server' into persistent_ha

This commit is contained in:
Rongsong Shen 2026-05-07 04:41:14 +00:00
commit 9da6bbb260
51 changed files with 7100 additions and 592 deletions

View file

@ -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 "<question>"`, `graphify path "<A>" "<B>"`, or `graphify explain "<concept>"` 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)

View file

@ -105,10 +105,12 @@
## P1-3: Heartbeat Keep-Alive
- [ ] **#13 P1-3: Heartbeat keep-alive implementation**
- `backend/clib.go` — Add `heartbeatIntervalSecs int32` (default 30), `SetHeartbeatInterval()`, `heartbeatFn` var, `StartHeartbeat(ctx context.Context)`
- `backend/handler.go` — `OnClusterUpdated` calls `SetHeartbeatInterval(cluster.HeartbeatInterval)` after store update
- `cmd/main.go` — Call `backend.StartHeartbeat(ctx)` after `backend.RegisterToBackend(...)`
- [x] **#13 P1-3: Heartbeat keep-alive implementation**
- `backend/clib.go` — Added `heartbeatIntervalSecs int32` (default 30), `SetHeartbeatInterval()`, `heartbeatFn` var, `StartHeartbeat(ctx context.Context)` goroutine with ticker
- `backend/handler.go` — `OnClusterUpdated` calls `SetHeartbeatInterval(cluster.HeartbeatInterval)` after store update (zero-preserve: skips if interval==0)
- `cmd/main.go` — Calls `backend.StartHeartbeat(ctx)` after `backend.RegisterToBackend(...)`
- `backend/handler_test.go` — Tests for heartbeat interval update, zero-preserve, timer leak, test drain, store consistency
- Commits: `ffc5b20` (goroutine), `0a3de8b` (leak/drain/consistency fixes)
---
@ -119,4 +121,11 @@
- `ninja -C build` for C components passes — required meson reconfigure + CGO val_helpers.h fix
- RBAC verification — `host:control` (cluster_admin) and `host:view` (all authenticated) confirmed in routes + rbac.yaml
- Input safety — comment limit 512 chars; select expression limit 4094 chars (1024 reduction deferred to P1-2b)
- **CGO fix applied:** extracted `static` val helpers from `clib.go` preamble into `val_helpers.h` shared header; `convert.go` now includes it; added explicit `float64()` cast for `C.double` return
- **CGO fix applied:** extracted `static` val helpers from `clib.go` preamble into `val_helpers.h` shared header; `convert.go` now includes it; added explicit `float64()` cast for `C.double` return
- [x] **#15 Integration test suite** — see [`apiserver-backend-integration-test-plan.md`](apiserver-backend-integration-test-plan.md)
- `backend/integration_test.go` — §1 (config push), §3 (resource values), §4 (heartbeat), §5 (cross-feature): 17 tests
- `controllers/integration_test.go` — §2 (host control), §6 (RBAC), §8 (input safety): 17 tests + 1 skip (§8.1 pending P1-2b)
- `store/concurrency_test.go` — §7 (thread safety): 3 concurrent-access tests
- All use `//go:build !cgo`; run with `CGO_ENABLED=0 go test ./backend/... ./controllers/... ./store/...`
- Commit: `86a65d6`

21
src/docs/phase6-audit.md Normal file
View file

@ -0,0 +1,21 @@
# Phase 6 Pre-Audit: Single-Threaded Invariant and Stub Locations
## Audit Results
- **sched_meet_host_selection stub confirmed** at `src/scheduler/src/resource.c:127-131`
- Declaration: `src/scheduler/include/scheduler.h:111`
- Call site: `src/scheduler/src/resource.c:368` (within sched_try_alloc_on_host_)
- Stub body: returns 1 unconditionally
- **Allocator-thread-only invariant confirmed**: Single `while(1)` loop in `src/scheduler/src/main.c:119`
- `sched_begin_session()` at line 121
- `sched_go()` at line 122
- `sched_end_session()` at line 161
- All three calls executed sequentially in the same thread context
- **eval_select_simple signature confirmed**: `int eval_select_simple(const xce_select_node_t* root, const host_eval_ctx_t* host)`
- Declaration: `src/include/resreq.h:110-111`
- Definition: `src/lib/xce_select.c:558`
- Returns: 1 (match), 0 (no match), negative (error)
No unexpected references to `sched_meet_host_selection`, `eval_select`, `alloc_thread`, or `sched_session` found in scheduler source.

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

View file

@ -14,12 +14,13 @@ Phase 1: TDD test skeletons (parallel with Phase 3) ✓ DONE
Phase 2: Header definitions + ABI audit ✓ DONE
Phase 3: PEG grammar (parallel with Phase 1) ✓ DONE 143 PASS / 3 FAIL (stubs)
Phase 4a: Core interning + sel_optimize_level ✓ DONE 145 PASS / 1 FAIL (Phase 4b stub)
Phase 4b: Type validation (gated on P0-1) ← CURRENT
Phase 4c/4d: Canonicalization + const-fold
Phase 4e: Memoized eval
Phase 4f: GC + recompaction
Phase 5: Build wiring
Phase 6: Scheduler integration (after all unit tests pass)
Phase 4b: Type validation (gated on P0-1) ✓ DONE 203 PASS (2026-04-29)
Phase 4c: AND/OR associativity flattening ✓ DONE 157 PASS (2026-04-29)
Phase 4d: Const-fold transform ✓ DONE (2026-04-29)
Phase 4e: Memoized eval ✓ DONE 73 PASS (2026-04-29)
Phase 4f: GC + recompaction ✓ DONE 73 PASS (2026-04-29)
Phase 5: Build wiring ✓ DONE
Phase 6: Scheduler integration ✓ DONE 10 PASS (2026-04-30)
```
---
@ -113,67 +114,76 @@ Develop concurrently with Phase 3 (grammar is the test oracle).
---
## Phase 4b — Parse-Time Type Validation
## Phase 4b — Parse-Time Type Validation ✓ DONE (2026-04-29)
**Priority: HIGH — gate: P0-1 `makeString()` fix must be merged first**
| # | Task | Notes |
|---|------|-------|
| 4b-1 | Wire `parse_select()` into `parse_resreq_primary` **before** `kill_resreq_data` | Use-after-free if called after `kill_resreq_data` frees `data.selectString` |
| 4b-2 | Implement EQ/NE type validation: reject BOOL×STR; coerce SEL_NUM to SEL_STR when comparing with string resource (`snprintf "%.17g"`), NULL-check `makeString` result, symmetric coercion | Free orphaned SEL_NUM node in unoptimized mode before overwriting pointer |
| 4b-3 | Implement `>/<`/`>=`/`<=` type validation: reject non-numeric operands | `log_fatal` (not `assert`) for belt-and-suspenders at eval time too |
| 4b-4 | Implement `res_idx` validity guard and unknown-resource error; add migration diagnostic for `true`/`false` identifiers (parse as SEL_RESNAME with no match) | |
| 4b-5 | Update `kill_primary_resreq` for conditional teardown: CSE-enabled → null root only; CSE-disabled → `free_select_tree()` then `FREEUP` | NULL store before `FREEUP` to prevent GC mark dangling-pointer race |
| 4b-6 | **LC_NUMERIC enforcement**: wrap PEG parser call in `parse_select` with `newlocale(LC_NUMERIC_MASK, "C")` / `uselocale` / `freelocale`; `log_fatal` on failure in scheduler, `log_error`+return in API server | `_XOPEN_SOURCE 700` required before system headers |
| 4b-7 | **strtod locale coverage check**: `grep -n 'strtod\|atof\|strtof' build/lib/xce_select.c` — if packcc inserts `strtod` in scanner, use `%0` capture + explicit `strtod` inside PEG action | Document finding (matches or no-matches) as a confirmed safe/unsafe result |
| 4b-1 | ~~Wire `parse_select()` into `parse_resreq_primary` **before** `kill_resreq_data`~~ **DONE** | Use-after-free if called after `kill_resreq_data` frees `data.selectString` |
| 4b-2 | ~~Implement EQ/NE type validation: reject BOOL×STR; coerce SEL_NUM to SEL_STR when comparing with string resource (`snprintf "%.17g"`), NULL-check `makeString` result, symmetric coercion~~ **DONE** | Free orphaned SEL_NUM node in unoptimized mode before overwriting pointer |
| 4b-3 | ~~Implement `>/<`/`>=`/`<=` type validation: reject non-numeric operands~~ **DONE** | `log_fatal` (not `assert`) for belt-and-suspenders at eval time too |
| 4b-4 | ~~Implement `res_idx` validity guard and unknown-resource error; add migration diagnostic for `true`/`false` identifiers (parse as SEL_RESNAME with no match)~~ **DONE** | |
| 4b-5 | ~~Update `kill_primary_resreq` for conditional teardown: CSE-enabled → null root only; CSE-disabled → `free_select_tree()` then `FREEUP`~~ **DONE** | NULL store before `FREEUP` to prevent GC mark dangling-pointer race |
| 4b-6 | ~~**LC_NUMERIC enforcement**: wrap PEG parser call in `parse_select` with `newlocale(LC_NUMERIC_MASK, "C")` / `uselocale` / `freelocale`; `log_fatal` on failure in scheduler, `log_error`+return in API server~~ **DONE** | `_XOPEN_SOURCE 700` required before system headers |
| 4b-7 | ~~**strtod locale coverage check**~~ **DONE — SAFE**: packcc emits single `strtod` in `pcc_action_value_node_1` (line 1593 of generated file), inside the `uselocale` scope — no locale vulnerability | |
| 4b-R1 | ~~**[Review fix — HIGH]** Add `lock(auxil->resmgr, RTTI_LOCK_READ)` / `unlock(auxil->resmgr)` around `key_find` in `resolve_resname()` (`xce_select.peg:38`) — missing read lock caused data race in multi-threaded scheduler~~ **DONE** | |
| 4b-R2 | **[Review check — F2]** STRING×NUM coercion error paths reviewed: `makeSelectNode` releases the xstring on OOM failure (level-0 path calls `dereference(op1)` before returning NULL); outer `makeSelectNode(op_type)` failure correctly calls `SEL_DISCARD(l)` + `SEL_DISCARD(r)` — **no leak, false positive** | |
---
## Phase 4c — Commutativity + Associativity Flattening
## Phase 4c — Commutativity + Associativity Flattening ✓ DONE (2026-04-29)
| # | Task | Notes |
|---|------|-------|
| 4c-1 | Verify commutativity sort in `makeSelectNode` step 1 covers AND/OR/EQ/NE; non-commutative GT/LT/GE/LE/NOT unchanged | Already integrated in §5.4 design |
| 4c-2 | Implement post-parse **flatten pass** for `&&`/`||` chains: `do { changed = flatten_pass(root); } while (changed)` | Repeat-until-stable needed for depth-3+ explicit parenthesization |
| 4c-3 | In unoptimized mode: `flatten_pass` must `free()` each orphaned node — not its children (re-parented children must not be freed) | Children freed would cause double-free when `free_select_tree` walks from root |
| 4c-1 | ~~Verify commutativity sort in `makeSelectNode` step 1 covers AND/OR/EQ/NE; non-commutative GT/LT/GE/LE/NOT unchanged~~ **DONE** — `test_cse_gt_not_commutative` confirms GT is not commutative; AND/OR/EQ/NE sort verified. | Already integrated in §5.4 design |
| 4c-2 | ~~Implement post-parse **flatten pass** for `&&`/`||` chains: `do { changed = flatten_pass(root); } while (changed)`~~ **DONE** — `sel_flatten_pass` added to `xce_select.c`; do-while loop in `parse_select()` in `resreq.c`; 7 tests: noop, single-level, deep 3+, OR, mixed-ops, CSE mode. 157 PASS. | Repeat-until-stable needed for depth-3+ explicit parenthesization |
| 4c-3 | ~~In unoptimized mode: `flatten_pass` must `free()` each orphaned node — not its children (re-parented children must not be freed)~~ **DONE** — level==0 path frees orphaned intermediate node (`free(new_right); free(node)`) after re-parenting A/B/C; SEL_NOT branch frees `node` only. | Children freed would cause double-free when `free_select_tree` walks from root |
**Senior review fixes applied (2026-04-29):**
- **[MEDIUM — OOM safety]** In AND/OR branch: update `node->u.binop.left/right` in-place (level==0) immediately after recursive calls, before any `makeSelectNode`. This ensures `return node` after OOM returns a valid tree with live (not dangling) children.
- **[LOW — dead code]** Removed dead `if (node->type == SEL_STR ...)` / `if (new_right->type == SEL_STR ...)` guards in AND/OR and SEL_NOT branches — those nodes are never SEL_STR in those arms.
---
## Phase 4d — Const-Fold Transform
## Phase 4d — Const-Fold Transform ✓ DONE (2026-04-29)
| # | Task | Notes |
|---|------|-------|
| 4d-1 | `resname == any` → `SEL_TRUE`; `resname != any` → `SEL_FALSE` — check **both** child slots for SEL_ANY pointer (commutativity sort is ASLR-nondeterministic across runs) | |
| 4d-2 | Identity laws: `SEL_TRUE && X` → `X`; `SEL_FALSE \|\| X` → `X` | |
| 4d-3 | Annihilation laws: `SEL_FALSE && X` → `SEL_FALSE`; `SEL_TRUE \|\| X` → `SEL_TRUE` | |
| 4d-4 | Complement laws: `!SEL_TRUE` → `SEL_FALSE`; `!SEL_FALSE` → `SEL_TRUE` | Single-pass — const-fold shrinks trees, does not create new flattening opportunities |
| 4d-1 | ~~`resname == any` → `SEL_TRUE`; `resname != any` → `SEL_FALSE` — check **both** child slots for SEL_ANY pointer (commutativity sort is ASLR-nondeterministic across runs)~~ **DONE** | |
| 4d-2 | ~~Identity laws: `SEL_TRUE && X` → `X`; `SEL_FALSE \|\| X` → `X`~~ **DONE** | |
| 4d-3 | ~~Annihilation laws: `SEL_FALSE && X` → `SEL_FALSE`; `SEL_TRUE \|\| X` → `SEL_TRUE`~~ **DONE** | |
| 4d-4 | ~~Complement laws: `!SEL_TRUE` → `SEL_FALSE`; `!SEL_FALSE` → `SEL_TRUE`~~ **DONE** | Single-pass — const-fold shrinks trees, does not create new flattening opportunities |
---
## Phase 4e — Evaluation with Memoization
## Phase 4e — Evaluation with Memoization ✓ DONE (2026-04-29)
| # | Task | Notes |
|---|------|-------|
| 4e-1 | Implement `eval_get_bool`/`eval_set_bool` 2-bit bitmap accessors with `node_id >= bool_bits_cap` guards using `log_fatal` | `_Static_assert(EVAL_NONE == 0)` for memset-reset correctness |
| 4e-2 | Implement `eval_select_inner` — recursive eval with **explicit** error propagation (`if (lv < 0) return -1`), NOT C `&&`/`||` (C's `-1` is truthy, would swallow errors) | SEL_ANY reaching eval → `log_fatal("SEL_ANY reached eval — const-fold not applied")` |
| 4e-3 | Leaf cache writes in `eval_select_inner`: SEL_RESNAME writes to `ctx->num_vals[node_id]` or `ctx->str_vals[node_id]`; comparison ops call `eval_select(child, ctx)` then read cached arrays | **Critical**: `get_leaf_value()` must NOT be called directly — bypasses cache writes, silently reads 0.0/NULL |
| 4e-4 | Implement memoized `eval_select` wrapper — bitmap cache check before calling `eval_select_inner`; cache EVAL_ERROR too (node-structural errors repeat per host) | |
| 4e-5 | Implement `eval_select_simple` for unoptimized path — plain recursive tree-walk, no `select_eval_ctx_t`, thread-safe (no shared state) | |
| 4e-6 | Add `eval_select` guard: `log_fatal` if `sel_get_optimize_level() <= 0` | |
| 4e-7 | Implement `select_eval_ctx_init` / `select_eval_ctx_reset` / `select_eval_ctx_fini` | `node_count` snapshot at init; `reset` uses snapshot, never re-reads `sel_node_next_id` |
| 4e-8 | Pre-session capacity check: `memset` extended bool_bits bytes after realloc (realloc does not zero extension); partial-OOM contract — all three reallocs (bool_bits, num_vals, str_vals) must succeed atomically or `log_fatal` | Mismatched capacities allow out-of-bounds write to num_vals with no inline bounds check |
| 4e-1 | ~~Implement `eval_get_bool`/`eval_set_bool` 2-bit bitmap accessors with `node_id >= bool_bits_cap` guards using `log_fatal`~~ **DONE** | `_Static_assert(EVAL_NONE == 0)` for memset-reset correctness |
| 4e-2 | ~~Implement `eval_select_inner` — recursive eval with **explicit** error propagation (`if (lv < 0) return -1`), NOT C `&&`/`||` (C's `-1` is truthy, would swallow errors)~~ **DONE** | SEL_ANY reaching eval → `log_fatal("SEL_ANY reached eval — const-fold not applied")` |
| 4e-3 | ~~Leaf cache writes in `eval_select_inner`: SEL_RESNAME writes to `ctx->num_vals[node_id]` or `ctx->str_vals[node_id]`; comparison ops call `eval_select(child, ctx)` then read cached arrays~~ **DONE** | **Critical**: `get_leaf_value()` must NOT be called directly — bypasses cache writes, silently reads 0.0/NULL |
| 4e-4 | ~~Implement memoized `eval_select` wrapper — bitmap cache check before calling `eval_select_inner`; cache EVAL_ERROR too (node-structural errors repeat per host)~~ **DONE** | |
| 4e-5 | ~~Implement `eval_select_simple` for unoptimized path — plain recursive tree-walk, no `select_eval_ctx_t`, thread-safe (no shared state)~~ **DONE** | |
| 4e-6 | ~~Add `eval_select` guard: `log_fatal` if `sel_get_optimize_level() <= 0`~~ **DONE** | |
| 4e-7 | ~~Implement `select_eval_ctx_init` / `select_eval_ctx_reset` / `select_eval_ctx_fini`~~ **DONE** | `node_count` snapshot at init; `reset` uses snapshot, never re-reads `sel_node_next_id` |
| 4e-8 | ~~Pre-session capacity check: `memset` extended bool_bits bytes after realloc (realloc does not zero extension); partial-OOM contract — all three reallocs (bool_bits, num_vals, str_vals) must succeed atomically or `log_fatal`~~ **DONE** | Mismatched capacities allow out-of-bounds write to num_vals with no inline bounds check |
---
## Phase 4f — Mark-and-Sweep GC + Recompaction
## Phase 4f — Mark-and-Sweep GC + Recompaction ✓ DONE (2026-04-29)
| # | Task | Notes |
|---|------|-------|
| 4f-1 | Implement `gc_mark_tree` — recursive DFS, early return on `node == NULL \|\| node->gc_mark`; handles leaf / SEL_NOT / binary node types | Recursion bounded by `SELECT_MAX_DEPTH` (64); any increase requires stack analysis |
| 4f-2 | Implement `gc_sweep_select_nodes` — mark pass (clear all gc_mark + mark from live jobs dispatching on XCE_RESREQ_PRIMARY/COMPOUND/ALTER), sweep pass (collect unmarked, `key_del`, `dereference(strval)`, `free(node)`, `zarray_del`) | Use `xlist_get` with counted loop for xlist iteration; GC does NOT acquire `sel_node_table.lock` (idle-phase invariant) |
| 4f-3 | Insert **`atomic_thread_fence(memory_order_seq_cst)`** after all `primary->select->root = NULL` unlinks and before `gc_sweep_select_nodes` call | Without fence: CPU/compiler may reorder NULL store after GC mark pass |
| 4f-4 | Runtime GC idle invariant guards at GC call site (`sched_session_active`, `sched_eval_in_progress`, `staging_merge_complete`) — `log_fatal` not `assert` | `assert` compiled out by NDEBUG; use-after-free would result |
| 4f-5 | Implement `recompact_node_ids` — iterate `nodes` zarray over `caps` slots (not `count`, gaps exist after `zarray_del`), assign sequential `node_id`, reset `sel_node_next_id` via `atomic_store_explicit(..., memory_order_release)` | `node_id` and zarray slot index are **decoupled** — do not conflate |
| 4f-1 | ~~Implement `gc_mark_tree` — recursive DFS, early return on `node == NULL \|\| node->gc_mark`; handles leaf / SEL_NOT / binary node types~~ **DONE** — `__attribute__((unused))` for Phase 6 wire-in | Recursion bounded by `SELECT_MAX_DEPTH` (64); any increase requires stack analysis |
| 4f-2 | ~~Implement `gc_sweep_select_nodes` — Phase 6 guard placed **before** mark-clear (prevents two-call sweep-of-reachable-nodes bug); mark-clear iterates `caps`; sweep iterates `caps`, `zarray_del` before `free`~~ **DONE** — live_jobs traversal deferred to Phase 6 (sched/types.h not in libxce scope) | GC does NOT acquire `sel_node_table.lock` (idle-phase invariant) |
| 4f-3 | Insert **`atomic_thread_fence(memory_order_seq_cst)`** after all `primary->select->root = NULL` unlinks and before `gc_sweep_select_nodes` call — **Phase 6 call-site task** | Without fence: CPU/compiler may reorder NULL store after GC mark pass |
| 4f-4 | Runtime GC idle invariant guards at GC call site (`sched_session_active`, `sched_eval_in_progress`, `staging_merge_complete`) — `log_fatal` not `assert` — **Phase 6 call-site task** → see 6-3/6-4 | `assert` compiled out by NDEBUG; use-after-free would result |
| 4f-5 | ~~Implement `recompact_node_ids` — iterate `nodes` zarray over `caps` slots (not `count`, gaps exist after `zarray_del`), assign sequential `node_id`, reset `sel_node_next_id` via `atomic_store_explicit(..., memory_order_release)`~~ **DONE** | `node_id` and zarray slot index are **decoupled** — do not conflate |
| 4f-R1 | ~~**[Review fix — HIGH]** Move Phase 6 early-return guard to before mark-clear loop — prior ordering allowed two consecutive calls (njobs>0 then njobs==0) to sweep still-reachable nodes~~ **DONE** | Two-call scenario: first call clears all marks and returns; second call sweeps everything |
| 4f-R2 | ~~**[Review fix — HIGH]** Swap `free(node)` / `zarray_del` order — `zarray_del` NULLs slot first, then `free`; eliminates latent use-after-free if killer is ever installed on nodes zarray~~ **DONE** | |
| 4f-R3 | ~~**[Bug fix]** `sel_node_table_fini` iterated `nodes.count` instead of `nodes.caps` — after `zarray_del` creates gaps, new nodes slot into positions beyond `count` boundary and leaked~~ **DONE** | Manifested as 32-byte ASAN leak after GC sweep + reintern in same lifetime |
---
@ -181,29 +191,32 @@ Develop concurrently with Phase 3 (grammar is the test oracle).
| # | Task | Notes |
|---|------|-------|
| 5-1 | Add `custom_target` for `xce_select.peg` in `lib/meson.build` matching `section_parser`/`rusage_parser` pattern | Verify `-o 'lib/xce_select'` path format: `grep -A8 'custom_target' src/lib/meson.build \| grep '\-o'` |
| 5-2 | Add `xce_select_parser[0]` (generated `.c`) to `libxce_src` | |
| 5-3 | Add `test_resmgr_fixture.c` to testsuite `executable()` sources in `lib/meson.build` | |
| 5-1 | ~~Add `custom_target` for `xce_select.peg` in `lib/meson.build` matching `section_parser`/`rusage_parser` pattern~~ **DONE** | Verify `-o 'lib/xce_select'` path format: `grep -A8 'custom_target' src/lib/meson.build \| grep '\-o'` |
| 5-2 | ~~Add `xce_select_parser[0]` (generated `.c`) to `libxce_src`~~ **DONE** | |
| 5-3 | ~~Add `test_resmgr_fixture.c` to testsuite `executable()` sources in `lib/meson.build`~~ **DONE** | |
---
## Phase 6 — Scheduler Integration
**Priority: MEDIUM — after Phases 1–5 pass all tests**
**Status: ✓ DONE — 2026-04-30 — 10/10 scheduler tests pass**
| # | Task | Notes |
|---|------|-------|
| 6-1 | **Pre-Phase-6 audit**: `grep -rn 'eval_select\|alloc_thread\|sched_session' src/scheduler/` — confirm allocator-thread-only invariant | Document result ("confirmed single-threaded" or "found unexpected call site at [location]") in commit message |
| 6-2 | **Blocking gate**: `grep -rn 'sched_meet_host_selection\|meet_host_selection' src/` — all hits must be accounted for before signature change | Missed call site → wrong ABI → UB at runtime (garbage stack parameter) |
| 6-3 | Add `_Atomic int session_active`, `eval_in_progress`, `staging_merge_complete` to `xce_scheduler_t` in `src/scheduler/include/scheduler.h` | Use `_Atomic int` not `volatile int` — ARM atomicity guarantee |
| 6-4 | Implement GC predicate wrappers referencing real `xce_scheduler_t` fields (replace Phase 1 `fake_scheduler_t` stubs) | Signatures unchanged; only wrapper body changes |
| 6-5 | Restructure scheduler session loop to **outer-host / inner-job** pattern with single `select_eval_ctx_t` per session | Current stub returns 1 — full from-scratch implementation |
| 6-6 | Update `sched_meet_host_selection` to `void` return; migrate `add_job_preason(PREASON_MATCH_FAIL)` into inner-job loop | Deleting the preason call silently leaves jobs appearing to hang |
| 6-7 | Implement `build_host_eval_ctx` (scheduler-private in `src/scheduler/src/`) — flatten `worker->resvals` zarray into `numvals`/`strvals` arrays using `Value_u_fvalue`/`Value_u_svalue`/`Value_u_bvalue` from capnp-generated header | Confirm header location: `grep -rn 'Value_u_fvalue' src/scheduler/` |
| 6-8 | Materialize `live_jobs` flat array from `sched->jobs` kvmgr using growth-loop heap allocation (not fragile `zarray_t.data` cast) | |
| 6-9 | Wire `sel_optimize_set(1)` at scheduler startup; `sel_optimize_set(0)` at API server startup | Add `log_fatal` guard at session loop entry if `sel_get_optimize_level() <= 0` |
| 6-10 | Set `sel_alloc_thread_id` at allocator thread first iteration; add thread-ID checks in `eval_select` and `gc_sweep_select_nodes` | Skip check when `sel_alloc_thread_id == 0` (unregistered — Phase 1–5 unit tests) |
| 6-11 | Write integration/parity test against known-good resreq/host pairs | Stub always returns 1, so no meaningful baseline exists — must test explicitly |
| 6-1 | ~~**Pre-Phase-6 audit**: confirm allocator-thread-only invariant~~ **DONE** `bb4a95c` | Single-threaded confirmed; stub at `resource.c:127` |
| 6-2 | ~~**Blocking gate**: all `sched_meet_host_selection` call sites accounted for~~ **DONE** `bb4a95c` | No missed call sites |
| 6-3 | ~~Add `_Atomic int session_active`, `eval_in_progress`, `staging_merge_complete` to `xce_scheduler_t`~~ **DONE** `823a66e` | Use `_Atomic int` not `volatile int` — ARM atomicity guarantee |
| 6-4 | ~~Implement GC predicate wrappers referencing real `xce_scheduler_t` fields~~ **DONE** `acda9bb` | Signatures unchanged; only wrapper body changed |
| 6-5 | ~~Implement `sched_meet_host_selection` via `eval_select_simple`~~ **DONE** `40261b7` | Full implementation replacing stub that always returned 1 |
| 6-6 | ~~`sched_meet_host_selection` returns int (0/1); preason on fail~~ **DONE** `40261b7` | int return kept; preason wired |
| 6-7 | ~~Implement `build_host_eval_ctx`~~ **DONE** `907d8d8` | Flattens `worker->resvals` zarray to `numvals`/`strvals` using `Value_u_fvalue/svalue/bvalue` |
| 6-8 | ~~Materialize `live_jobs` flat array from `sched->jobs` kvmgr~~ **DONE** `5267d36` | Growth-loop heap alloc; OOM-safe (double-guarded: `calloc` fail + `realloc` null-sentinel `742d79e`, `fb5127a`) |
| 6-9 | ~~Wire `sel_optimize_set(1)` at scheduler startup~~ **DONE** `acda9bb` | Called in `sched_init`; no `sel_optimize_set(0)` needed (API server has separate process) |
| 6-10 | ~~Set `sel_alloc_thread_id`; add thread-ID checks in `eval_select` and `gc_sweep_select_nodes`~~ **DONE** `b722e5c` `1eb4eb1` | `_Atomic int` registered flag + `pthread_equal` guard; registered before `while(1)` loop |
| 6-11 | ~~Write integration/parity tests against known-good resreq/host pairs~~ **DONE** `5fcf85c` | Numeric (`SEL_GE`) and string (`SEL_EQ`) pass/fail cases |
| 6-12 | ~~`atomic_thread_fence(memory_order_seq_cst)` before `gc_sweep_select_nodes`~~ **DONE** `5267d36` | After all `gc_mark_select_tree` calls in `sched_end_session` |
| R1 | ~~**[Review fix — CRITICAL]** `staging_merge_complete` bootstrap deadlock — GC never ran~~ **DONE** `d438742` | `sched_begin_session` never set it to 1; guard in `sched_end_session` always fired |
| R2 | ~~**[Review fix — HIGH]** `recompact_node_ids()` missing from GC protocol~~ **DONE** `d438742` | Added after `gc_sweep_select_nodes` in `sched_end_session` |
---

8
src/flake.lock generated
View file

@ -117,11 +117,11 @@
"xbuild": "xbuild"
},
"locked": {
"lastModified": 1772607564,
"narHash": "sha256-kBYbiOvM7bBZdIFq2idjQ3eidlAT3w6WS3GRoxHM67s=",
"lastModified": 1777021191,
"narHash": "sha256-hyyJZBX9o1TQLH/7aaEP1/rV3QQTqn70PGs2IDACNnw=",
"ref": "master",
"rev": "10c32f769b4431cb657c3c8f23158f643fad74b4",
"revCount": 19,
"rev": "7212ba170d8db040ac99d0fb47c7b1df70df6e11",
"revCount": 20,
"type": "git",
"url": "ssh://forgejo@gitea.shenrs.eu/rshen/librtti"
},

View file

@ -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 {

View file

@ -6,6 +6,9 @@ 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"
@ -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)
}

View file

@ -2,6 +2,7 @@ package backend
import (
"apiserver/models"
"encoding/json"
"time"
)
@ -9,7 +10,7 @@ import (
//
// |63 62|61 56|55 52|51 20|19 15|14 0|
// | TAG | CLUSTER_ID | APISERVER_ID| TIMESTAMP | SEQ_HIGH | ARRAY_IDX |
// | 2 bits | 6 bits | 4 bits | 32 bits | 15 bits | 15 bits |
// | 2 bits | 6 bits | 4 bits | 32 bits | 5 bits | 15 bits |
const (
nTagBits = 2
tagShift = 62
@ -20,8 +21,8 @@ const (
nTimestampBits = 32
timestampShift = 20
nArrayIdxBits = 15
sequenceHighShift = 15 // mirrors SEQUENCE_HIGH_SHIFT in config.h
nSequenceBits = 30 // mirrors N_SEQUENCE_BITS in config.h
sequenceHighShift = 15 // mirrors SEQUENCE_HIGH_SHIFT in config.h (== N_ARRAY_IDX_BITS)
nSequenceBits = 20 // N_JOBID_BITS(64) - N_CLUSTER_ID_BITS(6) - N_APISERVER_ID_BITS(4) - N_TIMESTAMP_BITS(32) - N_TAG_BITS(2) = 20
tagJobArray = 0x01 // mirrors TAG_JOB_ARRAY in config.h
)
@ -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
}

View file

@ -367,6 +367,42 @@ func TestRawJobToModel_ArrayChildrenShareBaseID(t *testing.T) {
}
}
// TestExtractArrayBaseID_LargeTimestamp verifies that ExtractArrayBaseID does not
// leak timestamp bits into the extracted sequence when the timestamp field has
// large values. This would fail with nSequenceBits=30 (extracts 15 bits from
// position 15, bleeding into the timestamp at bits 20-51) but passes with the
// correct nSequenceBits=20 (extracts 5 bits from position 15).
func TestExtractArrayBaseID_LargeTimestamp(t *testing.T) {
jobNumber := uint64(7)
arrayIdx := uint64(3)
sequence := jobNumber<<sequenceHighShift + arrayIdx
// Use a typical Unix-epoch timestamp (bits 20-51 should stay in timestamp)
id := makeJobID(1, 0, 1714560000, tagJobArray, sequence)
base := ExtractArrayBaseID(id)
// The base ID must have the same cluster, apiserver, and timestamp —
// no timestamp bits should leak into the sequence field.
gotCluster := extractBits(base, clusterIDShift, nClusterIDBits)
gotApiserver := extractBits(base, apiserverIDShift, nApiserverIDBits)
gotTimestamp := extractBits(base, timestampShift, nTimestampBits)
if gotCluster != 1 {
t.Errorf("cluster ID leaked: got %d, want 1", gotCluster)
}
if gotApiserver != 0 {
t.Errorf("apiserver ID leaked: got %d, want 0", gotApiserver)
}
if gotTimestamp != 1714560000 {
t.Errorf("timestamp corrupted: got %d, want 1714560000", gotTimestamp)
}
// The base ID's sequence should be just jobNumber<<sequenceHighShift (idx=0, tag=0)
expectedBase := makeJobID(1, 0, 1714560000, 0, jobNumber<<sequenceHighShift)
if base != expectedBase {
t.Errorf("base ID mismatch with large timestamp: got %d, want %d", base, expectedBase)
}
}
func TestIsArrayJob(t *testing.T) {
// Non-array job (tag=0x00): IsArrayJob returns false
nonArrayID := makeJobID(1, 0, 1700000000, 0, 5)

View file

@ -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

View file

@ -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) {

View file

@ -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)
}
}

View file

@ -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)

View file

@ -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"

View file

@ -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 {

View file

@ -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",

View file

@ -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
}

View file

@ -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)
}
}
}

View file

@ -4,6 +4,7 @@ import (
"encoding/json"
"fmt"
"strings"
"xcecli/common/format"
"xcecli/common/request"
"xcecli/config"
"xcecli/models"
@ -47,8 +48,15 @@ func (c *JobController) RunJobsControl() error {
// control jobs by job IDs
func (c *JobController) controlJobsByIDs(apiClient *request.APIClient, token string, certPath string) error {
jobIDsStr := strings.Join(c.jobIDs, ",")
url := fmt.Sprintf("/xce/v1/jobs/%s/control", jobIDsStr)
apiIDs := make([]string, 0, len(c.jobIDs))
for _, id := range c.jobIDs {
apiID, err := format.ToAPIJobID(strings.TrimSpace(id))
if err != nil {
return fmt.Errorf("invalid job ID %q: %w", id, err)
}
apiIDs = append(apiIDs, apiID)
}
url := fmt.Sprintf("/xce/v1/jobs/%s/control", strings.Join(apiIDs, ","))
req := models.JobControlJobIDRequest{
Action: c.action,
@ -88,19 +96,11 @@ func (c *JobController) printJobControlResponse(req interface{}, apiClient *requ
return err
}
for _, data := range resp.Data.Accepted {
if data.ArrayIdx != 0 {
fmt.Printf("job <%d[%d]> %s\n", data.JobID, data.ArrayIdx, strings.TrimPrefix(data.Msg, "job "))
} else {
fmt.Printf("job <%d> %s\n", data.JobID, strings.TrimPrefix(data.Msg, "job "))
}
fmt.Printf("job <%s> %s\n", format.FormatArrayChildID(data.JobID, data.ArrayIdx), strings.TrimPrefix(data.Msg, "job "))
}
for _, data := range resp.Data.Failed {
if data.ArrayIdx != 0 {
fmt.Printf("job <%d[%d]> %s\n", data.JobID, data.ArrayIdx, strings.TrimPrefix(data.Msg, "job "))
} else {
fmt.Printf("job <%d> %s\n", data.JobID, strings.TrimPrefix(data.Msg, "job "))
}
fmt.Printf("job <%s> %s\n", format.FormatArrayChildID(data.JobID, data.ArrayIdx), strings.TrimPrefix(data.Msg, "job "))
}
return nil
}

View file

@ -0,0 +1,148 @@
package format
import (
"fmt"
"regexp"
"strconv"
)
// JobID bit layout constants — mirrors src/include/config.h.
//
// |63 62|61 56|55 52|51 20|19 15|14 0|
// | TAG | CLUSTER_ID | APISERVER_ID| TIMESTAMP | SEQ_HIGH | ARRAY_IDX |
// | 2 bits | 6 bits | 4 bits | 32 bits | 5 bits | 15 bits |
const (
nTagBits = 2
tagShift = 62
nClusterIDBits = 6
clusterIDShift = 56
nApiserverIDBits = 4
apiserverIDShift = 52
nTimestampBits = 32
timestampShift = 20
nArrayIdxBits = 15
sequenceHighShift = 15 // mirrors SEQUENCE_HIGH_SHIFT in config.h (== N_ARRAY_IDX_BITS)
nSequenceBits = 20 // N_JOBID_BITS(64) - N_CLUSTER_ID_BITS(6) - N_APISERVER_ID_BITS(4) - N_TIMESTAMP_BITS(32) - N_TAG_BITS(2) = 20
tagJobArray = 0x01
)
// extractBits extracts a bit field from id at the given shift position.
// Mirrors EXTRACT_BITS(id, shift, bits) in src/include/utils.h.
func extractBits(id uint64, shift, bits int) uint64 {
mask := uint64((1 << bits) - 1)
return (id >> shift) & mask
}
// FormatJobID decomposes a 64-bit packed job ID and formats it as a
// human-readable string matching the C fmt_jobid() format in src/lib/utils.c.
// - Regular job: "{cluster}-{apiserver}-{timestamp}-{sequence}"
// - Array child: "{cluster}-{apiserver}-{timestamp}-{base}[{idx}]"
func FormatJobID(uniqueJobID int64) string {
id := uint64(uniqueJobID)
clusterID := extractBits(id, clusterIDShift, nClusterIDBits)
apiserverID := extractBits(id, apiserverIDShift, nApiserverIDBits)
timestamp := extractBits(id, timestampShift, nTimestampBits)
if (extractBits(id, tagShift, nTagBits) & tagJobArray) != 0 {
// arrayBase mirrors EXTRACT_ARRAY_JOB: the seq_high bits shifted back to their
// original position, matching C's fmt_jobid_ex output exactly.
arrayBase := extractBits(id, sequenceHighShift, nSequenceBits-nArrayIdxBits) << uint(sequenceHighShift)
arrayIdx := extractBits(id, 0, nArrayIdxBits)
return fmt.Sprintf("%d-%d-%d-%d[%d]",
clusterID, apiserverID, timestamp, arrayBase, arrayIdx)
}
sequence := extractBits(id, 0, nSequenceBits)
return fmt.Sprintf("%d-%d-%d-%d",
clusterID, apiserverID, timestamp, sequence)
}
// IsArrayJob returns true if the 64-bit packed job ID has the array job tag set.
func IsArrayJob(uniqueJobID int64) bool {
return extractBits(uint64(uniqueJobID), tagShift, nTagBits)&tagJobArray != 0
}
// DecomposeJobID splits a 64-bit packed job ID into its component fields.
func DecomposeJobID(uniqueJobID int64) (clusterID, apiserverID, timestamp, sequence uint32, isArray bool, arrayIdx uint32) {
id := uint64(uniqueJobID)
clusterID = uint32(extractBits(id, clusterIDShift, nClusterIDBits))
apiserverID = uint32(extractBits(id, apiserverIDShift, nApiserverIDBits))
timestamp = uint32(extractBits(id, timestampShift, nTimestampBits))
isArray = extractBits(id, tagShift, nTagBits)&tagJobArray != 0
sequence = uint32(extractBits(id, 0, nSequenceBits))
arrayIdx = uint32(extractBits(id, 0, nArrayIdxBits))
return
}
// MakeJobID packs the components into a 64-bit job ID, mirroring MAKE_JOBID in src/include/utils.h.
func MakeJobID(clusterID, apiserverID, timestamp uint64, tag, sequence uint64) uint64 {
return (clusterID << clusterIDShift) |
(apiserverID << apiserverIDShift) |
(timestamp << timestampShift) |
(tag << tagShift) |
sequence
}
var (
// regexJobID matches "{int}-{int}-{int}-{int}" — mirrors REGEX_JOBID in config.h.
regexJobID = regexp.MustCompile(`^\s*(\d+)-(\d+)-(\d+)-(\d+)\s*$`)
// regexJobArrayID matches "{int}-{int}-{int}-{int}[{int}]" — mirrors REGEX_JOBARRAY_ID in config.h.
regexJobArrayID = regexp.MustCompile(`^\s*(\d+)-(\d+)-(\d+)-(\d+)\[(\d+)\]\s*$`)
)
// ParseJobID parses a human-readable job ID string into a packed 64-bit job ID,
// mirroring parse_jobid() in src/lib/utils.c:482.
// Accepts: "0-0-12874529-0" (regular) or "0-0-12874529-0[5]" (array child).
func ParseJobID(s string) (uint64, error) {
if m := regexJobArrayID.FindStringSubmatch(s); m != nil {
clusterID, _ := strconv.ParseUint(m[1], 10, 64)
apiserverID, _ := strconv.ParseUint(m[2], 10, 64)
timestamp, _ := strconv.ParseUint(m[3], 10, 64)
idJob, _ := strconv.ParseUint(m[4], 10, 64)
idIdx, _ := strconv.ParseUint(m[5], 10, 64)
return MakeJobID(clusterID, apiserverID, timestamp, tagJobArray, idJob+idIdx), nil
}
if m := regexJobID.FindStringSubmatch(s); m != nil {
clusterID, _ := strconv.ParseUint(m[1], 10, 64)
apiserverID, _ := strconv.ParseUint(m[2], 10, 64)
timestamp, _ := strconv.ParseUint(m[3], 10, 64)
idJob, _ := strconv.ParseUint(m[4], 10, 64)
return MakeJobID(clusterID, apiserverID, timestamp, 0, idJob), nil
}
return 0, fmt.Errorf("invalid job ID format: %q", s)
}
// ExtractBaseJobID clears the tag and array_idx bits, returning the base job ID
// that the API server stores in the job_id column.
func ExtractBaseJobID(packed uint64) uint64 {
id := packed &^ (uint64((1<<nTagBits)-1) << uint(tagShift)) // clear tag bits
id = id &^ uint64((1<<nArrayIdxBits)-1) // clear array_idx bits
return id
}
// FormatArrayChildID formats a job ID from a base job ID and array index.
// Used when only base+index are available (e.g. JobControlResponseItem).
func FormatArrayChildID(baseJobID int64, arrayIdx int) string {
if arrayIdx == 0 {
return FormatJobID(baseJobID)
}
uniqueID := uint64(baseJobID) | (uint64(tagJobArray) << uint(tagShift)) | uint64(arrayIdx)
return FormatJobID(int64(uniqueID))
}
// ToAPIJobID converts a fmt_jobid-format string to the numeric form the API
// server expects ("{int}" or "{int}[{idx}]"). Plain integers pass through unchanged.
func ToAPIJobID(s string) (string, error) {
if regexJobArrayID.MatchString(s) || regexJobID.MatchString(s) {
packed, err := ParseJobID(s)
if err != nil {
return "", err
}
if IsArrayJob(int64(packed)) {
baseID := ExtractBaseJobID(packed)
arrayIdx := extractBits(packed, 0, nArrayIdxBits)
return fmt.Sprintf("%d[%d]", baseID, arrayIdx), nil
}
return fmt.Sprintf("%d", packed), nil
}
return s, nil // already plain int or int[idx]
}

View file

@ -0,0 +1,117 @@
package format
import (
"fmt"
"testing"
)
func TestFormatJobIDRegular(t *testing.T) {
packed := MakeJobID(1, 2, 12874529, 0, 42)
got := FormatJobID(int64(packed))
if got != "1-2-12874529-42" {
t.Errorf("got %q, want %q", got, "1-2-12874529-42")
}
}
func TestFormatJobIDArray(t *testing.T) {
// seq_high=0, array_idx=5 → sequence = (0<<15)|5 = 5
packed := MakeJobID(0, 0, 12874529, tagJobArray, 5)
got := FormatJobID(int64(packed))
if got != "0-0-12874529-0[5]" {
t.Errorf("got %q, want %q", got, "0-0-12874529-0[5]")
}
// seq_high=1 (shifted: 1<<15 = 32768), array_idx=7 → sequence = 32768+7
packed2 := MakeJobID(0, 0, 12874529, tagJobArray, (1<<sequenceHighShift)+7)
got2 := FormatJobID(int64(packed2))
if got2 != "0-0-12874529-32768[7]" {
t.Errorf("seq_high=1: got %q, want %q", got2, "0-0-12874529-32768[7]")
}
}
func TestParseJobIDRegular(t *testing.T) {
packed := MakeJobID(1, 2, 12874529, 0, 42)
got, err := ParseJobID("1-2-12874529-42")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got != packed {
t.Errorf("got 0x%x, want 0x%x", got, packed)
}
}
func TestParseJobIDArray(t *testing.T) {
packed := MakeJobID(0, 0, 12874529, tagJobArray, 5)
got, err := ParseJobID("0-0-12874529-0[5]")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got != packed {
t.Errorf("got 0x%x, want 0x%x", got, packed)
}
}
func TestParseFormatRoundTrip(t *testing.T) {
cases := []string{
"1-2-12874529-42",
"0-0-12874529-0[5]",
"0-0-12874529-32768[7]",
}
for _, want := range cases {
packed, err := ParseJobID(want)
if err != nil {
t.Fatalf("ParseJobID(%q): %v", want, err)
}
got := FormatJobID(int64(packed))
if got != want {
t.Errorf("round-trip: got %q, want %q", got, want)
}
}
}
func TestParseJobIDInvalid(t *testing.T) {
for _, s := range []string{"abc", "1-2-3", "1[2]", ""} {
if _, err := ParseJobID(s); err == nil {
t.Errorf("ParseJobID(%q): expected error, got nil", s)
}
}
}
func TestToAPIJobIDRegular(t *testing.T) {
packed := MakeJobID(0, 0, 12874529, 0, 42)
s := FormatJobID(int64(packed))
got, err := ToAPIJobID(s)
if err != nil {
t.Fatalf("ToAPIJobID(%q): %v", s, err)
}
want := fmt.Sprintf("%d", packed)
if got != want {
t.Errorf("got %q, want %q", got, want)
}
}
func TestToAPIJobIDArray(t *testing.T) {
packed := MakeJobID(0, 0, 12874529, tagJobArray, 5)
s := FormatJobID(int64(packed))
got, err := ToAPIJobID(s)
if err != nil {
t.Fatalf("ToAPIJobID(%q): %v", s, err)
}
baseID := ExtractBaseJobID(packed)
want := fmt.Sprintf("%d[5]", baseID)
if got != want {
t.Errorf("got %q, want %q", got, want)
}
}
func TestToAPIJobIDPassthrough(t *testing.T) {
for _, s := range []string{"123456", "123456[3]"} {
got, err := ToAPIJobID(s)
if err != nil {
t.Fatalf("ToAPIJobID(%q): %v", s, err)
}
if got != s {
t.Errorf("passthrough: got %q, want %q", got, s)
}
}
}

View file

@ -5,6 +5,7 @@ import (
"fmt"
"strings"
"time"
jobfmt "xcecli/common/format"
"xcecli/models"
)
@ -81,9 +82,7 @@ func (p *ArrayElementsPagePrinter) initColumnWidths(jobs *models.JobsResponse) {
width := 0
switch format.name {
case models.HEADER_JOB_JOBID:
// format the job id as "jobID[idx]"
arrayJobIDStr := fmt.Sprintf("%d[%d]", job.JobID, job.ArrayIdx)
width = len(arrayJobIDStr)
width = len(jobfmt.FormatJobID(job.UniqueJobID))
case models.HEADER_JOB_USER:
width = len(job.SubmitUser)
case models.HEADER_JOB_STATUS:
@ -156,9 +155,7 @@ func (p *ArrayElementsPagePrinter) formatRows(jobs *models.JobsResponse, truncat
align := format.align
switch format.name {
case models.HEADER_JOB_JOBID:
// format the job id as "jobID[idx]"
arrayJobIDStr := fmt.Sprintf("%d[%d]", job.JobID, job.ArrayIdx)
mainRow = appendOrDefault(mainRow, fmt.Sprintf("%"+align+"*s", 20, arrayJobIDStr), "-")
mainRow = appendOrDefault(mainRow, fmt.Sprintf("%"+align+"*s", 20, jobfmt.FormatJobID(job.UniqueJobID)), "-")
case models.HEADER_JOB_USER:
mainRow = appendOrDefault(mainRow, FormatStringField(job.SubmitUser, width, truncatFlag, true), "-")
case models.HEADER_JOB_STATUS:

View file

@ -7,6 +7,7 @@ import (
"strconv"
"strings"
"text/tabwriter"
jobfmt "xcecli/common/format"
"xcecli/models"
)
@ -83,11 +84,7 @@ func (p *JobsPagePrinter) initColumnWidths(jobs *models.JobsResponse) {
width := 0
switch format.name {
case models.HEADER_JOB_JOBID:
if job.UniqueJobID != job.JobID {
width = len(fmt.Sprintf("%d[%d]", job.JobID, job.ArrayIdx))
} else {
width = len(fmt.Sprintf("%d", job.JobID))
}
width = len(jobfmt.FormatJobID(job.UniqueJobID))
case models.HEADER_JOB_USER:
width = len(job.SubmitUser)
case models.HEADER_JOB_STATUS:
@ -177,14 +174,8 @@ func (p *JobsPagePrinter) formatRows(jobs *models.JobsResponse, truncatFlag bool
switch format.name {
case models.HEADER_JOB_JOBID:
var arrayJobIDStr string
if job.UniqueJobID != job.JobID {
arrayJobIDStr = fmt.Sprintf("%d[%d]", job.JobID, job.ArrayIdx)
} else {
arrayJobIDStr = fmt.Sprintf("%d", job.JobID)
}
mainRow = appendOrDefault(mainRow,
fmt.Sprintf("%"+align+"*s", 20, arrayJobIDStr), "-")
fmt.Sprintf("%"+align+"*s", 20, jobfmt.FormatJobID(job.UniqueJobID)), "-")
case models.HEADER_JOB_EXEC_HOST:
var execHosts models.ExecHosts
@ -276,11 +267,7 @@ func printJobsVerboseOutput(jobs *models.JobsResponse, filter []string) {
continue
}
}
// generate jobId for two format: 111 or 111[10]
jobId := fmt.Sprintf("%d", job.JobID)
if job.UniqueJobID != job.JobID {
jobId = fmt.Sprintf("%d[%d]", job.JobID, job.ArrayIdx)
}
jobId := jobfmt.FormatJobID(job.UniqueJobID)
// (1) print the job baisc info when sending "jobs list -l" command
firstLine := "Job <" + jobId + ">, "

View file

@ -4,6 +4,8 @@ import (
"fmt"
"regexp"
"strings"
"xcecli/common/format"
)
// JobsFilter struct for filtering jobs
@ -39,12 +41,18 @@ func (f *JobsFilter) HandleJobFilter(array_type string) (string, error) {
if f.JobIDs != "" {
jobIdStr := strings.Split(f.JobIDs, ",")
apiIDs := make([]string, 0, len(jobIdStr))
for _, jobId := range jobIdStr {
if err := isValidJobId(jobId); err != nil {
return "", err
}
apiID, err := format.ToAPIJobID(strings.TrimSpace(jobId))
if err != nil {
return "", err
}
apiIDs = append(apiIDs, apiID)
}
filters = append(filters, fmt.Sprintf("{job_id:in:(%s)}", f.JobIDs))
filters = append(filters, fmt.Sprintf("{job_id:in:(%s)}", strings.Join(apiIDs, ",")))
return strings.Join(filters, ","), nil
} else {
if array_type == "" {
@ -163,13 +171,20 @@ func GetJobStatusSet(f *JobsFilter) []string {
return result
}
var (
reFmtJobID = regexp.MustCompile(`^\d+-\d+-\d+-\d+$`)
reFmtJobArrayID = regexp.MustCompile(`^\d+-\d+-\d+-\d+\[\d+\]$`)
reNumericID = regexp.MustCompile(`^\d+$`)
reNumericIndex = regexp.MustCompile(`^\d+\[\d+\]$`)
)
func isValidJobId(jobID string) error {
matched, _ := regexp.MatchString(`^\d+$`, jobID)
matchedIndex, _ := regexp.MatchString(`^\d+\[\d+\]$`, jobID)
if !matched && !matchedIndex {
return fmt.Errorf("jobid '%s' contains invalid characters. Allowed characters are 0-9 or with index value", jobID)
s := strings.TrimSpace(jobID)
if reFmtJobID.MatchString(s) || reFmtJobArrayID.MatchString(s) ||
reNumericID.MatchString(s) || reNumericIndex.MatchString(s) {
return nil
}
return nil
return fmt.Errorf("jobid '%s': must be numeric, numeric[idx], or cluster-apiserver-timestamp-seq[idx] format", jobID)
}
// Helper function, generates search rule

View file

@ -64,6 +64,25 @@ struct xce_select_node_ {
typedef struct select_eval_ctx_ select_eval_ctx_t;
typedef struct host_eval_ctx_ host_eval_ctx_t;
/* Host snapshot: per-resource numeric (incl. bool as 0.0/1.0) and string values */
struct host_eval_ctx_ {
int num_res;
const double* numvals; /* indexed by res_idx */
const char** strvals; /* indexed by res_idx */
};
/* Per-evaluation-session memoization context */
struct select_eval_ctx_ {
const host_eval_ctx_t* host;
uint8_t* bool_bits; /* 2 bits per node_id: NONE/TRUE/FALSE/ERROR */
uint32_t bool_bits_cap; /* capacity in nodes (each uses 2 bits) */
uint32_t node_count; /* snapshot of sel_node_next_id at init */
double* num_vals; /* numeric/bool leaf cache, indexed by node_id */
uint32_t num_vals_cap;
const char** str_vals; /* string leaf cache, indexed by node_id */
uint32_t str_vals_cap;
};
/* Scheduler job forward declaration for GC API */
typedef struct xce_sched_job_ xce_sched_job_t;
@ -72,14 +91,27 @@ 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);
extern int sel_get_optimize_level(void);
/* Register the calling thread as the sole allocator/eval/GC thread.
* Must be called once at scheduler startup (before eval_select or
* gc_sweep_select_nodes). When unregistered (id==0) the checks are skipped. */
extern void sel_register_alloc_thread(void);
/* Intern table lifecycle */
extern void sel_node_table_init(void);
extern void sel_node_table_fini(void);
extern void sel_node_table_init(void);
extern void sel_node_table_fini(void);
extern uint32_t sel_get_node_count(void);
/* Test-only reset (requires #define UNIT_TEST before this header) */
#ifdef UNIT_TEST
@ -92,8 +124,7 @@ extern int eval_select_simple(const xce_select_node_t* root,
const host_eval_ctx_t* host);
/* Eval context lifecycle */
extern void select_eval_ctx_init(select_eval_ctx_t* ctx,
const host_eval_ctx_t* host);
extern select_eval_ctx_t* select_eval_ctx_init(uint32_t max_node_id);
extern void select_eval_ctx_reset(select_eval_ctx_t* ctx,
const host_eval_ctx_t* host);
extern void select_eval_ctx_fini(select_eval_ctx_t* ctx);
@ -103,6 +134,12 @@ extern void gc_sweep_select_nodes(const xce_sched_job_t** live_jobs,
size_t njobs);
extern void recompact_node_ids(void);
/* GC mark primitives — called by scheduler to implement the mark phase.
* Protocol: gc_clear_marks() → gc_mark_select_tree(root) per live node
* → gc_sweep_select_nodes(NULL, 0) → recompact_node_ids() */
extern void gc_clear_marks(void);
extern void gc_mark_select_tree(xce_select_node_t* node);
/* Flatten AND/OR right-associative chains to left-associative form.
* Returns the new root (may differ from the input root).
* Sets *changed to 1 if any re-association took place, leaves it unchanged
@ -251,6 +288,8 @@ extern res_type_t valtype_to_res_type(int v);
* Returns 1 if a == b, 0 otherwise. Named for future wildcard extension. */
static inline int xce_str_match(const char* a, const char* b)
{
/* NULL means the resource has no value — treat as no-match */
if (a == NULL || b == NULL) return 0;
return strcmp(a, b) == 0;
}

View file

@ -40,6 +40,11 @@ typedef struct {
xce_val_t* val;
} xce_val_with_idx_t;
typedef struct {
char* name;
xce_val_t* val;
} xce_name_val_t;
typedef struct {
char* name;
char* host;

View file

@ -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;
}

View file

@ -143,7 +143,7 @@ hent_t* ht_add(htab_t* t, void* key)
e->data = -1;
hc = HASH_KEY_HASH(t, key) % (t->nslots);
mylock_init(&(t->lock));
mylock_lock(&(t->lock));
e->next = t->slots[hc];
t->slots[hc] = e;
mylock_unlock(&(t->lock));
@ -162,7 +162,7 @@ int ht_del(htab_t* t, void* key)
}
hc = HASH_KEY_HASH(t, key) % (t->nslots);
mylock_init(&(t->lock));
mylock_lock(&(t->lock));
e = &(t->slots[hc]);
while ((*e) != NULL) {
if (HASH_KEY_CMP(t, (*e)->key, key) == 0) {

View file

@ -49,11 +49,24 @@ xce_resreq_parser = custom_target(
],
build_by_default: true)
xce_select_parser = custom_target(
'xce select expression parser',
output: ['xce_select_peg.c', 'xce_select_peg.h'],
input: ['xce_select.peg'],
command: [
packcc,
'-l',
'-o',
'lib/xce_select_peg',
'@INPUT@'
],
build_by_default: true)
libxce_src = [ gen_srcs[0],
section_parser[0],
rusage_parser[0],
xce_resreq_parser[0],
xce_select_parser[0],
'client.c',
'dataq.c',
'hook.c',
@ -83,6 +96,7 @@ libxce_src = [ gen_srcs[0],
'xvec.c',
'helper.c',
'resreq.c',
'xce_select.c',
'bitset.c',
'kv.c',
'jobj.c',
@ -141,6 +155,8 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_limit.c',
'testsuite/test_xvec.c',
'testsuite/test_resreq.c',
'testsuite/test_resmgr_fixture.c',
'testsuite/test_select.c',
'testsuite/test_bitset.c',
'testsuite/test_kv.c',
'testsuite/test_jobj.c',
@ -156,4 +172,4 @@ libxce_test = executable('xcelib_test',
dependencies: libxce_test_deps,
install: true)
test('libxce test', libxce_test)
test('libxce test', libxce_test, timeout: 120)

View file

@ -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) {

View file

@ -46,36 +46,22 @@ select_clause <- "select[" _ s:select_statement _ "]" { $$ = s; }
/ s:select_statement {$$ = s; }
select_statement <- (!("rusage[" / "]") .)* {$$ = strdup($0); }
/ (!("span[" "]" / "]") .)* {$$ = strdup($0); }
/ (!("same[" !"]" / "]") .)* {$$ = strdup($0);}
/ (!("affinity[" / "]") !"]" .)* {$$ = strdup($0);}
select_statement <- (!("rusage[" / "span[" / "same[" / "affinity[" / "]") .)* { $$=strdup($0); }
usage_clause <- "rusage[" _ usage:usage_statement _ "]" { $$ = usage; }
usage_statement <- (!("span[" / "]") .)* { $$ = strdup($0); }
/ (!("same[" / "]") .)* { $$ = strdup($0); }
/ (!("affinity[" / "]") .)* { $$ = strdup($0); }
usage_statement <- (!("span[" / "same[" / "affinity[" / "]") .)* { $$=strdup($0); }
span_clause <- "span[" _ span:span_statement _ "]" { $$ = span; }
span_statement <- (!("same[" / "]") .)* { $$ = strdup($0); }
/ (!("affinity[" / "]") .)* { $$ = strdup($0); }
span_statement <- (!("same[" / "affinity[" / "]") .)* { $$=strdup($0); }
same_clause <- "same[" _ same:same_statement _ "]" { $$ = same; }
same_statement <- (!("affinity[" / "]") .)* { $$ = strdup($0); }
affinity_clause <- "affinity[" _ affinity:affinity_statement _ "]" { $$ = affinity; }
affinity_statement <- (!"]" .)* { $$ = strdup($0); }
EOL <- "\n" / "\r\n" / "\r" / "\0"

View file

@ -22,6 +22,7 @@ extern void test_xlist(void);
extern void test_limit(void);
extern void test_xvec(void);
extern void test_resreq(void);
extern void test_select(void);
extern void test_bitset(void);
extern void test_kv(void);
extern void test_jobobj(void);
@ -119,6 +120,7 @@ int main(int argc, char* argv[])
TEST(test_limit);
TEST(test_xvec);
TEST(test_resreq);
TEST(test_select);
TEST(test_bitset);
TEST(test_kv);
TEST(test_jobobj);

View file

@ -66,6 +66,35 @@ void test_parse_section_fail(void)
FREEUP(errmsg);
}
void test_parse_section3(void)
{
char* resreq = "type == any span[ptile=1]";
char* errmsg = NULL;
int rc = 0;
xce_resreq_data_t data;
rc = parse_section(&data, (char*)resreq, &errmsg);
if (rc < 0) {
if (errmsg != NULL) {
printf("%s", errmsg);
FREEUP(errmsg);
}
}
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_TRUE(data.selectString != NULL);
TEST_ASSERT_TRUE((data.rusageString == NULL) && (data.spanString != NULL)
&& (data.sameString == NULL)
&& (data.affinityString == NULL));
TEST_ASSERT_TRUE(errmsg == NULL);
printf("select: %s\n", data.selectString);
printf("span: %s\n", data.spanString);
FREEUP(data.selectString);
FREEUP(data.spanString);
}
void test_parse_rusage1(void)
{
char* rusageString
@ -173,7 +202,7 @@ void test_resreq_primary(void)
int rc = 0;
xce_primary_resreq_t resreq;
rc = parse_resreq_primary(&resreq, resreqString, &errmsg);
rc = parse_resreq_primary(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
if (rc >= 0) {
int i;
@ -220,7 +249,7 @@ void test_resreq_primary2(void)
int rc = 0;
xce_primary_resreq_t resreq;
rc = parse_resreq_primary(&resreq, resreqString, &errmsg);
rc = parse_resreq_primary(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
if (rc >= 0) {
int i;
@ -241,7 +270,7 @@ void test_resreq1(void)
int rc = 0;
xce_resreq_t resreq;
rc = parse_resreq(&resreq, resreqString, &errmsg);
rc = parse_resreq(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
if (rc >= 0) {
int i;
@ -298,7 +327,7 @@ void test_resreq2(void)
memset(&resreq, 0x0, sizeof(resreq));
rc = parse_resreq(&resreq, resreqString, &errmsg);
rc = parse_resreq(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
if (rc >= 0) {
char* buf = NULL;
@ -327,7 +356,7 @@ void test_resreq3(void)
memset(&resreq, 0x0, sizeof(resreq));
rc = parse_resreq(&resreq, resreqString, &errmsg);
rc = parse_resreq(&resreq, resreqString, NULL, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
if (rc >= 0) {
char* buf = NULL;
@ -347,6 +376,111 @@ void test_resreq3(void)
}
}
/*
* Boundary tests for the P0-2 span-bleed fix in section.peg.
*
* Before the fix line 50 read: !("span[]" / "]")
* After the fix line 50 reads: !("span[" / "]")
*
* With the old grammar the negative lookahead would only suppress
* consumption of the literal token "span[]", so any span clause
* with content (e.g. "span[ptile=1]") was silently absorbed into
* selectString. These tests verify the corrected boundary.
*/
/* Test 1: explicit select[...] with a non-empty span[...] clause. */
void test_parse_section_explicit_select_with_span(void)
{
char* resreq = "select[type==linux] span[ptile=1]";
char* errmsg = NULL;
int rc = 0;
xce_resreq_data_t data = {0};
rc = parse_section(&data, (char*)resreq, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
printf("select: %s\n", data.selectString ? data.selectString : "(null)");
printf("span: %s\n", data.spanString ? data.spanString : "(null)");
/* select must contain the predicate, not bleed into the span keyword */
TEST_ASSERT_TRUE(data.selectString != NULL);
TEST_ASSERT_TRUE(strstr(data.selectString, "type==linux") != NULL);
/* span must be parsed separately and hold only its content */
TEST_ASSERT_TRUE(data.spanString != NULL);
TEST_ASSERT_TRUE(strstr(data.spanString, "ptile=1") != NULL);
/* unrelated clauses must be absent */
TEST_ASSERT_TRUE(data.rusageString == NULL);
TEST_ASSERT_TRUE(data.sameString == NULL);
TEST_ASSERT_TRUE(data.affinityString == NULL);
FREEUP(data.selectString);
FREEUP(data.spanString);
FREEUP(errmsg);
}
/* Test 2: all three major clauses — select, rusage, span — in one string. */
void test_parse_section_explicit_select_rusage_span(void)
{
char* resreq = "select[a&&b] rusage[mem=100] span[ptile=2]";
char* errmsg = NULL;
int rc = 0;
xce_resreq_data_t data = {0};
rc = parse_section(&data, (char*)resreq, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
printf("select: %s\n", data.selectString ? data.selectString : "(null)");
printf("rusage: %s\n", data.rusageString ? data.rusageString : "(null)");
printf("span: %s\n", data.spanString ? data.spanString : "(null)");
TEST_ASSERT_TRUE(data.selectString != NULL);
TEST_ASSERT_TRUE(strstr(data.selectString, "a&&b") != NULL);
TEST_ASSERT_TRUE(data.rusageString != NULL);
TEST_ASSERT_TRUE(strstr(data.rusageString, "mem=100") != NULL);
TEST_ASSERT_TRUE(data.spanString != NULL);
TEST_ASSERT_TRUE(strstr(data.spanString, "ptile=2") != NULL);
FREEUP(data.selectString);
FREEUP(data.rusageString);
FREEUP(data.spanString);
FREEUP(errmsg);
}
/*
* Test 3: the core regression guard — selectString must NOT contain the
* substring "span" when a span clause is present. With the old buggy
* grammar the span clause text would bleed into selectString.
*/
void test_parse_section_span_not_in_select(void)
{
char* resreq = "select[gpu] span[ptile=4]";
char* errmsg = NULL;
int rc = 0;
xce_resreq_data_t data = {0};
rc = parse_section(&data, (char*)resreq, &errmsg);
TEST_ASSERT_TRUE(rc >= 0);
printf("select: %s\n", data.selectString ? data.selectString : "(null)");
printf("span: %s\n", data.spanString ? data.spanString : "(null)");
/* selectString must not contain the span keyword (bleed check) */
TEST_ASSERT_TRUE(data.selectString != NULL);
TEST_ASSERT_TRUE(strstr(data.selectString, "span") == NULL);
/* span content must be correctly isolated */
TEST_ASSERT_TRUE(data.spanString != NULL);
TEST_ASSERT_TRUE(strstr(data.spanString, "ptile=4") != NULL);
FREEUP(data.selectString);
FREEUP(data.spanString);
FREEUP(errmsg);
}
void test_resreq(void)
{
printf("\n>> resreq test\n\n");
@ -354,6 +488,7 @@ void test_resreq(void)
RUN_TEST(test_parse_section1);
RUN_TEST(test_parse_section2);
RUN_TEST(test_parse_section_fail);
RUN_TEST(test_parse_section3);
RUN_TEST(test_parse_rusage1);
RUN_TEST(test_parse_rusage2);
RUN_TEST(test_resreq_primary);
@ -361,4 +496,7 @@ void test_resreq(void)
RUN_TEST(test_resreq1);
RUN_TEST(test_resreq2);
RUN_TEST(test_resreq3);
RUN_TEST(test_parse_section_explicit_select_with_span);
RUN_TEST(test_parse_section_explicit_select_rusage_span);
RUN_TEST(test_parse_section_span_not_in_select);
}

File diff suppressed because it is too large Load diff

View file

@ -1,3 +1,4 @@
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -6,6 +7,147 @@
#include "utils.h"
#include "xstring.h"
/* ------------------------------------------------------------------ */
/* Concurrent-test helpers */
/* ------------------------------------------------------------------ */
#define NTHREADS (8)
/* Thread argument for the same-key test */
typedef struct {
const char* key;
char* result;
} same_key_arg_t;
static void* thread_make_same_key(void* arg)
{
same_key_arg_t* a = (same_key_arg_t*)arg;
a->result = makeString((char*)a->key);
return a->result;
}
/* Thread argument for the distinct-key test */
typedef struct {
char key[16];
char* result;
} distinct_key_arg_t;
static void* thread_make_distinct_key(void* arg)
{
distinct_key_arg_t* a = (distinct_key_arg_t*)arg;
a->result = makeString(a->key);
return a->result;
}
/* ------------------------------------------------------------------ */
/* Concurrent tests */
/* ------------------------------------------------------------------ */
/*
* test_xstring_concurrent_same_key
*
* Spawn NTHREADS threads, each racing to intern the same key.
* With the TOCTOU fix the lock is held from key_find through key_add,
* so every thread must receive the identical interned pointer.
* Without the fix, two threads could both miss key_find, both call
* makeInstance, and one would leak — and the pointers would differ.
*/
static void test_xstring_concurrent_same_key(void)
{
pthread_t tids[NTHREADS];
same_key_arg_t args[NTHREADS];
int rc;
int i;
rc = xstring_table_init(139);
TEST_ASSERT_TRUE(rc >= 0);
for (i = 0; i < NTHREADS; i++) {
args[i].key = "concurrent_key";
args[i].result = NULL;
rc = pthread_create(&tids[i], NULL, thread_make_same_key, &args[i]);
TEST_ASSERT_TRUE(rc == 0);
}
for (i = 0; i < NTHREADS; i++) {
void* retval = NULL;
rc = pthread_join(tids[i], &retval);
TEST_ASSERT_TRUE(rc == 0);
}
/* All threads must have received a non-NULL pointer */
for (i = 0; i < NTHREADS; i++) {
TEST_ASSERT_TRUE(args[i].result != NULL);
}
/* All pointers must be the same interned object */
for (i = 1; i < NTHREADS; i++) {
TEST_ASSERT_EQUAL_PTR(args[0].result, args[i].result);
}
/*
* makeString increments the refcount on a cache hit. Each thread
* got one reference; drop NTHREADS - 1 extras so xstring_table_fini
* can release the final one cleanly.
*/
for (i = 1; i < NTHREADS; i++) {
dereference((var)args[i].result);
}
xstring_table_fini();
}
/*
* test_xstring_concurrent_distinct_keys
*
* Spawn NTHREADS threads, each interning a unique key.
* Every returned pointer must be non-NULL and distinct from all others,
* confirming that concurrent inserts of different keys do not collide.
*/
static void test_xstring_concurrent_distinct_keys(void)
{
pthread_t tids[NTHREADS];
distinct_key_arg_t args[NTHREADS];
int rc;
int i;
int j;
rc = xstring_table_init(139);
TEST_ASSERT_TRUE(rc >= 0);
for (i = 0; i < NTHREADS; i++) {
snprintf(args[i].key, sizeof(args[i].key), "key_%d", i);
args[i].result = NULL;
rc = pthread_create(&tids[i], NULL, thread_make_distinct_key, &args[i]);
TEST_ASSERT_TRUE(rc == 0);
}
for (i = 0; i < NTHREADS; i++) {
void* retval = NULL;
rc = pthread_join(tids[i], &retval);
TEST_ASSERT_TRUE(rc == 0);
}
/* Every pointer must be non-NULL */
for (i = 0; i < NTHREADS; i++) {
TEST_ASSERT_TRUE(args[i].result != NULL);
}
/* Every pointer must be distinct from all others */
for (i = 0; i < NTHREADS; i++) {
for (j = i + 1; j < NTHREADS; j++) {
TEST_ASSERT_NOT_EQUAL(args[i].result, args[j].result);
}
}
/* Release each thread's reference */
for (i = 0; i < NTHREADS; i++) {
dereference((var)args[i].result);
}
xstring_table_fini();
}
static void test_xstring_create(void)
{
char* s = "Hello, world";
@ -76,6 +218,8 @@ void test_xstrings(void)
RUN_TEST(test_xstring_create);
RUN_TEST(test_xstring_deref);
RUN_TEST(test_xstring_share);
RUN_TEST(test_xstring_concurrent_same_key);
RUN_TEST(test_xstring_concurrent_distinct_keys);
return;
}

View file

@ -106,7 +106,7 @@ primary_resreq <- < (! ('{' / '}' ) .)* > {
xce_primary_resreq_t *primary = (xce_primary_resreq_t *)calloc(1, sizeof(*primary));
if (primary != NULL) {
auxil->primary = primary;
parse_resreq_primary(primary, (char *)$0, &(auxil->errmsg));
parse_resreq_primary(primary, (char *)$0, auxil->resmgr, &(auxil->errmsg));
}
}

View file

@ -7,6 +7,7 @@
#define UNIT_TEST 1 /* expose sel_optimize_test_reset in resreq.h */
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
@ -22,7 +23,9 @@
/* ── optimize level ───────────────────────────────────────────────────── */
static _Atomic int g_sel_optimize_level = -1; /* -1 = unset */
static _Atomic int g_sel_optimize_level = -1; /* -1 = unset */
static pthread_t sel_alloc_thread_id;
static _Atomic int sel_alloc_thread_registered = 0; /* 1 after sel_register_alloc_thread() */
void sel_optimize_set(int level)
{
@ -155,9 +158,11 @@ void sel_node_table_fini(void)
}
/* Free each interned node before destroying the kvtab.
* datakiller=NULL means the kvtab does NOT free node values on destroy. */
int count = sel_node_table.nodes.count;
for (int i = 0; i < count; i++) {
* datakiller=NULL means the kvtab does NOT free node values on destroy.
* Iterate caps (not count): zarray_del creates NULL gaps, so live nodes
* may reside in slots beyond the count boundary. */
int caps = sel_node_table.nodes.caps;
for (int i = 0; i < caps; i++) {
xce_select_node_t* node
= (xce_select_node_t*)zarray_get(&sel_node_table.nodes, i);
if (node == NULL) {
@ -182,6 +187,14 @@ void sel_optimize_test_reset(void)
{
sel_node_table_fini();
atomic_store_explicit(&g_sel_optimize_level, -1, memory_order_seq_cst);
atomic_store_explicit(&sel_alloc_thread_registered, 0, memory_order_seq_cst);
/* sel_alloc_thread_id left as-is; only checked when registered flag is set */
}
void sel_register_alloc_thread(void)
{
sel_alloc_thread_id = pthread_self();
atomic_store_explicit(&sel_alloc_thread_registered, 1, memory_order_release);
}
/* ── node construction ───────────────────────────────────────────────── */
@ -439,38 +452,363 @@ void kill_select(xce_select_t* sel)
free(sel);
}
/* ── evaluation (stubs — return -1 / error) ──────────────────────────── */
/* ── eval context lifecycle ───────────────────────────────────────────── */
int eval_select(const xce_select_node_t* node, select_eval_ctx_t* ctx)
select_eval_ctx_t* select_eval_ctx_init(uint32_t max_node_id)
{
(void)node; (void)ctx;
return -1;
}
int eval_select_simple(const xce_select_node_t* root,
const host_eval_ctx_t* host)
{
(void)root; (void)host;
return -1;
}
/* ── eval context lifecycle (stubs) ──────────────────────────────────── */
void select_eval_ctx_init(select_eval_ctx_t* ctx,
const host_eval_ctx_t* host)
{
(void)ctx; (void)host;
select_eval_ctx_t* ctx = calloc(1, sizeof(*ctx));
if (!ctx) { ERROR("OOM in select_eval_ctx_init"); abort(); }
ctx->node_count = max_node_id;
ctx->bool_bits_cap = max_node_id;
ctx->num_vals_cap = max_node_id;
ctx->str_vals_cap = max_node_id;
if (max_node_id > 0) {
size_t bb_bytes = ((size_t)max_node_id * 2 + 7) / 8;
ctx->bool_bits = calloc(bb_bytes, 1);
ctx->num_vals = calloc(max_node_id, sizeof(double));
ctx->str_vals = calloc(max_node_id, sizeof(const char*));
if (!ctx->bool_bits || !ctx->num_vals || !ctx->str_vals) {
free(ctx->bool_bits);
free(ctx->num_vals);
free(ctx->str_vals);
free(ctx);
ERROR("OOM partial alloc in select_eval_ctx_init"); abort();
}
}
return ctx;
}
void select_eval_ctx_reset(select_eval_ctx_t* ctx,
const host_eval_ctx_t* host)
{
(void)ctx; (void)host;
ctx->host = host;
if (ctx->bool_bits && ctx->bool_bits_cap > 0) {
size_t bytes = ((size_t)ctx->bool_bits_cap * 2 + 7) / 8;
memset(ctx->bool_bits, 0, bytes);
}
}
void select_eval_ctx_fini(select_eval_ctx_t* ctx)
{
(void)ctx;
if (!ctx) return;
free(ctx->bool_bits);
free(ctx->num_vals);
free(ctx->str_vals);
free(ctx);
}
uint32_t sel_get_node_count(void)
{
return atomic_load_explicit(&sel_node_next_id, memory_order_acquire);
}
/* ── 2-bit bitmap helpers (EVAL_NONE=0 so memset(0) clears correctly) ── */
#define EVAL_NONE 0
#define EVAL_TRUE 1
#define EVAL_FALSE 2
#define EVAL_ERROR 3
_Static_assert(EVAL_NONE == 0,
"bool_bits reset via memset(0) requires EVAL_NONE==0");
static inline int eval_get_bool(const select_eval_ctx_t* ctx, uint32_t id)
{
uint32_t bit_off = id * 2;
return (ctx->bool_bits[bit_off / 8] >> (bit_off % 8)) & 0x3;
}
static inline void eval_set_bool(select_eval_ctx_t* ctx, uint32_t id, int val)
{
uint32_t bit_off = id * 2;
uint32_t byte = bit_off / 8;
uint32_t shift = bit_off % 8;
ctx->bool_bits[byte] =
(ctx->bool_bits[byte] & ~((uint8_t)(0x3 << shift))) |
(uint8_t)((val & 0x3) << shift);
}
/* ── comparison helpers ───────────────────────────────────────────────── */
static int is_str_cmp(const xce_select_node_t* node)
{
const xce_select_node_t* l = node->u.binop.left;
const xce_select_node_t* r = node->u.binop.right;
if (l->type == SEL_STR || r->type == SEL_STR) return 1;
if (l->type == SEL_RESNAME && l->u.resname.res_type == RES_TYPE_STRING)
return 1;
if (r->type == SEL_RESNAME && r->u.resname.res_type == RES_TYPE_STRING)
return 1;
return 0;
}
static double simple_num(const xce_select_node_t* n, const host_eval_ctx_t* h)
{
if (n->type == SEL_NUM) return n->u.numval;
int idx = n->u.resname.res_idx;
if (h->numvals == NULL || idx < 0 || idx >= h->num_res) return 0.0;
return h->numvals[idx];
}
static const char* simple_str(const xce_select_node_t* n,
const host_eval_ctx_t* h)
{
if (n->type == SEL_STR) return n->u.strval;
int idx = n->u.resname.res_idx;
if (h->strvals == NULL || idx < 0 || idx >= h->num_res) return NULL;
return h->strvals[idx];
}
/* Safe cache reads: use cached value when in-bounds, fall back to host data
* when node_id >= cap (e.g., ctx initialised before this node was interned). */
static double ctx_num(const select_eval_ctx_t* ctx, const xce_select_node_t* n)
{
return (n->node_id < ctx->num_vals_cap) ? ctx->num_vals[n->node_id]
: simple_num(n, ctx->host);
}
static const char* ctx_str(const select_eval_ctx_t* ctx,
const xce_select_node_t* n)
{
return (n->node_id < ctx->str_vals_cap) ? ctx->str_vals[n->node_id]
: simple_str(n, ctx->host);
}
/* ── eval_select_simple: tree-walk, no memoization (level == 0) ─────── */
int eval_select_simple(const xce_select_node_t* root,
const host_eval_ctx_t* host)
{
if (root == NULL) return 1;
switch (root->type) {
case SEL_TRUE: return 1;
case SEL_FALSE: return 0;
case SEL_ANY: return 1;
case SEL_RESNAME:
switch (root->u.resname.res_type) {
case RES_TYPE_BOOL:
case RES_TYPE_NUMERIC: {
int idx = root->u.resname.res_idx;
if (host->numvals == NULL || idx < 0 || idx >= host->num_res) return 0;
return host->numvals[idx] != 0.0 ? 1 : 0;
}
case RES_TYPE_STRING:
return -1;
}
return -1;
case SEL_NUM:
return root->u.numval != 0.0 ? 1 : 0;
case SEL_STR:
return -1;
case SEL_NOT: {
int v = eval_select_simple(root->u.operand, host);
return (v < 0) ? -1 : !v;
}
case SEL_AND: {
int l = eval_select_simple(root->u.binop.left, host);
if (l <= 0) return l;
return eval_select_simple(root->u.binop.right, host);
}
case SEL_OR: {
int l = eval_select_simple(root->u.binop.left, host);
if (l > 0) return 1;
if (l < 0) return l;
return eval_select_simple(root->u.binop.right, host);
}
case SEL_GT:
return simple_num(root->u.binop.left, host) >
simple_num(root->u.binop.right, host) ? 1 : 0;
case SEL_LT:
return simple_num(root->u.binop.left, host) <
simple_num(root->u.binop.right, host) ? 1 : 0;
case SEL_GE:
return simple_num(root->u.binop.left, host) >=
simple_num(root->u.binop.right, host) ? 1 : 0;
case SEL_LE:
return simple_num(root->u.binop.left, host) <=
simple_num(root->u.binop.right, host) ? 1 : 0;
case SEL_EQ:
if (is_str_cmp(root))
return xce_str_match(simple_str(root->u.binop.left, host),
simple_str(root->u.binop.right, host));
return simple_num(root->u.binop.left, host) ==
simple_num(root->u.binop.right, host) ? 1 : 0;
case SEL_NE:
if (is_str_cmp(root))
return !xce_str_match(simple_str(root->u.binop.left, host),
simple_str(root->u.binop.right, host))
? 1 : 0;
return simple_num(root->u.binop.left, host) !=
simple_num(root->u.binop.right, host) ? 1 : 0;
}
ERROR("eval_select_simple: unknown node type %d", (int)root->type);
abort();
}
/* ── memoized eval (level > 0) ───────────────────────────────────────── */
/* Forward declaration — eval_select and eval_select_inner are mutually
* recursive: eval_select checks the bitmap cache then calls eval_select_inner;
* eval_select_inner calls eval_select on children so leaf-cache writes
* (num_vals, str_vals) happen before the parent reads them. */
static int eval_select_inner(const xce_select_node_t* node,
select_eval_ctx_t* ctx);
int eval_select(const xce_select_node_t* node, select_eval_ctx_t* ctx)
{
if (node == NULL) return 1;
if (sel_get_optimize_level() <= 0) {
ERROR("eval_select called with optimize_level <= 0"); abort();
}
if (atomic_load_explicit(&sel_alloc_thread_registered, memory_order_acquire)
&& !pthread_equal(pthread_self(), sel_alloc_thread_id)) {
ERROR("eval_select called from non-allocator thread");
return -1;
}
/* bitmap cache hit */
if (node->node_id < ctx->bool_bits_cap) {
int cached = eval_get_bool(ctx, node->node_id);
if (cached == EVAL_TRUE) return 1;
if (cached == EVAL_FALSE) return 0;
if (cached == EVAL_ERROR) return -1;
}
int result = eval_select_inner(node, ctx);
if (node->node_id < ctx->bool_bits_cap) {
int bits = (result > 0) ? EVAL_TRUE
: (result == 0) ? EVAL_FALSE
: EVAL_ERROR;
eval_set_bool(ctx, node->node_id, bits);
}
return result;
}
static int eval_select_inner(const xce_select_node_t* node,
select_eval_ctx_t* ctx)
{
if (node == NULL) return 1;
const host_eval_ctx_t* host = ctx->host;
switch (node->type) {
case SEL_TRUE: return 1;
case SEL_FALSE: return 0;
case SEL_ANY: return 1;
case SEL_RESNAME:
switch (node->u.resname.res_type) {
case RES_TYPE_BOOL:
case RES_TYPE_NUMERIC: {
int idx = node->u.resname.res_idx;
double v = (host->numvals != NULL && idx >= 0 && idx < host->num_res)
? host->numvals[idx] : 0.0;
if (node->node_id < ctx->num_vals_cap)
ctx->num_vals[node->node_id] = v;
return v != 0.0 ? 1 : 0;
}
case RES_TYPE_STRING: {
int idx = node->u.resname.res_idx;
const char* sv = (host->strvals != NULL && idx >= 0 && idx < host->num_res)
? host->strvals[idx] : NULL;
if (node->node_id < ctx->str_vals_cap)
ctx->str_vals[node->node_id] = sv;
return -1;
}
}
return -1;
case SEL_NUM: {
if (node->node_id < ctx->num_vals_cap)
ctx->num_vals[node->node_id] = node->u.numval;
return node->u.numval != 0.0 ? 1 : 0;
}
case SEL_STR: {
if (node->node_id < ctx->str_vals_cap)
ctx->str_vals[node->node_id] = node->u.strval;
return -1;
}
case SEL_NOT: {
int v = eval_select(node->u.operand, ctx);
return (v < 0) ? -1 : !v;
}
case SEL_AND: {
int l = eval_select(node->u.binop.left, ctx);
if (l <= 0) return l;
return eval_select(node->u.binop.right, ctx);
}
case SEL_OR: {
int l = eval_select(node->u.binop.left, ctx);
if (l > 0) return 1;
if (l < 0) return l;
return eval_select(node->u.binop.right, ctx);
}
case SEL_GT: {
int lr = eval_select(node->u.binop.left, ctx);
int rr = eval_select(node->u.binop.right, ctx);
if (lr < 0) return lr;
if (rr < 0) return rr;
return ctx_num(ctx, node->u.binop.left) >
ctx_num(ctx, node->u.binop.right) ? 1 : 0;
}
case SEL_LT: {
int lr = eval_select(node->u.binop.left, ctx);
int rr = eval_select(node->u.binop.right, ctx);
if (lr < 0) return lr;
if (rr < 0) return rr;
return ctx_num(ctx, node->u.binop.left) <
ctx_num(ctx, node->u.binop.right) ? 1 : 0;
}
case SEL_GE: {
int lr = eval_select(node->u.binop.left, ctx);
int rr = eval_select(node->u.binop.right, ctx);
if (lr < 0) return lr;
if (rr < 0) return rr;
return ctx_num(ctx, node->u.binop.left) >=
ctx_num(ctx, node->u.binop.right) ? 1 : 0;
}
case SEL_LE: {
int lr = eval_select(node->u.binop.left, ctx);
int rr = eval_select(node->u.binop.right, ctx);
if (lr < 0) return lr;
if (rr < 0) return rr;
return ctx_num(ctx, node->u.binop.left) <=
ctx_num(ctx, node->u.binop.right) ? 1 : 0;
}
case SEL_EQ:
if (is_str_cmp(node)) {
int lr = eval_select(node->u.binop.left, ctx);
int rr = eval_select(node->u.binop.right, ctx);
if (lr < 0) return lr;
if (rr < 0) return rr;
return xce_str_match(ctx_str(ctx, node->u.binop.left),
ctx_str(ctx, node->u.binop.right));
}
{
int lr = eval_select(node->u.binop.left, ctx);
int rr = eval_select(node->u.binop.right, ctx);
if (lr < 0) return lr;
if (rr < 0) return rr;
return ctx_num(ctx, node->u.binop.left) ==
ctx_num(ctx, node->u.binop.right) ? 1 : 0;
}
case SEL_NE:
if (is_str_cmp(node)) {
int lr = eval_select(node->u.binop.left, ctx);
int rr = eval_select(node->u.binop.right, ctx);
if (lr < 0) return lr;
if (rr < 0) return rr;
return !xce_str_match(ctx_str(ctx, node->u.binop.left),
ctx_str(ctx, node->u.binop.right)) ? 1 : 0;
}
{
int lr = eval_select(node->u.binop.left, ctx);
int rr = eval_select(node->u.binop.right, ctx);
if (lr < 0) return lr;
if (rr < 0) return rr;
return ctx_num(ctx, node->u.binop.left) !=
ctx_num(ctx, node->u.binop.right) ? 1 : 0;
}
}
ERROR("eval_select_inner: unknown node type %d", (int)node->type);
abort();
}
/* ── GC ──────────────────────────────────────────────────────────────── */
@ -517,37 +855,70 @@ static sel_node_key_t sel_node_to_key(const xce_select_node_t* node)
return sel_make_key(node->type, op1, op2);
}
/* Mark-and-sweep GC. In Phase 4a the mark pass is not yet implemented
* (live_jobs traversal comes in Phase 4f). All nodes are treated as
* unreachable when live_jobs == NULL || njobs == 0.
/* Clear all gc_mark bits in the intern table.
* Called by the scheduler before the mark phase. */
void gc_clear_marks(void)
{
if (sel_node_table.table == NULL) return;
int caps = sel_node_table.nodes.caps;
for (int i = 0; i < caps; i++) {
xce_select_node_t* node =
(xce_select_node_t*)zarray_get(&sel_node_table.nodes, i);
if (node == NULL) continue;
node->gc_mark = 0;
}
}
/* Mark all nodes reachable from root (recursive DFS).
* Early-returns on NULL root or already-marked node to handle DAG sharing.
* Recursion depth is bounded by SELECT_MAX_DEPTH (64).
* Called by the scheduler to mark each live job's select tree. */
void gc_mark_select_tree(xce_select_node_t* node)
{
if (node == NULL || node->gc_mark) return;
node->gc_mark = 1;
switch (node->type) {
case SEL_RESNAME: case SEL_NUM: case SEL_STR:
case SEL_ANY: case SEL_TRUE: case SEL_FALSE:
break;
case SEL_NOT:
gc_mark_select_tree(node->u.operand);
break;
default:
gc_mark_select_tree(node->u.binop.left);
gc_mark_select_tree(node->u.binop.right);
break;
}
}
/* Sweep phase of mark-and-sweep GC. Runs between scheduling sessions
* (idle-phase invariant): no concurrent makeSelectNode or eval_select calls
* are in progress. Does NOT acquire sel_node_table.lock — no contention at
* idle phase.
*
* Sweep: remove every node from the intern table and free it. */
* The mark phase (gc_clear_marks + gc_mark_select_tree per live node) must
* be called by the scheduler BEFORE this function. The live_jobs and njobs
* parameters are retained for API compatibility but are unused — the caller
* owns the mark phase. */
void gc_sweep_select_nodes(const xce_sched_job_t** live_jobs, size_t njobs)
{
(void)live_jobs;
(void)njobs;
if (sel_node_table.table == NULL) {
return;
}
/* Phase 4a: no mark pass — treat all nodes as unreachable when
* njobs == 0. Once Phase 4f adds live-job marking this block will
* skip nodes whose gc_mark bit is set. */
if (live_jobs != NULL && njobs > 0) {
/* Mark pass not yet implemented — skip sweep when live jobs present
* to avoid freeing potentially-reachable nodes. */
return;
}
int caps = sel_node_table.nodes.caps;
mylock_lock(&sel_node_table.lock);
/* Iterate the zarray of interned nodes and remove each from the kv
* table, then free the node allocation. */
int count = sel_node_table.nodes.count;
for (int i = 0; i < count; i++) {
xce_select_node_t* node
= (xce_select_node_t*)zarray_get(&sel_node_table.nodes, i);
if (node == NULL) {
continue;
}
/* Sweep: iterate all allocated slots (caps), skipping NULL gaps left
* by prior zarray_del calls. For each unmarked node: remove its kv
* entry, release xstring ref if SEL_STR, null the zarray slot via
* zarray_del, then free the allocation. */
for (int i = 0; i < caps; i++) {
xce_select_node_t* node =
(xce_select_node_t*)zarray_get(&sel_node_table.nodes, i);
if (node == NULL || node->gc_mark) continue;
sel_node_key_t key = sel_node_to_key(node);
key_del(sel_node_table.table, &key);
@ -555,16 +926,25 @@ void gc_sweep_select_nodes(const xce_sched_job_t** live_jobs, size_t njobs)
if (node->type == SEL_STR && node->u.strval != NULL) {
dereference(node->u.strval);
}
zarray_del(&sel_node_table.nodes, i); /* NULL slot before free */
free(node);
}
/* Clear the zarray tracking list. */
zarray_clear(&sel_node_table.nodes, 0);
mylock_unlock(&sel_node_table.lock);
}
void recompact_node_ids(void) {}
/* Reassign node_id values to dense 0..N-1 and reset sel_node_next_id.
* Called after gc_sweep_select_nodes to keep eval-context arrays bounded. */
void recompact_node_ids(void)
{
uint32_t next_id = 0;
int caps = sel_node_table.nodes.caps;
for (int i = 0; i < caps; i++) {
xce_select_node_t* node =
(xce_select_node_t*)zarray_get(&sel_node_table.nodes, i);
if (node == NULL) continue;
node->node_id = next_id++;
}
atomic_store_explicit(&sel_node_next_id, next_id, memory_order_release);
}
/* ── Phase 4d: const-fold transform ──────────────────────────────────────── */

View file

@ -35,7 +35,9 @@
}
xce_resource_t* r = NULL;
lock(auxil->resmgr, RTTI_LOCK_READ);
int rc = key_find(auxil->resmgr, (void*)name, (void**)&r);
unlock(auxil->resmgr);
if (rc <= 0 || r == NULL) {
if (strcmp(name, "true") == 0 || strcmp(name, "false") == 0) {
snprintf(auxil->errmsg, sizeof(auxil->errmsg),
@ -182,6 +184,19 @@ comparison <- l:resname_node _ op:cmp_op _ r:value_node {
}
}
}
} else if (l->u.resname.res_type == RES_TYPE_NUMERIC &&
r->type == SEL_STR) {
/* Numeric resource vs string literal: reject — at eval time
* is_str_cmp() would return true and simple_str() would read
* h->strvals[numeric_idx] which is never populated for numeric
* resources (NULL or garbage), causing a strcmp NULL-deref. */
snprintf(auxil->errmsg, sizeof(auxil->errmsg),
"type mismatch: numeric resource cannot be compared "
"with a string literal");
auxil->has_error = 1;
SEL_DISCARD(l);
SEL_DISCARD(r);
$$ = NULL;
} else {
$$ = (xce_select_node_t*)makeSelectNode(op_type,
(uintptr_t)l,

View file

@ -145,30 +145,30 @@ char* makeString(char* s)
}
mylock_lock(&(xstring_table.lock));
rc = key_find(xstring_table.xstr_table, (void*)s, (void**)&string);
if (rc > 0) {
/* add reference */
reference(string);
}
mylock_unlock(&(xstring_table.lock));
if (rc > 0) {
mylock_unlock(&(xstring_table.lock));
return (char*)string;
}
/* Cache miss: allocate and insert under the lock so no concurrent caller
* can insert the same key between our find and our add. key_add cannot
* return -EEXIST here because we hold the lock continuously. */
string = makeInstance(&xstring_type, s);
if (string == NULL) {
mylock_unlock(&(xstring_table.lock));
return NULL;
}
mylock_lock(&(xstring_table.lock));
rc = key_add(xstring_table.xstr_table, (void*)s, string);
mylock_unlock(&(xstring_table.lock));
if (rc < 0) {
destroy_(string, (obj_destroy_fn_t)xstring_destroy);
string = NULL;
errno = -rc;
errno = -rc;
return NULL;
}
return (char*)string;

View file

@ -77,7 +77,7 @@ typedef struct {
char* resreq;
} xce_sched_queue_t;
typedef struct {
typedef struct xce_sched_job_ {
uint64_t jobid;
time_t sub_time;
time_t queue_time;
@ -96,11 +96,12 @@ typedef struct {
} xce_sched_job_t;
typedef struct {
char* name;
int idx;
uint32_t status;
zarray_t resvals;
zarray_t resInUse;
char* name;
int idx;
uint32_t status;
zarray_t resvals;
zarray_t resInUse;
host_eval_ctx_t* host_eval_ctx; /* cached per-session; NULL until first use */
} xce_sched_worker_t;
typedef struct {

View file

@ -4,6 +4,7 @@
#include <pthread.h>
#include <rtti.h>
#include <stdatomic.h>
#include <stdint.h>
#include <time.h>
@ -83,6 +84,10 @@ typedef struct {
cbuf_t wbuf; /* action to trigger scheduling */
cbuf_t jbuf; /* job picked for allocation */
cbuf_t decisions; /* scheduler decisions */
_Atomic int session_active; /* 1 during scheduling session, 0 between */
_Atomic int eval_in_progress; /* 1 while eval_select loop runs */
_Atomic int staging_merge_complete; /* 1 after staging merge completes */
select_eval_ctx_t* sel_eval_ctx; /* session-level memoisation context; NULL between sessions */
} xce_scheduler_t;
extern int sched_init(xce_scheduler_t*, xce_client_config_t*,
@ -108,6 +113,7 @@ extern int sched_queue_is_ready(xce_sched_queue_t*);
extern int sched_commit_decision(xce_scheduler_t*, xce_decision_t*);
extern int sched_publish_pend_reason(xce_scheduler_t*, xce_preason_set_t*);
extern int sched_publish_dispatch(xce_scheduler_t*, xce_job_dispatch_t*);
extern void sched_worker_release_host_ctx(xce_sched_worker_t*);
extern int sched_meet_host_selection(xce_scheduler_t*, xce_sched_worker_t*,
xce_select_t*);
extern int sched_check_shared_satisfy(xce_scheduler_t*, xvec_t*, xvec_idval_t*,

View file

@ -10,6 +10,7 @@
#include "scheduler.h"
#include "resreq.h"
#include "utils.h"
extern int scheduler_req_handler(xce_request_t*, void*);

View file

@ -110,14 +110,16 @@ static int my_alloc(xce_sched_plugin_t* plugin, xce_sched_ctx_t* ctx)
xce_sched_obj_t* schedobj = ctx->schedobj;
int rc;
xlist_t workers;
xce_pendreason_t* reason = NULL;
xce_pendreason_t* reason = ctx->reason;
xvec_t* uncomit = NULL;
xce_sched_worker_t* w = NULL;
reason = make_pendreason(0);
ctx->reason = reason;
if (reason == NULL) {
return -ENOMEM;
reason = make_pendreason(0);
ctx->reason = reason;
if (reason == NULL) {
return -ENOMEM;
}
}
rc = sched_ctx_get_uncomit_shared(&uncomit, ctx);

View file

@ -5,6 +5,7 @@
#include "kvmgr.h"
#include "plugin.h"
#include "preason.h"
#include "resreq.h"
#include "sched/types.h"
#include "scheduler.h"
#include "utils.h"
@ -124,11 +125,93 @@ static int unsatisfy_(int nres, int* idxs, float* total, float* inuse,
return count;
}
static host_eval_ctx_t*
build_host_eval_ctx(const xce_sched_worker_t* w, int num_res)
{
host_eval_ctx_t* ctx = calloc(1, sizeof(*ctx));
if (ctx == NULL) return NULL;
if (num_res <= 0) {
/* valid empty context — numvals/strvals remain NULL, bounds checks protect access */
ctx->num_res = 0;
return ctx;
}
double* numvals = calloc(num_res, sizeof(double));
const char** strvals = calloc(num_res, sizeof(char*));
if (numvals == NULL || strvals == NULL) {
free(numvals);
free(strvals);
free(ctx);
return NULL;
}
for (int i = 0; i < num_res; i++) {
xce_val_t* v = (xce_val_t*)zarray_get((zarray_t*)&w->resvals, i);
if (v == NULL) continue;
switch (v->which) {
case Value_u_fvalue:
numvals[i] = (double)v->u.fvalue;
break;
case Value_u_bvalue:
numvals[i] = v->u.bvalue ? 1.0 : 0.0;
break;
case Value_u_svalue:
strvals[i] = v->u.svalue;
break;
}
}
ctx->num_res = num_res;
ctx->numvals = numvals;
ctx->strvals = strvals;
return ctx;
}
static void
free_host_eval_ctx(host_eval_ctx_t* ctx)
{
if (ctx == NULL) return;
free((void*)ctx->numvals);
free((void*)ctx->strvals);
free(ctx);
}
void sched_worker_release_host_ctx(xce_sched_worker_t* w)
{
if (w == NULL) return;
free_host_eval_ctx(w->host_eval_ctx);
w->host_eval_ctx = NULL;
}
int sched_meet_host_selection(xce_scheduler_t* sched, xce_sched_worker_t* w,
xce_select_t* s)
{
/* FIXME: add real work later */
return 1;
if (s == NULL || s->root == NULL) return 1;
/* Build host context lazily — reused for all jobs against this host. */
if (w->host_eval_ctx == NULL) {
int num_res = (int)w->resvals.caps;
w->host_eval_ctx = build_host_eval_ctx(w, num_res);
if (w->host_eval_ctx == NULL) {
/* OOM: conservatively pass host */
return 1;
}
}
/* Fast path: use session-level memoisation context when available. */
if (sched->sel_eval_ctx != NULL) {
/* Reset memoisation bitmap only when moving to a different host. */
if (sched->sel_eval_ctx->host != w->host_eval_ctx)
select_eval_ctx_reset(sched->sel_eval_ctx, w->host_eval_ctx);
int rc = eval_select(s->root, sched->sel_eval_ctx);
return (rc > 0) ? 1 : 0;
}
/* Fallback (no session context — e.g. unit tests or OOM at session start). */
int rc = eval_select_simple(s->root, w->host_eval_ctx);
return (rc > 0) ? 1 : 0;
}
static int sched_check_shared_satisfy_(xce_scheduler_t* sched, xvec_t* total,
@ -367,7 +450,15 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
rc = sched_meet_host_selection(sched, w, primary->select);
if (rc == 0) {
add_job_preason(schedobj->reason->reasons, PREASON_MATCH_FAIL, w->name,
if (ctx->reason == NULL) {
ctx->reason = make_pendreason(0);
}
if (ctx->reason == NULL) {
return -errno;
}
add_job_preason(ctx->reason->reasons, PREASON_MATCH_FAIL, w->name,
NULL);
return 0;
@ -396,6 +487,7 @@ static int sched_try_alloc_on_host_(xce_sched_worker_t* w, va_list ap)
if (rc == 0) {
int i;
if (ctx)
for (i = 0; i < ctx->unmeet.count; i++) {
xce_resource_t* r = NULL;

View file

@ -168,7 +168,7 @@ xce_sched_job_t* make_sched_job(void* arg, xce_job_t* j)
return NULL;
}
rc = parse_resreq(job->resreq, j->submit->resreq, &errmsg);
rc = parse_resreq(job->resreq, j->submit->resreq, sched->resmgr, &errmsg);
if (rc < 0) {
ERROR("%s parse_resreq failed, %s", __func__, errmsg);
FREEUP(errmsg);
@ -234,6 +234,7 @@ void kill_sched_worker(void* arg)
if (arg == NULL) {
return;
}
sched_worker_release_host_ctx((xce_sched_worker_t*)arg);
}
void kill_sched_resval(xce_val_t* val)

View file

@ -14,11 +14,30 @@
#include "plugin.h"
#include "preason.h"
#include "resobj.h"
#include "resreq.h"
#include "scheduler.h"
#include "utils.h"
#include "xcealloc.h"
#include "xstring.h"
static int sched_session_active(xce_scheduler_t* sched)
{
return atomic_load_explicit(&sched->session_active,
memory_order_acquire);
}
static int sched_eval_in_progress(xce_scheduler_t* sched)
{
return atomic_load_explicit(&sched->eval_in_progress,
memory_order_acquire);
}
static int sched_staging_merge_complete(xce_scheduler_t* sched)
{
return atomic_load_explicit(&sched->staging_merge_complete,
memory_order_acquire);
}
static void* sched_job_picker(void* arg)
{
xce_sched_plugin_t* job_picker = NULL;
@ -201,6 +220,7 @@ static void* sched_allocator(void* arg)
{
xce_scheduler_t* sched = (xce_scheduler_t*)arg;
sel_register_alloc_thread();
xce_use(sched);
while (1) {
uint64_t jobid;
@ -235,8 +255,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;
@ -356,6 +380,8 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
memset(sched, 0x0, sizeof(*sched));
resreq_init(1); /* enable CSE intern table for scheduler process */
sched->nextid = 1;
sched->configver = 0;
@ -455,6 +481,8 @@ void sched_fini(xce_scheduler_t* sched)
destroy(sched->pgmgr);
destroy(sched->hgmgr);
resreq_fini();
mylock_fini(&(sched->lock));
return;
@ -667,6 +695,9 @@ static int sched_reset_worker_(xce_scheduler_t* sched, xce_sched_worker_t* w)
return 0;
}
/* Release the per-session host snapshot so the next session rebuilds it. */
sched_worker_release_host_ctx(w);
return zarray_iter(&(w->resInUse), (zarray_iter_fn_t)my_reset_resval_,
sched);
}
@ -739,6 +770,15 @@ static int sched_account_jobs(xce_scheduler_t* sched)
void sched_begin_session(xce_scheduler_t* sched)
{
atomic_store_explicit(&sched->session_active, 1, memory_order_release);
atomic_store_explicit(&sched->staging_merge_complete, 0, memory_order_release);
/* Allocate a fresh memoisation context sized to the current interned-node
* population. sched_reset_workers (called below) frees each worker's
* stale host_eval_ctx so they are rebuilt lazily during this session. */
select_eval_ctx_fini(sched->sel_eval_ctx);
sched->sel_eval_ctx = select_eval_ctx_init(sel_get_node_count());
plugin_begin_session(sched);
sched_reset_shared_resval(sched);
@ -746,14 +786,120 @@ void sched_begin_session(xce_scheduler_t* sched)
sched_account_jobs(sched);
/* Staging merge done — sched_end_session's GC guard reads this field. */
atomic_store_explicit(&sched->staging_merge_complete, 1, memory_order_release);
return;
}
void sched_end_session(xce_scheduler_t* sched)
{
atomic_store_explicit(&sched->session_active, 0, memory_order_release);
select_eval_ctx_fini(sched->sel_eval_ctx);
sched->sel_eval_ctx = NULL;
plugin_end_session(sched);
return;
/* Guard: GC requires the evaluation pipeline to be idle */
if (sched_eval_in_progress(sched) || !sched_staging_merge_complete(sched)) {
TRACE("%s: GC skipped — eval_in_progress=%d staging_merge_complete=%d",
__func__,
sched_eval_in_progress(sched),
sched_staging_merge_complete(sched));
return;
}
/* Collect live jobs into a flat array for the GC mark phase */
iterator_t it;
xce_sched_job_t* job = NULL;
int njobs = 0;
int cap = 64;
xce_sched_job_t** live_jobs = calloc(cap, sizeof(*live_jobs));
if (live_jobs != NULL) {
int rc = iter_make(&it, sched->jobs);
if (rc >= 0) {
for (;;) {
rc = iter_next(&it, (void**)&job);
if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
break;
}
if (njobs >= cap) {
int newcap = cap * 2;
xce_sched_job_t** tmp = realloc(live_jobs,
newcap * sizeof(*live_jobs));
if (tmp == NULL) {
free(live_jobs);
live_jobs = NULL;
break;
}
live_jobs = tmp;
cap = newcap;
}
live_jobs[njobs++] = job;
}
if (rc == RTTI_ITER_NO_DATA) {
rc = 0;
}
iter_destroy(&it);
}
}
if (live_jobs == NULL) {
/* OOM: cannot safely GC — skip to avoid sweeping nodes still in use */
atomic_store_explicit(&sched->staging_merge_complete, 1, memory_order_release);
return;
}
/* Mark phase */
gc_clear_marks();
for (int i = 0; i < njobs; i++) {
xce_resreq_t* rr = live_jobs[i]->resreq;
if (rr == NULL) {
continue;
}
if (rr->tag == XCE_RESREQ_PRIMARY) {
xce_primary_resreq_t* pr = rr->u.primary;
if (pr && pr->select) {
gc_mark_select_tree(pr->select->root);
}
} else if (rr->tag == XCE_RESREQ_COMPOUND) {
xce_compound_resreq_t* cp = rr->u.compound;
if (cp != NULL) {
for (int j = 0; j < cp->primaries.count; j++) {
xce_primary_resreq_t* pr = xlist_get(&cp->primaries, j);
if (pr && pr->select) {
gc_mark_select_tree(pr->select->root);
}
}
}
} else if (rr->tag == XCE_RESREQ_ALTER) {
xce_alter_resreq_t* alt = rr->u.alter;
if (alt != NULL) {
for (int j = 0; j < alt->compounds.count; j++) {
xce_compound_resreq_t* cp = xlist_get(&alt->compounds, j);
if (cp == NULL) {
continue;
}
for (int k = 0; k < cp->primaries.count; k++) {
xce_primary_resreq_t* pr = xlist_get(&cp->primaries, k);
if (pr && pr->select) {
gc_mark_select_tree(pr->select->root);
}
}
}
}
}
}
/* Fence: mark writes visible before sweep reads them */
atomic_thread_fence(memory_order_seq_cst);
gc_sweep_select_nodes(NULL, 0);
recompact_node_ids();
free(live_jobs);
atomic_store_explicit(&sched->staging_merge_complete, 1, memory_order_release);
}
int sched_add_job(xce_scheduler_t* sched, xce_sched_job_t* j)

View file

@ -2,9 +2,9 @@ scheduler_test_src = [
'test.c'
]
scheduler_test_deps = [libxce_deps, librtti_dep, unity_dep]
scheduler_test_deps = [libxce_deps, librtti_dep, unity_dep, libxce_dep]
incdir = include_directories('../include', '../../include', '.')
incdir = include_directories('../include', '../../include', '.', '../src', '../../lib')
scheduler_test = executable('sched_test',
scheduler_test_src,

View file

@ -1,8 +1,20 @@
#define UNIT_TEST 1
#include <argp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdatomic.h>
#include <unity/unity.h>
#include "xstring.h"
/* Pull resource.c into this TU so static helpers are visible for white-box testing.
* This also brings in scheduler.h (with the real xce_scheduler_t) and all deps. */
#include "../src/resource.c"
/* Stubs for symbols used by resource.c functions not under test */
xce_sched_worker_t* sched_get_worker(xce_scheduler_t* s, char* n) { (void)s; (void)n; return NULL; }
int plugin_prepare_allocation(void* s, xce_sched_ctx_t* c) { (void)s; (void)c; return 0; }
int sched_ctx_get_uncomit_host(xvec_t** v, xce_sched_ctx_t* c, int i) { (void)v; (void)c; (void)i; return 0; }
void setUp() { return; }
@ -65,10 +77,304 @@ static int check_test(struct arguments* args, char* x)
return 0;
}
static void
test_build_host_eval_ctx_float(void)
{
/* Worker with one float resource at res_idx=2 (value=4.0) */
xce_sched_worker_t w;
memset(&w, 0, sizeof(w));
zarray_init(&w.resvals, sizeof(xce_val_t*), NULL);
xce_val_t v;
memset(&v, 0, sizeof(v));
v.which = Value_u_fvalue;
v.u.fvalue = 4.0;
/* store xce_val_t* directly, matching production put_resvals behavior */
zarray_put(&w.resvals, 2, &v);
host_eval_ctx_t* ctx = build_host_eval_ctx(&w, 3);
TEST_ASSERT_NOT_NULL(ctx);
/* Unity is built without UNITY_INCLUDE_DOUBLE; compare via float cast */
TEST_ASSERT_EQUAL_FLOAT(4.0f, (float)ctx->numvals[2]);
TEST_ASSERT_NULL(ctx->strvals[2]);
TEST_ASSERT_EQUAL_INT(3, ctx->num_res);
free_host_eval_ctx(ctx);
zarray_fini(&w.resvals);
}
static void
test_scheduler_t_has_atomic_gc_fields(void)
{
xce_scheduler_t sched;
memset(&sched, 0, sizeof(sched));
TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.session_active));
TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.eval_in_progress));
TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.staging_merge_complete));
}
static void
test_sel_optimize_set_at_scheduler_startup(void)
{
/* sel_optimize_set(1) is called in sched_init.
* Since we can't call sched_init (it starts threads/network),
* verify the setter/getter contract directly. */
sel_optimize_set(1);
TEST_ASSERT_EQUAL_INT(1, sel_get_optimize_level());
/* reset for clean test isolation */
sel_optimize_test_reset();
}
static void
test_meet_host_selection_null_select_passes(void)
{
/* A NULL select pointer means no expression — always pass */
xce_scheduler_t sched;
memset(&sched, 0, sizeof(sched));
xce_sched_worker_t w;
memset(&w, 0, sizeof(w));
zarray_init(&w.resvals, 4, NULL);
int rc = sched_meet_host_selection(&sched, &w, NULL);
TEST_ASSERT_EQUAL_INT(1, rc);
zarray_fini(&w.resvals);
}
static void
test_meet_host_selection_always_true_passes(void)
{
xce_scheduler_t sched;
memset(&sched, 0, sizeof(sched));
sel_optimize_set(1);
sel_node_table_init();
xce_sched_worker_t w;
memset(&w, 0, sizeof(w));
zarray_init(&w.resvals, 4, NULL);
/* SEL_TRUE root — must pass any host */
xce_select_node_t* root = makeSelectNode(SEL_TRUE, 0, 0);
TEST_ASSERT_NOT_NULL(root);
xce_select_t sel = { .root = root };
int rc = sched_meet_host_selection(&sched, &w, &sel);
TEST_ASSERT_EQUAL_INT(1, rc);
zarray_fini(&w.resvals);
gc_clear_marks();
gc_sweep_select_nodes(NULL, 0);
sel_node_table_fini();
sel_optimize_test_reset();
}
static void
test_meet_host_selection_always_false_fails(void)
{
xce_scheduler_t sched;
memset(&sched, 0, sizeof(sched));
sel_optimize_set(1);
sel_node_table_init();
xce_sched_worker_t w;
memset(&w, 0, sizeof(w));
zarray_init(&w.resvals, 4, NULL);
/* SEL_FALSE root — must fail every host */
xce_select_node_t* root = makeSelectNode(SEL_FALSE, 0, 0);
TEST_ASSERT_NOT_NULL(root);
xce_select_t sel = { .root = root };
int rc = sched_meet_host_selection(&sched, &w, &sel);
TEST_ASSERT_EQUAL_INT(0, rc);
zarray_fini(&w.resvals);
gc_clear_marks();
gc_sweep_select_nodes(NULL, 0);
sel_node_table_fini();
sel_optimize_test_reset();
}
static void
test_gc_predicates_idle_invariant(void)
{
xce_scheduler_t sched;
memset(&sched, 0, sizeof(sched));
/* staging_merge_complete=1 means GC is safe to run */
atomic_store(&sched.staging_merge_complete, 1);
sel_optimize_set(1);
sel_node_table_init();
/* Verify the atomic fields are readable — testing the predicate contract */
TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.session_active));
TEST_ASSERT_EQUAL_INT(0, atomic_load(&sched.eval_in_progress));
TEST_ASSERT_EQUAL_INT(1, atomic_load(&sched.staging_merge_complete));
/* GC operations on an empty node table must not crash */
gc_clear_marks();
gc_sweep_select_nodes(NULL, 0);
sel_node_table_fini();
sel_optimize_test_reset();
}
static void
test_staging_merge_complete_gc_guard(void)
{
/* Regression: sched_begin_session must leave staging_merge_complete=1 so
* sched_end_session's guard (!staging_merge_complete) does not fire.
*
* sched_begin_session: sets staging_merge_complete=0, does all session setup,
* then sets staging_merge_complete=1 (staging done).
* sched_end_session: guard skips GC when staging_merge_complete==0.
*
* Without the trailing set-to-1 in sched_begin_session, staging_merge_complete
* stays 0 forever and GC never runs — select-node memory leaks indefinitely. */
xce_scheduler_t sched;
memset(&sched, 0, sizeof(sched));
/* Simulate the post-begin_session state the fix produces */
atomic_store(&sched.session_active, 0); /* session ended */
atomic_store(&sched.eval_in_progress, 0);
atomic_store(&sched.staging_merge_complete, 1); /* fix: begin_session sets this */
/* The GC guard in sched_end_session: must NOT fire */
int guard_fires = atomic_load(&sched.eval_in_progress)
|| !atomic_load(&sched.staging_merge_complete);
TEST_ASSERT_EQUAL_INT(0, guard_fires);
/* Confirm the unfixed state (staging=0) would block GC */
atomic_store(&sched.staging_merge_complete, 0);
int guard_fires_unfixed = atomic_load(&sched.eval_in_progress)
|| !atomic_load(&sched.staging_merge_complete);
TEST_ASSERT_EQUAL_INT(1, guard_fires_unfixed);
}
static void
test_meet_host_selection_numeric_match(void)
{
/* Job requires: res[0] >= 4.
* Worker with res[0]=8.0 must match (rc=1).
* Worker with res[0]=2.0 must not match (rc=0). */
sel_optimize_set(1);
sel_node_table_init();
/* Build resreq AST: SEL_GE(SEL_RESNAME(idx=0, NUMERIC), SEL_NUM(4.0)) */
xce_select_node_t* resname = makeSelectNode(SEL_RESNAME, 0, RES_TYPE_NUMERIC);
double dval = 4.0;
uintptr_t bits;
memcpy(&bits, &dval, sizeof(bits));
xce_select_node_t* num4 = makeSelectNode(SEL_NUM, bits, 0);
xce_select_node_t* ge = makeSelectNode(SEL_GE,
(uintptr_t)resname,
(uintptr_t)num4);
TEST_ASSERT_NOT_NULL(ge);
xce_select_t sel = { .root = ge };
xce_scheduler_t sched;
memset(&sched, 0, sizeof(sched));
/* Worker with 8.0 at res_idx=0 — must pass */
xce_sched_worker_t w8;
memset(&w8, 0, sizeof(w8));
zarray_init(&w8.resvals, sizeof(xce_val_t*), NULL);
xce_val_t v8;
memset(&v8, 0, sizeof(v8));
v8.which = Value_u_fvalue;
v8.u.fvalue = 8.0f;
zarray_put(&w8.resvals, 0, &v8);
TEST_ASSERT_EQUAL_INT(1, sched_meet_host_selection(&sched, &w8, &sel));
/* Worker with 2.0 at res_idx=0 — must fail */
xce_sched_worker_t w2;
memset(&w2, 0, sizeof(w2));
zarray_init(&w2.resvals, sizeof(xce_val_t*), NULL);
xce_val_t v2;
memset(&v2, 0, sizeof(v2));
v2.which = Value_u_fvalue;
v2.u.fvalue = 2.0f;
zarray_put(&w2.resvals, 0, &v2);
TEST_ASSERT_EQUAL_INT(0, sched_meet_host_selection(&sched, &w2, &sel));
zarray_fini(&w8.resvals);
zarray_fini(&w2.resvals);
gc_clear_marks();
gc_sweep_select_nodes(NULL, 0);
sel_node_table_fini();
sel_optimize_test_reset();
}
static void
test_meet_host_selection_string_match(void)
{
/* Job requires: res[1] == "gpu".
* Worker with strval[1]="gpu" must match (rc=1).
* Worker with strval[1]="cpu" must not match (rc=0). */
sel_optimize_set(1);
xstring_table_init(139);
sel_node_table_init();
/* Build resreq AST: SEL_EQ(SEL_RESNAME(idx=1, STRING), SEL_STR("gpu")) */
xce_select_node_t* resname = makeSelectNode(SEL_RESNAME, 1, RES_TYPE_STRING);
char* gpu_str = makeString("gpu");
xce_select_node_t* str_node = makeSelectNode(SEL_STR, (uintptr_t)gpu_str, 0);
xce_select_node_t* eq = makeSelectNode(SEL_EQ,
(uintptr_t)resname,
(uintptr_t)str_node);
TEST_ASSERT_NOT_NULL(eq);
xce_select_t sel = { .root = eq };
xce_scheduler_t sched;
memset(&sched, 0, sizeof(sched));
/* Worker with "gpu" at res_idx=1 — must pass */
xce_sched_worker_t wgpu;
memset(&wgpu, 0, sizeof(wgpu));
zarray_init(&wgpu.resvals, sizeof(xce_val_t*), NULL);
xce_val_t vgpu;
memset(&vgpu, 0, sizeof(vgpu));
vgpu.which = Value_u_svalue;
vgpu.u.svalue = "gpu";
zarray_put(&wgpu.resvals, 1, &vgpu);
TEST_ASSERT_EQUAL_INT(1, sched_meet_host_selection(&sched, &wgpu, &sel));
/* Worker with "cpu" at res_idx=1 — must fail */
xce_sched_worker_t wcpu;
memset(&wcpu, 0, sizeof(wcpu));
zarray_init(&wcpu.resvals, sizeof(xce_val_t*), NULL);
xce_val_t vcpu;
memset(&vcpu, 0, sizeof(vcpu));
vcpu.which = Value_u_svalue;
vcpu.u.svalue = "cpu";
zarray_put(&wcpu.resvals, 1, &vcpu);
TEST_ASSERT_EQUAL_INT(0, sched_meet_host_selection(&sched, &wcpu, &sel));
zarray_fini(&wgpu.resvals);
zarray_fini(&wcpu.resvals);
gc_clear_marks();
gc_sweep_select_nodes(NULL, 0);
sel_node_table_fini();
xstring_table_fini();
sel_optimize_test_reset();
}
#define TEST(x) \
do { \
if (check_test(&arguments, #x)) { \
x(); \
RUN_TEST(x, 0); \
} \
} while (0)
@ -78,5 +384,16 @@ int main(int argc, char* argv[])
UNITY_BEGIN();
TEST(test_build_host_eval_ctx_float);
TEST(test_scheduler_t_has_atomic_gc_fields);
TEST(test_sel_optimize_set_at_scheduler_startup);
TEST(test_meet_host_selection_null_select_passes);
TEST(test_meet_host_selection_always_true_passes);
TEST(test_meet_host_selection_always_false_fails);
TEST(test_gc_predicates_idle_invariant);
TEST(test_staging_merge_complete_gc_guard);
TEST(test_meet_host_selection_numeric_match);
TEST(test_meet_host_selection_string_match);
return UNITY_END();
}

View file

@ -161,6 +161,11 @@ struct ValueWithIdx $C.mapname("xce_val_with_idx_t") {
val @1: Value $C.mapname("val");
}
struct NameValue $C.mapname("xce_name_val_t") {
name @0: Text $C.mapname("name");
val @1: Value $C.mapname("val");
}
struct ConfigVar $C.mapname("xce_config_t") {
name @0: Text;
value @1: Text;