Spaces:
Paused
Paused
File size: 12,354 Bytes
5e0b58b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 | """Thread-safe runtime controller used by the interactive dashboard.
The controller intentionally does not know about HTTP, dashboard rendering or the
concrete NeuralNetwork implementation. It only coordinates a typed step callback.
This keeps the simulation core deterministic and allows the dashboard to request
bounded work without mutating the network from request-handler threads.
"""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from enum import StrEnum
from threading import Condition, RLock, Thread
from time import perf_counter, sleep
from typing import Final
from src.dashboard.models import JSONValue
StepCallback = Callable[[], None]
OptionalCallback = Callable[[], None]
class ControlMode(StrEnum):
"""Current execution state of the simulation controller."""
PAUSED = "paused"
RUNNING = "running"
STOPPED = "stopped"
FAULTED = "faulted"
class ControlCommand(StrEnum):
"""Commands exposed by the dashboard control plane."""
STEP = "step"
RUN = "run"
PAUSE = "pause"
STOP = "stop"
SNAPSHOT = "snapshot"
@dataclass(frozen=True, slots=True)
class ControlSnapshot:
"""Immutable public controller state."""
mode: ControlMode
ticks_executed: int
queued_ticks: int
loop_size: int
delay_ms: float
last_batch_ticks: int
last_batch_ms: float
total_runtime_ms: float
fault: str | None
can_snapshot: bool
def to_json(self) -> dict[str, JSONValue]:
"""Return a JSON-ready dashboard representation."""
return {
"mode": self.mode.value,
"ticks_executed": self.ticks_executed,
"queued_ticks": self.queued_ticks,
"loop_size": self.loop_size,
"delay_ms": self.delay_ms,
"last_batch_ticks": self.last_batch_ticks,
"last_batch_ms": self.last_batch_ms,
"total_runtime_ms": self.total_runtime_ms,
"fault": self.fault,
"can_snapshot": self.can_snapshot,
}
class RuntimeController:
"""Own the simulation loop and expose safe bounded control operations.
The simulation is stepped only by the controller worker. HTTP request threads
merely enqueue work or change controller state. This avoids concurrent calls to
``network.step()`` and makes step/run/pause semantics deterministic.
"""
_MIN_LOOP_SIZE: Final[int] = 1
_MAX_LOOP_SIZE: Final[int] = 1_000_000
_MAX_QUEUE: Final[int] = 10_000_000
_MAX_DELAY_MS: Final[float] = 60_000.0
def __init__(
self,
step_callback: StepCallback,
*,
snapshot_callback: OptionalCallback | None = None,
stop_callback: OptionalCallback | None = None,
initial_loop_size: int = 100,
initial_delay_ms: float = 0.0,
max_manual_ticks: int = 100_000,
) -> None:
self._step_callback = step_callback
self._snapshot_callback = snapshot_callback
self._stop_callback = stop_callback
self._lock = RLock()
self._condition = Condition(self._lock)
self._mode = ControlMode.PAUSED
self._queued_ticks = 0
self._ticks_executed = 0
self._loop_size = self._validate_loop_size(initial_loop_size)
self._delay_ms = self._validate_delay(initial_delay_ms)
self._max_manual_ticks = self._validate_tick_count(max_manual_ticks)
self._last_batch_ticks = 0
self._last_batch_ms = 0.0
self._total_runtime_ms = 0.0
self._fault: str | None = None
self._snapshot_requested = False
self._shutdown = False
self._worker = Thread(
target=self._worker_main,
name="brain5d-runtime-controller",
daemon=True,
)
self._worker.start()
def snapshot(self) -> ControlSnapshot:
"""Return a consistent immutable state snapshot."""
with self._lock:
return ControlSnapshot(
mode=self._mode,
ticks_executed=self._ticks_executed,
queued_ticks=self._queued_ticks,
loop_size=self._loop_size,
delay_ms=self._delay_ms,
last_batch_ticks=self._last_batch_ticks,
last_batch_ms=self._last_batch_ms,
total_runtime_ms=self._total_runtime_ms,
fault=self._fault,
can_snapshot=self._snapshot_callback is not None,
)
def step(self, ticks: int = 1) -> ControlSnapshot:
"""Queue an exact number of ticks while remaining in paused mode."""
count = self._validate_tick_count(ticks)
with self._condition:
self._ensure_operational()
if self._mode is ControlMode.RUNNING:
raise RuntimeError("Pause continuous execution before exact stepping.")
self._queued_ticks = min(self._MAX_QUEUE, self._queued_ticks + count)
self._condition.notify_all()
return self.snapshot()
def single_step(self) -> ControlSnapshot:
"""Queue exactly one operator-requested tick."""
return self.step(1)
def run_ticks(self, count: int) -> ControlSnapshot:
"""Queue a bounded number of operator-requested ticks."""
validated = self._validate_tick_count(count)
if validated > self._max_manual_ticks:
raise ValueError(f"count must be in [1, {self._max_manual_ticks}]")
return self.step(validated)
def run_loop(self, count: int | None = None) -> ControlSnapshot:
"""Start continuous execution or queue a bounded finite run."""
if count is None:
return self.run()
return self.run_ticks(count)
def run(self, *, loop_size: int | None = None) -> ControlSnapshot:
"""Start continuous execution in bounded loop batches."""
with self._condition:
self._ensure_operational()
if loop_size is not None:
self._loop_size = self._validate_loop_size(loop_size)
self._mode = ControlMode.RUNNING
self._condition.notify_all()
return self.snapshot()
def pause(self) -> ControlSnapshot:
"""Pause continuous execution after the current tick."""
with self._condition:
self._ensure_operational()
self._mode = ControlMode.PAUSED
self._condition.notify_all()
return self.snapshot()
def stop(self) -> ControlSnapshot:
"""Stop execution and invoke the optional application stop callback."""
callback: OptionalCallback | None
with self._condition:
if self._mode is ControlMode.STOPPED:
return self.snapshot()
self._mode = ControlMode.STOPPED
self._queued_ticks = 0
callback = self._stop_callback
self._condition.notify_all()
if callback is not None:
callback()
return self.snapshot()
def request_snapshot(self) -> ControlSnapshot:
"""Request a storage snapshot at the next safe worker boundary."""
if self._snapshot_callback is None:
raise RuntimeError("Snapshot capability is not configured.")
with self._condition:
self._ensure_operational()
self._snapshot_requested = True
self._condition.notify_all()
return self.snapshot()
def configure(
self,
*,
loop_size: int | None = None,
delay_ms: float | None = None,
) -> ControlSnapshot:
"""Adjust execution batch size and inter-tick pacing."""
with self._condition:
self._ensure_operational()
if loop_size is not None:
self._loop_size = self._validate_loop_size(loop_size)
if delay_ms is not None:
self._delay_ms = self._validate_delay(delay_ms)
self._condition.notify_all()
return self.snapshot()
def close(self, timeout: float = 2.0) -> None:
"""Terminate the worker thread during application shutdown."""
with self._condition:
self._shutdown = True
self._condition.notify_all()
self._worker.join(timeout=timeout)
def _worker_main(self) -> None:
while True:
with self._condition:
self._condition.wait_for(self._has_work_or_shutdown)
if self._shutdown:
return
mode = self._mode
if mode in {ControlMode.STOPPED, ControlMode.FAULTED}:
self._condition.wait(timeout=0.05)
continue
if self._queued_ticks > 0:
batch = min(self._queued_ticks, self._loop_size)
self._queued_ticks -= batch
elif mode is ControlMode.RUNNING:
batch = self._loop_size
elif self._snapshot_requested:
batch = 0
else:
continue
delay_ms = self._delay_ms
start = perf_counter()
completed = 0
try:
for _ in range(batch):
with self._lock:
if self._mode in {
ControlMode.STOPPED,
ControlMode.FAULTED,
}:
break
if (
mode is ControlMode.RUNNING
and self._mode is not ControlMode.RUNNING
):
break
self._step_callback()
completed += 1
if delay_ms > 0.0:
sleep(delay_ms / 1000.0)
self._flush_snapshot_request()
except Exception as exc: # controller boundary must contain core faults
with self._condition:
self._mode = ControlMode.FAULTED
self._queued_ticks = 0
self._fault = f"{type(exc).__name__}: {exc}"
self._condition.notify_all()
continue
elapsed_ms = (perf_counter() - start) * 1000.0
with self._condition:
self._ticks_executed += completed
self._last_batch_ticks = completed
self._last_batch_ms = elapsed_ms
self._total_runtime_ms += elapsed_ms
def _flush_snapshot_request(self) -> None:
with self._lock:
requested = self._snapshot_requested
self._snapshot_requested = False
callback = self._snapshot_callback
if requested and callback is not None:
callback()
def _has_work_or_shutdown(self) -> bool:
return (
self._shutdown
or self._snapshot_requested
or self._queued_ticks > 0
or self._mode is ControlMode.RUNNING
or self._mode in {ControlMode.STOPPED, ControlMode.FAULTED}
)
def _ensure_operational(self) -> None:
if self._mode is ControlMode.STOPPED:
raise RuntimeError("Runtime controller is stopped.")
if self._mode is ControlMode.FAULTED:
raise RuntimeError(self._fault or "Runtime controller is faulted.")
@classmethod
def _validate_tick_count(cls, ticks: int) -> int:
if isinstance(ticks, bool):
raise TypeError("ticks must be an integer")
if ticks < 1 or ticks > cls._MAX_QUEUE:
raise ValueError(f"ticks must be in [1, {cls._MAX_QUEUE}]")
return ticks
@classmethod
def _validate_loop_size(cls, loop_size: int) -> int:
if isinstance(loop_size, bool):
raise TypeError("loop_size must be an integer")
if not cls._MIN_LOOP_SIZE <= loop_size <= cls._MAX_LOOP_SIZE:
raise ValueError(
f"loop_size must be in [{cls._MIN_LOOP_SIZE}, {cls._MAX_LOOP_SIZE}]"
)
return loop_size
@classmethod
def _validate_delay(cls, delay_ms: float) -> float:
value = float(delay_ms)
if not 0.0 <= value <= cls._MAX_DELAY_MS:
raise ValueError(f"delay_ms must be in [0, {cls._MAX_DELAY_MS}]")
return value
|