refactor(perception): share bounded streaming scheduler
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,147 @@
|
||||
"""Bounded stream ingress shared by serial GPU and chronological CPU stages.
|
||||
|
||||
No source duration, recording path, model, lease acquisition, or output history.
|
||||
Active inputs remain byte-accounted until their owner releases them. A completed
|
||||
GPU result shares the pending budget with ingress; it is not a third queue.
|
||||
These are scheduler primitives, not a replacement for graph terminal outcomes.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter, deque
|
||||
from collections.abc import Mapping
|
||||
from threading import Condition
|
||||
from typing import Any
|
||||
|
||||
StreamBundle = Mapping[str, Any]
|
||||
DISCARD_REASONS = frozenset(("cancelled", "gpu-failed", "processing-failed"))
|
||||
|
||||
|
||||
class StreamMailbox:
|
||||
def __init__(self, capacity: int = 2, byte_limit: int = 16 * 1024 * 1024) -> None:
|
||||
if type(capacity) is not int or not 1 <= capacity <= 2:
|
||||
raise ValueError("pending capacity must be 1..2")
|
||||
if type(byte_limit) is not int or not 1 <= byte_limit <= 16 * 1024 * 1024:
|
||||
raise ValueError("inflight byte budget must be 1..16 MiB")
|
||||
self.capacity, self.byte_limit = capacity, byte_limit
|
||||
self.condition = Condition()
|
||||
self.pending: deque[StreamBundle] = deque()
|
||||
self.bytes = self.peak_bytes = self.peak_pending = self.external_pending = 0
|
||||
self.drop_counts: Counter[str] = Counter()
|
||||
self._recent_drops: deque[dict[str, Any]] = deque(maxlen=256)
|
||||
self._owned: dict[int, tuple[StreamBundle, int]] = {}
|
||||
self._active: set[int] = set()
|
||||
self._last_sequence = -1
|
||||
self.done = False
|
||||
self.error: str | None = None
|
||||
|
||||
@property
|
||||
def dropped(self) -> list[dict[str, Any]]:
|
||||
"""Bounded diagnostic tail, NOT an all-session terminal journal."""
|
||||
with self.condition:
|
||||
return list(self._recent_drops)
|
||||
|
||||
@property
|
||||
def dropped_count(self) -> int:
|
||||
with self.condition:
|
||||
return sum(self.drop_counts.values())
|
||||
|
||||
@property
|
||||
def quiescent(self) -> bool:
|
||||
with self.condition:
|
||||
return self.done and not self._owned and not self.external_pending
|
||||
|
||||
def _drop(self, bundle: StreamBundle, reason: str) -> None:
|
||||
self.drop_counts[reason] += 1
|
||||
self._recent_drops.append({"sequence": bundle["sequence"], "reason": reason})
|
||||
|
||||
def _discard_pending(self, reason: str) -> None:
|
||||
old = self.pending.popleft()
|
||||
_, size = self._owned.pop(id(old))
|
||||
self.bytes -= size
|
||||
self._drop(old, reason)
|
||||
|
||||
def put(self, bundle: StreamBundle) -> bool:
|
||||
size, sequence = bundle["payload_bytes"], bundle["sequence"]
|
||||
if type(size) is not int or size <= 0 or type(sequence) is not int or sequence < 0:
|
||||
raise ValueError("invalid source sequence or payload size")
|
||||
with self.condition:
|
||||
if id(bundle) in self._owned:
|
||||
raise ValueError("input is already owned")
|
||||
if sequence <= self._last_sequence:
|
||||
raise ValueError("stream sequence must strictly increase")
|
||||
self._last_sequence = sequence
|
||||
if self.done or size > self.byte_limit:
|
||||
self._drop(bundle, "ingress-closed" if self.done else "byte-budget")
|
||||
return False
|
||||
while self.pending and (
|
||||
len(self.pending) + self.external_pending >= self.capacity
|
||||
or self.bytes + size > self.byte_limit
|
||||
):
|
||||
self._discard_pending("pending-overflow")
|
||||
if len(self.pending) + self.external_pending >= self.capacity:
|
||||
self._drop(bundle, "pending-overflow")
|
||||
return False
|
||||
if self.bytes + size > self.byte_limit:
|
||||
self._drop(bundle, "byte-budget")
|
||||
return False
|
||||
self.pending.append(bundle)
|
||||
self._owned[id(bundle)] = (bundle, size)
|
||||
self.bytes += size
|
||||
self.peak_bytes = max(self.peak_bytes, self.bytes)
|
||||
self.peak_pending = max(self.peak_pending, len(self.pending) + self.external_pending)
|
||||
self.condition.notify_all()
|
||||
return True
|
||||
|
||||
def reserve_completed(self) -> None:
|
||||
with self.condition:
|
||||
if self.external_pending:
|
||||
raise ValueError("only one completed GPU slot is permitted")
|
||||
while len(self.pending) + 1 > self.capacity:
|
||||
self._discard_pending("handoff-overflow")
|
||||
self.external_pending = 1
|
||||
self.peak_pending = max(self.peak_pending, len(self.pending) + 1)
|
||||
|
||||
def take_completed(self) -> None:
|
||||
with self.condition:
|
||||
if self.external_pending != 1:
|
||||
raise ValueError("completed GPU slot accounting mismatch")
|
||||
self.external_pending = 0
|
||||
|
||||
def take(self) -> StreamBundle | None:
|
||||
with self.condition:
|
||||
self.condition.wait_for(lambda: self.pending or self.done)
|
||||
if not self.pending:
|
||||
return None
|
||||
bundle = self.pending.popleft()
|
||||
self._active.add(id(bundle))
|
||||
return bundle
|
||||
|
||||
def release(self, bundle: StreamBundle, *, discard_reason: str | None = None) -> None:
|
||||
if discard_reason is not None and discard_reason not in DISCARD_REASONS:
|
||||
raise ValueError("unknown discard reason")
|
||||
with self.condition:
|
||||
if id(bundle) not in self._active or self._owned[id(bundle)][0] is not bundle:
|
||||
raise ValueError("input release ownership mismatch")
|
||||
self._active.remove(id(bundle))
|
||||
_, size = self._owned.pop(id(bundle))
|
||||
self.bytes -= size
|
||||
if discard_reason is not None:
|
||||
self._drop(bundle, discard_reason)
|
||||
|
||||
def finish(self, error: str | None = None) -> None:
|
||||
"""Close ingress, allowing accepted pending work to drain."""
|
||||
with self.condition:
|
||||
self.done = True
|
||||
self.error = self.error or error
|
||||
self.condition.notify_all()
|
||||
|
||||
def cancel(self, reason: str = "cancelled") -> None:
|
||||
"""Discard pending work, but never free another stage's active input."""
|
||||
if reason not in DISCARD_REASONS:
|
||||
raise ValueError("unknown cancellation reason")
|
||||
with self.condition:
|
||||
self.done = True
|
||||
while self.pending:
|
||||
self._discard_pending(reason)
|
||||
self.condition.notify_all()
|
||||
@@ -0,0 +1,116 @@
|
||||
"""One serial GPU lane overlapped with one chronological CPU consumer.
|
||||
|
||||
The callback owns all GPU model calls for a frame (e.g. DDRNet then RF-DETR).
|
||||
This class does not acquire the Worker lease or prove external GPU exclusivity.
|
||||
Close is cooperative: False means an active callback still owns resources and
|
||||
the supervisor must not admit another profile. No thread is forcefully killed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import traceback
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
from .streaming_queue import StreamBundle, StreamMailbox
|
||||
|
||||
|
||||
class SerialGpuStage:
|
||||
def __init__(
|
||||
self, mailbox: StreamMailbox, compute: Callable[[StreamBundle], Any], stop: threading.Event
|
||||
) -> None:
|
||||
self.mailbox, self.compute, self.stop = mailbox, compute, stop
|
||||
self.output_condition = threading.Condition()
|
||||
self.completed: tuple[StreamBundle, Any] | None = None
|
||||
self.completed_reserved = self.consumer_waiting = False
|
||||
self.output_slot = threading.Semaphore(1)
|
||||
self.finished = threading.Event()
|
||||
self.error: str | None = None
|
||||
self.peak_pending = self.direct_handoffs = self.buffered_handoffs = 0
|
||||
self.thread = threading.Thread(target=self._run, name="perception-serial-gpu", daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
def _run(self) -> None:
|
||||
active = None
|
||||
try:
|
||||
while not self.stop.is_set():
|
||||
# Reserve output BEFORE inference: no hidden third pending result.
|
||||
if not self.output_slot.acquire(timeout=0.05):
|
||||
continue
|
||||
active = self.mailbox.take()
|
||||
if active is None:
|
||||
break
|
||||
if self.stop.is_set():
|
||||
break
|
||||
result = self.compute(active)
|
||||
with self.output_condition:
|
||||
if self.stop.is_set():
|
||||
break
|
||||
if self.completed is not None:
|
||||
raise ValueError("completed GPU slot is already occupied")
|
||||
# A blocked consumer accepts a synchronous rendezvous, not
|
||||
# a pending slot that would spuriously evict newer ingress.
|
||||
reserved = not self.consumer_waiting
|
||||
if reserved:
|
||||
self.mailbox.reserve_completed()
|
||||
self.buffered_handoffs += 1
|
||||
else:
|
||||
self.direct_handoffs += 1
|
||||
self.completed = (active, result)
|
||||
active = None
|
||||
self.completed_reserved = reserved
|
||||
self.peak_pending = max(self.peak_pending, int(reserved))
|
||||
self.output_condition.notify_all()
|
||||
except Exception:
|
||||
self.error = traceback.format_exc()
|
||||
self.mailbox.finish(self.error)
|
||||
self.mailbox.cancel("gpu-failed")
|
||||
finally:
|
||||
if active is not None:
|
||||
self.mailbox.release(
|
||||
active, discard_reason="gpu-failed" if self.error else "cancelled"
|
||||
)
|
||||
self.finished.set()
|
||||
with self.output_condition:
|
||||
self.output_condition.notify_all()
|
||||
|
||||
def take(self) -> tuple[StreamBundle, Any] | None:
|
||||
while True:
|
||||
with self.output_condition:
|
||||
if self.stop.is_set():
|
||||
self.consumer_waiting = False
|
||||
return None
|
||||
if self.completed is not None:
|
||||
result = self.completed
|
||||
reserved = self.completed_reserved
|
||||
self.completed = None
|
||||
self.completed_reserved = self.consumer_waiting = False
|
||||
elif self.finished.is_set():
|
||||
self.consumer_waiting = False
|
||||
if self.error:
|
||||
raise RuntimeError(self.error) from None
|
||||
return None
|
||||
else:
|
||||
self.consumer_waiting = True
|
||||
self.output_condition.wait(timeout=0.05)
|
||||
continue
|
||||
if reserved:
|
||||
self.mailbox.take_completed()
|
||||
self.output_slot.release()
|
||||
return result
|
||||
|
||||
def close(self, *, timeout: float = 2.0) -> bool:
|
||||
if not 0 <= timeout <= 2:
|
||||
raise ValueError("stop timeout must be 0..2 seconds")
|
||||
self.stop.set()
|
||||
self.mailbox.cancel()
|
||||
self.thread.join(timeout=timeout)
|
||||
with self.output_condition:
|
||||
if self.completed is not None:
|
||||
if self.completed_reserved:
|
||||
self.mailbox.take_completed()
|
||||
self.mailbox.release(self.completed[0], discard_reason="cancelled")
|
||||
self.completed = None
|
||||
self.completed_reserved = False
|
||||
return not self.thread.is_alive()
|
||||
Reference in New Issue
Block a user