Run matched HTTP measurements on assigned ATOM and RNGD adapters

opt_prime/examples/pd/http_matrix/.gitattributes

@@ -0,0 +1 @@
+evidence*/*.json linguist-generated=true

opt_prime/examples/pd/http_matrix/README.md

@@ -73,3 +73,13 @@ Where the SIGPIPE candidate is needed, pass `--http-module /tmp/pd-http-safe.py`
 Both caches are disabled and producer/local chunks are256. Warmup, loading and separate token-ID checks are excluded from timed batches. JSON retains raw per-request SSE events' first-arrival latency, E2E latency, estimated TPOT, reconstructed text equality, whole-batch throughput and final server metrics. SSE events can coalesce tokens: first-event latency and estimated TPOT are not per-token engine timings. `passed` records workflow/count success; `parity_passed` additionally requires reference token/text equality. A parity failure saves evidence and exits unsuccessfully.
 
 Repeat modes in reverse order against the same native reference and do not run CPU regression suites concurrently with final measurements. These Torch PD baselines gather and transfer the whole KV after prefill; they do not implement within-request compute/copy overlap. Existing HTTP transfer histograms do not observe the batched worker timing path; zero observations do not mean zero transfer cost. See issue249 for separate accumulated worker duration counters.
+
+### External backend comparison
+
+`benchmark.py --server-config config.json` uses the same client requests, reference checks and timing loop with explicitly supplied server commands. The file is trusted executable configuration: `servers` is a nonempty list of argv arrays (no shell expansion), each containing exactly one standalone `{port}` argument. `metadata` describes physical resources and execution policies; it replaces GPU-specific defaults rather than implying a GPU configuration for an NPU server. Only supported resource fields are accepted.
+
+`atom_server.py` connects one assigned ATOM to its existing B1/context9216 artifact. Omit `--worker-python`/`--engine-factory` for native ATOM; supply both for the established RNGD subprocess producer. Pass explicit trusted module candidates and artifact root. The allocator/container bootstrap must restrict visible devices; the server also rejects anything other than one ATOM node. Requests in this benchmark use `ignore_eos=true`, so fixed output lengths are comparable. Client concurrency above one is queued work, not physical batch size.
+
+For slow cold initialization, pass `--startup-timeout 1800` explicitly. Default180 remains unchanged; nonpositive values fail before model loading. Startup is excluded from measured latency/throughput. In the first hardware attempt, the shorter outer timeout terminated initialization before the existing subprocess worker's1800second deadline; that failure is retained alongside the passing nativeATOM baseline. A longer deadline is not evidence that initialization or PD succeeds.
+
+The initial Qwen3-0.6B nativeATOM run passed12batches at inputs128/8192,output32,concurrency1/4. RNGD-to-ATOM performance/parity remains under validation. Use the experiment record for pinned runtime modules, environment and commands.

opt_prime/examples/pd/http_matrix/STATUS.md

@@ -39,3 +39,11 @@ A real terminated subprocess producer previously left /ready200 and new requests
 Four models times native/PD/DP2 completed144measured batches with parity. Inputs128/8192, outputs32, concurrency1/4, three repeats. DP2 leads all four at8192/c4. BlockA is ordered and has a short concurrent CPU-test window during KONI; quiet reversed-order replication and longer outputs remain. No within-request compute/copy overlap in this Torch baseline. See evidence248 and matched-pd-performance248 record.
 
 BlockB reverse-model/mode repetition completed144additional batches with parity (288A/B total). Same-two-GPU DP2 leads all four at8192/c4. PD relative to native is variable: Gemma23.11->17.79tok/s,Qwen0.6B22.37->18.41,Qwen4B13.25->15.12,KONI14.47->14.02. No stable native-to-PD speedup claim. Runtime273952a adds stage counters; scheduling/compute unchanged. Longer-output and actual heterogeneous benchmarking ongoing.
+
+External-backend benchmark candidate:trusted argv config and explicit startup deadline. Real ATOM nativeQwen0.6B12batches pass parity; firstPDattempt timed out at180seconds before readiness, raw failure retained. Retry with1800seconds is live. No heterogeneous speedup claim.3harness tests pass; no core serving changes.
+
+Qwen0.6B actual RNGD-to-ATOM completed12PD batches with parity against nativeATOM;42remote/0localprefills. Readiness369.73s excluded from timing.8192input/c4 throughputnative22.25vsPD10.58tok/s. Current width8192producer is slower; shorter prompts also pad to8192. Other three model comparisons running. Failed180s startup record retained.
+
+Qwen3-4B and KONI8B RNGD-to-ATOM matched runs complete: each12native+12PD batches pass parity. Input8192/output32/c4: Qwen4B5.0853→5.4903tok/s;KONI3.3576→4.6973tok/s. Input128/c4:Qwen4B26.3399→7.4952;KONI14.6259→5.9530. Cold startup excluded (PD810.53s/432.86s). Both longer-model PD paths help on long input but regress on short input. Gemma native done,PD pending.
+
+Four-model heterogeneous matrix completed96batches. Gemma short forced-output parity fails afterEOS: firstEOS index2, divergence index5. Offline through-EOS prefixes match; not a new normal-EOS HTTP run. Long Gemma parity passes but PD14.94vsnative27.75tok/s. Each PD path42remote/0local. Raw failure retained; no all-model parity claim. Owned benchmark Pod deleted and NotFound verified. Earlier pre-EOS numerical issue#233 remains open.

opt_prime/examples/pd/http_matrix/atom_server.py

@@ -0,0 +1,74 @@
+"""Single assigned ATOM HTTP backend, optionally fed by the established RNGD subprocess adapter."""
+import argparse
+import importlib.util
+import signal
+import sys
+from pathlib import Path
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument('--model', required=True)
+    p.add_argument('--cache-dir', required=True)
+    p.add_argument('--port', required=True, type=int)
+    p.add_argument('--batching', required=True, help='trusted ATOM batching module')
+    p.add_argument('--http-module', required=True, help='trusted HTTP module')
+    p.add_argument('--pd-module', required=True, help='trusted subprocess worker module')
+    p.add_argument('--worker-python')
+    p.add_argument('--engine-factory')
+    p.add_argument('--width', type=int, default=8192)
+    p.add_argument('--kv-blocks', type=int, default=3)
+    p.add_argument('--max-output', type=int, default=128)
+    a = p.parse_args()
+    if bool(a.worker_python) != bool(a.engine_factory):
+        p.error('worker-python and engine-factory must be supplied together')
+    for name, path in [('opt_prime.accel.pd.pd_batching', a.pd_module),
+                       ('opt_prime.accel.platforms.atom.batching', a.batching),
+                       ('opt_prime.accel.serve.llmd', a.http_module)]:
+        spec = importlib.util.spec_from_file_location(name, path)
+        module = importlib.util.module_from_spec(spec)
+        sys.modules[name] = module
+        spec.loader.exec_module(module)
+    from opt_prime.accel.pd.pd_batching import SubprocessPrefillWorker
+    from opt_prime.accel.platforms.atom.engine import AtomRblnEngine
+    from opt_prime.accel.serve.batched import BatchedOwner, BatchedService
+    from opt_prime.accel.serve.llmd import serve_stage
+    from transformers import AutoTokenizer
+
+    nodes = list(Path('/dev').glob('rbln[0-9]*'))
+    if len(nodes) != 1:
+        raise RuntimeError('benchmark requires exactly one assigned ATOM device')
+    engine = AtomRblnEngine(devices=[int(nodes[0].name[4:])], max_seq_len=9216,
+                           batch_size=1, kv_blocks=a.kv_blocks, cache_dir=a.cache_dir)
+    worker = owner = server = http = None
+    try:
+        engine.start(a.model)
+        c = engine.model.config.get_text_config()
+        if a.worker_python:
+            worker = SubprocessPrefillWorker(a.worker_python,
+                ['--model', a.model, '--width', str(a.width), '--engine-factory', a.engine_factory],
+                slots=1, layers=c.num_hidden_layers, kv_heads=c.num_key_value_heads,
+                head_dim=getattr(c, 'head_dim', None) or c.hidden_size // c.num_attention_heads,
+                max_context=a.width)
+        tok = AutoTokenizer.from_pretrained(a.model, local_files_only=True)
+        batcher = engine.batcher(prefix_cache=False, prefill_worker=worker, local_prefill=worker is None)
+        service = BatchedService(batcher, model=a.model, vocab_size=c.vocab_size,
+                                 max_output=a.max_output, tokenizer=tok,
+                                 render=lambda ids: tok.decode(ids, skip_special_tokens=True))
+        owner = BatchedOwner(service)
+        # The benchmark uses fixed-length ignore_eos requests; no EOS-dependent timing comparisons.
+        signal.signal(signal.SIGTERM, lambda *_: owner.submit('stop'))
+        server, http = serve_stage(owner, 'decode', a.port)
+        owner.run()
+    finally:
+        if server:
+            server.shutdown()
+            server.server_close()
+            http.join(5)
+        if worker:
+            worker.close()
+        engine.close()
+
+
+if __name__ == '__main__':
+    main()

opt_prime/examples/pd/http_matrix/benchmark.py

@@ -28,11 +28,35 @@ def timing_summary(rows, wall_s):
     }
 
 
+RESOURCE_FIELDS = {"physical_gpus", "physical_atom", "physical_rngd", "slots_per_decoder",
+                   "producer_cache", "decoder_cache", "producer_chunk", "local_chunk", "context"}
+
+
+def load_server_config(path):
+    """Trusted executable argv arrays; no shell expansion. Resource facts are supplied by the experiment."""
+    config = json.loads(Path(path).read_text())
+    if not isinstance(config, dict):
+        raise TypeError("server config must be an object")
+    servers, metadata = config.get("servers"), config.get("metadata")
+    if not isinstance(servers, list) or not servers or not isinstance(metadata, dict):
+        raise ValueError("server config needs nonempty servers and metadata object")
+    if set(metadata) - RESOURCE_FIELDS:
+        raise ValueError("unsupported resource metadata key")
+    for argv in servers:
+        if not isinstance(argv, list) or not argv or not all(isinstance(x, str) and x for x in argv):
+            raise ValueError("each server must be a nonempty argv array")
+        if argv.count("{port}") != 1:
+            raise ValueError("each server needs exactly one standalone {port} argument")
+    return config
+
+
 def main():
     p = argparse.ArgumentParser(description=__doc__)
     p.add_argument("--model", required=True)
     p.add_argument("--mode", choices=["native", "dp2", "pd"], required=True)
     p.add_argument("--output", required=True)
+    p.add_argument("--startup-timeout", type=int, default=180, help="server readiness deadline in seconds; excluded from timing")
+    p.add_argument("--server-config", help="trusted JSON argv templates and resource metadata for external backends")
     p.add_argument("--reference", help="native benchmark JSON with identical token-ID prompts")
     p.add_argument("--http-module", help="trusted local HTTP candidate")
     p.add_argument("--lengths", nargs="+", type=int, default=[128, 8192])
@@ -42,6 +66,9 @@ def main():
     p.add_argument("--slots", type=int, default=4)
     p.add_argument("--dtype", choices=["bfloat16", "float32"], default="bfloat16")
     a = p.parse_args()
+    if a.startup_timeout <= 0:
+        p.error("startup timeout must be positive")
+    external = load_server_config(a.server_config) if a.server_config else None
     if min(*a.lengths, *a.outputs, *a.concurrency, a.repeats, a.slots) < 1:
         p.error("sizes must be positive")
     if max(a.lengths) + max(a.outputs) > 9216:
@@ -77,10 +104,19 @@ def main():
               "local_chunk": 256, "context": 9216, "prompts": prompts,
               "warmups": [], "correctness_batches": [], "batches": [],
               "harness_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
+              "startup_timeout_s": a.startup_timeout,
               "timing_scope": "Client first SSE event, full response e2e and estimated TPOT=(e2e-first_event)/(output_tokens-1). SSE can coalesce tokens; this is not per-token kernel timing. Loading/warmup/correctness checks excluded."}
+    devices = ["cuda:0", "cuda:1"] if a.mode == "dp2" else ["cuda:1"]
+    if external:
+        devices = [None] * len(external["servers"])
+        for key in RESOURCE_FIELDS:
+            result.pop(key, None)
+        result.update(external["metadata"])
+        result["replicas"] = len(devices)
+        result["server_config"] = external
     started = time.perf_counter()
     try:
-        for endpoint, device in enumerate(["cuda:0", "cuda:1"] if a.mode == "dp2" else ["cuda:1"]):
+        for endpoint, device in enumerate(devices):
             with socket.socket() as sock:
                 sock.bind(("127.0.0.1", 0))
                 port = sock.getsockname()[1]
@@ -95,11 +131,13 @@ def main():
             if a.mode == "pd":
                 cmd += ["--pd-prefill-device", "cuda:0", "--pd-prefill-no-prefix-cache",
                         "--pd-prefill-chunk", "256"]
+            if external:
+                cmd = [str(port) if arg == "{port}" else arg for arg in external["servers"][endpoint]]
             proc = subprocess.Popen(cmd, stdout=log, stderr=subprocess.STDOUT, start_new_session=True,
                                     env=dict(os.environ, OMP_NUM_THREADS="4", MKL_NUM_THREADS="4"))
             procs.append(proc)
             ports.append(port)
-            end = time.monotonic() + 180
+            end = time.monotonic() + a.startup_timeout
             while True:
                 if proc.poll() is not None:
                     raise RuntimeError(f"server{endpoint} exited{proc.returncode}; see server log")