| |
| """Out-of-process async proxy (VLLM_DSPARK_REMOTE_ASYNC_PROXY=process). torch-free: the proxy process imports this |
| package as a top-level package and never imports vllm or torch. |
| |
| Why a process (directive 2026-10-23 §3): the thread proxy's success path - wait for the gather, network round, |
| publish - is Python, so it needs the engine's GIL; during prefill the engine thread holds the GIL for long stretches |
| and the round's publication waits for it (GB10: +4-8 % TTFT, 31 watchdog-preempted rounds of 5,113 in the async |
| compact set). Here the same reviewed round code (roundcore.RoundCore, unchanged) runs in a child interpreter with its |
| own GIL; it shares nothing with the engine but the memory region (shmregion.py) and the C hand-off words (hostflag.c). |
| Its success path is: acquire-wait on the descriptor and gathered words, the round, one GIL-free C publish into the |
| shared page. No step needs the engine's GIL, a CUDA call or an engine-held lock. |
| |
| engine thread: gather -> D2H into the region -> descriptor t -> desc_head = t -> [stream: gathered = t, |
| wait released >= t, H2D reply] -> return (never blocks) |
| proxy process: wait desc_head >= t -> copy descriptor, desc_tail = t -> wait gathered >= t -> RoundCore round |
| -> publish(t) (claim, copy, release: hostflag.c) |
| engine watchdog (C thread in the engine process): a round gathered but unreleased past the deadline is claimed, |
| failed and released, exactly as with the thread proxy. |
| |
| Ownership of round t's completion is the claim word (one winner: proxy publish, watchdog, or a fail-all). Failure of |
| the proxy process is seen by the engine at the next round (Popen.poll): fail-all far ahead, then target-only rounds, |
| the same end state as a dead proxy thread (S9 F2). The proxy watches its parent (getppid, and PR_SET_PDEATHSIG when |
| started from the main thread) and exits when the engine is gone. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import ctypes |
| import dataclasses |
| import json |
| import logging |
| import os |
| import signal |
| import subprocess |
| import sys |
| import time |
|
|
| import numpy as np |
|
|
| from . import wire |
| from .breaker import CircuitBreaker |
| from .client import RemoteDraftClient |
| from .config import RemoteConfig |
| from .hostflag import _M32, _V, HostFlagCore, set_timer_slack |
| from .roundcore import STATUS_ROUND_FAILED, BatchView, ReplyLayout, RoundCore, collect_meta |
| from .shmregion import (C_DESC_HEAD, C_DESC_TAIL, C_DONE, C_ENGINE_SEQ, C_FLUSH_ACK, C_FLUSH_REQ, C_HEARTBEAT, C_PID, |
| C_START, C_STATE, C_STOP, C_WD, F_ENQUEUE_FAILED, R_FAILED, R_OK, R_PENDING, S_FAILED, |
| S_RUNNING, S_STOPPED, RegionLayout, SharedRegion) |
| from .state import SlotTable |
| from .stats import Stats |
| from .transport import Endpoint |
|
|
| log = logging.getLogger("vllm.remote_dspark") |
|
|
| PR_SET_PDEATHSIG = 1 |
| DESC_SPIN_NS = 200_000 |
| DESC_NAP_NS = 100_000 |
| POLL_S = 0.1 |
| START_TIMEOUT_S = 60.0 |
|
|
| |
|
|
|
|
| def _enc(v): |
| if isinstance(v, bytes): |
| return {"__hex__": v.hex()} |
| if isinstance(v, dict): |
| return {k: _enc(x) for k, x in v.items()} |
| if isinstance(v, (list, tuple)): |
| return [_enc(x) for x in v] |
| return v |
|
|
|
|
| def _dec(v): |
| if isinstance(v, dict): |
| if set(v) == {"__hex__"}: |
| return bytes.fromhex(v["__hex__"]) |
| return {k: _dec(x) for k, x in v.items()} |
| if isinstance(v, list): |
| return [_dec(x) for x in v] |
| return v |
|
|
|
|
| def encode_init(cfg: RemoteConfig, hello: wire.Hello, layout: RegionLayout, *, n_out_by_reqs, lib_path: str, |
| spin_us: int, nap_us: int, watchdog_ms: int, strict: bool, parent_pid: int, pdeathsig: bool, |
| start_round: int = 1, session: int | None = None, base_counters: dict | None = None, |
| generation: int = 0) -> bytes: |
| """start_round: the first round this proxy serves (a respawned proxy continues the engine's numbering); |
| session: its worker session id; base_counters: its predecessors' counters, carried on in its stats file; |
| generation: 0 for the first proxy, k for the k-th respawn.""" |
| return json.dumps(_enc({ |
| "cfg": dataclasses.asdict(cfg), "hello": dataclasses.asdict(hello), "layout": layout.to_dict(), |
| "n_out_by_reqs": list(n_out_by_reqs) if n_out_by_reqs else None, "lib_path": lib_path, |
| "spin_us": int(spin_us), "nap_us": int(nap_us), "watchdog_ms": int(watchdog_ms), "strict": bool(strict), |
| "parent_pid": int(parent_pid), "pdeathsig": bool(pdeathsig), "start_round": int(start_round), |
| "session": session, "base_counters": dict(base_counters or {}), "generation": int(generation)})).encode() |
|
|
|
|
| def decode_init(data: bytes) -> dict: |
| d = _dec(json.loads(data.decode())) |
| c = d["cfg"] |
| c["endpoint"] = Endpoint(**c["endpoint"]) |
| d["cfg"] = RemoteConfig(**c) |
| h = d["hello"] |
| h["tap_layer_ids"] = tuple(h["tap_layer_ids"]) |
| d["hello"] = wire.Hello(**h) |
| d["layout"] = RegionLayout(**d["layout"]) |
| return d |
|
|
|
|
| ENGINE_OWNED = ("async_dead",) |
|
|
|
|
| def merge_snapshots(proxy: dict, engine: dict | None) -> dict: |
| """One stats file from the proxy's snapshot and the engine side's (counters add up; histograms and extras are |
| disjoint by construction, the proxy's win on a clash except ENGINE_OWNED).""" |
| if not engine: |
| return proxy |
| out = {k: v for k, v in engine.items() if k not in ("counters", "latency", "uptime_s")} |
| out.update({k: v for k, v in proxy.items() if k not in ("counters", "latency")}) |
| for k in ENGINE_OWNED: |
| if k in engine: |
| out[k] = engine[k] |
| cnt = dict(engine.get("counters") or {}) |
| for k, v in (proxy.get("counters") or {}).items(): |
| cnt[k] = cnt.get(k, 0) + v |
| out["counters"] = dict(sorted(cnt.items())) |
| lat = dict(engine.get("latency") or {}) |
| lat.update(proxy.get("latency") or {}) |
| out["latency"] = dict(sorted(lat.items())) |
| |
| out["engine_share"] = {"counters": dict(engine.get("counters") or {}), |
| "latency": sorted((engine.get("latency") or {}).keys())} |
| return out |
|
|
|
|
| def unmerge_snapshot(merged: dict) -> dict: |
| """The proxy's part of a file merge_snapshots() wrote (the inverse of the engine share).""" |
| share = merged.get("engine_share") or {} |
| out = {k: v for k, v in merged.items() if k != "engine_share"} |
| eng_c = share.get("counters") or {} |
| out["counters"] = {k: v - eng_c.get(k, 0) for k, v in (merged.get("counters") or {}).items() |
| if v - eng_c.get(k, 0)} |
| eng_l = set(share.get("latency") or ()) |
| out["latency"] = {k: v for k, v in (merged.get("latency") or {}).items() if k not in eng_l} |
| return out |
|
|
|
|
| def write_json_atomic(path: str, obj: dict) -> None: |
| tmp = f"{path}.tmp.{os.getpid()}" |
| with open(tmp, "w") as f: |
| json.dump(obj, f, indent=1, default=str) |
| os.replace(tmp, path) |
|
|
|
|
| |
|
|
|
|
| class _ProxyFlags(HostFlagCore): |
| """The proxy process's view of the engine's hand-off page: wait, publish, fail-all. The watchdog runs in the |
| engine process; its stats are mirrored into the control block every round.""" |
|
|
| def __init__(self, region: SharedRegion, lib_path: str, spin_us: int, nap_us: int, reply_bytes: int, |
| watchdog_ms: int): |
| super().__init__(lib_path, spin_us, nap_us, page=region.page_addr) |
| self.configure_reply(reply_bytes, STATUS_ROUND_FAILED) |
| self.region = region |
| self.deadline_ms = int(watchdog_ms) |
|
|
| def attach(self, *a, **k) -> None: |
| raise RuntimeError("the watchdog belongs to the engine process") |
|
|
| def watchdog_stats(self) -> dict: |
| fires, last, late = (int(x) for x in self.region.u64(C_WD, 3)) |
| return {"running": True, "deadline_ms": self.deadline_ms, "fires": fires, "last_failed_round": last, |
| "max_late_ms": round(late / 1e6, 3), "mirrored": True} |
|
|
| def describe(self) -> dict: |
| d = super().describe() |
| d["mechanism"] = "host_flags_shared" |
| return d |
|
|
|
|
| class _RegionRows: |
| """What RoundCore reads of the gatherer, from the region: the STEP message buffer and the gather metadata.""" |
|
|
| def __init__(self, region: SharedRegion): |
| L = region.layout |
| self.region = region |
| self.msg_np = region.msg_np |
| self.max_rows = L.max_rows |
| self.row_bytes = L.row_bytes |
|
|
| def collect(self, n: int, n_candidates: int, meta_slot: int = 0): |
| return collect_meta(self.region.meta_np[meta_slot], n, n_candidates) |
|
|
| def gather_from_ring(self, n_req, spans): |
| raise RuntimeError("synchronous resync in the process proxy") |
|
|
|
|
| def monotonic_ms() -> int: |
| """CLOCK_MONOTONIC in ms as a u32 (the heartbeat's clock; system-wide, so the engine compares it with its own).""" |
| return int(time.monotonic() * 1000) & _M32 |
|
|
|
|
| class _Housekeeping: |
| """RoundCore._stop for the proxy process: polled at least every POLL_S while it waits and once per round. Writes |
| the heartbeat (C_HEARTBEAT, S11 P-3), serves stats flush requests; reports stop when the engine asked for it or |
| the engine process is gone.""" |
|
|
| def __init__(self, rnd: "ProcessRound"): |
| self.r = rnd |
| self.lib = rnd.sync.lib |
| self._stop_w = _V(rnd.region.word(C_STOP)) |
| self._flush_w = _V(rnd.region.word(C_FLUSH_REQ)) |
| self._hb_w = _V(rnd.region.word(C_HEARTBEAT)) |
| self._flushed = 0 |
|
|
| def beat(self) -> None: |
| self.lib.dsv41_hf_set(self._hb_w, monotonic_ms()) |
|
|
| def is_set(self) -> bool: |
| self.beat() |
| req = int(self.lib.dsv41_hf_load(self._flush_w)) |
| if req != self._flushed: |
| self._flushed = req |
| try: |
| self.r.write_stats() |
| finally: |
| self.lib.dsv41_hf_store(_V(self.r.region.word(C_FLUSH_ACK)), req) |
| if self.lib.dsv41_hf_load(self._stop_w): |
| return True |
| return os.getppid() != self.r.parent_pid |
|
|
|
|
| class ProcessRound(RoundCore): |
| """The proxy process's round: RoundCore fed from the descriptor ring instead of the thread proxy's queue.""" |
|
|
| def __init__(self, region: SharedRegion, cfg: RemoteConfig, hello: wire.Hello, *, n_out_by_reqs, lib_path: str, |
| spin_us: int, nap_us: int, watchdog_ms: int, parent_pid: int, background: bool, |
| transport_factory=None, start_round: int = 1, session: int | None = None, |
| base_counters: dict | None = None, generation: int = 0): |
| self.cfg = cfg |
| self.hello = hello |
| self.region = region |
| self.parent_pid = parent_pid |
| self.start_round = int(start_round) |
| self.stats = Stats() |
| for k, v in (base_counters or {}).items(): |
| self.stats.inc(k, int(v)) |
| self.N, self.W, self.M = hello.num_spec, hello.window, hello.max_slots |
| self.layout = ReplyLayout(self.M, self.N, cfg.sampled_top32) |
| self.bcast_np = np.zeros(self.layout.size, dtype=np.int32) |
| self.n_out_by_reqs = list(n_out_by_reqs) if n_out_by_reqs else None |
| self.rounds = self.start_round - 1 |
| self._last_epoch = 0 |
| self._last_log = time.monotonic() |
| self._err_logged = 0.0 |
| self.is_sender = True |
| self.async_mode = True |
| L = region.layout |
| if (L.max_slots, L.window, L.row_bytes, L.max_rows) != (self.M, self.W, hello.row_bytes, |
| hello.max_rows_per_step): |
| raise ValueError(f"region layout {L} does not match the HELLO") |
| self.gatherer = _RegionRows(region) |
| self._meta_slots = L.ring |
| self.resync_np = region.resync_np |
| self.slots = SlotTable(self.M, self.W) |
| self.breaker = CircuitBreaker(cfg.trip, cfg.cooldown) |
| self.stats.set("breaker", self.breaker.snapshot()) |
| self.sync = _ProxyFlags(region, lib_path, spin_us, nap_us, self.layout.size * 4, watchdog_ms) |
| self.lib = self.sync.lib |
| self._staged: frozenset[int] = frozenset() |
| self._dead: str | None = None |
| self.client = RemoteDraftClient(cfg, hello, stats=self.stats, transport_factory=transport_factory, |
| background=background, session=session) |
| self.stats.set("contract_version", wire.CONTRACT_VERSION) |
| self.stats.set("policy", cfg.policy) |
| self.stats.set("endpoint", str(cfg.endpoint)) |
| self.stats.set("n_out_by_reqs", self.n_out_by_reqs) |
| |
| self.stats.set("async_proxy_process", {"pid": os.getpid(), "generation": int(generation), |
| "start_round": self.start_round}) |
| self._stop = _Housekeeping(self) |
| self._engine_seq = 0 |
| self._engine_snap: dict | None = None |
|
|
| |
| def _request_resync(self, slots: set[int]) -> None: |
| f = self.region.resync_flags |
| for sl in slots: |
| f[sl] = 1 |
|
|
| def _clear_resync(self, staged: frozenset[int]) -> None: |
| f = self.region.resync_flags |
| for sl in staged: |
| f[sl] = 0 |
|
|
| def _wait_descriptor(self, t: int) -> bool: |
| head = _V(self.region.word(C_DESC_HEAD)) |
| while not self._stop.is_set(): |
| if self.lib.dsv41_hf_wait_geq(head, t & _M32, int(POLL_S * 1e9), DESC_SPIN_NS, DESC_NAP_NS) == 0: |
| return True |
| return False |
|
|
| def _take(self, t: int): |
| d = self.region.read_descriptor(t) |
| if d.flags & F_ENQUEUE_FAILED: |
| return (t, None, 0, frozenset()) |
| bv = BatchView(req_ids=d.owners, slots=d.slots, qsl=np.zeros(0, dtype=np.int64), |
| need_draft=d.need_draft, k_window=d.k_window, max_q=d.max_q) |
| return (t, bv, d.R, d.staged) |
|
|
| def serve(self) -> None: |
| """Rounds start_round, start_round + 1, ... until stopped. Raises on a broken hand-off (the caller fails |
| everything closed).""" |
| t = self.start_round |
| tail = _V(self.region.word(C_DESC_TAIL)) |
| done = _V(self.region.word(C_DONE)) |
| while True: |
| if not self._wait_descriptor(t): |
| return |
| item = self._take(t) |
| self.rounds = t |
| if not self._async_round(item): |
| return |
| |
| self.lib.dsv41_hf_store(tail, t & _M32) |
| self.lib.dsv41_hf_store(done, t & _M32) |
| t += 1 |
|
|
| def _engine_snapshot(self) -> dict | None: |
| seq = int(self.lib.dsv41_hf_load(_V(self.region.word(C_ENGINE_SEQ)))) |
| if seq and seq != self._engine_seq: |
| try: |
| self._engine_snap = json.loads(self.region.read_engine_blob(seq).decode()) |
| self._engine_seq = seq |
| except (ValueError, UnicodeDecodeError): |
| pass |
| return self._engine_snap |
|
|
| def write_stats(self) -> None: |
| if not self.cfg.stats_file: |
| return |
| self.stats.set("rounds", self.rounds) |
| self.stats.set("breaker", self.breaker.snapshot()) |
| self.stats.set("async_watchdog", self.sync.watchdog_stats()) |
| self.stats.set("proxy_torch_loaded", "torch" in sys.modules) |
| try: |
| write_json_atomic(self.cfg.stats_file, merge_snapshots(self.stats.snapshot(), self._engine_snapshot())) |
| except OSError as e: |
| log.warning("remote DSpark stats file %s: %s", self.cfg.stats_file, e) |
|
|
|
|
| def _pdeathsig() -> None: |
| try: |
| libc = ctypes.CDLL(None, use_errno=True) |
| libc.prctl(PR_SET_PDEATHSIG, signal.SIGKILL, 0, 0, 0) |
| except (OSError, AttributeError): |
| pass |
|
|
|
|
| def _store(region: SharedRegion, lib, off: int, value: int) -> None: |
| lib.dsv41_hf_set(_V(region.word(off)), value & _M32) |
|
|
|
|
| def main(fd: int, init: bytes | None = None, transport_factory=None) -> int: |
| """Proxy process entry (BOOT below): the init blob comes on stdin.""" |
| d = decode_init(init if init is not None else sys.stdin.buffer.read()) |
| if d["pdeathsig"]: |
| _pdeathsig() |
| set_timer_slack() |
| if os.getppid() != d["parent_pid"]: |
| return 5 |
| logging.basicConfig(level=logging.INFO, stream=sys.stderr, |
| format="(remote-dspark proxy pid=%(process)d) %(levelname)s %(asctime)s %(message)s") |
| region = SharedRegion.attach(d["layout"], fd) |
| region.u32(C_PID)[0] = os.getpid() |
| cfg: RemoteConfig = d["cfg"] |
| rnd = None |
| lib = None |
| try: |
| rnd = ProcessRound(region, cfg, d["hello"], n_out_by_reqs=d["n_out_by_reqs"], lib_path=d["lib_path"], |
| spin_us=d["spin_us"], nap_us=d["nap_us"], watchdog_ms=d["watchdog_ms"], |
| parent_pid=d["parent_pid"], background=not d["strict"], |
| transport_factory=transport_factory, start_round=d.get("start_round", 1), |
| session=d.get("session"), base_counters=d.get("base_counters"), |
| generation=d.get("generation", 0)) |
| lib = rnd.lib |
| if d["strict"]: |
| if not rnd.client.connect_now(): |
| raise RuntimeError(f"connect/HELLO to {cfg.endpoint} failed at startup " |
| f"({rnd.client.disabled_reason or rnd.client.last_error})") |
| rnd.client.start_background() |
| except BaseException as e: |
| region.set_start_reason(f"{type(e).__name__}: {e}") |
| if lib is not None: |
| _store(region, lib, C_START, R_FAILED) |
| _store(region, lib, C_STATE, S_FAILED) |
| else: |
| region.u32(C_START)[0] = R_FAILED |
| region.u32(C_STATE)[0] = S_FAILED |
| if rnd is not None: |
| rnd.client.close() |
| return 2 |
| rnd._stop.beat() |
| _store(region, lib, C_START, R_OK) |
| _store(region, lib, C_STATE, S_RUNNING) |
| log.info("remote DSpark proxy process running (engine pid %d, %s)", d["parent_pid"], rnd.sync.describe()) |
| code = 0 |
| try: |
| rnd.serve() |
| except BaseException as e: |
| rnd._die(f"proxy process: {type(e).__name__}: {e}") |
| code = 4 |
| finally: |
| try: |
| rnd.write_stats() |
| finally: |
| rnd.client.close() |
| _store(region, lib, C_STATE, S_FAILED if code else S_STOPPED) |
| return code |
|
|
|
|
| |
|
|
|
|
| class ProxyStartError(RuntimeError): |
| """The proxy process did not start (spawn, interpreter, imports, start timeout; strict: connect + HELLO). The engine |
| then runs synchronous rounds unless the start is strict (round.make_remote_round, S11 P-2).""" |
|
|
| |
| |
| BOOT = """import importlib.util, sys |
| pkg, fd = sys.argv[1], int(sys.argv[2]) |
| spec = importlib.util.spec_from_file_location("remote_dspark", pkg + "/__init__.py", submodule_search_locations=[pkg]) |
| mod = importlib.util.module_from_spec(spec) |
| sys.modules["remote_dspark"] = mod |
| spec.loader.exec_module(mod) |
| from remote_dspark.proxy_proc import main |
| sys.exit(main(fd)) |
| """ |
|
|
|
|
| class ProxyProcess: |
| """Engine side: start the proxy process on a region, watch it, stop it. CPU only.""" |
|
|
| def __init__(self, region: SharedRegion, lib, init: bytes, *, python: str | None = None): |
| self.region = region |
| self.lib = lib |
| pkg = os.path.dirname(os.path.abspath(__file__)) |
| |
| |
| self.proc = subprocess.Popen([python or sys.executable, "-I", "-c", BOOT, pkg, str(region.fd)], |
| stdin=subprocess.PIPE, pass_fds=(region.fd,), close_fds=True) |
| try: |
| self.proc.stdin.write(init) |
| self.proc.stdin.close() |
| except OSError: |
| pass |
| self._flush = 0 |
| self._engine_seq = 0 |
| self._blob_warned = False |
|
|
| @property |
| def pid(self) -> int: |
| return self.proc.pid |
|
|
| def word(self, off: int) -> int: |
| return int(self.lib.dsv41_hf_load(_V(self.region.word(off)))) |
|
|
| def start_status(self) -> tuple[str, str]: |
| """Without waiting: ("ok", ""), ("pending", "") or ("failed", why) (it reported a failure or exited).""" |
| r = self.word(C_START) |
| if r == R_OK: |
| return "ok", "" |
| if r == R_FAILED: |
| return "failed", f"remote DSpark proxy process failed to start: {self.region.start_reason()}" |
| rc = self.proc.poll() |
| if rc is not None: |
| return "failed", f"remote DSpark proxy process exited with {rc} before it started" |
| return "pending", "" |
|
|
| def wait_started(self, timeout_s: float = START_TIMEOUT_S) -> None: |
| """Until the proxy reports its start (client built; when strict: connected + HELLO). Raises ProxyStartError |
| when it failed, exited or did not report in time; the process is then gone or killed.""" |
| t_end = time.monotonic() + timeout_s |
| while True: |
| state, why = self.start_status() |
| if state == "ok": |
| return |
| if state == "failed": |
| self.kill(2.0) |
| raise ProxyStartError(why) |
| if time.monotonic() > t_end: |
| self.kill() |
| raise ProxyStartError(f"remote DSpark proxy process did not start within {timeout_s:.0f} s") |
| time.sleep(0.005) |
|
|
| def exit_code(self) -> int | None: |
| return self.proc.poll() |
|
|
| def tail(self) -> int: |
| return self.word(C_DESC_TAIL) |
|
|
| def heartbeat_age_ms(self) -> int: |
| """Milliseconds since the proxy's last housekeeping poll (0 if its clock read is a hair ahead of ours).""" |
| d = (monotonic_ms() - self.word(C_HEARTBEAT)) & _M32 |
| return d if d < (1 << 31) else 0 |
|
|
| def publish_head(self, t: int) -> None: |
| self.lib.dsv41_hf_store(_V(self.region.word(C_DESC_HEAD)), t & _M32) |
|
|
| def resync_requests(self) -> list[int]: |
| return [int(i) for i in np.flatnonzero(self.region.resync_flags)] |
|
|
| def publish_engine_stats(self, snapshot: dict) -> None: |
| data = json.dumps(snapshot, default=str).encode() |
| seq = self._engine_seq + 1 |
| if self.region.write_engine_blob(seq, data): |
| self._engine_seq = seq |
| self.lib.dsv41_hf_store(_V(self.region.word(C_ENGINE_SEQ)), seq & _M32) |
| elif not self._blob_warned: |
| self._blob_warned = True |
| log.warning("remote DSpark engine-side stats (%d B) do not fit the shared region's blob; the stats file " |
| "keeps the last engine snapshot that did", len(data)) |
|
|
| def flush(self, timeout_s: float = 2.0) -> bool: |
| """Ask the proxy to write the stats file now; True once it did.""" |
| if self.proc.poll() is not None: |
| return False |
| self._flush += 1 |
| self.lib.dsv41_hf_store(_V(self.region.word(C_FLUSH_REQ)), self._flush & _M32) |
| r = self.lib.dsv41_hf_wait_geq(_V(self.region.word(C_FLUSH_ACK)), self._flush & _M32, int(timeout_s * 1e9), |
| 200_000, 1_000_000) |
| return r == 0 |
|
|
| def stop(self, timeout_s: float = 5.0) -> int | None: |
| self.lib.dsv41_hf_store(_V(self.region.word(C_STOP)), 1) |
| try: |
| return self.proc.wait(timeout=timeout_s) |
| except subprocess.TimeoutExpired: |
| self.kill() |
| return None |
|
|
| def signal_kill(self) -> None: |
| """SIGKILL, without waiting: the engine thread must not block on a child that may sit in uninterruptible |
| sleep (S11 P-6). exit_code() reaps it later. A killed process runs no more user code, so it can no longer |
| write the region (it may still be finishing a system call).""" |
| if self.proc.poll() is None: |
| self.proc.kill() |
|
|
| def kill(self, wait_s: float = 5.0) -> None: |
| """SIGKILL and wait for the exit at most wait_s (startup and shutdown only; the round path uses signal_kill).""" |
| self.signal_kill() |
| if self.proc.returncode is None: |
| try: |
| self.proc.wait(timeout=wait_s) |
| except subprocess.TimeoutExpired: |
| log.error("remote DSpark proxy process %d did not exit within %.0f s of SIGKILL", self.pid, wait_s) |
|
|
|
|
| __all__ = ["ProcessRound", "ProxyProcess", "ProxyStartError", "encode_init", "decode_init", "merge_snapshots", "main", |
| "R_PENDING"] |
|
|
