diff --git a/.agents/skills/add-new-model/SKILL.md b/.agents/skills/add-new-model/SKILL.md index 5aa277a1..e6cacb92 100644 --- a/.agents/skills/add-new-model/SKILL.md +++ b/.agents/skills/add-new-model/SKILL.md @@ -50,12 +50,13 @@ redesign serving infrastructure or introduce a general model framework. | `recipe//` | Checkpoint preparation/export scripts, setup and launch instructions, and example requests. Reusable runtime implementation belongs in `src/`. | | `tests//` and `tests//` | All test bodies, test helpers, and fixtures, including contract, tokenizer, checkpoint, HTTP, and kernel tests. | -The [native runtime](../../../src/runtime/README.md) provides serial FIFO admission +The [native runtime](../../../src/runtime/README.md) provides bounded serial FIFO admission and blocking dispatch per loaded executor. Integrate with available worker interfaces while separating processors and batch adapters from forward code. Models declare their execution unit, layouts and compatibility constraints; -shared runtime code owns admission policy. Processing orchestration, queue/token -budgets and dynamic batching remain planned. The native +shared runtime code owns admission policy. Document pending-unit limits and the +worker's overload response; preserve capacity after dispatched cancellation. +Processing orchestration, token budgets and dynamic batching remain planned. The native target uses Rust host orchestration and bindings with CUDA/Metal device kernels. **Do not add test code or fixtures under `src/`.** This includes inline Rust test diff --git a/.agents/skills/precheck-pr/SKILL.md b/.agents/skills/precheck-pr/SKILL.md index 1598da02..6c72f038 100644 --- a/.agents/skills/precheck-pr/SKILL.md +++ b/.agents/skills/precheck-pr/SKILL.md @@ -34,7 +34,8 @@ user instructions for those actions. processing and scheduling, model-specific adapters and executors, and hardware operations. In the native target, Rust orchestrates host work and dispatch; CUDA/Metal execute device kernels. Serial runtime admission and dispatch are - implemented; further orchestration and dynamic batching remain planned. + implemented with bounded pending units; further orchestration, token budgets + and dynamic batching remain planned. Consult the relevant `src/frontend/`, `src/runtime/`, `src/models/`, `src/backends/`, or `recipe/` documentation for the changed component's contract and current status. - For frontend changes, inspect affected request validation, error handling, @@ -47,6 +48,9 @@ user instructions for those actions. - For processing/scheduling changes, apply the architecture contract's batching, output reconstruction, numerical, and lifetime invariants to the affected model's prepared-work and executor boundaries. + Verify overload rejects before dispatch, capacity recovers on queued cancellation + and completion, dispatched cancellation retains capacity, and worker status/body + mappings distinguish overload from inference failure. - Check that tests exercise the changed behavior. For bug fixes, look for a regression case that fails before the fix. Verify documentation and examples against the implementation. diff --git a/.agents/skills/self-review/SKILL.md b/.agents/skills/self-review/SKILL.md index 086ac31c..8790b2b7 100644 --- a/.agents/skills/self-review/SKILL.md +++ b/.agents/skills/self-review/SKILL.md @@ -25,6 +25,7 @@ For processing or batching changes, trace prepared inputs through executor layouts and output reconstruction. Apply the architecture contract's batching, numerical, and lifetime invariants to the affected model. For shared runtime admission changes, check FIFO execution units, cancellation before dispatch, +pending-unit limits, overload/error response mapping, startup configuration, permit/resource retention after dispatch, and release on errors or panics against the [runtime contract](../../../src/runtime/README.md). diff --git a/.agents/skills/system1-omni-review/SKILL.md b/.agents/skills/system1-omni-review/SKILL.md index 3f8d8c17..7f11f8e0 100644 --- a/.agents/skills/system1-omni-review/SKILL.md +++ b/.agents/skills/system1-omni-review/SKILL.md @@ -23,7 +23,7 @@ baseline; target code and configuration determine current paths and commands. - Check separate transport, processing, scheduling, model-executor, and backend ownership against the architecture contracts. Model-specific processors and batch adapters preserve semantics; shared runtime code owns orchestration and scheduling when implemented. Rust owns native host orchestration and bindings/dispatch; CUDA/Metal supply device operations, including optional processing transforms and packing. Serial runtime admission and dispatch are implemented; further orchestration and batching remain planned. CUDA and Metal need not have identical internals. A directory or pass-through HTTP support does not prove native support or batching. - For `src/frontend/`, inspect exact envelope/body and end-to-end header preservation, hop-by-hop filtering, backend path prefix/query handling, timeout/status mapping, health forwarding, cancellation and shutdown. The configured backend bypasses proxies, follows no redirects and retries no requests; changes need transport regression evidence, not a model benchmark. - For models, trace serialization → prompt/template → token IDs → one-pass prefill → scoring → response. Preserve the specific model's contract; do not impose Cua-S1's 26-letter choice restriction on a new model that implements score/noul. Verify pinned checkpoint/adapter/tokenizer/configuration identity, shapes/dtypes and readiness after real initialization. -- For processing or batching changes, apply the architecture contract's compatibility, reconstruction, numerical, and lifetime invariants to the affected model. For admission changes, inspect the [runtime contract](../../../src/runtime/README.md), execution-unit granularity, cancellation before/after dispatch and permit/resource lifetimes. Require executor/kernel evidence for batch capability. +- For processing or batching changes, apply the architecture contract's compatibility, reconstruction, numerical, and lifetime invariants to the affected model. For admission changes, inspect the [runtime contract](../../../src/runtime/README.md), execution-unit granularity, pending limits, overload versus inference-error responses, cancellation before/after dispatch and permit/resource lifetimes. Require executor/kernel evidence for batch capability. - For native/FFI/CUDA changes, inspect Rust and C ABI in lockstep, buffer and graph lifetimes, stream/device ownership, sizes/overflow, BF16 rounding locations, and error propagation/recovery. Compile success and ignored GPU tests do not validate kernels. Test changed inputs, reuse, growth, eviction and failure recovery when graph/cache behavior changes. - For performance or numerical claims, freeze baseline/candidate, hardware/toolchain, checkpoint, input manifest, precision, variables, tolerances and budget before comparing. Separate quality/parity from speed, native inference from frontend overhead, and cold/capture costs from warmed latency. Preserve errors and denominators; never relax tolerances or discard failures to produce a winner. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 4eae78b7..cd664112 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -15,7 +15,7 @@ Keep processors and batch adapters separate from the model's forward implementation, even when they live in the same model directory. The README's target architecture shares processing orchestration and scheduling across models. Native workers already reuse -[serial admission and dispatch](src/runtime/README.md); processing orchestration +[bounded serial admission and dispatch](src/runtime/README.md); processing orchestration and GPU batching remain planned. Model executors own weights, forward passes, learned heads, and device state. diff --git a/README.md b/README.md index 92ee71d0..065f2a16 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ shared CUDA kernels in this repository. model workers. - **Model-owned execution.** Model executors own weights, forward passes, learned heads, device state, and kernel selection. Native workers have separate - processing and executor modules, with shared FIFO admission and blocking + processing and executor modules, with shared bounded FIFO admission and blocking dispatch in the [native runtime](src/runtime/README.md). - **Native CUDA workers.** The Cua-S1 4B 0.2 `text` adapter and Open-Jev-27B-v1.1 run as native workers with shared CUDA kernels. Cua-S1 @@ -68,7 +68,8 @@ Share processing and scheduling; let each model own its execution. The diagram shows the **target architecture**. Today the frontend forwards HTTP requests to separately running workers, whose handlers coordinate independent processors and executors. Both native workers use shared FIFO admission and -blocking dispatch per loaded executor. Processing orchestration, batch budgets, +blocking dispatch, with configurable pending-unit limits per loaded executor. +Processing orchestration, token/batch budgets, compatibility grouping and dynamic batching remain planned. | Layer | Responsibility | Native target implementation | @@ -161,7 +162,7 @@ requests and a matched comparison with raw HF Transformers and OpenJev-Fast. The current focus is the native Cua-S1 and Open-Jev CUDA workers and the serving benchmark harness. Planned work extends the shared runtime with processing -orchestration, admission budgets, compatibility grouping and bounded dynamic +orchestration, token budgets, compatibility grouping and bounded dynamic batching with batch-capable executors, the in-repository LAYA model engine, additional model engines and GPU backends including Metal, and per-model diff --git a/docs/architecture.md b/docs/architecture.md index b1988e80..ca66de7a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -13,7 +13,8 @@ model-specific workers coordinate independent processor and executor modules through `prepare` → `execute` → `finish`. The shared Qwen executor accepts one prompt per forward call. Both workers use the [native runtime](../src/runtime/README.md) for FIFO admission and blocking dispatch -per loaded executor. Shared processing orchestration, batch budgets, +per loaded executor, with bounded pending execution units. Shared processing +orchestration, token/batch budgets, compatibility grouping and dynamic batching are planned. The native workers currently compute their decision heads on the CPU after @@ -92,7 +93,13 @@ executor, with FIFO waiting. Engines own scheduler instances; model-specific execution adapters declare the unit and submit owned closures. Waiting is async, so queued requests do not occupy blocking threads waiting for a model lock. Schedulers are local to each worker/executor; this does not coordinate separate -processes. Queue-length limits and token budgets are not implemented yet. +processes. Pending execution units (waiting plus dispatched) are bounded to 64 by +default, configured by `OMNI_NATIVE_MAX_PENDING`. Full capacity rejects the unit +before dispatch with a typed overload error; native workers map it to HTTP `503` +using their model-specific error body. Normal inference failures remain `500`. +The limit counts Cua-S1 question forwards and Open-Jev complete requests, not +prepared tokens or ingress requests. A later rejected Cua-S1 unit fails the +whole request without partial answers. Token budgets remain planned. - Batch only work accepted by the same loaded executor, with compatible checkpoint/adapter identity, device, dtype, and input layout. Models declare @@ -120,7 +127,8 @@ Keep Rust declarations and the native ABI in sync, and rebuild all consumers when the ABI changes. Buffer handles and captured graphs must outlive queued device operations; synchronize before reading host results or reclaiming storage. Cancellation while waiting for admission removes the queued caller -without dispatch. Once dispatched, the runtime retains admission and captured executor +without dispatch and releases pending capacity. Once dispatched, the runtime retains +both admission and capacity permits and captured executor resources until the blocking closure completes, even if its caller is cancelled. The executor must synchronize device work before returning; cancellation or timeout does not authorize freeing buffers still in use by submitted work. Keep diff --git a/docs/benchmarks/native-admission/README.md b/docs/benchmarks/native-admission/README.md new file mode 100644 index 00000000..eb9b2b5e --- /dev/null +++ b/docs/benchmarks/native-admission/README.md @@ -0,0 +1,151 @@ +# Bounded native admission validation + +Actual runs on 2026-10-05 compared merged main `9adb0701132d259a9fcc8a559472f1c3dc5590b0` +with runtime candidate `b5c443c9310f2118289500b6088822e7bba1fb72`. All 1,536 measured +requests succeeded with exact response/key-order parity, excluding Open-Jev +`metadata.inference_seconds`. All six predeclared performance gates passed. +The limit-one functional burst accepted one request and rejected seven with +HTTP 503 for each model; a subsequent inference returned 200 with exact parity. + +The [protocol](protocol.json) records source/build/binary hashes, pinned models, +controls, input hashes, budget and thresholds. The [compressed evidence](raw-results.json.gz) +contains every measured/feasibility response, warmup, first-inference, near-limit, +expected-error and overload/recovery record, plus fixed input manifests, +per-run summaries, startup and raw device-memory samples. + +## Change and controls + +The independent variable is bounded pending execution and its overload mapping. +Execution remains serial. Normal-load comparisons set `OMNI_NATIVE_MAX_PENDING=64` +(the baseline ignores this new setting); candidate-only functional bursts use 1. +The bound counts Cua-S1 question forwards and Open-Jev complete requests. It does +not bound ingress requests or prepared-token memory; GPU batching remains planned. + +Controls: one reserved SM90 NVIDIA L20X, exact GPU5, driver 570.133.20, CUDA driver +API 13000, NUMA memory node 1/CPUs 56–71, Rust 1.98.1, +Python 3.12, httpx 0.28.1, tokenizers 0.22.2; BF16 backbone, existing FP32 CPU heads/f64 +normalization, unchanged CUDA ABI4 library, eager CUDA, 16 Tokio workers and disabled +tokenizer parallelism. Checkpoint revisions and input hashes are in the protocol. +No caches were dropped, models downloaded or exports repeated. Baseline binaries +reuse an isolated build whose entire tree equals the merged baseline; all comparison +measurements are fresh. Candidate uses its own Cargo target and matching package flags. + +The PR later integrated upstream documentation/recipes and added this report. +All Rust sources, Cargo manifests/lockfile and CUDA build inputs still match the +measured candidate; `native_source_hashes` lets reviewers verify that equivalence. + +## Warmed native HTTP comparison + +Per model/revision: one real-warmed server, first inference validation, one excluded +64-request feasibility pass, then two measured 64-request passes at each concurrency +1/8/16, with three excluded warmups per pass. There are 1,536 measured requests and +256 excluded feasibility requests. All warmup and first-inference responses also +passed parity. Each configuration then validated a 16,368-token input and four +expected errors outside performance timing. No extra runs or discarded failures. + +Gates: mean of per-run mean HTTP latency ≤105% baseline, mean requests/s ≥95%, and +mean per-run p95 ≤110%, for each model/concurrency. Timing is direct native HTTP; +frontend overhead and individual kernel timings are not measured. + +| Model | Concurrency | Baseline mean ms | Candidate mean ms | Latency change | Throughput change | p95 change | Observed gate | +|---|---:|---:|---:|---:|---:|---:|---| +| cua | 1 | 23.47 | 23.53 | +0.27% | -0.27% | +0.39% | pass | +| cua | 8 | 167.52 | 167.81 | +0.17% | -0.19% | +0.09% | pass | +| cua | 16 | 320.86 | 321.56 | +0.22% | -0.25% | +0.43% | pass | +| jev | 1 | 108.68 | 108.69 | +0.00% | -0.00% | -0.77% | pass | +| jev | 8 | 812.37 | 812.04 | -0.04% | +0.04% | +3.42% | pass | +| jev | 16 | 1515.66 | 1516.07 | +0.03% | -0.02% | -0.08% | pass | + + +Relative ranges across the two repetitions, `(max - min) / mean`: + +| Model | Concurrency | Baseline mean run range | Candidate mean run range | Baseline p95 run range | Candidate p95 run range | +|---|---:|---:|---:|---:|---:| +| cua | 1 | 0.04% | 0.02% | 0.42% | 0.30% | +| cua | 8 | 0.04% | 0.36% | 0.32% | 0.18% | +| cua | 16 | 0.28% | 0.06% | 0.29% | 1.29% | +| jev | 1 | 0.04% | 0.18% | 0.21% | 1.25% | +| jev | 8 | 0.03% | 0.19% | 0.22% | 0.12% | +| jev | 16 | 0.00% | 0.06% | 0.04% | 0.08% | + +## Functional overload and recovery + +After normal-load comparisons, one candidate server per model used limit 1, with +real warmup and a validated first inference. One burst of eight concurrent +16,368-token requests yielded one 200 and seven 503 responses per model. The +successful response matched the ordinary reference; each 503 had the native worker +error shape and `native execution capacity exhausted`. The next ordinary request +returned 200 with exact parity; `/health` remained ready. These 16 burst requests +and recovery/startup validation are outside measured performance denominators. +Queued/dispatched cancellation and error/panic release are covered by CPU tests. + +## Startup and memory + +CPU builds/copies and existing checkpoint exports are preparation, excluded from +process-to-readiness and replay. Startup and first inference are separate: + +| Configuration | Process to readiness (s) | First inference (ms) | Sampled GPU peak incl. near-limit (MiB) | +|---|---:|---:|---:| +| cua-baseline | 3.38 | 143.32 | 12417 | +| cua-candidate | 3.67 | 50.16 | 12417 | +| jev-baseline | 15.71 | 189.11 | 55465 | +| jev-candidate | 15.31 | 199.80 | 55465 | + +Memory is device footprint sampled every 200 ms, including near-limit validation; +it is not an allocator peak or a measured-only peak. All 40 task-owned processes +exited and the scheduler reservation was released. Full cleanup/probes remain in +the private original archive; machine/user locations are omitted from public records. + +## Reproduce and inspect + +Build each frozen source in a separate checkout/target using: + +```sh +cargo build --release --locked -j 16 -p omni-cua-s1-native -p omni-open-jev-native +``` + +Prepare the pinned checkpoints with the +[Cua-S1 recipe](../../../recipe/cua_s1/native.md) and +[Open-Jev recipe](../../../recipe/open_jev/native.md). Reuse one unchanged CUDA +library for both revisions and honor the execution host's GPU reservation rules. +The archive can be inspected and its fixed input files extracted without GPU use: + +```python +import gzip, json +from pathlib import Path +root = Path("docs/benchmarks/native-admission") +evidence = json.loads(gzip.decompress((root / "raw-results.json.gz").read_bytes())) +for name, text in evidence["inputs"].items(): + Path(name).write_text(text) +print(evidence["summary"]["comparisons"]) +``` + +For each reserved model/revision server, use the +[repository runner](../../../benchmarks/bench.py), metadata requirements and +[comparison protocol](../../../benchmarks/README.md). Run a feasibility pass +and then two passes at each concurrency, reusing that server: + +```sh +python benchmarks/bench.py run cua-requests.jsonl --endpoint http://127.0.0.1:8000/v1/systemone --model cua-s1-4b-0.2 --metadata metadata.json --output run-c1-r1 --phase measured --concurrency 1 --warmup 3 --timeout 180 +``` + +Use `jev-requests.jsonl` with model `open-jev` for Open-Jev. Create metadata from +actual reserved hardware and the protocol's pinned revisions/precision; readiness +requires the worker's successful real warmup. Preserve every repeat and failure. +The private launcher used `gpu run --gpu-ids 5 --wait 10m --timeout 30m` and +`numactl --membind=1 --physcpubind=56-71`. For the functional check, restart each +candidate with limit 1, validate its first inference, send the archive's near-limit +request eight times concurrently, and validate the following ordinary inference. + +## Limits and disclosure + +Two repetitions provide descriptive ranges, not a confidence interval or +statistically proven equivalence. This synthetic workload tests regression/parity, +not model accuracy or general serving capacity; no speed winner is claimed. +Sequential configuration blocks leave time/order, CPU and thermal effects possible. +Metal, CUDA Graphs, full frontend performance, aggregate token budgets and GPU +batching are unverified here. Labelled accuracy metrics and demo video are N/A for +this admission change. Inputs are task-generated synthetic data; the public archive +retains original response bodies and timings while removing machine-local identity, +paths and process/reservation details. No third-party images or private/customer +inputs are included. diff --git a/docs/benchmarks/native-admission/protocol.json b/docs/benchmarks/native-admission/protocol.json new file mode 100644 index 00000000..19c33b23 --- /dev/null +++ b/docs/benchmarks/native-admission/protocol.json @@ -0,0 +1,259 @@ +{ + "format": "native-admission-protocol/1", + "experiment_date_utc": "2026-10-05", + "source_revisions": { + "baseline": "9adb0701132d259a9fcc8a559472f1c3dc5590b0", + "candidate": "b5c443c9310f2118289500b6088822e7bba1fb72" + }, + "native_source_hashes": { + "Cargo.lock": "34f8ea26fefc050f35247581dac31546409f6146685bd3871907713644ff8b7f", + "Cargo.toml": "3e823af28ffba170d50adebf0624e6eba03b7d130e0479624a9c152dfac6cda3", + "src/backends/cuda/qwen3_5/attention.cu": "3d6935a8a550314815ebe6575be03ce1a7e0b1d5bd611508a4bffbc24cff3132", + "src/backends/cuda/qwen3_5/build.sh": "178d94396fdb46ee929f0da919f120683990ed7af8fd079796106ab5995b3f0f", + "src/backends/cuda/qwen3_5/common.cuh": "ea924f7c903b6d14b0e1f10c8c33a1a9ca23806df093628a178661a303d54d55", + "src/backends/cuda/qwen3_5/elementwise.cu": "b739f1bc2f08ae67399366063cc6c42bc0c5e65e98da23d8b425f9f9ea53ea59", + "src/backends/cuda/qwen3_5/gdn_prefill.cu": "9f2642a5f6e69cd22055419ecf1f799f1ccd811f7453763a41adf4f15b07a41d", + "src/backends/cuda/qwen3_5/gemm.cu": "a82b6d6012c6b13d5ec8dd8ce7107bd95e17f588c76dae1d314e9cff1585274a", + "src/backends/cuda/qwen3_5/mma.cuh": "64654f8b937585805daa72b692c243ae35d34b746e2cb6b0b55c255748495764", + "src/backends/cuda/qwen3_5/norm.cu": "756c9d4023787cc3961ec72bcc96b3f715e705d5dfcb3bf77622879980df82ab", + "src/backends/cuda/qwen3_5/ops.h": "776376f82c8a01aacea4ec29178714c95079ec6454dc0466a483d073fee3e06c", + "src/backends/cuda/qwen3_5/runtime.cu": "3c19d2a1167b44d3a84ffe480605e14a72c775d2c2e51d0c594af9d0db6cb69e", + "src/frontend/Cargo.toml": "595105b609cee3d7bc45c9d23c8cb1e9e37b381007fcbc6b37f6581bcbd74c3e", + "src/frontend/src/lib.rs": "79aada9b5d0acc335d6dcb5008f1b7964a0542f324528c46a40aef6a1f8a3597", + "src/frontend/src/main.rs": "53e5872105f85148727d9e0c6f840f4599b160570bbbe22a279249622b6550d7", + "src/frontend/tests/frontend.rs": "502e373ea4e7a31c965b24c1820dc17b526ea553d4aa6b3babc1e7dec85643bd", + "src/models/cua_s1/native/Cargo.toml": "d0afeefe2d795658c205a0cafa7f7ee2b2006e900637a807c4739943288b74b6", + "src/models/cua_s1/native/src/contract.rs": "c51564ca3c2b87d8f756f17cfbbe6f2bfb86e612a4085237df3e8ceee587486b", + "src/models/cua_s1/native/src/engine.rs": "d5970a1f0ebcb8dab04ebc6ef06de891d54015a4c812630fac88fa650edcefd9", + "src/models/cua_s1/native/src/executor.rs": "e1e962887003c123b7a229b7070843d58295a8247d32135273fed0b645dfd5f1", + "src/models/cua_s1/native/src/lib.rs": "2fbe67252a15f7fa062faf82434d4fbfbf682cefcc96d765360d7c21daabc634", + "src/models/cua_s1/native/src/main.rs": "f48bbe3608054dfe05dc807465d95e31e9a2015272cda1a95ef4f5096ff65d60", + "src/models/cua_s1/native/src/processing.rs": "ce11ac7789c9a9454f61c33b7d00093f868121f2ac4ff09ab5964a2230632615", + "src/models/laya/Cargo.toml": "1c71204adde7c04f9c7bf247139d936d9be257b12c3f19524176ae93aa49f835", + "src/models/laya/src/config.rs": "a8743612808bfa80fb948cd4d55c12d0dd2fbed80f5d540a48009e9ad58f6d5e", + "src/models/laya/src/lib.rs": "52ac328f629edb979cb470553112e7cb49b10caf75fdb7be0c854f96ff960ccf", + "src/models/laya/src/weights.rs": "72949828f4136da31b0a92c0d2d09c72af8b0547c288e419c1cea3b76facf467", + "src/models/open_jev/native/Cargo.toml": "801d38df41f46585c785d604cfab3a644b93b737dc43203c47ef7af087d379af", + "src/models/open_jev/native/src/contract.rs": "e70b779123b2aa27c83826f268f960fca444aad344d85bfc4ffea3b2ea4bb3f8", + "src/models/open_jev/native/src/engine.rs": "fecb9fbef1c5e2058ca366ed3f6b0968dd36d32d1f18ef1058f2cd6eadc85161", + "src/models/open_jev/native/src/executor.rs": "c354c86dfdeb6cf3665dd88197c5692ab7f8abbc643ecce59fa86eade1ab27c9", + "src/models/open_jev/native/src/lib.rs": "aa4a8203388865f116ff43e21fec6bc4e7aa1c9372c854ffc03ac1244db81796", + "src/models/open_jev/native/src/main.rs": "ca34e6d96d06bad0d188c27c78a039091e0ba2480302d4666633d0947ea17c8c", + "src/models/open_jev/native/src/processing.rs": "db66e17ea96ea85f2c47c20e1051f9388326af86ddb6d801e475c6a182317ae7", + "src/models/qwen3_5/native/Cargo.toml": "8c6e8121ab2dd8cf6e7991629b88d97ba2e52d8e305a62e57e7cb4b72f7ccf34", + "src/models/qwen3_5/native/src/cuda.rs": "959f8f513adb63510a7ab6d214596fdec5efeb12e9fe6d4a7df665614b15940d", + "src/models/qwen3_5/native/src/json.rs": "a3933459e7ccfcc8ae0782329c2607d75da0d567f1f341c1e59185aa433432ec", + "src/models/qwen3_5/native/src/lib.rs": "4dc5a2c1babb0e714396d9fe47f3750843036d4b7c5519a8ffe710e61ed88ade", + "src/models/qwen3_5/native/src/model.rs": "fa0754478c35f412365d682635063cc94cf6688d656e10f515fe05600494c6d7", + "src/runtime/Cargo.toml": "8124514a95adecb84447d39e2dde24f427c220458c15549a8b366b3b03015345", + "src/runtime/src/lib.rs": "2a11a78f48b3dc43e46ffc3b3766d833a3196749e043ea26ee4d17a497b99690" + }, + "hypothesis": "Bounding pending native execution units preserves successful/error responses below capacity with no material warmed HTTP regression; a one-unit limit rejects concurrent excess work as 503 and recovers afterward.", + "variable": "Native Rust runtime pending-capacity gate plus startup configuration and typed HTTP overload mapping: merged main 9adb0701 versus b5c443c9. Baseline has unbounded pending units, candidate has 64; execution remains serial. Capacity bounds and overload rejection are the intended consequences. CUDA, Qwen forward, numerical heads, tokenizer, checkpoints, request data, affinity, device and dependencies are fixed.", + "models": { + "cua": { + "base_repository": "Qwen/Qwen3.5-4B", + "adapter_repository": "cua-ai/cua-s1-4b-0.2", + "adapter_subfolder": "text", + "base": "851bf6e806efd8d0a36b00ddf55e13ccb7b8cd0a", + "adapter": "16818868b0cc7813808aae4e87b417657046ab79" + }, + "jev": { + "base_repository": "Qwen/Qwen3.8-27B", + "checkpoint_repository": "ZefanCai/Open-Jev-27B-v1.1", + "base": "1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0", + "checkpoint": "28cf73067d5b337860bbef3c85b8b82ba8730956", + "temperature": 2.5343690298472983, + "max_length": 16384 + } + }, + "aliases": { + "cua": "cua-s1-4b-0.2", + "jev": "open-jev" + }, + "hardware": { + "reported_device_name": "NVIDIA L20X", + "compute_capability": [ + 9, + 0 + ], + "device_count": 1, + "fixed_gpu_id": 5, + "driver_version": "570.133.20", + "cuda_driver_api_version": 13000, + "cpu_affinity": [ + 56, + 57, + 58, + 59, + 60, + 61, + 62, + 63, + 64, + 65, + 66, + 67, + 68, + 69, + 70, + 71 + ], + "numa_node": 1 + }, + "precision": "BF16 backbone; existing FP32 CPU heads; f64 normalization", + "toolchain": { + "rustc": "rustc 1.98.1 (48a229cea 2026-09-01)", + "python": "3.12", + "httpx": "0.28.1", + "tokenizers": "0.22.2" + }, + "build": { + "command": "cargo build --release --locked -j 16 -p omni-cua-s1-native -p omni-open-jev-native", + "rustflags": null, + "separate_target_directories": true, + "baseline_binary_source_revision": "887b29f1e58f99382c5b4c3528e2819158826294", + "baseline_full_tree_equals_merged_baseline": true, + "cargo_lock_sha256": { + "baseline": "34f8ea26fefc050f35247581dac31546409f6146685bd3871907713644ff8b7f", + "candidate": "34f8ea26fefc050f35247581dac31546409f6146685bd3871907713644ff8b7f" + } + }, + "cuda_library": { + "sha256": "80bb218b6dc7d9ae6fcab3ab762924595db0d0bebec7dbf331a4f4dc1b4fc0fa", + "abi": 4, + "source_hashes": { + "src/backends/cuda/qwen3_5/gemm.cu": "a82b6d6012c6b13d5ec8dd8ce7107bd95e17f588c76dae1d314e9cff1585274a", + "src/backends/cuda/qwen3_5/elementwise.cu": "b739f1bc2f08ae67399366063cc6c42bc0c5e65e98da23d8b425f9f9ea53ea59", + "src/backends/cuda/qwen3_5/norm.cu": "756c9d4023787cc3961ec72bcc96b3f715e705d5dfcb3bf77622879980df82ab", + "src/backends/cuda/qwen3_5/attention.cu": "3d6935a8a550314815ebe6575be03ce1a7e0b1d5bd611508a4bffbc24cff3132", + "src/backends/cuda/qwen3_5/gdn_prefill.cu": "9f2642a5f6e69cd22055419ecf1f799f1ccd811f7453763a41adf4f15b07a41d", + "src/backends/cuda/qwen3_5/runtime.cu": "3c19d2a1167b44d3a84ffe480605e14a72c775d2c2e51d0c594af9d0db6cb69e", + "src/backends/cuda/qwen3_5/common.cuh": "ea924f7c903b6d14b0e1f10c8c33a1a9ca23806df093628a178661a303d54d55", + "src/backends/cuda/qwen3_5/mma.cuh": "64654f8b937585805daa72b692c243ae35d34b746e2cb6b0b55c255748495764", + "src/backends/cuda/qwen3_5/ops.h": "776376f82c8a01aacea4ec29178714c95079ec6454dc0466a483d073fee3e06c" + }, + "build_target": "sm90" + }, + "binary_sha256": { + "baseline-cua": "fae5ad89c79f73c1f06a95bd5a560cf0d6dbc88e1d536c66ba14bb988d13df65", + "candidate-cua": "f4ecab69adb88a62544baa3587f201ba8940241ad301bcc7aa78bb4d3096fc79", + "baseline-jev": "9cc08be8d502dddf4ab447a976df272dcb977338011e85c98c78e8b8c8d1b14a", + "candidate-jev": "184cfb900ec0dc692385413ed6266bbb8d29358cdf2b15326c9c30313c629fcc" + }, + "controls": { + "graph": "CUA_S1_GRAPH=0", + "frontend": "Direct native worker HTTP; no frontend overhead comparison", + "cache": "Reuse prepared exports; real startup warmup before health; first inference, one full excluded feasibility pass, then 3 warmup requests before each measured pass. No shared caches dropped.", + "server": "One worker per model/revision normal-load configuration, reused across passes; after all four comparisons, one candidate worker per model with pending limit1 is reused for functional saturation and recovery.", + "request_timeout_s": 180, + "readiness_timeout_s": 600, + "cpu_threads": "TOKIO_WORKER_THREADS=16; TOKENIZERS_PARALLELISM=false, identical for both revisions", + "memory_method": "nvidia-smi GPU memory.used sampled every 0.2s, scoped to reserved GPU5; sampled peak rather than allocator peak.", + "pending": "OMNI_NATIVE_MAX_PENDING=64 for normal-load comparisons; baseline ignores the new setting. Candidate-only functional overload configuration uses 1. This is an explicit consequence of the declared variable, not a performance comparison." + }, + "budget": { + "feasibility_per_configuration": 1, + "measured_per_configuration_per_concurrency": 2, + "concurrency": [ + 1, + 8, + 16 + ], + "extra_validation_per_configuration": "One near-limit request and four expected error requests after measurements; not timed as performance.", + "gpu_timeout": "30m", + "extra_overload_validation": { + "candidate_only_per_model": { + "pending_limit": 1, + "first_inference_after_real_readiness": 1, + "concurrent_near_limit_requests": 8, + "recovery_request": 1 + }, + "measured_as_performance": false, + "stop_on_first_invalid_status_or_parity": true, + "no_extra_bursts_or_repetitions": true + } + }, + "order": [ + "cua-baseline", + "cua-candidate", + "jev-candidate", + "jev-baseline" + ], + "success": { + "parity": "No failures in successful workload; exact JSON equality including key order except metadata.inference_seconds; zero probability/score drift and decision flips. Exact expected error response parity.", + "performance": "For every model/concurrency, mean of per-run mean HTTP latency <= 1.05x baseline, mean requests/s >= 0.95x baseline, mean per-run p95 <= 1.10x baseline. Report both measured runs and their relative range; if variability overlaps observed differences, no performance winner is claimed.", + "overload": "Per candidate/model limit1 burst: at least one valid200 and one503, no other statuses; successful near-limit outputs equal the normal-load reference exactly except timing, every503 uses the native model error shape with native execution capacity exhausted; next ordinary inference returns200 with exact parity." + }, + "stop": "Stop at first CPU probe/build/readiness/response/parity or overload-validation failure, reservation interruption, 30m timeout, or declared budget completion. Expected503 stress responses are preserved outside performance denominators. No extra repeats or tolerance changes.", + "workload": { + "source": "Synthetic deterministic regression inputs, not an accuracy dataset", + "seed": 78, + "requests_per_model": 64, + "cua_prompt_lengths": [ + 146, + 2079 + ], + "jev_prompt_lengths": [ + 50, + 1971 + ], + "near_limit_tokens": 16368, + "input_sha256": { + "jev-lengths.json": "9a4ec56d98e61b3403fe81f70a515df0535207ace434518f3c1defd370dc498e", + "cua-lengths.json": "325163f0569335d04997866ad5ceca81ccf4f37a97e1599410caed17b7300b58", + "cua-near-limit.json": "f56587edd89e3febb9818d6c542b86298eda4c13c6772b04e14bfd88edef8b43", + "cua-requests.jsonl": "42fc2c78464f4a4e6f639e0470bca0ca31c936d963baf40b643405d09822cf93", + "jev-requests.jsonl": "8b44c0c0e90c264681beb5431222879961ef6ab8649415c4205419df9ec9930e", + "jev-errors.json": "6548838d3ad1e6a86c1aab4d9920f1673ff81b6c4c73c78b6ae39a72284b0cb2", + "cua-errors.json": "78eb34ba00c89a5410eecd42a1e6f3cb2517011571e207ddf0439efec206ddc6", + "jev-near-limit.json": "6ed69ea5580dc65d80210dfadea63aaaa39007fa76ce17b897c2725147c1b706" + } + }, + "frozen_private_protocol_sha256": "c3fa2907d36906a235336c885830f62f2a6c8982d1bdf7bd88ca8f27736d7d9b", + "archive": { + "file": "raw-results.json.gz", + "sha256": "ebacc7e2312b0fd3f8018c225e6d9431ce05573cdc9e6fd4ec4c61c0076e2b31", + "compressed_bytes": 352885, + "uncompressed_bytes": 3379336 + }, + "normal_load_counts": { + "measured_requests": 1536, + "excluded_feasibility_requests": 256, + "failed_measured_requests": 0 + }, + "overload_validation": { + "cua": { + "status": "passed", + "pending_limit": 1, + "burst_responses": { + "200": 1, + "503": 7 + }, + "exact_successful_response_parity": true, + "overload_body_checked": true, + "recovery_status": 200, + "health_after_recovery": { + "status": "ready", + "model": "cua-ai/cua-s1-4b-0.2@16818868b0cc7813808aae4e87b417657046ab79:text" + } + }, + "jev": { + "status": "passed", + "pending_limit": 1, + "burst_responses": { + "200": 1, + "503": 7 + }, + "exact_successful_response_parity": true, + "overload_body_checked": true, + "recovery_status": 200, + "health_after_recovery": { + "status": "ready", + "model": "Qwen/Qwen3.8-27B" + } + } + }, + "metadata_redaction": "Machine-local paths, host/user identity, process IDs and reservation metadata are omitted. Local-worker URLs in runner error descriptions are replaced; original model response strings, statuses, IDs and timings are retained.", + "runner_sha256": "4e5ff61d38239cdcd68e8b74f5cae09a03cb17699b7fc1147e0842397cff115d" +} diff --git a/docs/benchmarks/native-admission/raw-results.json.gz b/docs/benchmarks/native-admission/raw-results.json.gz new file mode 100644 index 00000000..718ceea8 Binary files /dev/null and b/docs/benchmarks/native-admission/raw-results.json.gz differ diff --git a/recipe/cua_s1/native.md b/recipe/cua_s1/native.md index 0e352007..fc57ca30 100644 --- a/recipe/cua_s1/native.md +++ b/recipe/cua_s1/native.md @@ -23,6 +23,14 @@ Start the worker (`CUA_S1_HOST` and `CUA_S1_PORT` default to `127.0.0.1` and `80 CUA_S1_MODEL=weights/cua-s1-4b-0.2-text-merged target/release/omni-cua-s1-native ``` +The shared runtime permits up to 64 pending question forwards, including the +running forward. Set `OMNI_NATIVE_MAX_PENDING` on the worker to change this +positive integer limit. At capacity, a question is rejected without dispatch and +the whole request returns HTTP `503` with +`{"detail":"native execution capacity exhausted"}`; earlier questions may have +run, but no partial answer is returned. See the +[runtime contract](../../src/runtime/README.md) for cancellation and limit scope. + For the local CUDA Graph experiment, also set `CUA_S1_GRAPH=1`. The first use of each exact prompt length warms the GEMM plans and captures the forward pass; later requests replay it with freshly uploaded token ids. At most eight lengths diff --git a/recipe/open_jev/native.md b/recipe/open_jev/native.md index 7537d48f..5c1f945f 100644 --- a/recipe/open_jev/native.md +++ b/recipe/open_jev/native.md @@ -61,6 +61,13 @@ The worker loads all text weights onto visible CUDA device 0, performs a real warmup inference, then exposes `/health` and `/v1/systemone`. Use a reservation before any GPU command on hosts with a GPU scheduler. +The shared runtime permits up to 64 pending complete requests, including the +running request. Set `OMNI_NATIVE_MAX_PENDING` on the worker to change this +positive integer limit. At capacity, the worker rejects a request before its +first forward with HTTP `503` and +`{"error":"native execution capacity exhausted"}`. See the +[runtime contract](../../src/runtime/README.md) for cancellation and limit scope. + In another terminal, start the existing Rust frontend: ```sh diff --git a/src/frontend/README.md b/src/frontend/README.md index 81acef10..7a154155 100644 --- a/src/frontend/README.md +++ b/src/frontend/README.md @@ -7,7 +7,8 @@ inference. In the [target architecture](../../docs/architecture.md), a shared Rust worker runtime coordinates independent processors, a scheduler/batcher, and model executors. Rust handles host orchestration; CUDA/Metal handle device operations -through backend bindings. That shared runtime is planned; the current frontend +through backend bindings. Native workers share bounded serial admission and +dispatch; further orchestration and GPU batching remain planned. The frontend continues to forward requests to existing model-specific worker pipelines. ## Run @@ -39,6 +40,11 @@ connections bypass system HTTP proxies. - Uploads are streamed. Responses are buffered so a body timeout can still return `504`. Set request size and concurrency limits at the ingress or worker. +Native Cua-S1 and Open-Jev workers configure pending execution-unit capacity with +`OMNI_NATIVE_MAX_PENDING` (default 64). Their HTTP `503` overload responses pass +through unchanged, with no frontend retry. See the +[native runtime](../runtime/README.md) for each worker's unit and cancellation rules. + Text, image, audio, video and mixed payloads pass through as bytes. Actual inference support depends on the worker. The [Laya recipe](../../recipe/laya/README.md) verifies text decisions against a real backend. diff --git a/src/models/cua_s1/README.md b/src/models/cua_s1/README.md index 36530206..6f9522bd 100644 --- a/src/models/cua_s1/README.md +++ b/src/models/cua_s1/README.md @@ -18,6 +18,12 @@ with a model-state mutex and CPU answer-letter projection after CUDA prefill, outside runtime admission. Processing orchestration and GPU batching remain planned. The contracts below describe its existing behavior. +The shared scheduler bounds pending question forwards (64 by default, +`OMNI_NATIVE_MAX_PENDING`). At capacity, the worker returns HTTP `503` with +`{"detail":"native execution capacity exhausted"}` and no partial answers; +earlier questions may have already executed. Preparation still validates the +whole request before inference. This is a unit-count limit, not a token budget. + ## Pinned revisions | Artifact | Revision | diff --git a/src/models/cua_s1/native/Cargo.toml b/src/models/cua_s1/native/Cargo.toml index a8eb3ae1..aa1cd7a1 100644 --- a/src/models/cua_s1/native/Cargo.toml +++ b/src/models/cua_s1/native/Cargo.toml @@ -22,3 +22,7 @@ serde_json = { version = "1.0.149", features = ["float_roundtrip", "preserve_ord # the onig regex backend, as in the Python tokenizers wheel tokenizers = { version = "=0.22.2", default-features = false, features = ["onig"] } tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread", "sync"] } + +[[test]] +name = "admission_config" +path = "../../../../tests/runtime/configuration.rs" diff --git a/src/models/cua_s1/native/src/engine.rs b/src/models/cua_s1/native/src/engine.rs index 21e2fd18..ec122417 100644 --- a/src/models/cua_s1/native/src/engine.rs +++ b/src/models/cua_s1/native/src/engine.rs @@ -17,6 +17,7 @@ pub struct Engine { impl Engine { pub async fn load(dir: &Path, library: &Path) -> Result { + let scheduler = SerialScheduler::from_env()?; // Written by export_text_merged.py; without it `dir` may hold the base model alone. ensure!( dir.join("cua_s1_export.json").exists(), @@ -36,7 +37,7 @@ impl Engine { let executor = Executor::load(dir, library, &ids).await?; Ok(Self { processor, - scheduler: SerialScheduler::default(), + scheduler, executor, }) } diff --git a/src/models/cua_s1/native/src/main.rs b/src/models/cua_s1/native/src/main.rs index f2912b9f..56cc8e16 100644 --- a/src/models/cua_s1/native/src/main.rs +++ b/src/models/cua_s1/native/src/main.rs @@ -20,6 +20,7 @@ use serde_json::{Value, json}; use omni_cua_s1_native::contract::MODEL_ID; use omni_cua_s1_native::cuda; use omni_cua_s1_native::engine::Engine; +use omni_runtime::Overloaded; const MAX_BODY_BYTES: usize = 4 << 20; const WARMUP: &[u8] = br#"{"model": "cua-s1-4b-0.2", "state": "Dialog: Update installed.", "questions": {"q": {"type": "choice", "instructions": "Close it.", "criteria": {"ok": "OK", "wait": "Wait"}}}}"#; @@ -28,6 +29,14 @@ fn reply(status: u16, body: Value) -> Response { (StatusCode::from_u16(status).unwrap(), Json(body)).into_response() } +fn inference_error(e: anyhow::Error) -> Response { + if let Some(overload) = e.downcast_ref::() { + return reply(503, json!({"detail": overload.to_string()})); + } + eprintln!("inference failed: {e:#}"); + reply(500, json!({"detail": "inference failed"})) +} + /// Every prompt is tokenized and checked against the limit before any forward pass. async fn decide(engine: &Engine, raw: &[u8]) -> Response { let prepared = match engine.processor.prepare(raw) { @@ -44,13 +53,14 @@ async fn decide(engine: &Engine, raw: &[u8]) -> Response { .await; match result { Ok(body) => reply(200, body), - Err(e) => { - eprintln!("inference failed: {e:#}"); - reply(500, json!({"detail": "inference failed"})) - } + Err(e) => inference_error(e), } } +#[cfg(test)] +#[path = "../../../../../tests/runtime/http_errors.rs"] +mod http_tests; + async fn systemone( State(engine): State>, body: Result, diff --git a/src/models/open_jev/README.md b/src/models/open_jev/README.md index a6b22822..36433b7d 100644 --- a/src/models/open_jev/README.md +++ b/src/models/open_jev/README.md @@ -28,3 +28,8 @@ retains request-wide model locking and independent single-prompt execution, with the scalar head on the CPU after CUDA prefill. The engine owns a [shared serial scheduler](../../runtime/README.md) that admits the complete request before blocking dispatch. GPU batching and batch budgets remain planned. + +Pending complete request units are bounded to 64 by default, configured by +`OMNI_NATIVE_MAX_PENDING`. Full capacity rejects the request before its first +forward with HTTP `503` and `{"error":"native execution capacity exhausted"}`. +Preparation precedes admission; aggregate token budgets remain planned. diff --git a/src/models/open_jev/native/Cargo.toml b/src/models/open_jev/native/Cargo.toml index 346440c6..457a8555 100644 --- a/src/models/open_jev/native/Cargo.toml +++ b/src/models/open_jev/native/Cargo.toml @@ -17,3 +17,7 @@ tokio = { version = "1.49.0", features = ["macros", "net", "rt-multi-thread", "s [[test]] name = "contract" path = "../../../../tests/open_jev/contract.rs" + +[[test]] +name = "admission_config" +path = "../../../../tests/runtime/configuration.rs" diff --git a/src/models/open_jev/native/src/engine.rs b/src/models/open_jev/native/src/engine.rs index 5e5d3d66..01887678 100644 --- a/src/models/open_jev/native/src/engine.rs +++ b/src/models/open_jev/native/src/engine.rs @@ -20,6 +20,7 @@ pub struct Engine { impl Engine { pub async fn load(dir: &Path, library: &Path) -> Result { + let scheduler = SerialScheduler::from_env()?; let manifest: Value = serde_json::from_slice( &std::fs::read(dir.join("open_jev_export.json")) .context("export the merged checkpoint; see recipe/open_jev/native.md")?, @@ -54,7 +55,7 @@ impl Engine { let executor = Executor::load(dir, library, head).await?; Ok(Self { processor, - scheduler: SerialScheduler::default(), + scheduler, executor, }) } diff --git a/src/models/open_jev/native/src/main.rs b/src/models/open_jev/native/src/main.rs index 7352bff5..c93424b2 100644 --- a/src/models/open_jev/native/src/main.rs +++ b/src/models/open_jev/native/src/main.rs @@ -13,6 +13,7 @@ use axum::{ }; use omni_open_jev_native::{contract::MODEL_ID, engine::Engine}; use omni_qwen3_5_native::cuda; +use omni_runtime::Overloaded; use serde_json::json; const WARMUP: &[u8] = br#"{"state":"Dialog: Update installed.","questions":{"q":{"type":"choice","instructions":"Close it.","criteria":{"ok":"OK","wait":"Wait"}}}}"#; @@ -21,6 +22,14 @@ fn error(status: StatusCode, message: impl ToString) -> Response { (status, Json(json!({"error": message.to_string()}))).into_response() } +fn inference_error(e: anyhow::Error) -> Response { + if let Some(overload) = e.downcast_ref::() { + return error(StatusCode::SERVICE_UNAVAILABLE, overload); + } + eprintln!("inference failed: {e:#}"); + error(StatusCode::INTERNAL_SERVER_ERROR, "model inference failed") +} + async fn decide(engine: &Engine, raw: &[u8]) -> Response { let prepared = match engine.processor.prepare(raw) { Ok(prepared) => prepared, @@ -36,13 +45,14 @@ async fn decide(engine: &Engine, raw: &[u8]) -> Response { .await; match result { Ok(body) => Json(body).into_response(), - Err(e) => { - eprintln!("inference failed: {e:#}"); - error(StatusCode::INTERNAL_SERVER_ERROR, "model inference failed") - } + Err(e) => inference_error(e), } } +#[cfg(test)] +#[path = "../../../../../tests/runtime/http_errors.rs"] +mod http_tests; + async fn systemone( State(engine): State>, headers: HeaderMap, diff --git a/src/runtime/README.md b/src/runtime/README.md index c06c3036..b9b99267 100644 --- a/src/runtime/README.md +++ b/src/runtime/README.md @@ -11,38 +11,62 @@ Cloning a scheduler shares that admission queue; separate loaded executors have separate queues. The model mutex remains a guard for Rust's mutable executor state. Waiting requests no longer occupy blocking threads waiting for that mutex. +Pending work is bounded to 64 execution units per loaded executor by default. +`OMNI_NATIVE_MAX_PENDING` sets a positive integer limit for either native worker; +invalid values fail startup before checkpoint or CUDA loading. The limit counts +waiting and dispatched units together, including work whose caller stopped +waiting. It is separate from the execution concurrency, which remains one. +At capacity, the scheduler immediately returns `Overloaded` without dispatching +the rejected unit. Workers return HTTP `503` with +`native execution capacity exhausted` in their existing error shape (`detail` +for Cua-S1, `error` for Open-Jev). The frontend forwards this response without +retrying. `SerialScheduler::new(limit)` configures the same bound for Rust callers; +`Default` uses 64 independently of environment variables. + | Worker | Admitted unit | Work outside admission | | --- | --- | --- | | Cua-S1 | One question's unpadded Qwen forward. Release before admitting its next question. | Request preparation, CPU letter projection and response finishing. | | Open-Jev | One complete request's independent candidate forwards and CPU scalar heads. | Request preparation and calibrated response finishing. | +The limit bounds execution units, not ingress requests or prepared-token memory. +Preparation still precedes admission. Cua-S1 can be rejected on a later question +after earlier forwards; the entire request fails with no partial answer body. +Open-Jev is accepted or rejected as one complete request unit. Individual prompt +and HTTP body limits remain model-specific; aggregate token budgets are planned. + Models retain weights, heads, device state, scratch buffers and graph caches. The scheduler takes an owned closure and returns its result; it does not know model identifiers, question/candidate mappings, tensor layouts or backend APIs. The existing numerical, ordering, whole-request failure and readiness contracts remain in the processors, executors and worker handlers. -Cancellation while waiting removes the caller from admission without dispatch. -After dispatch, the blocking closure retains the permit and its captured -resources until completion, even if the caller disconnects or stops waiting. +Cancellation while waiting removes the caller and releases pending capacity +without dispatch. After dispatch, the blocking closure retains both permits and +its captured resources until completion, even if the caller stops waiting. Executors must finish queued device work before returning; the current Qwen forward synchronizes the final hidden-state download. Errors and panics release -admission when the blocking task finishes. Model mutex poisoning retains its -existing failure behavior. +admission and pending capacity when the blocking task finishes. Model mutex +poisoning retains its existing failure behavior. This implements serial admission per executor within a worker process. Admitted -concurrency is one. Queue-length limits, token budgets, compatibility grouping, +concurrency is one. Token budgets, compatibility grouping, shared processing orchestration and GPU batching remain planned. The Qwen executor still accepts one prompt per forward; this scheduler does not pack inputs or coordinate separately running workers. CPU tests cover FIFO serialization, queued cancellation, retention of admission and resources after dispatched cancellation, errors/panics, and independent -loaded executors: +loaded executors, full-capacity rejection and recovery. Native worker tests also +cover startup configuration and overload/error HTTP responses: ```sh cargo test --locked -p omni-runtime +cargo test --locked -p omni-cua-s1-native -p omni-open-jev-native ``` +The [native admission comparison](../../docs/benchmarks/native-admission/README.md) +records normal-load parity and performance ranges, plus limit-one overload and +recovery on both real workers. + See the [architecture contracts](../../docs/architecture.md) and [benchmark protocol](../../benchmarks/README.md) when changing admission policy. diff --git a/src/runtime/src/lib.rs b/src/runtime/src/lib.rs index 85c4e1db..24b7a6c3 100644 --- a/src/runtime/src/lib.rs +++ b/src/runtime/src/lib.rs @@ -2,34 +2,75 @@ use std::sync::Arc; -use anyhow::{Context, Result}; +use anyhow::{Context, Result, ensure}; use tokio::sync::Semaphore; +/// All pending execution slots are occupied; the submitted work was not dispatched. +#[derive(Debug)] +pub struct Overloaded; + +impl std::fmt::Display for Overloaded { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("native execution capacity exhausted") + } +} + +impl std::error::Error for Overloaded {} + /// FIFO admission of one execution unit at a time. Clones share the same queue. /// Each loaded executor gets its own scheduler; models define the unit of work. #[derive(Clone)] pub struct SerialScheduler { admission: Arc, + capacity: Arc, } impl Default for SerialScheduler { fn default() -> Self { - Self { - admission: Arc::new(Semaphore::new(1)), - } + Self::new(64).expect("default pending limit is valid") } } impl SerialScheduler { + /// Bound waiting and dispatched units together; execution remains serial. + pub fn new(max_pending: usize) -> Result { + ensure!( + (1..=Semaphore::MAX_PERMITS).contains(&max_pending), + "pending limit must be within 1..={}", + Semaphore::MAX_PERMITS + ); + Ok(Self { + admission: Arc::new(Semaphore::new(1)), + capacity: Arc::new(Semaphore::new(max_pending)), + }) + } + + /// Configure a worker with OMNI_NATIVE_MAX_PENDING (default: 64 units). + /// Invalid values fail startup before loading the model. + pub fn from_env() -> Result { + match std::env::var("OMNI_NATIVE_MAX_PENDING") { + Ok(value) => Self::new(value.parse().context("OMNI_NATIVE_MAX_PENDING")?) + .context("OMNI_NATIVE_MAX_PENDING"), + Err(std::env::VarError::NotPresent) => Ok(Self::default()), + Err(e) => Err(e).context("OMNI_NATIVE_MAX_PENDING"), + } + } + /// Wait asynchronously before dispatching blocking model work. + /// Full capacity returns `Overloaded` immediately without waiting or dispatch. /// Cancellation while waiting removes the work from admission. Once dispatched, - /// the work retains its permit and captured resources until the closure returns, + /// the work retains both permits and captured resources until the closure returns, /// even when its caller stops waiting. Device work must complete before return. pub async fn run(&self, work: F) -> Result where F: FnOnce() -> Result + Send + 'static, T: Send + 'static, { + let capacity = self + .capacity + .clone() + .try_acquire_owned() + .map_err(|_| Overloaded)?; let permit = self .admission .clone() @@ -37,6 +78,7 @@ impl SerialScheduler { .await .expect("private admission semaphore is never closed"); tokio::task::spawn_blocking(move || { + let _capacity = capacity; let _permit = permit; work() }) diff --git a/tests/runtime/configuration.rs b/tests/runtime/configuration.rs new file mode 100644 index 00000000..7221e038 --- /dev/null +++ b/tests/runtime/configuration.rs @@ -0,0 +1,49 @@ +use std::process::Command; + +fn worker() -> (&'static str, &'static str) { + match ( + option_env!("CARGO_BIN_EXE_omni-cua-s1-native"), + option_env!("CARGO_BIN_EXE_omni-open-jev-native"), + ) { + (Some(binary), None) => (binary, "CUA_S1"), + (None, Some(binary)) => (binary, "OPEN_JEV"), + other => panic!("expected one native worker binary, got {other:?}"), + } +} + +#[test] +fn invalid_capacity_fails_startup_before_model_or_cuda_loading() { + let (binary, prefix) = worker(); + for limit in ["0", "-1", "invalid", "", "18446744073709551615"] { + let output = Command::new(binary) + .env("OMNI_NATIVE_MAX_PENDING", limit) + .env(format!("{prefix}_MODEL"), "/nonexistent-model") + .env(format!("{prefix}_CUDA_LIB"), "/nonexistent-cuda-library") + .output() + .unwrap(); + assert!(!output.status.success()); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!(stderr.contains("OMNI_NATIVE_MAX_PENDING"), "{stderr}"); + assert!(!stderr.contains("libcuda"), "{stderr}"); + } +} + +#[test] +fn valid_capacity_and_default_proceed_to_checkpoint_validation() { + let (binary, prefix) = worker(); + for limit in [None, Some("1"), Some("64")] { + let mut command = Command::new(binary); + command + .env_remove("OMNI_NATIVE_MAX_PENDING") + .env(format!("{prefix}_MODEL"), "/nonexistent-model") + .env(format!("{prefix}_CUDA_LIB"), "/nonexistent-cuda-library"); + if let Some(limit) = limit { + command.env("OMNI_NATIVE_MAX_PENDING", limit); + } + let output = command.output().unwrap(); + assert!(!output.status.success()); + let stderr = String::from_utf8(output.stderr).unwrap(); + assert!(stderr.contains("checkpoint"), "{stderr}"); + assert!(!stderr.contains("OMNI_NATIVE_MAX_PENDING"), "{stderr}"); + } +} diff --git a/tests/runtime/http_errors.rs b/tests/runtime/http_errors.rs new file mode 100644 index 00000000..d434008f --- /dev/null +++ b/tests/runtime/http_errors.rs @@ -0,0 +1,42 @@ +use super::*; + +fn expected_body(message: &str) -> serde_json::Value { + match env!("CARGO_PKG_NAME") { + "omni-cua-s1-native" => json!({"detail": message}), + "omni-open-jev-native" => json!({"error": message}), + other => panic!("unexpected worker {other}"), + } +} + +#[tokio::test] +async fn admission_overload_returns_503_with_the_worker_error_shape() { + for error in [ + anyhow::Error::new(Overloaded), + anyhow::Error::new(Overloaded).context("executor context"), + ] { + let response = inference_error(error); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(response.headers()["content-type"], "application/json"); + let bytes = axum::body::to_bytes(response.into_body(), 1024) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body, expected_body("native execution capacity exhausted")); + } +} + +#[tokio::test] +async fn executor_failures_keep_the_existing_500_response() { + let response = inference_error(anyhow::anyhow!("private forward failure")); + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + let bytes = axum::body::to_bytes(response.into_body(), 1024) + .await + .unwrap(); + let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + let message = match env!("CARGO_PKG_NAME") { + "omni-cua-s1-native" => "inference failed", + "omni-open-jev-native" => "model inference failed", + other => panic!("unexpected worker {other}"), + }; + assert_eq!(body, expected_body(message)); +} diff --git a/tests/runtime/scheduler.rs b/tests/runtime/scheduler.rs index b5999340..047357f2 100644 --- a/tests/runtime/scheduler.rs +++ b/tests/runtime/scheduler.rs @@ -6,7 +6,7 @@ use std::task::Poll; use std::time::Duration; use anyhow::Result; -use omni_runtime::SerialScheduler; +use omni_runtime::{Overloaded, SerialScheduler}; use tokio::sync::oneshot; use tokio::task::JoinHandle; @@ -42,7 +42,7 @@ async fn blocked(scheduler: SerialScheduler) -> (JoinHandle>, mpsc::S #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn queued_units_run_serially_in_fifo_order() { - let scheduler = SerialScheduler::default(); + let scheduler = SerialScheduler::new(3).unwrap(); let (first, release) = blocked(scheduler.clone()).await; let order = Arc::new(Mutex::new(Vec::new())); let mut second = Box::pin(scheduler.run({ @@ -85,7 +85,7 @@ fn cancelling_queued_work_never_dispatches_it() { .build() .unwrap() .block_on(async { - let scheduler = SerialScheduler::default(); + let scheduler = SerialScheduler::new(2).unwrap(); let (first, release) = blocked(scheduler.clone()).await; let calls = Arc::new(AtomicUsize::new(0)); let mut cancelled = Box::pin(scheduler.run({ @@ -96,10 +96,19 @@ fn cancelling_queued_work_never_dispatches_it() { } })); poll_waiting(cancelled.as_mut()).await; + assert!( + deadline(scheduler.run(|| Ok(()))) + .await + .unwrap_err() + .is::() + ); drop(cancelled); + // Cancelling the waiter returns capacity before the running unit ends. + let mut replacement = Box::pin(scheduler.run(|| Ok(42))); + poll_waiting(replacement.as_mut()).await; release.send(()).unwrap(); deadline(first).await.unwrap().unwrap(); - assert_eq!(deadline(scheduler.run(|| Ok(42))).await.unwrap(), 42); + assert_eq!(deadline(replacement).await.unwrap(), 42); assert_eq!(calls.load(Ordering::SeqCst), 0); }); } @@ -114,7 +123,7 @@ impl Drop for DeviceState { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn cancelling_dispatched_work_retains_admission_and_resources_until_completion() { - let scheduler = SerialScheduler::default(); + let scheduler = SerialScheduler::new(2).unwrap(); let dropped = Arc::new(AtomicBool::new(false)); let state = Arc::new(DeviceState(dropped.clone())); let (started, ready) = oneshot::channel(); @@ -151,6 +160,13 @@ async fn cancelling_dispatched_work_retains_admission_and_resources_until_comple .is_err() ); assert!(!dropped.load(Ordering::SeqCst)); + // The cancelled caller's dispatched unit still consumes pending capacity. + assert!( + deadline(scheduler.run(|| Ok(()))) + .await + .unwrap_err() + .is::() + ); release.send(()).unwrap(); deadline(next).await.unwrap(); assert!(dropped.load(Ordering::SeqCst)); @@ -158,7 +174,7 @@ async fn cancelling_dispatched_work_retains_admission_and_resources_until_comple #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn errors_and_panics_release_admission() { - let scheduler = SerialScheduler::default(); + let scheduler = SerialScheduler::new(1).unwrap(); let error = scheduler .run(|| -> Result<()> { anyhow::bail!("forward failed") }) .await; @@ -177,10 +193,43 @@ async fn errors_and_panics_release_admission() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn loaded_executors_have_independent_admission() { - let first_scheduler = SerialScheduler::default(); - let second_scheduler = SerialScheduler::default(); + let first_scheduler = SerialScheduler::new(1).unwrap(); + let second_scheduler = SerialScheduler::new(1).unwrap(); let (first, release) = blocked(first_scheduler).await; assert_eq!(deadline(second_scheduler.run(|| Ok(42))).await.unwrap(), 42); release.send(()).unwrap(); deadline(first).await.unwrap().unwrap(); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn full_capacity_rejects_without_dispatch_and_recovers_after_completion() { + let scheduler = SerialScheduler::new(1).unwrap(); + let (first, release) = blocked(scheduler.clone()).await; + let dropped = Arc::new(AtomicBool::new(false)); + let state = DeviceState(dropped.clone()); + let calls = Arc::new(AtomicUsize::new(0)); + let error = deadline(scheduler.clone().run({ + let calls = calls.clone(); + move || { + drop(state); + calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + })) + .await + .unwrap_err(); + assert!(error.is::()); + assert_eq!(calls.load(Ordering::SeqCst), 0); + assert!(dropped.load(Ordering::SeqCst)); + release.send(()).unwrap(); + deadline(first).await.unwrap().unwrap(); + assert_eq!(deadline(scheduler.run(|| Ok(42))).await.unwrap(), 42); +} + +#[test] +fn pending_limit_must_be_nonzero_and_fit_the_semaphore() { + assert!(SerialScheduler::new(0).is_err()); + assert!(SerialScheduler::new(tokio::sync::Semaphore::MAX_PERMITS + 1).is_err()); + assert!(SerialScheduler::new(1).is_ok()); + assert!(SerialScheduler::new(tokio::sync::Semaphore::MAX_PERMITS).is_ok()); +}