feat(perception): fence streaming lifecycle with worker-local ownership
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,257 @@
|
||||
"""Lifecycle/fencing for one subprocess-backed full perception profile.
|
||||
|
||||
Uses existing GraphState, StreamStart and bounded scheduler primitives. The
|
||||
trusted controller supplies child commands and one Worker-wide lease directory;
|
||||
neither is accepted from stream payloads. No backend queue or model is created
|
||||
here. External-client inventory and real network authentication remain separate.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from collections.abc import Callable, Iterator
|
||||
from contextlib import contextmanager, suppress
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
from .graph_contracts import GraphState
|
||||
from .realtime_contract import StreamStart
|
||||
from .realtime_scene import _wire_integer
|
||||
from .streaming_queue import StreamBundle, StreamMailbox
|
||||
from .worker_lease import WorkerLease, WorkerLeaseError
|
||||
|
||||
|
||||
def _group_exists(pgid: int) -> bool:
|
||||
try:
|
||||
os.killpg(pgid, 0)
|
||||
return True
|
||||
except ProcessLookupError:
|
||||
return False
|
||||
except PermissionError:
|
||||
return True # Unknown is not a release proof.
|
||||
|
||||
|
||||
class StreamingLifecycle:
|
||||
def __init__(
|
||||
self,
|
||||
start: StreamStart,
|
||||
lease_root: Path,
|
||||
mailbox: StreamMailbox,
|
||||
stop: threading.Event,
|
||||
*,
|
||||
ttl_seconds: float = 2.0,
|
||||
clock_ns: Callable[[], int] = time.monotonic_ns,
|
||||
) -> None:
|
||||
self.start, self.mailbox, self.stop_event = start, mailbox, stop
|
||||
self._lock = threading.RLock()
|
||||
self._cleanup_lock = threading.Lock()
|
||||
self._children: list[subprocess.Popen[bytes]] = []
|
||||
self._threads: list[threading.Thread] = []
|
||||
self._active = {"gpu": 0, "cpu": 0}
|
||||
self.state = GraphState.CREATED
|
||||
self.reason: str | None = None
|
||||
self.stop_requested_ns: int | None = None
|
||||
self.retired_ns: int | None = None
|
||||
self.lease = WorkerLease(lease_root, start, ttl_seconds=ttl_seconds, clock_ns=clock_ns)
|
||||
self.state = GraphState.STARTING
|
||||
self._watchdog_stop = threading.Event()
|
||||
self._watchdog = threading.Thread(
|
||||
target=self._watch, name="perception-lease-watchdog", daemon=True
|
||||
)
|
||||
self._watchdog.start()
|
||||
|
||||
def _check(self, start: StreamStart, *, starting: bool = False) -> None:
|
||||
allowed = (GraphState.STARTING, GraphState.RUNNING) if starting else (GraphState.RUNNING,)
|
||||
if self.state not in allowed or self.stop_event.is_set():
|
||||
raise WorkerLeaseError("runtime is not accepting work")
|
||||
try:
|
||||
self.lease.check(start)
|
||||
except WorkerLeaseError:
|
||||
# A stale client is rejected but cannot cancel the current owner.
|
||||
if start == self.start:
|
||||
self.request_stop("lease-lost")
|
||||
raise
|
||||
|
||||
def renew(self, start: StreamStart) -> None:
|
||||
with self._lock:
|
||||
self._check(start, starting=True)
|
||||
self.lease.renew(start)
|
||||
|
||||
def ready(self) -> None:
|
||||
with self._lock:
|
||||
self._check(self.start, starting=True)
|
||||
if self.state != GraphState.STARTING:
|
||||
raise WorkerLeaseError("warmup completion already consumed")
|
||||
if any(p.poll() is not None for p in self._children):
|
||||
self.request_stop("child-exited")
|
||||
raise WorkerLeaseError("owned child exited during warmup")
|
||||
self.state = GraphState.RUNNING
|
||||
|
||||
def spawn(self, factory: Callable[[], subprocess.Popen[bytes]]) -> subprocess.Popen[bytes]:
|
||||
"""Trusted, bounded process creation + registration, atomic with stop."""
|
||||
with self._lock:
|
||||
self._check(self.start, starting=True)
|
||||
if self.state != GraphState.STARTING or len(self._children) >= 8:
|
||||
raise WorkerLeaseError("model processes may start only once during warmup")
|
||||
process = factory()
|
||||
self._children.append(process)
|
||||
if process.poll() is None and os.getpgid(process.pid) != process.pid:
|
||||
self.request_stop("invalid-process-group")
|
||||
raise WorkerLeaseError("runtime children need dedicated process groups")
|
||||
return process
|
||||
|
||||
def track_thread(self, thread: threading.Thread) -> None:
|
||||
with self._lock:
|
||||
self._check(self.start, starting=True)
|
||||
if len(self._threads) >= 8 or thread in self._threads:
|
||||
raise WorkerLeaseError("runtime thread registration outside bound")
|
||||
self._threads.append(thread)
|
||||
|
||||
def admit(self, start: StreamStart, bundle: StreamBundle) -> bool:
|
||||
with self._lock:
|
||||
self._check(start)
|
||||
return self.mailbox.put(bundle)
|
||||
|
||||
@contextmanager
|
||||
def work(self, start: StreamStart, lane: Literal["gpu", "cpu"]) -> Iterator[None]:
|
||||
with self._lock:
|
||||
self._check(start)
|
||||
if lane not in self._active or self._active[lane]:
|
||||
raise WorkerLeaseError("only one active call per compute lane")
|
||||
self._active[lane] += 1
|
||||
try:
|
||||
yield
|
||||
self.check_current(start)
|
||||
except BaseException:
|
||||
self.request_stop("failed")
|
||||
raise
|
||||
finally:
|
||||
with self._lock:
|
||||
self._active[lane] -= 1
|
||||
|
||||
def check_current(self, start: StreamStart) -> None:
|
||||
with self._lock:
|
||||
self._check(start)
|
||||
|
||||
def validate_result_binding(self, value: object) -> None:
|
||||
"""Recheck at receipt/use; accepting a hash alone cannot renew authority."""
|
||||
start = StreamStart.from_dict(value)
|
||||
self.check_current(start)
|
||||
|
||||
def request_stop(self, reason: str = "cancelled") -> None:
|
||||
if reason not in (
|
||||
"completed",
|
||||
"cancelled",
|
||||
"lease-lost",
|
||||
"child-exited",
|
||||
"invalid-process-group",
|
||||
"failed",
|
||||
):
|
||||
raise ValueError("unknown runtime stop reason")
|
||||
with self._lock:
|
||||
if self.state in (GraphState.STOPPED, GraphState.CANCELLED, GraphState.FAILED):
|
||||
return
|
||||
self.reason = self.reason or reason
|
||||
if self.stop_requested_ns is None:
|
||||
self.stop_requested_ns = time.monotonic_ns()
|
||||
self.state = GraphState.STOPPING
|
||||
self.stop_event.set()
|
||||
self.mailbox.cancel()
|
||||
|
||||
def _watch(self) -> None:
|
||||
while not self._watchdog_stop.wait(0.05):
|
||||
try:
|
||||
with self._lock:
|
||||
if self.state in (GraphState.STARTING, GraphState.RUNNING):
|
||||
self._check(self.start, starting=True)
|
||||
if any(p.poll() is not None for p in self._children):
|
||||
self.request_stop("child-exited")
|
||||
stopping = self.state == GraphState.STOPPING
|
||||
if stopping:
|
||||
self.stop_children()
|
||||
return
|
||||
except (WorkerLeaseError, OSError):
|
||||
self.request_stop("lease-lost")
|
||||
self.stop_children()
|
||||
return
|
||||
|
||||
def stop_children(self) -> bool:
|
||||
"""Stop only owned groups. No Boolean resource-release attestation input."""
|
||||
with self._cleanup_lock:
|
||||
with self._lock:
|
||||
children = tuple(self._children)
|
||||
for sent_signal, grace in ((signal.SIGTERM, 4.0), (signal.SIGKILL, 1.0)):
|
||||
for process in reversed(children):
|
||||
if _group_exists(process.pid):
|
||||
with suppress(ProcessLookupError):
|
||||
os.killpg(process.pid, sent_signal)
|
||||
deadline = time.monotonic() + grace
|
||||
while True:
|
||||
complete = all(
|
||||
p.poll() is not None and not _group_exists(p.pid) for p in children
|
||||
)
|
||||
if complete or time.monotonic() >= deadline:
|
||||
break
|
||||
time.sleep(0.01)
|
||||
if complete:
|
||||
return True
|
||||
return False
|
||||
|
||||
def close(self, reason: str = "completed") -> bool:
|
||||
self.request_stop(reason)
|
||||
self._watchdog_stop.set()
|
||||
children_stopped = self.stop_children()
|
||||
if self._watchdog is not threading.current_thread():
|
||||
self._watchdog.join(timeout=0.1)
|
||||
with self._lock:
|
||||
if self.lease.released:
|
||||
return True
|
||||
# Caller joins/finishes its adapters; active callbacks and payloads
|
||||
# continue to fence even if all GPU subprocesses have died already.
|
||||
if (
|
||||
not children_stopped
|
||||
or any(self._active.values())
|
||||
or any(t.is_alive() for t in self._threads)
|
||||
or self._watchdog.is_alive()
|
||||
or not self.mailbox.quiescent
|
||||
):
|
||||
return False
|
||||
self.lease._retire_after_verified_stop()
|
||||
self.retired_ns = time.monotonic_ns()
|
||||
self.state = (
|
||||
GraphState.STOPPED
|
||||
if self.reason == "completed"
|
||||
else GraphState.CANCELLED
|
||||
if self.reason == "cancelled"
|
||||
else GraphState.FAILED
|
||||
)
|
||||
return True
|
||||
|
||||
def snapshot(self) -> dict[str, Any]:
|
||||
with self._lock:
|
||||
return {
|
||||
"state": self.state.value,
|
||||
"reason": self.reason,
|
||||
"start": self.start.to_dict(),
|
||||
"active_calls": dict(self._active),
|
||||
"lease_released": self.lease.released,
|
||||
"input_payloads_released": self.mailbox.quiescent,
|
||||
"owned_children": len(self._children),
|
||||
"live_children": sum(p.poll() is None for p in self._children),
|
||||
"actuation_allowed": False,
|
||||
"lease_renewals": self.lease.renewals,
|
||||
"lease_deadline_monotonic_ns": _wire_integer(self.lease.deadline_ns),
|
||||
"stop_requested_monotonic_ns": (
|
||||
_wire_integer(self.stop_requested_ns)
|
||||
if self.stop_requested_ns is not None
|
||||
else None
|
||||
),
|
||||
"retired_monotonic_ns": (
|
||||
_wire_integer(self.retired_ns) if self.retired_ns is not None else None
|
||||
),
|
||||
"event_clock": "worker-process-monotonic",
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
"""Worker-local cooperative fence for managed perception runtimes (POSIX).
|
||||
|
||||
The controller selects one fixed private directory per physical Worker, shared
|
||||
by ALL its managed profile containers. It must not come from a job/stream.
|
||||
An OS lock serializes processes; a durable active record quarantines a crashed
|
||||
owner even after the kernel releases its lock. Time expiry never frees the GPU.
|
||||
This does not police unmanaged GPU clients or replace backend job claims.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fcntl
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
from k1link.artifacts import write_json_atomic
|
||||
|
||||
from .realtime_contract import StreamStart, _integer, _number
|
||||
|
||||
LEASE_SCHEMA = "missioncore.perception-worker-lease/v1"
|
||||
|
||||
|
||||
class WorkerLeaseError(RuntimeError):
|
||||
"""Unknown, occupied, expired or quarantined Worker ownership."""
|
||||
|
||||
|
||||
class WorkerLease:
|
||||
"""Owned by the trusted local supervisor, never deserialized from a client.
|
||||
|
||||
Heartbeats are bounded local receipt-time renewals. A late heartbeat cannot
|
||||
resurrect authority. Only the supervisor may retire after checking actual
|
||||
process/thread/input release. There is deliberately no force/takeover API.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
root: Path,
|
||||
start: StreamStart,
|
||||
*,
|
||||
ttl_seconds: float = 2.0,
|
||||
clock_ns: Callable[[], int] = time.monotonic_ns,
|
||||
) -> None:
|
||||
_number(ttl_seconds, "lease ttl", minimum=0.1)
|
||||
if ttl_seconds > 30:
|
||||
raise WorkerLeaseError("lease ttl exceeds 30 seconds")
|
||||
if root.is_symlink():
|
||||
raise WorkerLeaseError("lease directory must not be a symlink")
|
||||
root.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||||
self.root = root.resolve()
|
||||
metadata = self.root.stat()
|
||||
if metadata.st_uid != os.geteuid() or metadata.st_mode & 0o022:
|
||||
raise WorkerLeaseError(
|
||||
"lease directory must be controller-owned and not writable by peers"
|
||||
)
|
||||
self.start, self.clock_ns = start, clock_ns
|
||||
self.ttl_ns = int(ttl_seconds * 1e9)
|
||||
self.path = self.root / "worker.lock"
|
||||
self.record_path = self.root / "owner.json"
|
||||
self.descriptor = os.open(
|
||||
self.path, os.O_RDWR | os.O_CREAT | os.O_CLOEXEC | os.O_NOFOLLOW, 0o600
|
||||
)
|
||||
self.released = False
|
||||
self.fenced = False
|
||||
self.renewals = 0
|
||||
self.pid = os.getpid()
|
||||
try:
|
||||
opened = os.fstat(self.descriptor)
|
||||
if not stat.S_ISREG(opened.st_mode) or opened.st_nlink != 1:
|
||||
raise WorkerLeaseError("lease lock must be one regular file")
|
||||
if opened.st_uid != os.geteuid() or opened.st_mode & 0o077:
|
||||
raise WorkerLeaseError("lease lock must be private")
|
||||
self.identity = (opened.st_dev, opened.st_ino)
|
||||
self._check_path()
|
||||
try:
|
||||
fcntl.flock(self.descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except BlockingIOError as exc:
|
||||
raise WorkerLeaseError("worker already owned") from exc
|
||||
self._check_path()
|
||||
self._check_previous()
|
||||
self.last_ns = self.clock_ns()
|
||||
_integer(self.last_ns, "worker monotonic time")
|
||||
self.deadline_ns = self.last_ns + self.ttl_ns
|
||||
write_json_atomic(self.record_path, self._record("active"))
|
||||
except BaseException:
|
||||
os.close(self.descriptor)
|
||||
self.released = self.fenced = True
|
||||
raise
|
||||
|
||||
def _check_path(self) -> None:
|
||||
current = self.path.lstat()
|
||||
if not stat.S_ISREG(current.st_mode) or (current.st_dev, current.st_ino) != self.identity:
|
||||
raise WorkerLeaseError("worker lock identity changed")
|
||||
|
||||
def _check_previous(self) -> None:
|
||||
try:
|
||||
metadata = self.record_path.lstat()
|
||||
except FileNotFoundError:
|
||||
return
|
||||
if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > 8192:
|
||||
raise WorkerLeaseError("invalid worker ownership record")
|
||||
if metadata.st_uid != os.geteuid() or metadata.st_mode & 0o077 or metadata.st_nlink != 1:
|
||||
raise WorkerLeaseError("worker ownership record must be private")
|
||||
try:
|
||||
record = json.loads(self.record_path.read_bytes())
|
||||
if (
|
||||
set(record) != {"schema_version", "state", "start"}
|
||||
or record["schema_version"] != LEASE_SCHEMA
|
||||
):
|
||||
raise ValueError("schema")
|
||||
previous = StreamStart.from_dict(record["start"])
|
||||
except (ValueError, TypeError, KeyError) as exc:
|
||||
raise WorkerLeaseError("corrupt worker ownership record") from exc
|
||||
if previous.worker_id != self.start.worker_id:
|
||||
raise WorkerLeaseError("controller directory belongs to another worker")
|
||||
if record["state"] != "released":
|
||||
raise WorkerLeaseError("previous owner unretired; resource-release recovery required")
|
||||
if self.start.lease_generation <= previous.lease_generation:
|
||||
raise WorkerLeaseError("lease generation must strictly increase")
|
||||
if self.start.epoch_id == previous.epoch_id:
|
||||
raise WorkerLeaseError("a new activation needs a new epoch")
|
||||
|
||||
def _record(self, state: str) -> dict[str, object]:
|
||||
return {"schema_version": LEASE_SCHEMA, "state": state, "start": self.start.to_dict()}
|
||||
|
||||
def check(self, start: StreamStart) -> None:
|
||||
if start != self.start:
|
||||
raise WorkerLeaseError("stream identity or lease generation mismatch")
|
||||
if self.released or self.fenced or os.getpid() != self.pid:
|
||||
raise WorkerLeaseError("worker authority is fenced")
|
||||
try:
|
||||
self._check_path()
|
||||
now = self.clock_ns()
|
||||
_integer(now, "worker monotonic time")
|
||||
if now < self.last_ns or now >= self.deadline_ns:
|
||||
raise WorkerLeaseError("worker lease expired or clock moved backwards")
|
||||
self.last_ns = now
|
||||
except (OSError, ValueError, WorkerLeaseError) as exc:
|
||||
self.fenced = True
|
||||
raise WorkerLeaseError("worker lease no longer current") from exc
|
||||
|
||||
def renew(self, start: StreamStart) -> None:
|
||||
self.check(start)
|
||||
self.deadline_ns = self.last_ns + self.ttl_ns
|
||||
self.renewals += 1
|
||||
|
||||
def _retire_after_verified_stop(self) -> None:
|
||||
"""Private supervisor boundary, not a caller-supplied release boolean."""
|
||||
if self.released:
|
||||
return
|
||||
self.fenced = True
|
||||
if os.getpid() != self.pid:
|
||||
raise WorkerLeaseError("only the owning controller can retire")
|
||||
self._check_path()
|
||||
write_json_atomic(self.record_path, self._record("released"))
|
||||
# Keep the stable file: unlink/recreate would split the lock domain.
|
||||
fcntl.flock(self.descriptor, fcntl.LOCK_UN)
|
||||
os.close(self.descriptor)
|
||||
self.released = True
|
||||
Reference in New Issue
Block a user