From 7b28974481529588d2e781a4f086d1afcd813465 Mon Sep 17 00:00:00 2001 From: gspark Date: Sat, 24 Oct 2026 04:44:49 +0900 Subject: [PATCH 1/3] exp: verify live RNGD GPU cancellation and producer recovery (#201) --- .../pd/http_matrix/.research/events.jsonl | 1 + opt_prime/examples/pd/http_matrix/STATUS.md | 2 + .../http_matrix/evidence201/llmd_readiness.py | 707 ++ .../evidence201/local201-qwen06b-faults.json | 66 + .../evidence201/local201-qwen06b-faults.log | 3 + .../local201-qwen06b-matrix-a-reference.json | 9757 +++++++++++++++++ .../evidence201/local201-qwen4b-faults.json | 66 + .../evidence201/local201-qwen4b-faults.log | 3 + .../local201-qwen4b-matrix-a-reference.json | 9757 +++++++++++++++++ .../2026-10-24-live-gpu-fault201.md | 33 + .../pd/http_matrix/live_gpu_faults.py | 192 + .../pd/serving_matrix/observer_factory.py | 37 + 12 files changed, 20624 insertions(+) create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/llmd_readiness.py create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-faults.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-faults.log create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-matrix-a-reference.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen4b-faults.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen4b-faults.log create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen4b-matrix-a-reference.json create mode 100644 opt_prime/examples/pd/http_matrix/experiments/2026-10-24-live-gpu-fault201.md create mode 100644 opt_prime/examples/pd/http_matrix/live_gpu_faults.py create mode 100644 opt_prime/examples/pd/serving_matrix/observer_factory.py diff --git a/opt_prime/examples/pd/http_matrix/.research/events.jsonl b/opt_prime/examples/pd/http_matrix/.research/events.jsonl index 2e372e17..beb6029d 100644 --- a/opt_prime/examples/pd/http_matrix/.research/events.jsonl +++ b/opt_prime/examples/pd/http_matrix/.research/events.jsonl @@ -5,3 +5,4 @@ {"id": "328b8349e9974a059ddc7e1a8a485e2c", "ts": "2026-10-23T21:32:23+09:00", "task": "Tokenizer EOS in batched serving (aicomp #229)", "result": "Real ATOM KONI HTTP also passes5/5 baseline/fixed/ignore/stream/reuse checks using same production factory: config-only length vs union stop. GPU+ATOM now10/10 expected outcomes; no SDK/artifact rebuild.", "next": "Update PR230 with NPU evidence; finish KONI medium-context RNGD handoff and broader PD work.", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {"backend": "atom", "kv_blocks": "1"}, "artifacts": [{"path": "analysis/serving-matrix-201/eos229-koni-atom-http.json", "size": 5199, "sha256": "4735255db379ff00"}], "experiments": ["experiments/2026-10-23-serving-eos229.md"], "labels": ["bug"], "review": true, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} {"id": "1020452aea8445f080108c1ed7f6d643", "ts": "2026-10-24T04:35:37+09:00", "task": "Dense hybrid GPU KV import (aicomp #258)", "result": "Dense hybrid receiver fix:31focused and505full CPU tests passed,34skipped,253subtests,151.69s. Preparing isolated PR and fresh Gemma live rerun; hardware and numerical acceptance not yet claimed.", "next": "Run corrected Gemma RNGD/GPU matrix preserving original failure; complete lifecycle and numerical checks", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [], "experiments": ["experiments/2026-10-24-dense-hybrid258.md"], "labels": ["bug"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} {"id": "551af4e77bca4e67bf0512304142ac48", "ts": "2026-10-24T04:41:11+09:00", "task": "Dense hybrid GPU KV import (aicomp #258)", "result": "Corrected live Gemma completed24batches;20/20token checks and60/60timed texts exact to nativeGPU,84remote/0local/0errors. Shape crash fixed on hardware. Qwen0.6 GPU lifecycle passed; Qwen4 next. Owned ATOM Pod confirmed absent.", "next": "Publish hardware evidence PR259; finish remaining GPU fault coverage and numerical boundary work", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [{"path": "analysis/serving-matrix-201/local258-gemma-summary.json", "size": 1288, "sha256": "3afbcc7e61bb3239"}], "experiments": ["experiments/2026-10-24-dense-hybrid258.md"], "labels": ["bug"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} +{"id": "c41ea1b2753146b091f85134832aa680", "ts": "2026-10-24T04:44:49+09:00", "task": "Four-model serving matrix (aicomp #201)", "result": "Prepared independent live GPU lifecycle validation branch on32f18ff with executed harness, native-call observer and completed Qwen0.6/Qwen4 raw evidence; KONI remains live. CLI help verified; production runtime unchanged.", "next": "Publish draft evidence PR and complete KONI/Gemma lifecycle runs", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [], "experiments": ["experiments/2026-10-24-live-gpu-fault201.md"], "labels": ["benchmark"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} diff --git a/opt_prime/examples/pd/http_matrix/STATUS.md b/opt_prime/examples/pd/http_matrix/STATUS.md index 76a8b514..05b568e6 100644 --- a/opt_prime/examples/pd/http_matrix/STATUS.md +++ b/opt_prime/examples/pd/http_matrix/STATUS.md @@ -41,3 +41,5 @@ Candidate exports airm_pd_producer_seconds_total{stage="prefill"|"transfer"} fro #258 dense hybrid KV: actual Gemma failure reproduced; mapping/group validation corrected.31focused and505full CPU tests pass; live rerun pending. #258 corrected live Gemma32ce5d5:24batches,20/20token checks,60/60timed texts exact;84remote/0local/0errors. Shape crash fixed in recorded hardware matrix; overall numerical/quality boundaries remain. + +Live GPU faults#201: Qwen0.6/Qwen4 cancellation, producer-exit503/readiness, SHM unlink and replacement exact-output checks pass. KONI/Gemma pending; long-output numerical failures remain. diff --git a/opt_prime/examples/pd/http_matrix/evidence201/llmd_readiness.py b/opt_prime/examples/pd/http_matrix/evidence201/llmd_readiness.py new file mode 100644 index 00000000..52910e21 --- /dev/null +++ b/opt_prime/examples/pd/http_matrix/evidence201/llmd_readiness.py @@ -0,0 +1,707 @@ +"""Optional local llm-d shared-storage control plane, separate from the engine. + +No KV bytes travel over HTTP. One serialized owner executes all device work; +run OwnerExecutor.run on the main thread for the checked RNGD PDMA backend. +""" + +from concurrent.futures import Future, TimeoutError as FutureTimeout +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +from io import BufferedIOBase +import queue +import select +import socket +import threading +import time +from opt_prime.accel.pd.contracts import RetryableError, SamplingParams +from opt_prime.accel.pd.runtime import PDRequest, PDRuntime, CancellationToken, RequestCancelled +from opt_prime.accel.serve.metrics import StageMetrics + + +class StageError(Exception): + """HTTP status + message. retryable: the request may be retried (here later, or on another replica).""" + def __init__(self, status, message, retryable=False): + super().__init__(message) + self.status, self.retryable = status, retryable + + +@dataclass +class Prepared: + request: PDRequest + result: object + created: float + spec: object + + def recover(self): + # host-side record of a fenced prefill; the source engine's own recovery is checked by PDStageService.reset + return True + + def prefill(self, request): + if ( + request.request_id != self.request.request_id + or request.input_ids != self.request.input_ids + ): + raise ValueError("prepared continuation belongs to another request") + return self.result + + +class CompletionAPI: + """OpenAI-style request parsing and responses shared by the stage services: token-ID / text prompts and chat + messages, sampling and stop fields, finish_reason and usage.""" + + def __init__(self, *, model, vocab_size, context, max_output, render, eos_token_ids=(), tokenizer=None): + if min(vocab_size, context, max_output) <= 0: + raise ValueError("positive service limits required") + self.model, self.vocab_size, self.context, self.max_output = model, vocab_size, context, max_output + self.render = render + self.eos_token_ids = tuple(eos_token_ids) + self.tokenizer = tokenizer # text prompts and chat messages (token-ID prompts need none) + self.metrics = StageMetrics(model) + + def _request(self, rid, body, chat=False): + allowed = { + "messages", + "stream_options", + "model", + "prompt", + "max_tokens", + "max_completion_tokens", + "temperature", + "top_p", + "top_k", + "seed", + "stop", + "ignore_eos", + "stream", + "cache_hit_threshold", + } + if not isinstance(rid, str) or not rid.strip() or len(rid) > 128: + raise StageError(400, "valid x-request-id required") + if ( + not isinstance(body, dict) + or set(body) - allowed + or body.get("model") != self.model + ): + raise StageError(400, "invalid model or unsupported request fields") + prompt = self._prompt(body, chat) + if ( + not isinstance(prompt, list) + or not prompt + or any(type(t) is not int or not 0 <= t < self.vocab_size for t in prompt) + ): + raise StageError(400, "one token-ID prompt required") + count = body.get("max_completion_tokens", body.get("max_tokens", 32)) + if ( + type(count) is not int + or not 1 <= count <= self.max_output + or len(prompt) + count > self.context + ): + raise StageError(400, "output or context capacity exceeded") + if ( + "max_tokens" in body + and "max_completion_tokens" in body + and body["max_tokens"] != body["max_completion_tokens"] + ): + raise StageError(400, "conflicting output limits") + if not isinstance(body.get("stream", False), bool): + raise StageError(400, "stream must be a boolean") + stop = body.get("stop", []) + stop = [stop] if isinstance(stop, str) else stop + if (not isinstance(stop, list) or len(stop) > 4 + or any(not isinstance(x, str) or not 0 < len(x) <= 64 for x in stop)): + raise StageError(400, "stop must be up to 4 non-empty strings") + ignore_eos = body.get("ignore_eos", False) + if not isinstance(ignore_eos, bool): + raise StageError(400, "ignore_eos must be a boolean") + temperature, seed = body.get("temperature", 0), body.get("seed") + if isinstance(temperature, bool) or isinstance(body.get("top_p", 1), bool): + raise StageError(400, "invalid sampling parameters") + try: + sampling = SamplingParams(float(temperature) if isinstance(temperature, (int, float)) else temperature, + body.get("top_p", 1.0), body.get("top_k", 0), seed, + () if ignore_eos else self.eos_token_ids) + except ValueError as error: + raise StageError(400, str(error)) + if not sampling.greedy and not all(getattr(e, "supports_sampling", False) for e in self._engines()): + raise StageError(400, "engine supports greedy decoding only") + if "cache_hit_threshold" in body and body["cache_hit_threshold"] != 0: + raise StageError(400, "unsupported cache threshold") + return PDRequest(rid, tuple(prompt), count, sampling), tuple(stop) + + def expire(self): + """Release a fenced prefill whose decode did not arrive within the TTL (owner thread only).""" + if self.state == "ready" and self.clock() - self.prepared.created > self.ttl: + old = self.prepared.request.request_id + self.prepared = None + self.state = "idle" + self._event("expired", old) + return old + return None + + def _prompt(self, body, chat): + """token IDs of the request: `messages` through the chat template (chat), a text `prompt` through the + tokenizer, or a token-ID `prompt` as is""" + if chat: + messages = body.get("messages") + if "prompt" in body or not isinstance(messages, list) or not messages or any( + not isinstance(m, dict) or not isinstance(m.get("role"), str) + or not isinstance(m.get("content"), str) for m in messages): + raise StageError(400, "chat requires messages with role and text content") + if self.tokenizer is None: + raise StageError(400, "this stage has no tokenizer for chat messages") + ids = self.tokenizer.apply_chat_template(messages, add_generation_prompt=True, tokenize=True) + ids = ids["input_ids"] if isinstance(ids, dict) or hasattr(ids, "keys") else ids + return [int(t) for t in ids] + if "messages" in body: + raise StageError(400, "messages belong to /v1/chat/completions") + prompt = body.get("prompt") + if isinstance(prompt, str): + if self.tokenizer is None: + raise StageError(400, "this stage has no tokenizer for text prompts") + return [int(t) for t in self.tokenizer(prompt, add_special_tokens=True)["input_ids"]] + return prompt + + def _finish(self, request, stop_strings, tokens): + """(tokens, text, finish_reason): cut at the first stop string (the text excludes it, the tokens end with the + token that completed it); 'stop' also when the last token is a stop token.""" + text = self.render(tokens) + stops = [s for s in stop_strings if s in text] + if stops: + for i in range(1, len(tokens) + 1): + partial = self.render(tokens[:i]) + hit = min((partial.find(s) for s in stops if s in partial), default=-1) + if hit >= 0: + return tokens[:i], partial[:hit], "stop" + stop_ids = request.sampling.stop_token_ids + return tokens, text, "stop" if tokens and tokens[-1] in stop_ids else "length" + + def _prefill_reply(self, rid, chat=False): + if chat: + return dict(id=rid, object="chat.completion", model=self.model, + choices=[dict(index=0, message=dict(role="assistant", content=""), finish_reason="stop")]) + return dict( + id=rid, + object="text_completion", + model=self.model, + choices=[dict(index=0, text="", finish_reason="stop")], + ) + + def _engines(self): + """the engines that must support non-greedy sampling for it to be accepted""" + return () + + def _response(self, rid, request, tokens, text, finish, chat): + usage = dict(prompt_tokens=len(request.input_ids), completion_tokens=len(tokens), + total_tokens=len(request.input_ids) + len(tokens)) + if chat: + choice = dict(index=0, message=dict(role="assistant", content=text), token_ids=tokens, finish_reason=finish) + else: + choice = dict(index=0, text=text, token_ids=tokens, finish_reason=finish) + return dict(id=rid, object="chat.completion" if chat else "text_completion", created=int(time.time()), + model=self.model, choices=[choice], usage=usage) + + +class PDStageService(CompletionAPI): + """Single prepared source slot. Cancel/TTL can release only fenced prefill.""" + + def __init__( + self, + prefill, + decode, + transport, + *, + model, + vocab_size, + context, + max_output, + render, + ttl=60, + clock=time.monotonic, + eos_token_ids=(), + tokenizer=None + ): + if prefill.spec != decode.spec: + raise ValueError("engine specifications differ") + if ttl <= 0: + raise ValueError("positive service limits required") + super().__init__(model=model, vocab_size=vocab_size, context=context, max_output=max_output, render=render, + eos_token_ids=eos_token_ids, tokenizer=tokenizer) + self.prefill, self.decode, self.transport = prefill, decode, transport + self.ttl, self.clock = ttl, clock + self.prepared = None + self.runtime = None + self.state = "idle" + self.events = [] + + def _engines(self): + return self.prefill, self.decode + + def gauges(self, waiting=0): + """(running, waiting, KV lane usage) for /metrics: one lane, occupied unless idle""" + running = 1 if self.state in ("prefilling", "decoding") else 0 + return running, waiting, 0.0 if self.state == "idle" else 1.0 + + def _event(self, stage, request_id): + self.events.append(dict(stage=stage, request_id=request_id, time=self.clock())) + if len(self.events) > 1024: + del self.events[:-1024] + + def cancel(self, rid): + if self.state != "ready" or self.prepared.request.request_id != rid: + raise StageError(409, "only a matching fenced prefill can be cancelled") + self.prepared = None + self.state = "idle" + self._event("cancelled", rid) + + def execute(self, stage, rid, body, cancellation=None, on_token=None, chat=False): + """on_token(token) streams output tokens on the owner thread (first token after the fenced prefill).""" + if stage not in ("prefill", "decode"): + raise StageError(400, "unknown stage") + request, stops = self._request(rid, body, chat) + if cancellation is not None and cancellation.cancelled: + raise StageError(499, "request cancelled before execution") + if self.state == "poisoned": + raise StageError(503, "owner quarantined; recovery required", retryable=True) + self.expire() + if self.prepared is not None and self.prepared.request.request_id == rid: + if self.prepared.request.input_ids != request.input_ids: + raise StageError(409, "request ID reused with a different prompt") + if stage == "prefill": + if self.state == "ready" and self.prepared.request.request_id == rid: + return self._prefill_reply(rid, chat) + if self.state != "idle": + raise StageError(429, "single cache lane occupied") + self.state = "prefilling" + prefill_fenced = False + started = self.clock() + try: + result = self.prefill.prefill(request) + prefill_fenced = True + self.prepared = Prepared( + request, result, self.clock(), self.prefill.spec + ) + result.validate(request) + if result.cache.spec != self.prefill.spec: + raise ValueError("prefill cache specification mismatch") + if cancellation is not None and cancellation.cancelled: + raise RequestCancelled(rid) + self.state = "ready" + self._event("prefill_ready", rid) + elapsed = self.clock() - started + self.metrics.observe("vllm:time_to_first_token_seconds", elapsed) # prefill emits the first token + self.metrics.observe("vllm:e2e_request_latency_seconds", elapsed) + self.metrics.finished("stop", len(request.input_ids), 1) + if on_token is not None: + for t in result.output_prefix: + on_token(t) + return self._prefill_reply(rid, chat) + except RequestCancelled: + if not prefill_fenced: + self.state = "poisoned" + raise + self.prepared = None + self.state = "idle" + self._event("cancelled", rid) + raise StageError(499, "prefill cancelled after completion") + except RetryableError as error: + self.prepared = None # the engine left nothing outstanding + self.state = "idle" + self.metrics.failed() + self._event("retryable_failure", rid) + raise StageError(503, f"retryable: {error}", retryable=True) + except BaseException: + self.state = "poisoned" + self.metrics.failed() + raise + if self.state != "ready" or self.prepared.request.request_id != rid: + raise StageError(409, "matching completed prefill required") + self.state = "decoding" + self._event("decode_start", rid) + started = self.clock() + try: + # PDRuntime reuses Prepared.prefill; the model is not prefetched again. + self.runtime = PDRuntime(self.prepared, self.decode, self.transport) + tokens = self.runtime.generate(request, cancellation=cancellation, on_token=on_token) + tokens, text, finish = self._finish(request, stops, tokens) + response = self._response(rid, request, tokens, text, finish, chat) + self.prepared = None + self.state = "idle" + self._event("decode_complete", rid) + t = self.runtime.last_timings + if "transfer_seconds" in t: + self.metrics.observe("airm_kv_transfer_seconds", t["transfer_seconds"]) + if len(tokens) > 1 and "decode_seconds" in t: + self.metrics.observe("vllm:time_per_output_token_seconds", t["decode_seconds"] / (len(tokens) - 1)) + self.metrics.observe("vllm:e2e_request_latency_seconds", self.clock() - started) + self.metrics.finished(finish, len(request.input_ids), len(tokens)) + return response + except RequestCancelled: + if self.runtime.state != "idle": + self.state = "poisoned" + raise + self.prepared = None + self.state = "idle" + self._event("cancelled", rid) + raise StageError(499, "decode cancelled after drain") + except RetryableError as error: + self.prepared = None # runtime released the leases; retry from prefill + self.state = "idle" + self.metrics.failed() + self._event("retryable_failure", rid) + raise StageError(503, f"retryable: {error}", retryable=True) + except BaseException: + self.state = "poisoned" + self.metrics.failed() + raise + + def reset(self): + """Recovery from quarantine: every endpoint must confirm a known-good device state (`recover() -> True`); + only then are the retained allocations released and the lane reopened. Owner thread only.""" + if self.state != "poisoned": + raise StageError(409, "owner is not quarantined") + for part in (self.prefill, self.decode, self.transport): + fn = getattr(part, "recover", None) + if fn is None: + raise StageError(501, f"{type(part).__name__} has no recovery; restart the stage") + if fn() is not True: + raise StageError(503, f"{type(part).__name__} not known-good; still quarantined") + if self.runtime is not None and not self.runtime.recover(): + raise StageError(503, "cache lane not recovered; still quarantined") + self.prepared = None + self.state = "idle" + self._event("recovered", None) + return dict(state=self.state) + + +class OwnerExecutor: + """HTTP threads enqueue only. Timeouts never cancel or release device work. + + While idle, the owner sweeps the TTL every `sweep` seconds, so an abandoned prepared prefill frees the lane + without waiting for the next request.""" + + def __init__(self, service, capacity=64, sweep=1.0): + self.service, self.sweep = service, sweep + self.queue = queue.Queue(capacity) + self.started = False + self.accepting = True + self._submit_lock = threading.Lock() + self._pending = {} + + def submit(self, operation, *args, **kwargs): + return self.submit_async(operation, *args, **kwargs).result() + + def submit_async(self, operation, *args, **kwargs): + future = Future() + token = CancellationToken() if operation == "execute" else None + rid = args[1] if token is not None else None + if token is not None and ( + not isinstance(rid, str) or not rid.strip() or len(rid) > 128 + ): + raise StageError(400, "valid x-request-id required") + with self._submit_lock: + if not self.started or not self.accepting: + raise StageError(503, "owner not accepting work") + try: + self.queue.put_nowait((operation, args, kwargs, future, token)) + if token is not None: + self._pending.setdefault(rid, []).append(token) + except queue.Full: + raise StageError(429, "owner command queue full") + if operation == "stop": + self.accepting = False + return future + + def cancel_request(self, rid): + if not isinstance(rid, str) or not rid.strip() or len(rid) > 128: + raise StageError(400, "valid x-request-id required") + with self._submit_lock: + if not self.started or not self.accepting: + raise StageError(503, "owner not accepting work") + tokens = self._pending.get(rid, ()) + if tokens: + for token in tokens: + token.cancel() + return dict(id=rid, cancellation_requested=True, drained=False) + # No active/queued call: only the owner may release a prepared cache. + self.submit("cancel", rid) + return dict(id=rid, cancellation_requested=True, drained=True) + + def run(self): + self.started = True + while True: + try: + operation, args, kwargs, future, token = self.queue.get(timeout=self.sweep) + except queue.Empty: + self.service.expire() + continue + error = None + stopping = operation == "stop" + try: + if stopping: + if self.service.state not in ("idle", "ready"): + raise StageError(409, "owner not quiescent") + if self.service.state == "ready": + self.service.cancel(self.service.prepared.request.request_id) + self.started = False + result = None + elif operation == "execute": + result = self.service.execute(*args, cancellation=token, **kwargs) + elif operation == "cancel": + result = self.service.cancel(*args) + elif operation == "reset": + result = self.service.reset() + else: + raise StageError(400, "unknown owner operation") + except BaseException as exc: + error = exc + # Complete pending-token bookkeeping atomically with cancellation. + # A cancel accepted after execute() returns must still drain a ready + # prefill; otherwise the client sees 202 but the slot remains occupied. + with self._submit_lock: + if token is not None: + try: + if error is None and token.cancelled: + if (self.service.state == "ready" and + self.service.prepared.request.request_id == args[1]): + self.service.cancel(args[1]) + error = StageError(499, "request cancelled after completion") + except BaseException as exc: + error = exc + finally: + pending = self._pending[args[1]] + pending.remove(token) + if not pending: + del self._pending[args[1]] + if error is not None: + if stopping: + self.accepting = True # the stop was refused: reset and a retried stop must still get in + future.set_exception(error) + else: + future.set_result(result) + if stopping: + return + + +POLL = 0.5 # seconds between client-disconnect checks of a waiting handler (results wake it at once) +_DONE = object() # end-of-stream marker in a handler's token queue + +class _NoSignalSocketWriter(BufferedIOBase): + """Unbuffered HTTP writes that retain EPIPE without invoking native SIGPIPE handlers.""" + def __init__(self, sock): + self._sock = sock + + def writable(self): + return True + + def write(self, data): + self._sock.sendall(data, getattr(socket, "MSG_NOSIGNAL", 0)) + with memoryview(data) as view: + return view.nbytes + + def fileno(self): + return self._sock.fileno() + + +def serve_stage(owner, stage, port): + if stage not in ("prefill", "decode"): + raise ValueError("invalid stage") + + class Handler(BaseHTTPRequestHandler): + def setup(self): + super().setup() + self.wfile = _NoSignalSocketWriter(self.connection) + + def log_message(self, *args): + pass + + def reply(self, status, payload, retry_after=None): + data = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + if retry_after is not None: + self.send_header("Retry-After", str(retry_after)) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + try: + self.wfile.write(data) + except (BrokenPipeError, ConnectionResetError): + pass # The completed owner operation is not cancelled by disconnect. + + def client_gone(self): + try: + readable, _, _ = select.select([self.connection], [], [], 0) + return bool(readable) and self.connection.recv(1, socket.MSG_PEEK) == b"" + except OSError: + return True + + def do_GET(self): + if self.path == "/metrics": + data = owner.service.metrics.render(*owner.service.gauges(owner.queue.qsize())).encode() + self.send_response(200) + self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + return + if self.path not in ("/ready", "/health"): + self.reply(404, dict(error="not found")) + return + ready = (owner.started and getattr(owner, "accepting", True) + and owner.service.state != "poisoned" and getattr(owner.service, "ready", True)) + self.reply(200 if ready else 503, dict(ready=ready, stage=stage)) + + def cancel_if_gone(self, rid, state): + # A closed client connection is a cancel request through the same fenced path. + if not state["cancelled"] and self.client_gone(): + state["cancelled"] = True + try: + owner.cancel_request(rid) + except StageError: + pass + + def do_POST(self): + if self.path not in ("/v1/completions", "/v1/chat/completions", "/v1/cancel", "/v1/admin/reset"): + self.reply(404, dict(error="not found")) + return + try: + if self.path == "/v1/admin/reset": + self.reply(200, owner.submit("reset")) + return + if self.path == "/v1/cancel": + result = owner.cancel_request(self.headers.get("x-request-id")) + self.reply(200 if result["drained"] else 202, result) + return + if not owner.started: + raise StageError(503, "owner not running") + length = int(self.headers.get("Content-Length", "0")) + if not 0 < length <= 1048576: + raise StageError(413, "invalid body size") + body = json.loads(self.rfile.read(length)) + rid = self.headers.get("x-request-id") + chat = self.path == "/v1/chat/completions" + if isinstance(body, dict) and body.get("stream") is True: + self.stream(rid, body, chat) + return + future = owner.submit_async("execute", stage, rid, body, chat=chat) + state = dict(cancelled=False) + while True: + try: + response = future.result(timeout=POLL) # returns at completion; POLL checks the client + break + except FutureTimeout: + self.cancel_if_gone(rid, state) + self.reply(200, response) + except StageError as error: + self.reply(error.status, dict(error=str(error), retryable=error.retryable), + retry_after=1 if error.retryable else None) + except (ValueError, TypeError): + failed = owner.service.state == "poisoned" + self.reply( + 500 if failed else 400, + dict(error="engine failed" if failed else "invalid request", retryable=False), + ) + except Exception: + self.reply(500, dict(error="engine failed; owner retained", retryable=False)) + + def stream(self, rid, body, chat): + """OpenAI-compatible SSE: a chunk per token as the owner produces it (the first right after the fenced + prefill), text held back while it could still be the start of a stop string, then a chunk with + finish_reason, a usage chunk and [DONE]. Errors before the first chunk are plain JSON replies.""" + tokens = queue.Queue() + future = owner.submit_async("execute", stage, rid, body, chat=chat, on_token=tokens.put) + stops = body.get("stop") or [] + stops = [stops] if isinstance(stops, str) else stops + hold = max((len(x) for x in stops), default=1) - 1 + render, created = owner.service.render, int(time.time()) + kind = "chat.completion.chunk" if chat else "text_completion" + state, out, sent, started = dict(cancelled=False), [], "", False + + def chunk(delta=None, finish=None, usage=None): + if chat: + choice = dict(index=0, delta={} if delta is None else delta, finish_reason=finish) + else: + choice = dict(index=0, text="" if delta is None else delta, finish_reason=finish) + payload = dict(id=rid, object=kind, created=created, model=owner.service.model, + choices=[] if usage else [choice]) + if usage: + payload["usage"] = usage + return payload + + def send(payload): + nonlocal started + if not started: + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + started = True + if chat: + send(chunk(dict(role="assistant", content=""))) + data = payload if isinstance(payload, str) else json.dumps(payload) + try: + self.wfile.write(f"data: {data}\n\n".encode()) + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError): + self.cancel_if_gone(rid, state) + + def text_delta(final=None): + nonlocal sent + text = render(out) if final is None else final + if final is None: + cut = min((text.find(x) for x in stops if x in text), default=-1) + text = text[:cut] if cut >= 0 else text[:max(len(sent), len(text) - hold)] + if not text.startswith(sent) or len(text) == len(sent): + return # nothing new, or a multi-byte character still open + delta, sent = text[len(sent):], text + send(chunk(dict(content=delta) if chat else delta)) + + response, error = None, None + future.add_done_callback(lambda _f: tokens.put(_DONE)) # tokens are queued before completion + while True: + try: + item = tokens.get(timeout=POLL) # wakes on a token; POLL only checks the client + except queue.Empty: + self.cancel_if_gone(rid, state) + continue + done = item is _DONE + if not done: + out.append(item) + while True: # drain: one detokenization per wake-up + try: + item = tokens.get_nowait() + except queue.Empty: + break + if item is _DONE: + done = True + else: + out.append(item) + text_delta() + if done: + try: + response = future.result() + except BaseException as exc: + error = exc + break + if error is not None: + if started: + send(dict(error=str(error), status=getattr(error, "status", 500))) + send("[DONE]") + return + raise error + choice = response["choices"][0] + if "usage" in response: # decode stage: the final, stop-cut text + text_delta(choice["message"]["content"] if chat else choice["text"]) + else: # prefill stage: its token(s), cut at a stop string + text = render(out) + cut = min((text.find(x) for x in stops if x in text), default=-1) + text_delta(text[:cut] if cut >= 0 else text) + send(chunk(finish=choice["finish_reason"])) + if "usage" in response: + send(chunk(usage=response["usage"])) + send("[DONE]") + + server = ThreadingHTTPServer(("127.0.0.1", port), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, thread diff --git a/opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-faults.json b/opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-faults.json new file mode 100644 index 00000000..8b30cff5 --- /dev/null +++ b/opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-faults.json @@ -0,0 +1,66 @@ +{ + "passed": true, + "lifecycle_passed": true, + "stage": "complete", + "model": "/home/gspark/.cache/huggingface/hub/models--Qwen--Qwen3-0.6B/snapshots/c1899de289a04d12100db370d81485cdf75e47ca", + "faults": { + "cancel": { + "entry": { + "sequence": 2, + "phase": "start", + "tokens": 8192, + "time": 296514.717422046 + }, + "status": 499, + "reuse_exact": true + }, + "worker_exit": { + "entry": { + "sequence": 4, + "phase": "start", + "tokens": 8192, + "time": 296518.273910603 + }, + "pending_status": 500, + "later_status": 503, + "readiness": 503, + "slab_unlinked": true, + "replacement_exact": true + } + }, + "control_tokens": [ + 9625, + 198, + 1986, + 374, + 4004, + 2266, + 13, + 576, + 2518, + 3745, + 5610, + 40676, + 13, + 576, + 6303, + 3745, + 5610, + 6467, + 624, + 1986, + 374, + 4004, + 2266, + 13, + 576, + 2518, + 3745, + 5610, + 40676, + 13, + 576, + 6303 + ], + "native_tokens_equal": true +} diff --git a/opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-faults.log b/opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-faults.log new file mode 100644 index 00000000..586863ba --- /dev/null +++ b/opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-faults.log @@ -0,0 +1,3 @@ + +Loading weights: 0%| | 0/311 [00:00 analysis/serving-matrix-201/local258-faults-queue.log 2>&1 +``` +This sequentially runs check_gpu201_faults.py; exact per-model argv retained in queue log. Environment/runtime paths are recorded in the launcher. + +## Results +Qwen3-0.6B and Qwen3-4B completed: lifecycle_passed=true,passed=true,native_tokens_equal=true. Cancel during actual prefill returns499 and then identical recovery tokens. Producer termination yields pending500, later admission503,/ready503; old shared-memory slab is unlinked. Replacement makes readiness200 and returns identical32tokens. Both raw local201-{qwen06b,qwen4b}-faults.json retained. +KONI now loading; Gemma pending. No total-four-model claim yet. + +## Conclusion +Two models pass cancellation/drain/death/readiness/replacement checks. Qwen4 long-output numerical failure remains independently failed. This suite does not test another already-decoding GPU request surviving producer death (covered only by separate ATOM-prefetch suite). Performance is not measured by these fault tests. + +## Portable invocation +```bash +export PYTHONPATH=opt_prime:opt_prime/examples/pd/serving_matrix:/path/to/research_producer_factories:/path/to/rngd_research_src +export OMP_NUM_THREADS=4 MKL_NUM_THREADS=4 HF_HUB_OFFLINE=1 TRANSFORMERS_OFFLINE=1 +python opt_prime/examples/pd/http_matrix/live_gpu_faults.py --model /path/to/model --factory long_width_variant:make_prefill --worker-python /path/to/furiosa-worker-python --reference opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-matrix-a-reference.json --http-module opt_prime/examples/pd/http_matrix/evidence201/llmd_readiness.py --markers /tmp/fault-markers --output /tmp/fault-result.json +``` +Worker interpreter must expose Furiosa Torch2026.4, research producer modules/factories from the live PD work (#235/#245), and the bundled observer_factory. Its model/factory must match the supplied native reference. The proprietary SDK and model weights are not bundled. Evidence contains completed first two models only at this stage. Long-output numerical failures remain separate. Markers neither modify KV nor impose an artificial execution gate. diff --git a/opt_prime/examples/pd/http_matrix/live_gpu_faults.py b/opt_prime/examples/pd/http_matrix/live_gpu_faults.py new file mode 100644 index 00000000..8882fb39 --- /dev/null +++ b/opt_prime/examples/pd/http_matrix/live_gpu_faults.py @@ -0,0 +1,192 @@ +"""Live RNGD/GPU HTTP lifecycle test; run separately from performance measurements.""" +import argparse +import importlib.util +import json +import os +import sys +import threading +import time +import traceback +import urllib.error +import urllib.request +from concurrent.futures import ThreadPoolExecutor +from multiprocessing import shared_memory +from pathlib import Path + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--model', required=True) + parser.add_argument('--factory', required=True) + parser.add_argument('--worker-python', required=True) + parser.add_argument('--reference', type=Path, required=True) + parser.add_argument('--http-module', required=True) + parser.add_argument('--markers', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + args = parser.parse_args() + if args.output.exists(): + raise FileExistsError(args.output) + spec = importlib.util.spec_from_file_location('opt_prime.accel.serve.llmd', args.http_module) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + import torch + from transformers import AutoModelForCausalLM, AutoTokenizer + from opt_prime.accel.pd.pd_batching import PDBatcher, SubprocessPrefillWorker + from opt_prime.accel.serve.batched import BatchedOwner, BatchedService + from opt_prime.accel.serve.llmd import serve_stage + + reference = json.loads(args.reference.read_text()) + result = dict(passed=False, lifecycle_passed=False, stage='loading', model=args.model, faults={}) + worker = owner = server = http = client = None + errors = [] + + def save(): + args.output.write_text(json.dumps(result, indent=2)+'\n') + + def wait_for(predicate, label, timeout=240): + end = time.monotonic()+timeout + while time.monotonic() < end: + value = predicate() + if value: + return value + time.sleep(.002) + raise TimeoutError(label) + + save() + try: + model = AutoModelForCausalLM.from_pretrained(args.model, dtype=torch.bfloat16, + local_files_only=True).to('cuda:0').eval() + torch.cuda.synchronize() + c = model.config.get_text_config() + tokenizer = AutoTokenizer.from_pretrained(args.model, local_files_only=True) + + def new_worker(): + return SubprocessPrefillWorker(args.worker_python, + ['--model', args.model, '--width', '8192', '--engine-factory', 'observer_factory:make_prefill'], + slots=4, 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=9216, env=dict(os.environ, PREFILL_OBSERVER_DIR=str(args.markers.resolve()), + PREFILL_OBSERVER_FACTORY=args.factory)) + + worker = new_worker() + batcher = PDBatcher(model, worker, slots=4, max_context=9216, kv='paged', page_size=64, prefix_cache=False) + service = BatchedService(batcher, model=args.model, vocab_size=c.vocab_size, max_output=32, + tokenizer=tokenizer, render=lambda ids: tokenizer.decode(ids, skip_special_tokens=True)) + owner = BatchedOwner(service) + server, http = serve_stage(owner, 'decode', 0) + result['stage'] = 'running' + save() + + def request(path, body, rid): + req = urllib.request.Request(f'http://127.0.0.1:{server.server_port}{path}', + data=None if body is None else json.dumps(body).encode(), headers={'x-request-id': rid}) + try: + with urllib.request.urlopen(req, timeout=240) as response: + return response.status, json.load(response) + except urllib.error.HTTPError as error: + return error.code, error.read().decode() + + def generate(rid, n=8192): + return request('/v1/completions', dict(model=args.model, prompt=reference['prompts'][str(n)], + temperature=0, max_tokens=32, ignore_eos=True, stream=False), rid) + + control_tokens = [] + + def check_output(rid): + status, response = generate(rid, 128) + assert status == 200, (status, response) + tokens = response['choices'][0]['token_ids'] + assert len(tokens) == 32 + if not control_tokens: + control_tokens.extend(tokens) + result['control_tokens'] = tokens + result['native_tokens_equal'] = tokens == reference['references']['128:32']['tokens'] + else: + assert tokens == control_tokens, 'fault changed output relative to the same PD route' + + def entered(rid): + path = args.markers/f'{worker._proc.pid}.json' + if not path.exists(): + return False + event = json.loads(path.read_text()) + with batcher._lock: + state = batcher.by_id.get(rid) + pending = state is not None and state.pending is not None and not state.pending.done() + return event if pending and event['phase'] == 'start' else False + + def exercise(): + nonlocal worker + try: + wait_for(lambda: owner.started, 'owner startup') + check_output('before-faults') + with ThreadPoolExecutor(max_workers=2) as clients: + future = clients.submit(generate, 'pending-cancel') + event = wait_for(lambda: entered('pending-cancel'), 'actual producer entry before cancel') + status, _ = request('/v1/cancel', {}, 'pending-cancel') + assert status in [200, 202], status + assert future.result(timeout=240)[0] == 499 + wait_for(lambda: batcher.idle, 'cancel drain') + check_output('after-cancel') + result['faults']['cancel'] = dict(entry=event, status=499, reuse_exact=True) + save() + future = clients.submit(generate, 'pending-death') + event = wait_for(lambda: entered('pending-death'), 'actual producer entry before death') + worker._proc.terminate() # Only this harness-owned producer. + status, _ = future.result(timeout=240) + assert status in [500, 503], status + wait_for(lambda: batcher.idle, 'failed producer drain') + assert generate('after-death', 128)[0] == 503 + assert request('/ready', None, 'readiness-dead')[0] == 503 + old_slab = worker._shm.name + worker.close() + try: + leaked = shared_memory.SharedMemory(name=old_slab) + except FileNotFoundError: + pass + else: + leaked.close() + raise AssertionError('worker slab still linked') + worker = new_worker() + with batcher._lock: + assert not (batcher.waiting or batcher.prefilling or batcher.decoding or batcher.by_id) + assert sorted(batcher.free) == [0, 1, 2, 3] + batcher.worker = worker + assert request('/ready', None, 'readiness-replaced')[0] == 200 + check_output('after-replacement') + result['faults']['worker_exit'] = dict(entry=event, pending_status=status, + later_status=503, readiness=503, slab_unlinked=True, replacement_exact=True) + result.update(lifecycle_passed=True, passed=result['native_tokens_equal'], stage='complete') + except BaseException: + errors.append(traceback.format_exc()) + result.update(stage='failed', error=errors[-1]) + finally: + if errors: + worker.close() # Settle futures before stopping/draining owner. + save() + owner.submit('stop') + + client = threading.Thread(target=exercise, daemon=True) + client.start() + owner.run() + client.join(10) + assert not client.is_alive() + if errors: + raise RuntimeError(errors[0]) + if not result['native_tokens_equal']: + raise AssertionError('lifecycle passed, but native GPU token comparison failed; see recorded control') + except BaseException: + result.update(stage='failed', error=traceback.format_exc()) + save() + raise + finally: + if server: + server.shutdown() + server.server_close() + http.join(5) + if worker: + worker.close() + + +if __name__ == '__main__': + main() diff --git a/opt_prime/examples/pd/serving_matrix/observer_factory.py b/opt_prime/examples/pd/serving_matrix/observer_factory.py new file mode 100644 index 00000000..30d47b46 --- /dev/null +++ b/opt_prime/examples/pd/serving_matrix/observer_factory.py @@ -0,0 +1,37 @@ +"""Test-only native-call entry/completion markers; no execution gate or KV replay.""" +import importlib +import json +import os +import time +from pathlib import Path + + +def make_prefill(config, weights, width, chip=0): + module, name = os.environ.get( + "PREFILL_OBSERVER_FACTORY", "long_width_variant:make_prefill" + ).split(":") + engine = getattr(importlib.import_module(module), name)(config, weights, width, chip) + directory = Path(os.environ["PREFILL_OBSERVER_DIR"]) + directory.mkdir(parents=True, exist_ok=True) + marker = directory / f"{os.getpid()}.json" + original = engine.prefill + sequence = 0 + + def prefill(tokens): + nonlocal sequence + sequence += 1 + + def record(phase): + temporary = marker.with_suffix(".tmp") + temporary.write_text(json.dumps({"sequence": sequence, "phase": phase, + "tokens": len(tokens), "time": time.monotonic()})) + temporary.replace(marker) + + record("start") + try: + return original(tokens) + finally: + record("done") + + engine.prefill = prefill + return engine From f2de635f221b260825d4a8c438af89237a64e05c Mon Sep 17 00:00:00 2001 From: gspark Date: Sat, 24 Oct 2026 04:48:27 +0900 Subject: [PATCH 2/3] exp: complete all-four-model GPU producer recovery evidence (#201) --- .../pd/http_matrix/.research/events.jsonl | 1 + opt_prime/examples/pd/http_matrix/STATUS.md | 2 + .../evidence201/local201-gemma-faults.json | 66 + .../evidence201/local201-gemma-faults.log | 3 + .../local201-gemma-matrix-a-reference.json | 9757 +++++++++++++++++ .../evidence201/local201-koni-faults.json | 66 + .../evidence201/local201-koni-faults.log | 16 + .../local201-koni-matrix-a-reference.json | 9757 +++++++++++++++++ .../2026-10-24-live-gpu-fault201.md | 6 +- 9 files changed, 19671 insertions(+), 3 deletions(-) create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-faults.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-faults.log create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-matrix-a-reference.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-koni-faults.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-koni-faults.log create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-koni-matrix-a-reference.json diff --git a/opt_prime/examples/pd/http_matrix/.research/events.jsonl b/opt_prime/examples/pd/http_matrix/.research/events.jsonl index beb6029d..a4d98359 100644 --- a/opt_prime/examples/pd/http_matrix/.research/events.jsonl +++ b/opt_prime/examples/pd/http_matrix/.research/events.jsonl @@ -6,3 +6,4 @@ {"id": "1020452aea8445f080108c1ed7f6d643", "ts": "2026-10-24T04:35:37+09:00", "task": "Dense hybrid GPU KV import (aicomp #258)", "result": "Dense hybrid receiver fix:31focused and505full CPU tests passed,34skipped,253subtests,151.69s. Preparing isolated PR and fresh Gemma live rerun; hardware and numerical acceptance not yet claimed.", "next": "Run corrected Gemma RNGD/GPU matrix preserving original failure; complete lifecycle and numerical checks", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [], "experiments": ["experiments/2026-10-24-dense-hybrid258.md"], "labels": ["bug"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} {"id": "551af4e77bca4e67bf0512304142ac48", "ts": "2026-10-24T04:41:11+09:00", "task": "Dense hybrid GPU KV import (aicomp #258)", "result": "Corrected live Gemma completed24batches;20/20token checks and60/60timed texts exact to nativeGPU,84remote/0local/0errors. Shape crash fixed on hardware. Qwen0.6 GPU lifecycle passed; Qwen4 next. Owned ATOM Pod confirmed absent.", "next": "Publish hardware evidence PR259; finish remaining GPU fault coverage and numerical boundary work", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [{"path": "analysis/serving-matrix-201/local258-gemma-summary.json", "size": 1288, "sha256": "3afbcc7e61bb3239"}], "experiments": ["experiments/2026-10-24-dense-hybrid258.md"], "labels": ["bug"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} {"id": "c41ea1b2753146b091f85134832aa680", "ts": "2026-10-24T04:44:49+09:00", "task": "Four-model serving matrix (aicomp #201)", "result": "Prepared independent live GPU lifecycle validation branch on32f18ff with executed harness, native-call observer and completed Qwen0.6/Qwen4 raw evidence; KONI remains live. CLI help verified; production runtime unchanged.", "next": "Publish draft evidence PR and complete KONI/Gemma lifecycle runs", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [], "experiments": ["experiments/2026-10-24-live-gpu-fault201.md"], "labels": ["benchmark"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} +{"id": "2919d0420ffc49ceacc9aa8982bf4859", "ts": "2026-10-24T04:48:11+09:00", "task": "Four-model serving matrix (aicomp #201)", "result": "All4 live GPU lifecycle runs completed passed=true, including KONI/Gemma: cancellation/reuse, producer death503/readiness, slab unlink and replacement exact32-token controls. Started separate512-token concurrent-decode isolation suite after prior queue terminal and device-idle checks.", "next": "Publish remaining lifecycle evidence; verify currently decoding requests survive producer failure across4models", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [{"path": "analysis/serving-matrix-201/local201-koni-faults.json", "size": 1003, "sha256": "5ef1011fb2c2a4c6"}, {"path": "analysis/serving-matrix-201/local201-gemma-faults.json", "size": 1016, "sha256": "e707fb1f52adbb1a"}], "experiments": ["experiments/2026-10-24-live-gpu-fault201.md"], "labels": ["benchmark"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} diff --git a/opt_prime/examples/pd/http_matrix/STATUS.md b/opt_prime/examples/pd/http_matrix/STATUS.md index 05b568e6..87e17355 100644 --- a/opt_prime/examples/pd/http_matrix/STATUS.md +++ b/opt_prime/examples/pd/http_matrix/STATUS.md @@ -43,3 +43,5 @@ Candidate exports airm_pd_producer_seconds_total{stage="prefill"|"transfer"} fro #258 corrected live Gemma32ce5d5:24batches,20/20token checks,60/60timed texts exact;84remote/0local/0errors. Shape crash fixed in recorded hardware matrix; overall numerical/quality boundaries remain. Live GPU faults#201: Qwen0.6/Qwen4 cancellation, producer-exit503/readiness, SHM unlink and replacement exact-output checks pass. KONI/Gemma pending; long-output numerical failures remain. + +KONI/Gemma GPU lifecycle completed; allfour pass cancellation/reuse, producer death/readiness503, slab unlink and producer replacement. Concurrent-decode isolation is being tested separately and not claimed here. diff --git a/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-faults.json b/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-faults.json new file mode 100644 index 00000000..f0010084 --- /dev/null +++ b/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-faults.json @@ -0,0 +1,66 @@ +{ + "passed": true, + "lifecycle_passed": true, + "stage": "complete", + "model": "/tmp/serving-matrix-models/Gemma3-1B", + "faults": { + "cancel": { + "entry": { + "sequence": 2, + "phase": "start", + "tokens": 8192, + "time": 296913.945863423 + }, + "status": 499, + "reuse_exact": true + }, + "worker_exit": { + "entry": { + "sequence": 4, + "phase": "start", + "tokens": 8192, + "time": 296917.991428454 + }, + "pending_status": 500, + "later_status": 503, + "readiness": 503, + "slab_unlinked": true, + "replacement_exact": true + } + }, + "control_tokens": [ + 9079, + 107, + 106, + 107, + 106, + 130614, + 107, + 106, + 107, + 106, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695 + ], + "native_tokens_equal": true +} diff --git a/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-faults.log b/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-faults.log new file mode 100644 index 00000000..39c07150 --- /dev/null +++ b/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-faults.log @@ -0,0 +1,3 @@ + +Loading weights: 0%| | 0/340 [00:00 Date: Sat, 24 Oct 2026 04:58:19 +0900 Subject: [PATCH 3/3] test: verify active GPU decode survives producer death on four models (#201) --- .../pd/http_matrix/.research/events.jsonl | 1 + opt_prime/examples/pd/http_matrix/STATUS.md | 2 + .../evidence201/local201-gemma-isolation.json | 584 ++++++++++++++++++ .../evidence201/local201-gemma-isolation.log | 3 + .../evidence201/local201-koni-isolation.json | 584 ++++++++++++++++++ .../evidence201/local201-koni-isolation.log | 16 + .../local201-qwen06b-isolation.json | 584 ++++++++++++++++++ .../local201-qwen06b-isolation.log | 3 + .../local201-qwen4b-isolation.json | 584 ++++++++++++++++++ .../evidence201/local201-qwen4b-isolation.log | 3 + .../2026-10-24-live-gpu-fault201.md | 4 + .../pd/http_matrix/live_gpu_faults.py | 27 +- 12 files changed, 2391 insertions(+), 4 deletions(-) create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-isolation.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-isolation.log create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-koni-isolation.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-koni-isolation.log create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-isolation.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen06b-isolation.log create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen4b-isolation.json create mode 100644 opt_prime/examples/pd/http_matrix/evidence201/local201-qwen4b-isolation.log diff --git a/opt_prime/examples/pd/http_matrix/.research/events.jsonl b/opt_prime/examples/pd/http_matrix/.research/events.jsonl index a4d98359..f883ccc0 100644 --- a/opt_prime/examples/pd/http_matrix/.research/events.jsonl +++ b/opt_prime/examples/pd/http_matrix/.research/events.jsonl @@ -7,3 +7,4 @@ {"id": "551af4e77bca4e67bf0512304142ac48", "ts": "2026-10-24T04:41:11+09:00", "task": "Dense hybrid GPU KV import (aicomp #258)", "result": "Corrected live Gemma completed24batches;20/20token checks and60/60timed texts exact to nativeGPU,84remote/0local/0errors. Shape crash fixed on hardware. Qwen0.6 GPU lifecycle passed; Qwen4 next. Owned ATOM Pod confirmed absent.", "next": "Publish hardware evidence PR259; finish remaining GPU fault coverage and numerical boundary work", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [{"path": "analysis/serving-matrix-201/local258-gemma-summary.json", "size": 1288, "sha256": "3afbcc7e61bb3239"}], "experiments": ["experiments/2026-10-24-dense-hybrid258.md"], "labels": ["bug"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} {"id": "c41ea1b2753146b091f85134832aa680", "ts": "2026-10-24T04:44:49+09:00", "task": "Four-model serving matrix (aicomp #201)", "result": "Prepared independent live GPU lifecycle validation branch on32f18ff with executed harness, native-call observer and completed Qwen0.6/Qwen4 raw evidence; KONI remains live. CLI help verified; production runtime unchanged.", "next": "Publish draft evidence PR and complete KONI/Gemma lifecycle runs", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [], "experiments": ["experiments/2026-10-24-live-gpu-fault201.md"], "labels": ["benchmark"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} {"id": "2919d0420ffc49ceacc9aa8982bf4859", "ts": "2026-10-24T04:48:11+09:00", "task": "Four-model serving matrix (aicomp #201)", "result": "All4 live GPU lifecycle runs completed passed=true, including KONI/Gemma: cancellation/reuse, producer death503/readiness, slab unlink and replacement exact32-token controls. Started separate512-token concurrent-decode isolation suite after prior queue terminal and device-idle checks.", "next": "Publish remaining lifecycle evidence; verify currently decoding requests survive producer failure across4models", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [{"path": "analysis/serving-matrix-201/local201-koni-faults.json", "size": 1003, "sha256": "5ef1011fb2c2a4c6"}, {"path": "analysis/serving-matrix-201/local201-gemma-faults.json", "size": 1016, "sha256": "e707fb1f52adbb1a"}], "experiments": ["experiments/2026-10-24-live-gpu-fault201.md"], "labels": ["benchmark"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} +{"id": "eb53d939aca44b85ab4b6fd602754651", "ts": "2026-10-24T04:58:02+09:00", "task": "Four-model serving matrix (aicomp #201)", "result": "All4 GPU concurrent-decode isolation tests passed512exact tokens after producer killed at token2, plus cancellation/503/slab-unlink/replacement. KONI paddedGPU control completed: changes output but does NOT reproduce short RNGD output; padding-only explanation rejected.", "next": "Publish all4isolation evidence; capture short KONI KV for identical-prefix numerical analysis", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {}, "artifacts": [{"path": "analysis/serving-matrix-201/koni-short-padding-diagnostic.json", "size": 5903, "sha256": "f67039817a78849a"}], "experiments": ["experiments/2026-10-24-live-gpu-fault201.md"], "labels": ["benchmark"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"} diff --git a/opt_prime/examples/pd/http_matrix/STATUS.md b/opt_prime/examples/pd/http_matrix/STATUS.md index 87e17355..4664d251 100644 --- a/opt_prime/examples/pd/http_matrix/STATUS.md +++ b/opt_prime/examples/pd/http_matrix/STATUS.md @@ -45,3 +45,5 @@ Candidate exports airm_pd_producer_seconds_total{stage="prefill"|"transfer"} fro Live GPU faults#201: Qwen0.6/Qwen4 cancellation, producer-exit503/readiness, SHM unlink and replacement exact-output checks pass. KONI/Gemma pending; long-output numerical failures remain. KONI/Gemma GPU lifecycle completed; allfour pass cancellation/reuse, producer death/readiness503, slab unlink and producer replacement. Concurrent-decode isolation is being tested separately and not claimed here. + +Allfour live GPU isolation runs pass: active512-token decode survives other-request producer death with exact output; all baseline lifecycle gates retained. diff --git a/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-isolation.json b/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-isolation.json new file mode 100644 index 00000000..f2eed53f --- /dev/null +++ b/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-isolation.json @@ -0,0 +1,584 @@ +{ + "passed": true, + "lifecycle_passed": true, + "stage": "complete", + "model": "/tmp/serving-matrix-models/Gemma3-1B", + "faults": { + "cancel": { + "entry": { + "sequence": 3, + "phase": "start", + "tokens": 8192, + "time": 297463.302226823 + }, + "status": 499, + "reuse_exact": true + }, + "worker_exit": { + "entry": { + "sequence": 6, + "phase": "start", + "tokens": 8192, + "time": 297468.743740514 + }, + "pending_status": 500, + "later_status": 503, + "readiness": 503, + "slab_unlinked": true, + "replacement_exact": true, + "current_status": 200, + "current_output_tokens": 512, + "current_tokens_at_termination": 2, + "current_exact": true + } + }, + "control_tokens": [ + 9079, + 107, + 106, + 107, + 106, + 130614, + 107, + 106, + 107, + 106, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695 + ], + "native_tokens_equal": true, + "decode_control_tokens": [ + 9079, + 107, + 106, + 107, + 106, + 130614, + 107, + 106, + 107, + 106, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097, + 6249, + 236761, + 107, + 2094, + 563, + 1695, + 4403, + 236761, + 669, + 2604, + 3673, + 6097, + 36157, + 236761, + 669, + 3730, + 3673, + 6097 + ] +} diff --git a/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-isolation.log b/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-isolation.log new file mode 100644 index 00000000..71a3858e --- /dev/null +++ b/opt_prime/examples/pd/http_matrix/evidence201/local201-gemma-isolation.log @@ -0,0 +1,3 @@ + +Loading weights: 0%| | 0/340 [00:00