From d84250d5504afecaac697e60034e0474386e5062 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 11 Jul 2026 14:44:00 -0500 Subject: [PATCH 1/4] feat:types: first layer of adding typing --- .github/workflows/python.yml | 36 ++++++++++++ README.md | 26 ++++++++- SmallPackage/OSlist.py | 38 ++++++++----- SmallPackage/SmallConfig.py | 51 +++++++++++------ SmallPackage/SmallIO.py | 35 ++++++++---- SmallPackage/SmallOS.py | 67 +++++++++++++++------- SmallPackage/SmallPID.py | 13 +++-- SmallPackage/SmallTask.py | 69 ++++++++++++++-------- SmallPackage/TaskState.py | 25 +++++--- SmallPackage/_types.py | 76 +++++++++++++++++++++++++ SmallPackage/awaitables.py | 39 +++++++++---- SmallPackage/list_util/binSearchList.py | 31 +++++++++- SmallPackage/list_util/linkedList.py | 18 +++--- SmallPackage/py.typed | 1 + pyproject.toml | 64 +++++++++++++++++++++ setup.py | 41 ++----------- 16 files changed, 469 insertions(+), 161 deletions(-) create mode 100644 .github/workflows/python.yml create mode 100644 SmallPackage/_types.py create mode 100644 SmallPackage/py.typed create mode 100644 pyproject.toml diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml new file mode 100644 index 0000000..3fd9862 --- /dev/null +++ b/.github/workflows/python.yml @@ -0,0 +1,36 @@ +name: Python + +on: + push: + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.10", "3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - run: python -m pip install --upgrade pip + - run: python -m pip install . + - run: python -m unittest discover -s tests -v + + typing-and-package: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - uses: actions/setup-python@v5 + with: + python-version: "3.10" + - run: python -m pip install --upgrade pip + - run: python -m pip install ".[dev]" + - run: pyright + - run: python -m build + diff --git a/README.md b/README.md index 48bf207..52e6099 100644 --- a/README.md +++ b/README.md @@ -73,17 +73,39 @@ That makes it a good fit for: Desktop development: ```bash -python3 -m venv .venv +python3.10 -m venv .venv source .venv/bin/activate -pip install -e . +python -m pip install -e ".[dev]" ``` +smallOS supports CPython 3.10 and newer. Python 3.6 through 3.9 are no +longer supported. MicroPython compatibility is maintained separately because +its language and standard-library support do not map directly to a CPython +release number; checker-only imports are kept off embedded runtime paths. + Run the test suite: ```bash python3 -m unittest discover -s tests -v ``` +Run the static type checker: + +```bash +pyright +``` + +Build the wheel and source distribution: + +```bash +python -m build +``` + +The package ships a `py.typed` marker. Type coverage is being tightened by +subsystem: configuration, awaitables, task lifecycle, scheduling, and core +utilities form the first checked boundary, while platform and protocol-client +implementations remain on the incremental typing backlog. + ## Quick Start Minimal desktop runtime: diff --git a/SmallPackage/OSlist.py b/SmallPackage/OSlist.py index dc60ab5..3c96836 100644 --- a/SmallPackage/OSlist.py +++ b/SmallPackage/OSlist.py @@ -10,9 +10,21 @@ queue per priority, and sleeping tasks live in a wake-time heap. """ +from __future__ import annotations + from collections import deque import heapq +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover + TYPE_CHECKING = False + +if TYPE_CHECKING: + from typing import Literal + + from .SmallTask import SmallTask + from .list_util.binSearchList import insert, search from .SmallPID import SmallPID @@ -22,7 +34,7 @@ class OSList(SmallPID): Combined PID registry and queue manager for the cooperative scheduler. """ - def __init__(self, priors=5, length=2**12): + def __init__(self, priors: int = 5, length: int = 2**12) -> None: """Create the PID registry plus ready/sleep queue structures.""" SmallPID.__init__(self, length) self.num_priorities = priors @@ -37,7 +49,7 @@ def resetCatSel(self): """Compatibility no-op kept for older callers.""" return - def insert(self, task): + def insert(self, task: SmallTask) -> int: """Assign a PID and register a task in the PID-sorted backing list.""" priority = task.priority if not 0 < priority < self.num_priorities: @@ -55,7 +67,7 @@ def insert(self, task): self.tasks.insert(index, task) return pid - def search(self, pid): + def search(self, pid: int) -> SmallTask | Literal[-1]: """Look up a task by PID.""" length = len(self.tasks) index = search(self.tasks, pid, 0, length, self.func) @@ -63,7 +75,7 @@ def search(self, pid): return -1 return self.tasks[index] - def delete(self, pid): + def delete(self, pid: int) -> int: """Remove a task from PID storage and watcher accounting.""" length = len(self.tasks) index = search(self.tasks, pid, 0, length, self.func) @@ -77,7 +89,7 @@ def delete(self, pid): self.freePID(pid) return 0 - def enqueue(self, task, front=False): + def enqueue(self, task: SmallTask, front: bool = False) -> int: """ Put a runnable task on its per-priority ready queue. @@ -99,7 +111,7 @@ def enqueue(self, task, front=False): task._queued = True return 0 - def pop(self): + def pop(self) -> SmallTask | None: """ Return the next runnable task. @@ -118,12 +130,12 @@ def pop(self): return task return None - def add_sleeping(self, task, wake_time): + def add_sleeping(self, task: SmallTask, wake_time: int) -> None: """Push a sleeping task onto the wake-time heap.""" self._sleep_seq += 1 heapq.heappush(self.sleeping, (wake_time, self._sleep_seq, task)) - def wake_sleeping(self, now): + def wake_sleeping(self, now: int) -> list[SmallTask]: """ Return every task whose scheduled wake time has arrived. @@ -140,7 +152,7 @@ def wake_sleeping(self, now): ready.append(task) return ready - def next_wake_time(self): + def next_wake_time(self) -> int | None: """Peek at the next valid wake time, discarding stale heap entries.""" while self.sleeping: wake_time, _, task = self.sleeping[0] @@ -150,18 +162,18 @@ def next_wake_time(self): return wake_time return None - def list(self): + def list(self) -> list[SmallTask]: """Return a snapshot list of currently registered tasks.""" return [task for task in self.tasks] - def isOnlyWatchers(self): + def isOnlyWatchers(self) -> bool: """Report whether every remaining task is marked as a watcher.""" return len(self.tasks) == self.numWatchers - def __len__(self): + def __len__(self) -> int: """Return the number of registered tasks.""" return len(self.tasks) - def __str__(self): + def __str__(self) -> str: """Return a newline-separated dump of all known tasks.""" return "\n".join([str(x) for x in self.tasks]) diff --git a/SmallPackage/SmallConfig.py b/SmallPackage/SmallConfig.py index e94b841..f3a051a 100644 --- a/SmallPackage/SmallConfig.py +++ b/SmallPackage/SmallConfig.py @@ -7,6 +7,19 @@ points. """ +from __future__ import annotations + +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover - exercised on constrained runtimes + TYPE_CHECKING = False + +if TYPE_CHECKING: + from os import PathLike + from typing import Any, Mapping + + from ._types import SmallOSConfigData + _DEFAULT_CLIENT_DEFAULTS = { "stream": { @@ -64,12 +77,12 @@ class SmallOSConfig: def __init__( self, - task_capacity=2**10, - priority_levels=10, - io_buffer_length=1024, - eternal_watchers=False, - client_defaults=None, - ): + task_capacity: int = 2**10, + priority_levels: int = 10, + io_buffer_length: int = 1024, + eternal_watchers: bool = False, + client_defaults: Mapping[str, Mapping[str, int]] | None = None, + ) -> None: self.task_capacity = self._validate_positive_int("task_capacity", task_capacity) self.priority_levels = self._validate_positive_int("priority_levels", priority_levels) if self.priority_levels < 2: @@ -79,7 +92,7 @@ def __init__( self.client_defaults = self._normalize_client_defaults(client_defaults) @staticmethod - def _validate_positive_int(name, value): + def _validate_positive_int(name: str, value: int) -> int: if not isinstance(value, int): raise TypeError("{} must be an int.".format(name)) if value <= 0: @@ -87,7 +100,7 @@ def _validate_positive_int(name, value): return value @staticmethod - def _validate_non_negative_int(name, value): + def _validate_non_negative_int(name: str, value: int) -> int: if not isinstance(value, int): raise TypeError("{} must be an int.".format(name)) if value < 0: @@ -95,14 +108,16 @@ def _validate_non_negative_int(name, value): return value @classmethod - def _default_client_defaults(cls): + def _default_client_defaults(cls) -> dict[str, dict[str, int]]: defaults = {} for section, values in _DEFAULT_CLIENT_DEFAULTS.items(): defaults[section] = dict(values) return defaults @classmethod - def _normalize_client_defaults(cls, client_defaults): + def _normalize_client_defaults( + cls, client_defaults: Mapping[str, Mapping[str, int]] | None + ) -> dict[str, dict[str, int]]: if client_defaults is None: return cls._default_client_defaults() if not isinstance(client_defaults, dict): @@ -128,12 +143,14 @@ def _normalize_client_defaults(cls, client_defaults): return normalized @classmethod - def default(cls): + def default(cls) -> SmallOSConfig: """Return a fresh config populated with the runtime defaults.""" return cls() @classmethod - def from_dict(cls, data): + def from_dict( + cls, data: SmallOSConfig | Mapping[str, Any] | None + ) -> SmallOSConfig: """Build a config from canonical keys or supported aliases.""" if data is None: return cls() @@ -151,19 +168,19 @@ def from_dict(cls, data): ) @classmethod - def from_json_file(cls, path): + def from_json_file(cls, path: str | PathLike[str]) -> SmallOSConfig: """Load a config from a JSON file on desktop Python or MicroPython.""" json_mod = _import_json_module() with open(path, "r") as handle: return cls.from_dict(json_mod.load(handle)) - def copy(self, **updates): + def copy(self, **updates: Any) -> SmallOSConfig: """Return a new config with a few fields replaced.""" - data = self.to_dict() + data: dict[str, Any] = dict(self.to_dict()) data.update(updates) return type(self).from_dict(data) - def to_dict(self): + def to_dict(self) -> SmallOSConfigData: """Return a plain-serializable dictionary with canonical config keys.""" return { "task_capacity": self.task_capacity, @@ -175,7 +192,7 @@ def to_dict(self): }, } - def client_defaults_for(self, section): + def client_defaults_for(self, section: str) -> dict[str, int]: """ Return one client section merged with the shared stream-level defaults. diff --git a/SmallPackage/SmallIO.py b/SmallPackage/SmallIO.py index 883d49a..187c3f7 100644 --- a/SmallPackage/SmallIO.py +++ b/SmallPackage/SmallIO.py @@ -7,8 +7,20 @@ back to app view flushes the buffered output in order. """ +from __future__ import annotations + from collections import deque +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover + TYPE_CHECKING = False + +if TYPE_CHECKING: + from typing import Any + + from ._types import TerminalStatus + class SmallIO: """ @@ -22,25 +34,26 @@ class SmallIO: - flush buffered output """ - def __init__(self, buffer_length): + def __init__(self, buffer_length: int) -> None: """Set up the shell/app terminal split plus the app output buffer.""" + self.kernel: Any = None self.terminalToggle = False self.buffer_length = max(0, int(buffer_length)) self.appPrintQueue = deque(maxlen=self.buffer_length) if self.buffer_length else deque() return - def _coerce_message(self, *args): + def _coerce_message(self, *args: object) -> str: """Join arbitrary print arguments into one text payload.""" return "".join(str(arg) for arg in args) - def _write_direct(self, msg): + def _write_direct(self, msg: str) -> bool: """Write straight to the active kernel when one is attached.""" if not getattr(self, "kernel", None): return False self.kernel.write(msg) return True - def print(self, *args): + def print(self, *args: object) -> None: """ Write application output. @@ -57,7 +70,7 @@ def print(self, *args): self.appPrintQueue.append(msg) return - def sPrint(self, *args, force=False): + def sPrint(self, *args: object, force: bool = False) -> None: """ Write shell or OS output. @@ -70,7 +83,7 @@ def sPrint(self, *args, force=False): self._write_direct(msg) return - def terminalStatus(self): + def terminalStatus(self) -> TerminalStatus: """Return a small snapshot of the terminal/buffer state.""" return { "terminal_visible": bool(self.terminalToggle), @@ -78,17 +91,17 @@ def terminalStatus(self): "buffer_length": self.buffer_length, } - def getBufferedOutput(self): + def getBufferedOutput(self) -> list[str]: """Return a copy of the buffered app output.""" return list(self.appPrintQueue) - def clearBufferedOutput(self): + def clearBufferedOutput(self) -> int: """Drop every buffered app message and return how many were removed.""" removed = len(self.appPrintQueue) self.appPrintQueue.clear() return removed - def flushBufferedOutput(self): + def flushBufferedOutput(self) -> int: """Write all buffered app output immediately and return the flush count.""" flushed = 0 while self.appPrintQueue: @@ -97,7 +110,7 @@ def flushBufferedOutput(self): flushed += 1 return flushed - def setTerminalMode(self, enabled): + def setTerminalMode(self, enabled: bool) -> TerminalStatus: """Explicitly switch between shell view and app view.""" enabled = bool(enabled) if self.terminalToggle == enabled: @@ -109,6 +122,6 @@ def setTerminalMode(self, enabled): self.flushBufferedOutput() return self.terminalStatus() - def toggleTerminal(self): + def toggleTerminal(self) -> TerminalStatus: """Toggle between shell view and app view.""" return self.setTerminalMode(not self.terminalToggle) diff --git a/SmallPackage/SmallOS.py b/SmallPackage/SmallOS.py index d1713aa..9f95180 100644 --- a/SmallPackage/SmallOS.py +++ b/SmallPackage/SmallOS.py @@ -11,6 +11,21 @@ - a runtime shape that is easier to port to MicroPython """ +from __future__ import annotations + +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover + TYPE_CHECKING = False + +if TYPE_CHECKING: + from collections.abc import Iterable + from typing import Any + + from .Kernel import Kernel + from .SmallTask import SmallTask + from ._types import ErrorHandler, RuntimeErrorEvent, SmallOSConfigData + from .awaitables import TaskInstruction from .SmallIO import SmallIO from .SmallConfig import SmallOSConfig @@ -31,7 +46,12 @@ class SmallOS(SmallIO): and easier to customize. """ - def __init__(self, size=None, config=None, **kwargs): + def __init__( + self, + size: int | None = None, + config: SmallOSConfig | SmallOSConfigData | None = None, + **kwargs: Any, + ) -> None: """ Create the runtime shell plus its task and shell registries. @@ -79,11 +99,11 @@ def __init__(self, size=None, config=None, **kwargs): shells.setOS(self) self.shells.append(shells) - def startOS(self): + def startOS(self) -> None: """Compatibility entrypoint kept from the earlier project API.""" return self.start() - def start(self): + def start(self) -> None: """ Run the scheduler until no live tasks remain. @@ -113,12 +133,12 @@ def start(self): return return - def next(self): + def next(self) -> SmallTask | None: """Return the next runnable task without advancing the main loop.""" self.cursor = self.tasks.pop() return self.cursor - def fork(self, children): + def fork(self, children: SmallTask | list[SmallTask]) -> int | list[int]: """Register one task or a list of tasks with the runtime.""" if isinstance(children, list): ids = [] @@ -127,7 +147,7 @@ def fork(self, children): return ids return self._fork_one(children) - def _fork_one(self, task): + def _fork_one(self, task: SmallTask) -> int: """Assign a PID, attach the runtime, and enqueue the task if runnable.""" pid = self.tasks.insert(task) if pid == -1: @@ -138,17 +158,19 @@ def _fork_one(self, task): self.tasks.enqueue(task) return pid - def setKernel(self, kernel): + def setKernel(self, kernel: Kernel) -> SmallOS: """Attach the platform abstraction used for time and output.""" self.kernel = kernel return self - def setEternalWatchers(self, isEternalWatcherPresent): + def setEternalWatchers(self, isEternalWatcherPresent: bool) -> SmallOS: """Control whether the runtime exits once only watcher tasks remain.""" self.eternalWatchers = isEternalWatcherPresent return self - def setErrorHandler(self, handler, include_cancelled=False): + def setErrorHandler( + self, handler: ErrorHandler | None, include_cancelled: bool = False + ) -> SmallOS: """ Install a best-effort runtime error observer for failed tasks. @@ -196,7 +218,13 @@ def _idle_until_next_task(self): self.kernel.sleep_ms(timeout) return True - def resume_task(self, task, value=_MISSING, exc=None, front=False): + def resume_task( + self, + task: SmallTask, + value: Any = _MISSING, + exc: BaseException | None = None, + front: bool = False, + ) -> int: """ Requeue a blocked task with the value or exception that completed it. @@ -213,13 +241,13 @@ def resume_task(self, task, value=_MISSING, exc=None, front=False): self.tasks.enqueue(task, front=front) return 0 - def on_signal(self, task, sig): + def on_signal(self, task: SmallTask, sig: int) -> None: """Wake a task immediately if it is actively waiting on ``sig``.""" if task._blocked_reason == "signal" and task._waiting_signal == sig: task.signals[sig] = 0 self.resume_task(task, value=sig, front=True) - def _handle_yield(self, task, yielded): + def _handle_yield(self, task: SmallTask, yielded: Any) -> None: """ Interpret one scheduler instruction emitted by a task. @@ -320,13 +348,14 @@ def _handle_yield(self, task, yielded): task.fail(UnsupportedAwaitableError("Unknown instruction {!r}".format(operation))) self._finalize_task(task) - def _resolve_task(self, target): + def _resolve_task(self, target: int | SmallTask) -> SmallTask | None: """Normalize either a task object or a PID to a task object.""" - if hasattr(target, "getID"): + if not isinstance(target, int): return target - return self.tasks.search(target) + found = self.tasks.search(target) + return None if found == -1 else found - def _normalize_targets(self, targets): + def _normalize_targets(self, targets: Iterable[int | SmallTask]) -> list[SmallTask] | None: """Resolve a join target list while preserving caller-specified order.""" normalized = [] seen = set() @@ -423,7 +452,7 @@ def _register_io_wait(self, task, io_obj, mode): if task not in waiters[io_obj]: waiters[io_obj].append(task) - def _wake_io_tasks(self, timeout_ms=0): + def _wake_io_tasks(self, timeout_ms: int | None = 0): """ Ask the kernel which I/O objects are ready and resume their waiters. @@ -647,7 +676,7 @@ def _finalize_task(self, task): if failure_event is not None: self._dispatch_error_handler(failure_event) - def cancel_task(self, task, recursive=False): + def cancel_task(self, task: int | SmallTask, recursive: bool = False) -> int: """Cancel a task by object or PID and optionally cancel its descendants.""" target = self._resolve_task(task) if target is None or target == -1: @@ -667,7 +696,7 @@ def cancel_task(self, task, recursive=False): self._finalize_task(target) return 0 - def __str__(self): + def __str__(self) -> str: """Return a human-readable dump of the currently registered tasks.""" all_tasks = list(self.tasks.tasks) string = "SmallOS\n" diff --git a/SmallPackage/SmallPID.py b/SmallPackage/SmallPID.py index 5d28a45..b38e762 100644 --- a/SmallPackage/SmallPID.py +++ b/SmallPackage/SmallPID.py @@ -1,13 +1,16 @@ -class SmallPID(): +from __future__ import annotations + + +class SmallPID: ''' @class smallPID() - Creates and keeps track of avialable Process ID's ''' - def __init__(self, max=2**16): + def __init__(self, max: int = 2**16) -> None: ''' @function __init__() - Initializes essential maintenence structures. @@ -21,7 +24,7 @@ def __init__(self, max=2**16): return - def newPID(self): + def newPID(self) -> int: ''' @function newPID() - Creates a new valid PID. @return - int - positive integer on success @@ -40,7 +43,7 @@ def newPID(self): return -1 - def freePID(self,pid): + def freePID(self, pid: int) -> None: ''' @function freePID - removes the pid from the used pid set() making the pid available for future @@ -49,4 +52,4 @@ def freePID(self,pid): ''' if pid in self.usedPID: self.usedPID.remove(pid) - return \ No newline at end of file + return diff --git a/SmallPackage/SmallTask.py b/SmallPackage/SmallTask.py index 2a88215..91a47a4 100644 --- a/SmallPackage/SmallTask.py +++ b/SmallPackage/SmallTask.py @@ -7,8 +7,24 @@ small because most task-local lifecycle details live here. """ +from __future__ import annotations + import inspect +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover - exercised on constrained runtimes + TYPE_CHECKING = False + +if TYPE_CHECKING: + from collections.abc import Iterable + from typing import Any, Generic, TypeVar + + from .SmallOS import SmallOS + from ._types import TaskRoutine + + T = TypeVar("T") + from .awaitables import join_instruction, join_all_instruction from .SmallErrors import PIDError, TaskCancelledError from .SmallSignals import SmallSignals @@ -19,7 +35,7 @@ _MISSING = object() -class SmallTask(SmallSignals, Node): +class SmallTask(SmallSignals, Node, Generic[T] if TYPE_CHECKING else object): """ Priority-scheduled unit of execution managed by ``SmallOS``. @@ -29,7 +45,7 @@ class SmallTask(SmallSignals, Node): to ``asyncio``. """ - def __init__(self, priority, routine, **kwargs): + def __init__(self, priority: int, routine: TaskRoutine[T], **kwargs: Any) -> None: """ Create a task shell around a routine or coroutine object. @@ -51,7 +67,7 @@ def __init__(self, priority, routine, **kwargs): self.updateFunc = None self.routine = routine - self._coroutine = None + self._coroutine: Any = None self._done = False self._result = None self._exception = None @@ -87,17 +103,17 @@ def __init__(self, priority, routine, **kwargs): self.args = kwargs["args"] @property - def done(self): + def done(self) -> bool: """Whether the task has reached a terminal state.""" return self._done @property - def result(self): + def result(self) -> T | None: """Return the stored result after successful completion.""" return self._result @property - def exception(self): + def exception(self) -> BaseException | None: """Return the stored terminal exception, if any.""" return self._exception @@ -193,7 +209,7 @@ def update(self): return 0 return -1 - def complete(self, result): + def complete(self, result: T | None) -> T | None: """Mark the task as successfully finished and store its result.""" self._done = True self._result = result @@ -203,7 +219,7 @@ def complete(self, result): self.state.update({"return_status": 0, "result": result}, "system") return result - def fail(self, exc): + def fail(self, exc: BaseException) -> BaseException: """Mark the task as failed and store its terminal exception.""" self._done = True self._exception = exc @@ -213,7 +229,7 @@ def fail(self, exc): self.state.update({"return_status": -1, "exception": exc}, "system") return exc - def cancel(self, message="Task cancelled"): + def cancel(self, message: str = "Task cancelled") -> None: """ Force the task into a cancelled terminal state. @@ -252,7 +268,7 @@ def block(self, reason): self.isSleep = 1 if reason == "sleep" else 0 self.state.update({"return_status": 1, "blocked_reason": reason}, "system") - def setID(self, pid): + def setID(self, pid: int) -> None: """Assign the PID chosen by ``SmallOS`` exactly once.""" if isinstance(pid, int): if self.pid == -1: @@ -263,11 +279,11 @@ def setID(self, pid): raise TypeError("PID must be type Int") return - def getID(self): + def getID(self) -> int: """Return the task PID.""" return self.pid - def setOS(self, OS): + def setOS(self, OS: SmallOS) -> None: """Attach the task to its owning runtime.""" self.OS = OS @@ -281,7 +297,9 @@ def build(self, priority, task, ready=1, name="", parent=None): parent=parent, ) - def spawn(self, routine, priority=None, **kwargs): + def spawn( + self, routine: TaskRoutine[Any] | SmallTask[Any], priority: int | None = None, **kwargs: Any + ) -> SmallTask[Any]: """ Create and register a child task on the same runtime. @@ -291,11 +309,12 @@ def spawn(self, routine, priority=None, **kwargs): if not self.OS: raise RuntimeError("Task must belong to an OS before it can spawn children.") - child = routine - if not isinstance(child, SmallTask): + if isinstance(routine, SmallTask): + child: SmallTask[Any] = routine + if priority is not None: + child.priority = priority + else: child = SmallTask(priority or self.priority, routine, **kwargs) - elif priority is not None: - child.priority = priority child.parent = self self.OS.fork(child) @@ -307,20 +326,20 @@ def fork(self, new_task): child = self.spawn(new_task) return child.getID() - def join(self, child): + def join(self, child: int | SmallTask[Any]): """Return the awaitable used to wait for one child task.""" return join_instruction(child) - def join_all(self, children): + def join_all(self, children: Iterable[int | SmallTask[Any]]): """Return the awaitable used to wait for several child tasks.""" return join_all_instruction(children) - def add_join_waiter(self, waiter): + def add_join_waiter(self, waiter: SmallTask[Any]) -> None: """Register a task that is currently waiting on this task.""" if waiter not in self._join_waiters: self._join_waiters.append(waiter) - def discard_join_waiter(self, waiter): + def discard_join_waiter(self, waiter: SmallTask[Any]) -> None: """Remove a waiter once it no longer depends on this task.""" while waiter in self._join_waiters: self._join_waiters.remove(waiter) @@ -333,15 +352,15 @@ def kill(self, flags=None): return -1 return self.OS.cancel_task(self, recursive="-r" in flags) - def getExeStatus(self): + def getExeStatus(self) -> bool: """Report whether the scheduler may run this task right now.""" return bool(self.isReady) and not self.isLocked and not self._done - def getDelStatus(self): + def getDelStatus(self) -> bool: """Report whether the task is terminal and removable.""" return self._done and not self.isWatcher - def stat(self): + def stat(self) -> str: """Return an expanded debug dump for shell-style inspection.""" msg = "\nisReady={}\nisWaiting={}\nisSleep={}\n".format( self.isReady, @@ -354,7 +373,7 @@ def stat(self): msg += "exception={!r}\n".format(self.exception) return str(self) + msg - def __str__(self): + def __str__(self) -> str: """Return a compact single-line summary of the task state.""" name = self.name or "Unamed Process" return ( diff --git a/SmallPackage/TaskState.py b/SmallPackage/TaskState.py index cedef44..2cf8195 100644 --- a/SmallPackage/TaskState.py +++ b/SmallPackage/TaskState.py @@ -1,25 +1,35 @@ +from __future__ import annotations + import copy +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover + TYPE_CHECKING = False + +if TYPE_CHECKING: + from typing import Any + ''' File Deprecated as of concurrency branch update. ''' -class TaskState(): +class TaskState: ''' @class taskState - state management class for keeping track of system stats and data of task. ''' - def __init__(self): + def __init__(self) -> None: self._state = dict() self._state['system'] = dict() self._state['data'] = dict() - def update(self,dict_blob,namespace='data'): + def update(self, dict_blob: dict[Any, Any], namespace: str = 'data') -> None: ''' @function updateState() - updates the contents of the state object. @@ -32,7 +42,7 @@ def update(self,dict_blob,namespace='data'): return - def isFree(self,key,namespace='data'): + def isFree(self, key: Any, namespace: str = 'data') -> bool: ''' @function isFree() - checks to see if the key is free for use in the state dict. @@ -50,7 +60,7 @@ def isFree(self,key,namespace='data'): return True - def free(self,key,namespace='data'): + def free(self, key: Any, namespace: str = 'data') -> int: ''' @function free() - Will free the selected key from the state. @param - key - key to be deleted. @@ -65,7 +75,7 @@ def free(self,key,namespace='data'): return -1 - def getState(self,key=None,namespace='data'): + def getState(self, key: Any = None, namespace: str = 'data') -> tuple[Any, int]: ''' @function getState - can return the entire state or just one element of the state returns a -1 if the requested key does not exist. @@ -83,6 +93,3 @@ def getState(self,key=None,namespace='data'): return copy.deepcopy(self._state[namespace][key]), 0 else: return None, -1 - - - diff --git a/SmallPackage/_types.py b/SmallPackage/_types.py new file mode 100644 index 0000000..b217170 --- /dev/null +++ b/SmallPackage/_types.py @@ -0,0 +1,76 @@ +"""Shared static types for smallOS. + +Runtime modules import these definitions only while type checking so embedded +targets do not need to provide the :mod:`typing` module. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence +from typing import Any, Protocol, TypeAlias, TypedDict, TypeVar + + +T = TypeVar("T") + + +class TaskLike(Protocol): + """Minimum task interface used by queues and scheduler integrations.""" + + priority: int + isWatcher: bool + @property + def done(self) -> bool: ... + + def getID(self) -> int: ... + + +class KernelLike(Protocol): + """Platform operations consumed by the scheduler and output layer.""" + + def write(self, msg: str) -> Any: ... + def scheduler_now_ms(self) -> int: ... + def sleep_ms(self, delay_ms: int) -> None: ... + def io_wait( + self, + readables: Iterable[Any], + writables: Iterable[Any], + timeout_ms: int | None = None, + ) -> tuple[Sequence[Any], Sequence[Any]]: ... + + +class RuntimeErrorEvent(TypedDict): + task_id: int + task_name: str + parent_id: int | None + exception: BaseException + exception_type: type[BaseException] + exception_repr: str + is_cancelled: bool + blocked_reason: str | None + waiting_signal: int | None + io_wait_mode: str | None + join_target_id: int | None + join_pending_ids: list[int] + traceback_text: str + + +class SmallOSConfigData(TypedDict, total=False): + task_capacity: int + oslist_length: int + priority_levels: int + num_categories: int + io_buffer_length: int + eternal_watchers: bool + client_defaults: Mapping[str, Mapping[str, int]] + clients: Mapping[str, Mapping[str, int]] + + +class TerminalStatus(TypedDict): + terminal_visible: bool + buffered_messages: int + buffer_length: int + + +TaskRoutine: TypeAlias = Callable[..., T | Awaitable[T]] +ErrorHandler: TypeAlias = Callable[[RuntimeErrorEvent], None] +TaskTarget: TypeAlias = int | TaskLike diff --git a/SmallPackage/awaitables.py b/SmallPackage/awaitables.py index 5a08fe9..d3cb58e 100644 --- a/SmallPackage/awaitables.py +++ b/SmallPackage/awaitables.py @@ -11,6 +11,21 @@ - the scheduler decides when that wait is satisfied """ +from __future__ import annotations + +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover - exercised on constrained runtimes + TYPE_CHECKING = False + +if TYPE_CHECKING: + from collections.abc import Generator, Iterable + from typing import Any, Generic, TypeVar + + from ._types import TaskTarget + + T = TypeVar("T") + class TaskInstruction: """ @@ -21,18 +36,18 @@ class TaskInstruction: easier to debug and more realistic to port to constrained runtimes. """ - def __init__(self, operation, **payload): + def __init__(self, operation: str, **payload: Any) -> None: self.operation = operation self.payload = payload - def __repr__(self): + def __repr__(self) -> str: return "TaskInstruction(operation={!r}, payload={!r})".format( self.operation, self.payload, ) -class _InstructionAwaitable: +class _InstructionAwaitable(Generic[T] if TYPE_CHECKING else object): """ Tiny awaitable wrapper shared by the public helpers below. @@ -41,45 +56,45 @@ class _InstructionAwaitable: ``await`` expression in user code. """ - def __init__(self, instruction): + def __init__(self, instruction: TaskInstruction) -> None: self.instruction = instruction - def __await__(self): + def __await__(self) -> Generator[TaskInstruction, Any, T]: """Yield one scheduler instruction and then return the resume value.""" result = yield self.instruction return result -def sleep_instruction(seconds): +def sleep_instruction(seconds: float) -> _InstructionAwaitable[None]: """Create the awaitable used for cooperative sleeping.""" return _InstructionAwaitable(TaskInstruction("sleep", seconds=seconds)) -def wait_signal_instruction(signal): +def wait_signal_instruction(signal: int) -> _InstructionAwaitable[int]: """Create the awaitable used for waiting on a task signal.""" return _InstructionAwaitable(TaskInstruction("wait_signal", signal=signal)) -def yield_now_instruction(): +def yield_now_instruction() -> _InstructionAwaitable[None]: """Create the awaitable used for a voluntary scheduler yield.""" return _InstructionAwaitable(TaskInstruction("yield_now")) -def join_instruction(target): +def join_instruction(target: TaskTarget) -> _InstructionAwaitable[Any]: """Create the awaitable used for waiting on a single task.""" return _InstructionAwaitable(TaskInstruction("join", target=target)) -def join_all_instruction(targets): +def join_all_instruction(targets: Iterable[TaskTarget]) -> _InstructionAwaitable[list[Any]]: """Create the awaitable used for waiting on several tasks at once.""" return _InstructionAwaitable(TaskInstruction("join_all", targets=list(targets))) -def wait_readable_instruction(io_obj): +def wait_readable_instruction(io_obj: Any) -> _InstructionAwaitable[Any]: """Create the awaitable used for waiting until an I/O object is readable.""" return _InstructionAwaitable(TaskInstruction("wait_readable", io_obj=io_obj)) -def wait_writable_instruction(io_obj): +def wait_writable_instruction(io_obj: Any) -> _InstructionAwaitable[Any]: """Create the awaitable used for waiting until an I/O object is writable.""" return _InstructionAwaitable(TaskInstruction("wait_writable", io_obj=io_obj)) diff --git a/SmallPackage/list_util/binSearchList.py b/SmallPackage/list_util/binSearchList.py index 2c55f6a..8eb90f3 100644 --- a/SmallPackage/list_util/binSearchList.py +++ b/SmallPackage/list_util/binSearchList.py @@ -1,3 +1,14 @@ +from __future__ import annotations + +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover + TYPE_CHECKING = False + +if TYPE_CHECKING: + from collections.abc import Callable, Sequence + from typing import Any + # from numpy import log as ln # import math # import random @@ -5,7 +16,13 @@ -def search(data,target,l,r, func=None): +def search( + data: Sequence[Any], + target: Any, + l: int, + r: int, + func: Callable[[Sequence[Any], int], Any] | None = None, +) -> int: ''' @function search - Performs a binary search on a sorted list and returns the index of the found element. @@ -26,6 +43,7 @@ def search(data,target,l,r, func=None): if func == None: func = lambda dat,index: dat[index] + assert func is not None mid = ((r - l) / 2) mid = int(mid) @@ -42,7 +60,13 @@ def search(data,target,l,r, func=None): return -1 -def insert(data,target,l,r,func=None): +def insert( + data: Sequence[Any], + target: Any, + l: int, + r: int, + func: Callable[[Sequence[Any], int], Any] | None = None, +) -> int: ''' @function insert - Performs a binary search on a sorted list and returns the index of where the element should be added. @@ -63,6 +87,7 @@ def insert(data,target,l,r,func=None): if func == None: func = lambda dat,index: dat[index] + assert func is not None mid = (r - l)/2 mid = int(mid) @@ -143,4 +168,4 @@ def insert(data,target,l,r,func=None): # count = 0 # test_search() -# test_insert() \ No newline at end of file +# test_insert() diff --git a/SmallPackage/list_util/linkedList.py b/SmallPackage/list_util/linkedList.py index 16c414a..9003906 100644 --- a/SmallPackage/list_util/linkedList.py +++ b/SmallPackage/list_util/linkedList.py @@ -3,7 +3,9 @@ @file linkedList - modules to create and manipulate doubly linked list. ''' -def insertPrev(root, newNode): +from __future__ import annotations + +def insertPrev(root: Node, newNode: Node) -> None: ''' @function insertPrev() - takes in a rootNode and newNode and inserts the newNode behind the rootNode. @@ -24,7 +26,7 @@ def insertPrev(root, newNode): return -def insertNext(root,newNode): +def insertNext(root: Node, newNode: Node) -> None: ''' @function insertNext() - takes in a rootNode and newNode and inserts the newNode infront of the rootNode. @@ -45,7 +47,7 @@ def insertNext(root,newNode): return -def removeNode(node): +def removeNode(node: Node) -> None: ''' @function removeNode() - Removes the Node() from the doubly-linked-list @@ -67,7 +69,7 @@ def removeNode(node): -class Node(): +class Node: ''' @class Node() - Holds no data on its own, developed with the intent of being a super class @@ -79,13 +81,11 @@ class Node(): are not taken. *** ''' - next=None - prev=None + next: Node | None = None + prev: Node | None = None - def __init__(self): + def __init__(self) -> None: self.next = None self.prev = None return - - diff --git a/SmallPackage/py.typed b/SmallPackage/py.typed new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/SmallPackage/py.typed @@ -0,0 +1 @@ + diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..81c74ca --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,64 @@ +[build-system] +requires = ["setuptools>=77", "setuptools-scm>=8"] +build-backend = "setuptools.build_meta" + +[project] +name = "SmallPackage" +dynamic = ["version"] +description = "Concurrent and priority-oriented task management system" +readme = "README.md" +requires-python = ">=3.10" +license = "MIT" +license-files = ["LICENSE"] +authors = [{ name = "Michael E" }] +dependencies = [] +classifiers = [ + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Operating System :: OS Independent", + "Typing :: Typed", +] + +[project.optional-dependencies] +dev = [ + "build>=1.2", + "pyright>=1.1.390", +] + +[project.urls] +Documentation = "https://github.com/MikiEEE/SmallOS" +Source = "https://github.com/MikiEEE/SmallOS" +Issues = "https://github.com/MikiEEE/SmallOS/issues" + +[tool.setuptools.packages.find] +include = ["SmallPackage*"] + +[tool.setuptools.package-data] +SmallPackage = ["py.typed"] +"SmallPackage.clients" = ["README.md"] + +[tool.setuptools_scm] +fallback_version = "0.0.0" + +[tool.pyright] +pythonVersion = "3.10" +typeCheckingMode = "basic" +include = ["SmallPackage", "demos"] +exclude = ["build", "dist"] +ignore = [ + "SmallPackage/Kernel.py", + "SmallPackage/SmallSignals.py", + "SmallPackage/clients", + "SmallPackage/shells.py", + "demos", +] +reportMissingTypeStubs = false +reportUnknownMemberType = false +reportUnknownArgumentType = false +reportUnknownVariableType = false +reportUnknownLambdaType = false +reportMissingImports = false diff --git a/setup.py b/setup.py index cf6dbd4..a1a8789 100644 --- a/setup.py +++ b/setup.py @@ -1,40 +1,9 @@ -from pathlib import Path +"""Compatibility entry point for tools that have not adopted PEP 517 yet. -from setuptools import find_packages, setup +Project metadata lives in ``pyproject.toml``. +""" +from setuptools import setup -README = Path(__file__).resolve().parent / "README.md" -try: - from setuptools_scm import get_version -except ImportError: - VERSION = "0.0.0" -else: - VERSION = get_version(root=".", fallback_version="0.0.0") - -setup( - name="SmallPackage", - version=VERSION, - packages=find_packages(), - include_package_data=True, - package_data={"SmallPackage.clients": ["README.md"]}, - install_requires=[], - author="Michael E", - description="Concurrent and Priority oriented Task Management System", - long_description=README.read_text(encoding="utf-8"), - long_description_content_type="text/markdown", - url="https://github.com/MikiEEE/SmallOS", - project_urls={ - "Documentation": "https://github.com/MikiEEE/SmallOS", - "Source": "https://github.com/MikiEEE/SmallOS", - "Issues": "https://github.com/MikiEEE/SmallOS/issues", - }, - license="MIT", - license_files=("LICENSE",), - classifiers=[ - "Programming Language :: Python :: 3", - "License :: OSI Approved :: MIT License", - "Operating System :: OS Independent", - ], - python_requires=">=3.6", -) +setup() From efdfe34076a44f4cf9c2fc91e2e7a3b57d756a51 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 11 Jul 2026 15:17:53 -0500 Subject: [PATCH 2/4] feat:types: second layer of adding typing --- .github/workflows/python.yml | 72 ++++++++++++++++++++++--- .gitignore | 9 ++++ README.md | 18 +++++-- SmallPackage/Kernel.py | 102 ++++++++++++++++++++++------------- SmallPackage/SmallSignals.py | 66 ++++++++++++++++------- pyproject.toml | 15 +++++- 6 files changed, 215 insertions(+), 67 deletions(-) create mode 100644 .gitignore diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index 3fd9862..c50eb5b 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -1,11 +1,41 @@ -name: Python +name: Python CI on: push: pull_request: +permissions: + contents: read + +concurrency: + group: python-ci-${{ github.ref }} + cancel-in-progress: true + jobs: + docs-index: + name: Documentation index + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: "22" + - run: node docs/build-index.mjs --check + + typing: + name: Static typing + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.10" + cache: pip + - run: python -m pip install ".[dev]" + - run: pyright + test: + name: Unit tests (Python ${{ matrix.python-version }}) runs-on: ubuntu-latest strategy: fail-fast: false @@ -16,11 +46,31 @@ jobs: - uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} - - run: python -m pip install --upgrade pip + cache: pip - run: python -m pip install . - run: python -m unittest discover -s tests -v - typing-and-package: + coverage: + name: Coverage + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.10" + cache: pip + - run: python -m pip install ".[dev]" + - run: coverage run -m unittest discover -s tests -v + - run: coverage report + - run: coverage xml + - uses: actions/upload-artifact@v4 + with: + name: coverage-report + path: coverage.xml + + package: + name: Build and verify distribution + needs: [docs-index, typing, test, coverage] runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -29,8 +79,16 @@ jobs: - uses: actions/setup-python@v5 with: python-version: "3.10" - - run: python -m pip install --upgrade pip - - run: python -m pip install ".[dev]" - - run: pyright + cache: pip + - run: python -m pip install build - run: python -m build - + - run: python -m pip install --force-reinstall dist/*.whl + - name: Smoke-test the installed wheel outside the checkout + working-directory: /tmp + run: >- + python -c "from SmallPackage import SmallOS, SmallOSConfig, SmallTask, Unix; + assert SmallOS(config=SmallOSConfig()).setKernel(Unix())" + - uses: actions/upload-artifact@v4 + with: + name: python-distributions + path: dist/ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..659e075 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +__pycache__/ +*.py[cod] +.coverage +coverage.xml +htmlcov/ +build/ +dist/ +*.egg-info/ + diff --git a/README.md b/README.md index 52e6099..31b65a7 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,13 @@ Run the test suite: python3 -m unittest discover -s tests -v ``` +Run the tests with the same branch-coverage gate used by CI: + +```bash +coverage run -m unittest discover -s tests -v +coverage report +``` + Run the static type checker: ```bash @@ -101,10 +108,15 @@ Build the wheel and source distribution: python -m build ``` +The GitHub Actions pipeline runs five gates: documentation-index validation, +Pyright, unit tests across Python 3.10–3.13, branch coverage with a 60% floor, +and distribution verification. Packaging runs only after the earlier gates +pass, installs the built wheel, and smoke-tests it outside the source checkout. + The package ships a `py.typed` marker. Type coverage is being tightened by -subsystem: configuration, awaitables, task lifecycle, scheduling, and core -utilities form the first checked boundary, while platform and protocol-client -implementations remain on the incremental typing backlog. +subsystem: configuration, awaitables, task lifecycle, scheduling, signals, +platform kernels, and core utilities form the current checked boundary, while +protocol clients, shells, and demos remain on the incremental typing backlog. ## Quick Start diff --git a/SmallPackage/Kernel.py b/SmallPackage/Kernel.py index 1a34fbf..d51a16d 100644 --- a/SmallPackage/Kernel.py +++ b/SmallPackage/Kernel.py @@ -14,11 +14,22 @@ shared MicroPython transport/time API. """ +from __future__ import annotations + +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover - exercised on constrained runtimes + TYPE_CHECKING = False + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping, Sequence + from typing import Any + _UNSET = object() -def _import_first(*module_names): +def _import_first(*module_names: str) -> Any | None: """ Import the first available module name from a list. @@ -35,7 +46,7 @@ def _import_first(*module_names): return None -def _portable_shell_split(line): +def _portable_shell_split(line: str) -> list[str]: """ Tokenize one shell command line without relying on desktop-only helpers. @@ -90,7 +101,7 @@ def _portable_shell_split(line): return tokens -def detect_micropython_machine_name(sys_mod=None, os_mod=None): +def detect_micropython_machine_name(sys_mod: Any = None, os_mod: Any = None) -> str: """ Best-effort lookup of the active board/firmware machine name. @@ -122,7 +133,9 @@ def detect_micropython_machine_name(sys_mod=None, os_mod=None): return '' -def build_micropython_kernel(machine_name=None, **kwargs): +def build_micropython_kernel( + machine_name: str | None = None, **kwargs: Any +) -> MicroPythonKernel: """ Return the best built-in MicroPython kernel profile for the current board. @@ -157,14 +170,14 @@ class Kernel: requiring protocol-specific kernel methods. ''' - def __init__(self): + def __init__(self) -> None: self._scheduler_anchor_tick = None self._scheduler_elapsed_ms = 0 - def write(self, msg): + def write(self, msg: str) -> None: pass - def shell_split(self, line): + def shell_split(self, line: str) -> list[str]: """ Tokenize one shell command line. @@ -174,22 +187,22 @@ def shell_split(self, line): """ return _portable_shell_split(line) - def time_epoch(self): - pass + def time_epoch(self) -> float: + raise NotImplementedError - def ticks_ms(self): - pass + def ticks_ms(self) -> int: + raise NotImplementedError - def ticks_add(self, base, delta_ms): + def ticks_add(self, base: int, delta_ms: int) -> int: return base + delta_ms - def ticks_diff(self, end, start): + def ticks_diff(self, end: int, start: int) -> int: return end - start - def time_monotonic(self): + def time_monotonic(self) -> float: return self.scheduler_now_ms() / 1000 - def scheduler_now_ms(self): + def scheduler_now_ms(self) -> int: ''' Returns a non-decreasing scheduler clock in milliseconds. @@ -208,17 +221,24 @@ def scheduler_now_ms(self): self._scheduler_anchor_tick = current return self._scheduler_elapsed_ms - def sleep(self, secs): + def sleep(self, secs: float) -> None: self.sleep_ms(int(max(0, secs) * 1000)) return - def sleep_ms(self, delay_ms): + def sleep_ms(self, delay_ms: int) -> None: pass - def io_wait(self, readables, writables, timeout_ms=None): + def io_wait( + self, + readables: Iterable[Any], + writables: Iterable[Any], + timeout_ms: int | None = None, + ) -> tuple[list[Any], list[Any]]: return [], [] - def validate_io_wait_object(self, obj): + def validate_io_wait_object( + self, obj: Any + ) -> tuple[bool, BaseException | None]: """ Return whether ``obj`` still looks safe to hand to poll/select. @@ -245,28 +265,28 @@ def validate_io_wait_object(self, obj): ) return True, None - def resolve_address(self, host, port): + def resolve_address(self, host: str, port: int) -> Any: return None - def socket_open(self, address_info): + def socket_open(self, address_info: Any) -> Any: return None - def socket_setblocking(self, sock, flag): + def socket_setblocking(self, sock: Any, flag: bool) -> None: return - def socket_connect(self, sock, sockaddr): + def socket_connect(self, sock: Any, sockaddr: Any) -> bool: return True - def socket_connection_error(self, sock): + def socket_connection_error(self, sock: Any) -> int: return 0 - def socket_send(self, sock, data): + def socket_send(self, sock: Any, data: bytes) -> int: return 0 - def socket_recv(self, sock, buffer_size): + def socket_recv(self, sock: Any, buffer_size: int) -> bytes: return b'' - def socket_close(self, sock): + def socket_close(self, sock: Any) -> None: return def socket_wrap_tls_client( @@ -280,10 +300,10 @@ def socket_wrap_tls_client( ): return sock - def socket_do_handshake(self, sock): + def socket_do_handshake(self, sock: Any) -> None: return - def _extract_errno(self, exc): + def _extract_errno(self, exc: BaseException) -> int | None: errno_value = getattr(exc, 'errno', None) if errno_value is not None: return errno_value @@ -291,13 +311,13 @@ def _extract_errno(self, exc): return exc.args[0] return None - def socket_needs_read(self, exc): + def socket_needs_read(self, exc: BaseException) -> bool: return isinstance(exc, BlockingIOError) - def socket_needs_write(self, exc): + def socket_needs_write(self, exc: BaseException) -> bool: return False - def _poll_lookup_key(self, obj): + def _poll_lookup_key(self, obj: Any) -> Any: """ Return the identity key used to map poll events back to registered objects. @@ -506,9 +526,15 @@ def __init__( modules = modules or {} - self._time = modules.get('time') or _import_first('time', 'utime') + time_mod = modules.get('time') or _import_first('time', 'utime') + if time_mod is None: + raise ImportError('MicroPythonKernel requires time or utime.') + self._time = time_mod self._select = modules.get('select') or _import_first('select', 'uselect') - self._socket = modules.get('socket') or _import_first('socket', 'usocket') + socket_mod = modules.get('socket') or _import_first('socket', 'usocket') + if socket_mod is None: + raise ImportError('MicroPythonKernel requires socket or usocket.') + self._socket = socket_mod self._ssl = modules.get('ssl') if self._ssl is None: self._ssl = _import_first('ssl', 'ussl') @@ -680,6 +706,8 @@ def socket_wrap_tls_client( except TypeError as exc: last_error = exc else: + if last_error is None: + raise RuntimeError('TLS wrapper did not produce a socket.') raise last_error if hasattr(wrapped, 'setblocking'): wrapped.setblocking(False) @@ -694,12 +722,14 @@ def socket_needs_read(self, exc): err = self._extract_errno(exc) if err in {11, 115}: return True - if self._ssl and hasattr(self._ssl, 'SSLWantReadError') and isinstance(exc, self._ssl.SSLWantReadError): + want_read_error = getattr(self._ssl, 'SSLWantReadError', None) + if isinstance(want_read_error, type) and isinstance(exc, want_read_error): return True return isinstance(exc, BlockingIOError) def socket_needs_write(self, exc): - if self._ssl and hasattr(self._ssl, 'SSLWantWriteError') and isinstance(exc, self._ssl.SSLWantWriteError): + want_write_error = getattr(self._ssl, 'SSLWantWriteError', None) + if isinstance(want_write_error, type) and isinstance(exc, want_write_error): return True return False diff --git a/SmallPackage/SmallSignals.py b/SmallPackage/SmallSignals.py index 825b37a..9063cbe 100644 --- a/SmallPackage/SmallSignals.py +++ b/SmallPackage/SmallSignals.py @@ -7,6 +7,22 @@ ``SmallOS`` later interprets those awaitables and moves tasks between queues. """ +from __future__ import annotations + +try: + from typing import TYPE_CHECKING +except ImportError: # pragma: no cover - exercised on constrained runtimes + TYPE_CHECKING = False + +if TYPE_CHECKING: + from collections.abc import Callable + from typing import Any, ClassVar, cast + + from .SmallOS import SmallOS + from .SmallTask import SmallTask + from .TaskState import TaskState + from .awaitables import _InstructionAwaitable + from .awaitables import ( sleep_instruction, wait_readable_instruction, @@ -31,15 +47,19 @@ class SmallSignals: and app code should use named constants instead of bare integers. """ - SIGNAL_CAPACITY = 32 - CORE_SIGNAL_MEANINGS = {} - LEGACY_SIGNAL_MEANINGS = { + SIGNAL_CAPACITY: ClassVar[int] = 32 + CORE_SIGNAL_MEANINGS: ClassVar[dict[int, str]] = {} + LEGACY_SIGNAL_MEANINGS: ClassVar[dict[int, str]] = { 5: "Legacy waitOnAsync parent-wait signal from the pre-native-async runtime.", 6: "Legacy waitOnAsync child-complete signal from the pre-native-async runtime.", 7: "Legacy waitOnAsync all-children-complete signal from the pre-native-async runtime.", } - def __init__(self, OS, kwargs): + if TYPE_CHECKING: + OS: SmallOS | None + state: TaskState + + def __init__(self, OS: SmallOS | None, kwargs: dict[str, Any]) -> None: """ Initialize per-task signal state. @@ -52,14 +72,14 @@ def __init__(self, OS, kwargs): self.wakeSigs = [] self.sleepTime = 0 self.timeOfSleep = 0 - self.handlers = None + self.handlers: Callable[[SmallSignals], Any] | None = None super().__init__() if kwargs and kwargs.get("handlers", False): self.handlers = kwargs["handlers"] - def getSignals(self): + def getSignals(self) -> list[int]: """Return the list of currently latched signal numbers.""" received = [] for num, sig in enumerate(self.signals): @@ -68,7 +88,7 @@ def getSignals(self): return received @classmethod - def describeSignal(cls, sig): + def describeSignal(cls, sig: int) -> str: """ Return the documented meaning of a signal slot. @@ -82,7 +102,7 @@ def describeSignal(cls, sig): return cls.LEGACY_SIGNAL_MEANINGS[sig] + " Not used by the current runtime." return "Application-defined signal slot." - def sendSignal(self, pid, sig): + def sendSignal(self, pid: int, sig: int) -> int: """ Deliver a signal to another task by PID. @@ -102,7 +122,7 @@ def sendSignal(self, pid, sig): task.acceptSignal(sig) return 0 - def acceptSignal(self, sig): + def acceptSignal(self, sig: int) -> int: """ Latch an incoming signal and notify the scheduler. @@ -114,13 +134,16 @@ def acceptSignal(self, sig): self.signals[sig] = 1 if self.OS: - self.OS.on_signal(self, sig) + task = cast("SmallTask", self) if TYPE_CHECKING else self + self.OS.on_signal(task, sig) if self.handlers: self.handlers(self) return 0 - def sleep(self, secs, state_blob=None): + def sleep( + self, secs: float, state_blob: dict[Any, Any] | None = None + ) -> _InstructionAwaitable[None]: """ Return the awaitable used for cooperative sleeping. @@ -131,35 +154,40 @@ def sleep(self, secs, state_blob=None): self.state.update(state_blob) return sleep_instruction(secs) - def wait_signal(self, sig, state_blob=None): + def wait_signal( + self, sig: int, state_blob: dict[Any, Any] | None = None + ) -> _InstructionAwaitable[int]: """Return the awaitable used to wait until ``sig`` is delivered.""" if state_blob is not None: self.state.update(state_blob) return wait_signal_instruction(sig) - def sigSuspendV2(self, sig, state_blob=None): + def sigSuspendV2( + self, sig: int, state_blob: dict[Any, Any] | None = None + ) -> _InstructionAwaitable[int]: """Compatibility alias for the older generator-era suspension name.""" return self.wait_signal(sig, state_blob) - def yield_now(self): + def yield_now(self) -> _InstructionAwaitable[None]: """Return the awaitable used for an explicit cooperative yield.""" return yield_now_instruction() - def wait_readable(self, io_obj): + def wait_readable(self, io_obj: Any) -> _InstructionAwaitable[Any]: """Return the awaitable used to wait until ``io_obj`` is readable.""" return wait_readable_instruction(io_obj) - def wait_writable(self, io_obj): + def wait_writable(self, io_obj: Any) -> _InstructionAwaitable[Any]: """Return the awaitable used to wait until ``io_obj`` is writable.""" return wait_writable_instruction(io_obj) - def wake(self): + def wake(self) -> None: """Force the task back onto the ready queue if it belongs to an OS.""" if self.OS: - self.OS.resume_task(self) + task = cast("SmallTask", self) if TYPE_CHECKING else self + self.OS.resume_task(task) return - def checkSignal(self, sig): + def checkSignal(self, sig: int) -> bool: """ Consume and clear a previously received signal if it exists. diff --git a/pyproject.toml b/pyproject.toml index 81c74ca..3df95ae 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ classifiers = [ [project.optional-dependencies] dev = [ "build>=1.2", + "coverage>=7.6", "pyright>=1.1.390", ] @@ -50,8 +51,6 @@ typeCheckingMode = "basic" include = ["SmallPackage", "demos"] exclude = ["build", "dist"] ignore = [ - "SmallPackage/Kernel.py", - "SmallPackage/SmallSignals.py", "SmallPackage/clients", "SmallPackage/shells.py", "demos", @@ -62,3 +61,15 @@ reportUnknownArgumentType = false reportUnknownVariableType = false reportUnknownLambdaType = false reportMissingImports = false + +[tool.coverage.run] +branch = true +source = ["SmallPackage"] + +[tool.coverage.report] +fail_under = 60 +show_missing = true +skip_covered = true + +[tool.coverage.xml] +output = "coverage.xml" From 1645260ec29c1395db04b361c83b4044ec5bac1e Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 11 Jul 2026 15:21:36 -0500 Subject: [PATCH 3/4] feat:types: second layer of adding typing --- .github/workflows/python.yml | 12 +----------- README.md | 8 ++++---- 2 files changed, 5 insertions(+), 15 deletions(-) diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index c50eb5b..911a072 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -12,16 +12,6 @@ concurrency: cancel-in-progress: true jobs: - docs-index: - name: Documentation index - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: actions/setup-node@v4 - with: - node-version: "22" - - run: node docs/build-index.mjs --check - typing: name: Static typing runs-on: ubuntu-latest @@ -70,7 +60,7 @@ jobs: package: name: Build and verify distribution - needs: [docs-index, typing, test, coverage] + needs: [typing, test, coverage] runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 diff --git a/README.md b/README.md index 31b65a7..a8ef905 100644 --- a/README.md +++ b/README.md @@ -108,10 +108,10 @@ Build the wheel and source distribution: python -m build ``` -The GitHub Actions pipeline runs five gates: documentation-index validation, -Pyright, unit tests across Python 3.10–3.13, branch coverage with a 60% floor, -and distribution verification. Packaging runs only after the earlier gates -pass, installs the built wheel, and smoke-tests it outside the source checkout. +The GitHub Actions pipeline runs four gates: Pyright, unit tests across Python +3.10–3.13, branch coverage with a 60% floor, and distribution verification. +Packaging runs only after the earlier gates pass, installs the built wheel, +and smoke-tests it outside the source checkout. The package ships a `py.typed` marker. Type coverage is being tightened by subsystem: configuration, awaitables, task lifecycle, scheduling, signals, From f6c0a713cbd7c0d7523343fbaf4351f88e160df5 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 18 Jul 2026 16:03:03 -0500 Subject: [PATCH 4/4] feat: adapter - initial runtime adapter implementation --- README.md | 112 ++++ SmallPackage/Kernel.py | 10 + SmallPackage/SmallOS.py | 503 ++++++++++++++++- SmallPackage/SmallSignals.py | 14 +- SmallPackage/SmallTask.py | 10 +- SmallPackage/_types.py | 54 +- SmallPackage/adapters/__init__.py | 29 + SmallPackage/adapters/_completion.py | 108 ++++ SmallPackage/adapters/asyncio_loop.py | 373 +++++++++++++ SmallPackage/adapters/base.py | 169 ++++++ SmallPackage/adapters/errors.py | 56 ++ SmallPackage/adapters/threads.py | 195 +++++++ SmallPackage/awaitables.py | 50 +- demos/adapters_asyncio_demo.py | 107 ++++ demos/adapters_demo.py | 77 +++ demos/adapters_sqlite_demo.py | 96 ++++ demos/common.py | 21 +- pyproject.toml | 20 +- tests/test_adapters.py | 772 ++++++++++++++++++++++++++ tests/typing/adapter_contract.py | 33 ++ 20 files changed, 2760 insertions(+), 49 deletions(-) create mode 100644 SmallPackage/adapters/__init__.py create mode 100644 SmallPackage/adapters/_completion.py create mode 100644 SmallPackage/adapters/asyncio_loop.py create mode 100644 SmallPackage/adapters/base.py create mode 100644 SmallPackage/adapters/errors.py create mode 100644 SmallPackage/adapters/threads.py create mode 100644 demos/adapters_asyncio_demo.py create mode 100644 demos/adapters_demo.py create mode 100644 demos/adapters_sqlite_demo.py create mode 100644 tests/test_adapters.py create mode 100644 tests/typing/adapter_contract.py diff --git a/README.md b/README.md index a8ef905..22045ac 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ The project is currently experimental but usable. The runtime core supports: - time-based sleeping - readiness-based socket/I/O waiting - generic TCP/TLS kernel hooks for higher-level protocols +- dependency-free thread and asyncio execution adapters for user libraries - smallOS-native HTTP, Redis, MQTT, SSE, and WebSocket helper clients ## Why smallOS? @@ -45,6 +46,8 @@ That makes it a good fit for: buffered app/shell output routing and terminal-mode helpers - [SmallPackage/clients](SmallPackage/clients): protocol client package for cooperative network integrations +- [SmallPackage/adapters](SmallPackage/adapters): + dependency-free escape hatches for blocking and asyncio-owned user code - [SmallPackage/clients/README.md](SmallPackage/clients/README.md): detailed client-specific guide and API notes - [SmallPackage/clients/SmallHTTP.py](SmallPackage/clients/SmallHTTP.py): @@ -174,6 +177,8 @@ task has been finalized. Current event fields include: - `io_wait_mode` - `join_target_id` - `join_pending_ids` +- `adapter_name` +- `adapter_job_id` - `traceback_text` By default, `TaskCancelledError` does not trigger the handler. Pass @@ -311,6 +316,12 @@ The new demos live in [demos](demos): cooperative single-thread web app demo with HTTP routes, live browser UI, and shell-driven server shutdown - [demos/mqtt_demo.py](demos/mqtt_demo.py): MQTT example built on the native cooperative client +- [demos/adapters_demo.py](demos/adapters_demo.py): + thread and asyncio escape hatches running beside a regular SmallOS task +- [demos/adapters_sqlite_demo.py](demos/adapters_sqlite_demo.py): + a user-owned `sqlite3` connection kept on one thread-adapter worker +- [demos/adapters_asyncio_demo.py](demos/adapters_asyncio_demo.py): + a persistent asyncio queue and background task reused across adapter calls All of the shared demo entry points now install a default runtime error handler through [demos/common.py](demos/common.py). That means network failures, @@ -336,6 +347,107 @@ This means arbitrary `asyncio` libraries are not drop-in compatible with the runtime, but it also means scheduling policy and portability stay under your control. +## Execution Adapters + +Execution adapters let a SmallOS task yield while user-supplied code runs under +a different execution model. They use only the Python standard library and do +not install, import, configure, or wrap database drivers, ORMs, SDKs, or other +third-party packages. + +Use `ThreadAdapter` for a synchronous blocking callable: + +```python +from SmallPackage.adapters.threads import ThreadAdapter + + +def load_record(user_library, settings, record_id): + connection = user_library.connect(**settings) + try: + return connection.load(record_id) + finally: + connection.close() + + +with ThreadAdapter(max_workers=4, max_pending=64) as blocking: + async def load(task): + return await blocking.call( + load_record, + user_selected_library, + connection_settings, + 42, + ) + + runtime.fork(SmallTask(2, load, name="load")) + runtime.start() +``` + +Use `AsyncioAdapter` for an async callable that must run on asyncio. The +adapter owns one persistent event loop in a dedicated thread, allowing +loop-affine clients to be reused when all their operations are routed through +the same adapter: + +```python +from SmallPackage.adapters.asyncio_loop import AsyncioAdapter + + +async def fetch_record(user_library, settings, record_id): + async with user_library.Client(**settings) as client: + return await client.fetch(record_id) + + +with AsyncioAdapter(max_pending=64) as foreign_async: + async def load(task): + return await foreign_async.call( + fetch_record, + user_selected_async_library, + connection_settings, + 42, + ) + + runtime.fork(SmallTask(2, load, name="load")) + runtime.start() +``` + +Important behavior: + +- adapters bind to the first SmallOS runtime that uses them; +- adapter shutdown is explicit, so a context manager should wrap + `runtime.start()`; +- `max_pending` rejects excess work with `AdapterCapacityError` instead of + blocking the scheduler; +- cancelling a SmallTask can cancel queued thread work, but cannot forcibly + stop a running Python thread; +- asyncio cancellation is requested on the adapter loop, but a user library + may delay or suppress it; +- pass an async callable to `AsyncioAdapter.call()`, not a `Task` or `Future` + already owned by another loop; +- inspect `AsyncioAdapter.shutdown_error` after shutdown when application + diagnostics need to detect an unexpected loop stop or library teardown + failure; normal shutdown leaves it as `None`; +- use `ThreadAdapter(max_workers=1)` when user resources require a serialized, + thread-affine execution lane; create, use, and close those resources through + calls on that same adapter rather than creating them on the SmallOS thread. + +### Standard-library examples + +All adapter demos run without installing anything beyond SmallOS: + +```bash +python3 demos/adapters_demo.py +python3 demos/adapters_sqlite_demo.py +python3 demos/adapters_asyncio_demo.py +``` + +- [demos/adapters_demo.py](demos/adapters_demo.py) runs both adapters beside an + ordinary cooperative SmallOS task. +- [demos/adapters_sqlite_demo.py](demos/adapters_sqlite_demo.py) creates, uses, + and closes an in-memory `sqlite3` connection through + `ThreadAdapter(max_workers=1)`. This is the ownership pattern to adapt for a + thread-affine PostgreSQL driver or ORM session supplied by the user. +- [demos/adapters_asyncio_demo.py](demos/adapters_asyncio_demo.py) creates an + `asyncio.Queue`, `Future` objects, and a background task, performs multiple + operations, and cleans them up on the same persistent adapter event loop. + ## Running on MicroPython For MicroPython targets, the intended flow is: diff --git a/SmallPackage/Kernel.py b/SmallPackage/Kernel.py index d51a16d..50af9f3 100644 --- a/SmallPackage/Kernel.py +++ b/SmallPackage/Kernel.py @@ -236,6 +236,10 @@ def io_wait( ) -> tuple[list[Any], list[Any]]: return [], [] + def supports_external_wait_objects(self) -> bool: + """Whether ``io_wait`` can wake on adapter-owned readiness objects.""" + return False + def validate_io_wait_object( self, obj: Any ) -> tuple[bool, BaseException | None]: @@ -428,6 +432,9 @@ def io_wait(self, readables, writables, timeout_ms=None): ready_read, ready_write, _ = self._select.select(readables, writables, [], timeout) return ready_read, ready_write + def supports_external_wait_objects(self): + return True + def resolve_address(self, host, port): return self._socket.getaddrinfo(host, port, type=self._socket.SOCK_STREAM)[0] @@ -619,6 +626,9 @@ def io_wait(self, readables, writables, timeout_ms=None): self.sleep_ms(timeout_ms) return [], [] + def supports_external_wait_objects(self): + return bool(self._poll_factory) + def resolve_address(self, host, port): return self._socket.getaddrinfo(host, port)[0] diff --git a/SmallPackage/SmallOS.py b/SmallPackage/SmallOS.py index 9f95180..4ddcda1 100644 --- a/SmallPackage/SmallOS.py +++ b/SmallPackage/SmallOS.py @@ -24,13 +24,25 @@ from .Kernel import Kernel from .SmallTask import SmallTask - from ._types import ErrorHandler, RuntimeErrorEvent, SmallOSConfigData + from ._types import ( + ErrorHandler, + ExecutionAdapterLike, + RuntimeErrorEvent, + SmallOSConfigData, + ) from .awaitables import TaskInstruction from .SmallIO import SmallIO from .SmallConfig import SmallOSConfig from .OSlist import OSList from .SmallErrors import MaxProcessError, TaskCancelledError, UnsupportedAwaitableError +from .adapters.errors import ( + AdapterCancelledError, + AdapterClosedError, + AdapterProtocolError, + AdapterUnavailableError, + normalize_adapter_exception, +) _MISSING = object() @@ -78,6 +90,11 @@ def __init__( self.wakeUpdate = [] self.ioReadWaiters = {} self.ioWriteWaiters = {} + self._adapter_jobs: dict[ + int, + tuple[ExecutionAdapterLike, SmallTask[Any]], + ] = {} + self._next_adapter_job_id = 0 self.shells = [] self.tasks = OSList(self.config.priority_levels, self.config.task_capacity) self.kernel = None @@ -127,6 +144,10 @@ def start(self) -> None: # bookkeeping always see a consistent terminal state. self._finalize_task(self.cursor) else: + # Adapter failure diagnostics only describe the coroutine step + # resumed by that adapter. Once the step yields successfully, + # a later failure should not be attributed to the old job. + self._clear_adapter_resume_origin(self.cursor) self._handle_yield(self.cursor, yielded) if not self.eternalWatchers and len(self.tasks) != 0 and self.tasks.isOnlyWatchers(): @@ -138,7 +159,10 @@ def next(self) -> SmallTask | None: self.cursor = self.tasks.pop() return self.cursor - def fork(self, children: SmallTask | list[SmallTask]) -> int | list[int]: + def fork( + self, + children: SmallTask[Any] | list[SmallTask[Any]], + ) -> int | list[int]: """Register one task or a list of tasks with the runtime.""" if isinstance(children, list): ids = [] @@ -147,7 +171,7 @@ def fork(self, children: SmallTask | list[SmallTask]) -> int | list[int]: return ids return self._fork_one(children) - def _fork_one(self, task: SmallTask) -> int: + def _fork_one(self, task: SmallTask[Any]) -> int: """Assign a PID, attach the runtime, and enqueue the task if runnable.""" pid = self.tasks.insert(task) if pid == -1: @@ -208,11 +232,13 @@ def _idle_until_next_task(self): if next_wake is not None: timeout = max(0, next_wake - self.kernel.scheduler_now_ms()) - has_io_waiters = bool(self.ioReadWaiters or self.ioWriteWaiters) - if next_wake is None and not has_io_waiters: + has_wait_sources = bool( + self.ioReadWaiters or self.ioWriteWaiters or self._adapter_jobs + ) + if next_wake is None and not has_wait_sources: return False - if has_io_waiters and hasattr(self.kernel, "io_wait"): + if has_wait_sources and hasattr(self.kernel, "io_wait"): self._wake_io_tasks(timeout_ms=timeout) elif timeout is not None and timeout > 0 and hasattr(self.kernel, "sleep_ms"): self.kernel.sleep_ms(timeout) @@ -306,6 +332,10 @@ def _handle_yield(self, task: SmallTask, yielded: Any) -> None: self._enter_io_wait(task, payload["io_obj"], "write") return + if operation == "adapter_call": + self._handle_adapter_call(task, payload) + return + if operation == "join": target = self._resolve_task(payload["target"]) if target is None: @@ -348,6 +378,167 @@ def _handle_yield(self, task: SmallTask, yielded: Any) -> None: task.fail(UnsupportedAwaitableError("Unknown instruction {!r}".format(operation))) self._finalize_task(task) + def _adapter_name(self, adapter: object) -> str: + """Return a stable best-effort name for adapter diagnostics.""" + try: + name = getattr(adapter, "name", None) + except BaseException: + name = None + if isinstance(name, str) and name: + return name + return type(adapter).__name__ + + def _safe_adapter_exception( + self, + adapter: object, + exc: object, + cancelled: bool = False, + ) -> Exception: + """Normalize foreign exceptions before they enter a SmallTask.""" + if not isinstance(exc, BaseException): + exc = AdapterProtocolError( + "{} produced a non-exception failure value".format( + self._adapter_name(adapter) + ) + ) + return normalize_adapter_exception( + exc, + self._adapter_name(adapter), + cancelled=cancelled, + ) + + def _validate_adapter(self, adapter: Any) -> ExecutionAdapterLike: + """Validate and bind the structural adapter contract on submission.""" + if self.kernel is None or not hasattr(self.kernel, "io_wait"): + raise AdapterUnavailableError( + "execution adapters require a kernel with io_wait()" + ) + supports_external_waits = getattr( + self.kernel, + "supports_external_wait_objects", + None, + ) + if callable(supports_external_waits) and not supports_external_waits(): + raise AdapterUnavailableError( + "{} cannot wait on adapter completion objects".format( + type(self.kernel).__name__ + ) + ) + + for method_name in ( + "_bind_runtime", + "submit", + "cancel", + "drain_completions", + ): + if not callable(getattr(adapter, method_name, None)): + raise AdapterProtocolError( + "adapter is missing callable {}()".format(method_name) + ) + + try: + is_closed = adapter.closed + except BaseException as exc: + raise AdapterProtocolError( + "{} closed-state check failed: {}".format( + self._adapter_name(adapter), + exc, + ) + ) from exc + if type(is_closed) is not bool: + raise AdapterProtocolError( + "{} closed state must be boolean".format( + self._adapter_name(adapter) + ) + ) + if is_closed: + raise AdapterClosedError( + "{} is closed".format(self._adapter_name(adapter)) + ) + + adapter._bind_runtime(self) + try: + wait_object = adapter.wait_object + hash(wait_object) + except BaseException as exc: + raise AdapterProtocolError( + "{} has no usable completion wait object: {}".format( + self._adapter_name(adapter), + exc, + ) + ) from exc + + validator = getattr(self.kernel, "validate_io_wait_object", None) + if validator is not None: + try: + is_valid, exc = validator(wait_object) + except BaseException as exc: + raise AdapterUnavailableError( + "{} could not validate the completion wait object: {}".format( + type(self.kernel).__name__, + exc, + ) + ) from exc + if not is_valid: + raise AdapterUnavailableError( + "{} completion wait object is invalid: {}".format( + self._adapter_name(adapter), + exc, + ) + ) + return adapter + + def _handle_adapter_call( + self, + task: SmallTask[Any], + payload: dict[str, Any], + ) -> None: + """Submit one foreign call and block ``task`` for its completion.""" + adapter_candidate = payload.get("adapter") + callable_obj = payload.get("callable") + args = payload.get("args", ()) + kwargs = payload.get("kwargs", {}) + + try: + adapter = self._validate_adapter(adapter_candidate) + if not callable(callable_obj): + raise TypeError("adapter call target must be callable") + if not isinstance(args, tuple): + raise AdapterProtocolError("adapter call args must be a tuple") + if not isinstance(kwargs, dict): + raise AdapterProtocolError("adapter call kwargs must be a dict") + except BaseException as exc: + safe_exc = self._safe_adapter_exception(adapter_candidate, exc) + self.resume_task(task, exc=safe_exc, front=True) + return + + self._next_adapter_job_id += 1 + job_id = self._next_adapter_job_id + self._enter_adapter_wait(task, adapter, job_id) + + try: + adapter.submit(job_id, callable_obj, args, kwargs) + except BaseException as exc: + self._adapter_jobs.pop(job_id, None) + safe_exc = self._safe_adapter_exception(adapter, exc) + self._record_adapter_resume_origin(task, adapter, job_id) + self.resume_task(task, exc=safe_exc, front=True) + + def _record_adapter_resume_origin( + self, + task: SmallTask[Any], + adapter: ExecutionAdapterLike, + job_id: int, + ) -> None: + """Preserve adapter identity until the resumed coroutine step finishes.""" + task._adapter_resume_name = self._adapter_name(adapter) + task._adapter_resume_job_id = job_id + + def _clear_adapter_resume_origin(self, task: SmallTask[Any]) -> None: + """Clear diagnostics associated with a completed resume step.""" + task._adapter_resume_name = None + task._adapter_resume_job_id = None + def _resolve_task(self, target: int | SmallTask) -> SmallTask | None: """Normalize either a task object or a PID to a task object.""" if not isinstance(target, int): @@ -404,6 +595,8 @@ def _clear_wait_metadata(self, task): task._join_pending = set() task._io_wait_obj = None task._io_wait_mode = None + task._adapter = None + task._adapter_job_id = None def _begin_wait(self, task, reason): """Prepare a runnable task to transition into one blocked wait state.""" @@ -429,6 +622,18 @@ def _enter_io_wait(self, task, io_obj, mode): task._io_wait_mode = mode self._register_io_wait(task, io_obj, mode) + def _enter_adapter_wait( + self, + task: SmallTask[Any], + adapter: ExecutionAdapterLike, + job_id: int, + ) -> None: + """Block ``task`` on one adapter-owned external operation.""" + self._begin_wait(task, "adapter") + task._adapter = adapter + task._adapter_job_id = job_id + self._adapter_jobs[job_id] = (adapter, task) + def _enter_join_wait(self, task, target): """Block ``task`` until ``target`` finishes.""" self._begin_wait(task, "join") @@ -454,27 +659,271 @@ def _register_io_wait(self, task, io_obj, mode): def _wake_io_tasks(self, timeout_ms: int | None = 0): """ - Ask the kernel which I/O objects are ready and resume their waiters. + Poll user I/O and adapter completion sources in one kernel wait. - This keeps the runtime single-threaded: tasks suspend on readiness - events and the scheduler wakes them when the kernel reports the socket - or stream can make progress. + Adapter worker threads only make their completion socket readable. All + queue mutation and task resumption still happens on this scheduler + thread. """ if not self.kernel or not hasattr(self.kernel, "io_wait"): return - if not self.ioReadWaiters and not self.ioWriteWaiters: + if not self.ioReadWaiters and not self.ioWriteWaiters and not self._adapter_jobs: return + self._fail_invalid_io_waiters() - if not self.ioReadWaiters and not self.ioWriteWaiters: + adapter_sources = self._collect_adapter_sources() + if not self.ioReadWaiters and not self.ioWriteWaiters and not adapter_sources: return - readable, writable = self.kernel.io_wait( - list(self.ioReadWaiters.keys()), - list(self.ioWriteWaiters.keys()), - timeout_ms, - ) + readables = list(self.ioReadWaiters.keys()) + for wait_object in adapter_sources: + if wait_object not in self.ioReadWaiters: + readables.append(wait_object) + + adapter_job_count = len(self._adapter_jobs) + io_waiter_count = len(self.ioReadWaiters) + len(self.ioWriteWaiters) + try: + readable, writable = self.kernel.io_wait( + readables, + list(self.ioWriteWaiters.keys()), + timeout_ms, + ) + except Exception: + # A completion channel may close after validation but before poll + # registration. Revalidate once and recover if that race removed a + # broken source; otherwise preserve the kernel's original failure. + self._fail_invalid_io_waiters() + self._collect_adapter_sources() + if ( + len(self._adapter_jobs) < adapter_job_count + or len(self.ioReadWaiters) + len(self.ioWriteWaiters) < io_waiter_count + ): + return + raise self._resume_io_waiters(readable, self.ioReadWaiters) self._resume_io_waiters(writable, self.ioWriteWaiters) + for ready_object in readable: + adapter = adapter_sources.get(ready_object) + if adapter is not None: + self._drain_adapter_completions(adapter) + + def _collect_adapter_sources(self) -> dict[Any, ExecutionAdapterLike]: + """Return valid completion wait objects for adapters with live jobs.""" + adapters: list[ExecutionAdapterLike] = [] + for adapter, _task in list(self._adapter_jobs.values()): + if not any(existing is adapter for existing in adapters): + adapters.append(adapter) + + sources: dict[Any, ExecutionAdapterLike] = {} + validator = getattr(self.kernel, "validate_io_wait_object", None) + for adapter in adapters: + try: + is_closed = adapter.closed + if type(is_closed) is not bool: + raise AdapterProtocolError( + "{} closed state must be boolean".format( + self._adapter_name(adapter) + ) + ) + except BaseException as exc: + self._fail_adapter_jobs( + adapter, + AdapterProtocolError( + "{} closed-state check failed: {}".format( + self._adapter_name(adapter), + exc, + ) + ), + ) + continue + if is_closed: + self._fail_adapter_jobs( + adapter, + AdapterClosedError( + "{} closed with jobs still pending".format( + self._adapter_name(adapter) + ) + ), + ) + continue + try: + wait_object = adapter.wait_object + hash(wait_object) + except BaseException as exc: + self._fail_adapter_jobs( + adapter, + AdapterProtocolError( + "{} completion wait object failed: {}".format( + self._adapter_name(adapter), + exc, + ) + ), + ) + continue + + if validator is not None: + try: + is_valid, exc = validator(wait_object) + except BaseException as exc: + self._fail_adapter_jobs( + adapter, + AdapterUnavailableError( + "{} could not validate its completion wait object: {}".format( + self._adapter_name(adapter), + exc, + ) + ), + ) + continue + if not is_valid: + self._fail_adapter_jobs( + adapter, + AdapterUnavailableError( + "{} completion wait object became invalid: {}".format( + self._adapter_name(adapter), + exc, + ) + ), + ) + continue + if wait_object in sources and sources[wait_object] is not adapter: + self._fail_adapter_jobs( + adapter, + AdapterProtocolError( + "adapter completion wait objects must be unique" + ), + ) + continue + sources[wait_object] = adapter + return sources + + def _drain_adapter_completions(self, adapter: ExecutionAdapterLike) -> None: + """Resume SmallTasks from every completion currently queued.""" + try: + completions = adapter.drain_completions() + if completions is None: + raise AdapterProtocolError( + "{} drain_completions() returned None".format( + self._adapter_name(adapter) + ) + ) + completions = list(completions) + except BaseException as exc: + self._fail_adapter_jobs( + adapter, + self._safe_adapter_exception(adapter, exc), + ) + return + + for completion in completions: + try: + job_id = completion.job_id + if type(job_id) is not int or job_id <= 0: + raise AdapterProtocolError( + "adapter completion job_id must be a positive integer" + ) + cancelled = completion.cancelled + if type(cancelled) is not bool: + raise AdapterProtocolError( + "adapter completion cancelled state must be boolean" + ) + exception = completion.exception + if exception is not None and not isinstance( + exception, + BaseException, + ): + raise AdapterProtocolError( + "adapter completion exception must derive from BaseException" + ) + has_value = completion.has_value + if type(has_value) is not bool: + raise AdapterProtocolError( + "adapter completion has_value state must be boolean" + ) + value = completion.value if has_value else _MISSING + outcome_count = int(cancelled) + int(exception is not None) + int(has_value) + if outcome_count != 1: + raise AdapterProtocolError( + "adapter completion must contain exactly one outcome" + ) + except BaseException as exc: + self._fail_adapter_jobs( + adapter, + self._safe_adapter_exception(adapter, exc), + ) + return + + entry = self._adapter_jobs.get(job_id) + if entry is None: + # A cancelled SmallTask may leave a late foreign completion. + continue + entry_adapter, task = entry + if entry_adapter is not adapter: + self._fail_adapter_jobs( + adapter, + AdapterProtocolError( + "{} completed job {} owned by another adapter".format( + self._adapter_name(adapter), + job_id, + ) + ), + ) + return + + self._adapter_jobs.pop(job_id, None) + if task.done or self.tasks.search(task.getID()) == -1: + continue + + if cancelled: + exc = AdapterCancelledError( + "{} job {} was cancelled by the foreign runtime".format( + self._adapter_name(adapter), + job_id, + ) + ) + self._record_adapter_resume_origin(task, adapter, job_id) + self.resume_task(task, exc=exc, front=True) + continue + + if exception is not None: + safe_exc = self._safe_adapter_exception(adapter, exception) + self._record_adapter_resume_origin(task, adapter, job_id) + self.resume_task(task, exc=safe_exc, front=True) + continue + + self.resume_task(task, value=value, front=True) + + def _fail_adapter_jobs( + self, + adapter: ExecutionAdapterLike, + exc: BaseException, + ) -> None: + """Fail every live SmallTask waiting on a broken adapter source.""" + jobs = [ + (job_id, task) + for job_id, (job_adapter, task) in list(self._adapter_jobs.items()) + if job_adapter is adapter + ] + for job_id, task in jobs: + self._adapter_jobs.pop(job_id, None) + try: + adapter.cancel(job_id) + except BaseException: + pass + if task.done or self.tasks.search(task.getID()) == -1: + continue + task_exc = self._clone_adapter_error(exc) + self._record_adapter_resume_origin(task, adapter, job_id) + self.resume_task(task, exc=task_exc, front=True) + + def _clone_adapter_error(self, exc: BaseException) -> Exception: + """Create a per-task adapter exception when one source fails broadly.""" + safe_exc = self._safe_adapter_exception(None, exc) + args = getattr(safe_exc, "args", ()) + try: + return safe_exc.__class__(*args) + except Exception: + return AdapterUnavailableError(str(safe_exc)) def _fail_invalid_io_waiters(self): """ @@ -544,6 +993,18 @@ def _clear_wait_registration(self, task): if not waiters and task._io_wait_obj in waiters_map: del waiters_map[task._io_wait_obj] + if task._adapter_job_id is not None: + job_id = task._adapter_job_id + entry = self._adapter_jobs.pop(job_id, None) + adapter = task._adapter + if entry is not None: + adapter = entry[0] + if adapter is not None and entry is not None: + try: + adapter.cancel(job_id) + except BaseException: + pass + def _should_dispatch_failure(self, task): """Return whether ``task`` should produce a runtime failure event.""" exc = task.exception @@ -583,6 +1044,11 @@ def _format_exception_traceback(self, exc): def _build_failure_event(self, task): """Snapshot the task failure context before finalization clears wait state.""" exc = task.exception + adapter_name = task._adapter_resume_name + adapter_job_id = task._adapter_resume_job_id + if adapter_name is None and task._adapter is not None: + adapter_name = self._adapter_name(task._adapter) + adapter_job_id = task._adapter_job_id return { "task_id": task.getID(), "task_name": task.name, @@ -597,6 +1063,8 @@ def _build_failure_event(self, task): "join_target_id": self._snapshot_task_id(task._join_target), "join_pending_ids": sorted(task._join_pending) if task._join_pending else [], "traceback_text": self._format_exception_traceback(exc), + "adapter_name": adapter_name, + "adapter_job_id": adapter_job_id, } def _write_runtime_diagnostic(self, message): @@ -673,6 +1141,7 @@ def _finalize_task(self, task): self._notify_waiters(task) self._detach_from_parent(task) self.tasks.delete(task.getID()) + self._clear_adapter_resume_origin(task) if failure_event is not None: self._dispatch_error_handler(failure_event) diff --git a/SmallPackage/SmallSignals.py b/SmallPackage/SmallSignals.py index 9063cbe..2be3ce7 100644 --- a/SmallPackage/SmallSignals.py +++ b/SmallPackage/SmallSignals.py @@ -21,7 +21,7 @@ from .SmallOS import SmallOS from .SmallTask import SmallTask from .TaskState import TaskState - from .awaitables import _InstructionAwaitable + from .awaitables import InstructionAwaitable from .awaitables import ( sleep_instruction, @@ -143,7 +143,7 @@ def acceptSignal(self, sig: int) -> int: def sleep( self, secs: float, state_blob: dict[Any, Any] | None = None - ) -> _InstructionAwaitable[None]: + ) -> InstructionAwaitable[None]: """ Return the awaitable used for cooperative sleeping. @@ -156,7 +156,7 @@ def sleep( def wait_signal( self, sig: int, state_blob: dict[Any, Any] | None = None - ) -> _InstructionAwaitable[int]: + ) -> InstructionAwaitable[int]: """Return the awaitable used to wait until ``sig`` is delivered.""" if state_blob is not None: self.state.update(state_blob) @@ -164,19 +164,19 @@ def wait_signal( def sigSuspendV2( self, sig: int, state_blob: dict[Any, Any] | None = None - ) -> _InstructionAwaitable[int]: + ) -> InstructionAwaitable[int]: """Compatibility alias for the older generator-era suspension name.""" return self.wait_signal(sig, state_blob) - def yield_now(self) -> _InstructionAwaitable[None]: + def yield_now(self) -> InstructionAwaitable[None]: """Return the awaitable used for an explicit cooperative yield.""" return yield_now_instruction() - def wait_readable(self, io_obj: Any) -> _InstructionAwaitable[Any]: + def wait_readable(self, io_obj: Any) -> InstructionAwaitable[Any]: """Return the awaitable used to wait until ``io_obj`` is readable.""" return wait_readable_instruction(io_obj) - def wait_writable(self, io_obj: Any) -> _InstructionAwaitable[Any]: + def wait_writable(self, io_obj: Any) -> InstructionAwaitable[Any]: """Return the awaitable used to wait until ``io_obj`` is writable.""" return wait_writable_instruction(io_obj) diff --git a/SmallPackage/SmallTask.py b/SmallPackage/SmallTask.py index 91a47a4..6ce42ca 100644 --- a/SmallPackage/SmallTask.py +++ b/SmallPackage/SmallTask.py @@ -21,7 +21,7 @@ from typing import Any, Generic, TypeVar from .SmallOS import SmallOS - from ._types import TaskRoutine + from ._types import ExecutionAdapterLike, TaskRoutine T = TypeVar("T") @@ -59,7 +59,7 @@ def __init__(self, priority: int, routine: TaskRoutine[T], **kwargs: Any) -> Non self.isLocked = 0 self.isWatcher = False self.parent = None - self.OS = None + self.OS: SmallOS | None = None self.state = TaskState() self.children = [] self.name = "" @@ -83,6 +83,10 @@ def __init__(self, priority: int, routine: TaskRoutine[T], **kwargs: Any) -> Non self._join_waiters = [] self._io_wait_obj = None self._io_wait_mode = None + self._adapter: ExecutionAdapterLike | None = None + self._adapter_job_id: int | None = None + self._adapter_resume_name: str | None = None + self._adapter_resume_job_id: int | None = None self.state.update({"return_status": 0}, "system") @@ -264,7 +268,7 @@ def block(self, reason): """Record why the task is no longer runnable.""" self._blocked_reason = reason self.isReady = 0 - self.isWaiting = 1 if reason in ("signal", "join", "join_all") else 0 + self.isWaiting = 1 if reason in ("signal", "join", "join_all", "adapter") else 0 self.isSleep = 1 if reason == "sleep" else 0 self.state.update({"return_status": 1, "blocked_reason": reason}, "system") diff --git a/SmallPackage/_types.py b/SmallPackage/_types.py index b217170..6589d5e 100644 --- a/SmallPackage/_types.py +++ b/SmallPackage/_types.py @@ -7,7 +7,10 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence -from typing import Any, Protocol, TypeAlias, TypedDict, TypeVar +from typing import TYPE_CHECKING, Any, Protocol, TypeAlias, TypedDict, TypeVar + +if TYPE_CHECKING: + from .SmallOS import SmallOS T = TypeVar("T") @@ -30,6 +33,7 @@ class KernelLike(Protocol): def write(self, msg: str) -> Any: ... def scheduler_now_ms(self) -> int: ... def sleep_ms(self, delay_ms: int) -> None: ... + def supports_external_wait_objects(self) -> bool: ... def io_wait( self, readables: Iterable[Any], @@ -38,12 +42,54 @@ def io_wait( ) -> tuple[Sequence[Any], Sequence[Any]]: ... +class AdapterCompletionLike(Protocol): + """Structural completion record consumed by the scheduler.""" + + job_id: int + value: Any + exception: BaseException | None + cancelled: bool + + @property + def has_value(self) -> bool: ... + + +class ExecutionAdapterLike(Protocol): + """Structural execution-adapter boundary consumed by ``SmallOS``.""" + + @property + def name(self) -> str: ... + + @property + def wait_object(self) -> Any: ... + + @property + def pending_count(self) -> int: ... + + @property + def closed(self) -> bool: ... + + def _bind_runtime(self, runtime: SmallOS) -> None: ... + + def submit( + self, + job_id: int, + callable_obj: Callable[..., Any], + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: ... + + def cancel(self, job_id: int) -> bool: ... + + def drain_completions(self) -> Iterable[AdapterCompletionLike]: ... + + class RuntimeErrorEvent(TypedDict): task_id: int task_name: str parent_id: int | None exception: BaseException - exception_type: type[BaseException] + exception_type: str | None exception_repr: str is_cancelled: bool blocked_reason: str | None @@ -51,7 +97,9 @@ class RuntimeErrorEvent(TypedDict): io_wait_mode: str | None join_target_id: int | None join_pending_ids: list[int] - traceback_text: str + traceback_text: str | None + adapter_name: str | None + adapter_job_id: int | None class SmallOSConfigData(TypedDict, total=False): diff --git a/SmallPackage/adapters/__init__.py b/SmallPackage/adapters/__init__.py new file mode 100644 index 0000000..a9ff3ba --- /dev/null +++ b/SmallPackage/adapters/__init__.py @@ -0,0 +1,29 @@ +"""Dependency-free contracts and errors for optional execution adapters. + +Desktop adapter implementations are intentionally not imported here. Import +``ThreadAdapter`` or ``AsyncioAdapter`` from their explicit modules so core and +MicroPython-oriented package imports do not load desktop concurrency modules. +""" + +from .base import AdapterCompletion, ExecutionAdapter +from .errors import ( + AdapterCancelledError, + AdapterCapacityError, + AdapterClosedError, + AdapterError, + AdapterExecutionError, + AdapterProtocolError, + AdapterUnavailableError, +) + +__all__ = [ + "AdapterCancelledError", + "AdapterCapacityError", + "AdapterClosedError", + "AdapterCompletion", + "AdapterError", + "AdapterExecutionError", + "AdapterProtocolError", + "AdapterUnavailableError", + "ExecutionAdapter", +] diff --git a/SmallPackage/adapters/_completion.py b/SmallPackage/adapters/_completion.py new file mode 100644 index 0000000..9a8e1f6 --- /dev/null +++ b/SmallPackage/adapters/_completion.py @@ -0,0 +1,108 @@ +"""Thread-safe completion queue with a readiness notification socket.""" + +from __future__ import annotations + +import queue +import socket +import threading + +from .base import AdapterCompletion +from .errors import AdapterUnavailableError + + +class CompletionChannel: + """Move completions between threads without touching scheduler state.""" + + def __init__(self) -> None: + socket_pair = getattr(socket, "socketpair", None) + if socket_pair is None: + raise AdapterUnavailableError( + "this Python runtime does not provide socket.socketpair()" + ) + try: + reader, writer = socket_pair() + except (OSError, TypeError) as exc: + raise AdapterUnavailableError( + "could not create the adapter completion socket pair: {}".format(exc) + ) from exc + self._reader: socket.socket = reader + self._writer: socket.socket = writer + self._reader.setblocking(False) + self._writer.setblocking(False) + self._queue: queue.SimpleQueue[AdapterCompletion] = queue.SimpleQueue() + self._lock: threading.Lock = threading.Lock() + self._closed: bool = False + + @property + def wait_object(self) -> socket.socket: + """Return the socket watched by ``Kernel.io_wait()``.""" + return self._reader + + @property + def closed(self) -> bool: + with self._lock: + return self._closed + + def post(self, completion: AdapterCompletion) -> bool: + """Enqueue a completion and make the reader socket ready.""" + notification_failed: bool = False + with self._lock: + if self._closed: + return False + self._queue.put(completion) + try: + self._writer.send(b"\x01") + except BlockingIOError: + # A full socket is already readable, so no wakeup is lost. + pass + except OSError: + notification_failed = True + + if notification_failed: + # A completion without a wakeup could strand the scheduler. Mark + # the channel unusable so the runtime fails affected jobs instead. + self.close() + return False + return True + + def drain(self) -> list[AdapterCompletion]: + """Drain notification bytes and all completions available now.""" + with self._lock: + if self._closed: + return [] + + while True: + try: + if not self._reader.recv(4096): + break + except BlockingIOError: + break + except OSError: + break + + completions: list[AdapterCompletion] = [] + while True: + try: + completions.append(self._queue.get_nowait()) + except queue.Empty: + break + return completions + + def close(self) -> None: + """Close both notification sockets once no further posts are needed.""" + with self._lock: + if self._closed: + return + self._closed = True + reader = self._reader + writer = self._writer + for sock in (reader, writer): + try: + sock.close() + except OSError: + pass + while True: + try: + self._queue.get_nowait() + except queue.Empty: + break diff --git a/SmallPackage/adapters/asyncio_loop.py b/SmallPackage/adapters/asyncio_loop.py new file mode 100644 index 0000000..306c0a2 --- /dev/null +++ b/SmallPackage/adapters/asyncio_loop.py @@ -0,0 +1,373 @@ +"""Persistent-loop escape hatch for user-supplied asyncio libraries.""" + +from __future__ import annotations + +import asyncio +import inspect +import threading +from typing import cast + +try: + from typing import TYPE_CHECKING as _type_checking +except ImportError: # pragma: no cover - desktop-only module + _type_checking = False + +if _type_checking: + from collections.abc import Awaitable, Callable + import socket + from typing import Any, TypeVar + + from ..awaitables import InstructionAwaitable + + _ResultT = TypeVar("_ResultT") + +from ..awaitables import adapter_call_instruction +from ._completion import CompletionChannel +from .base import AdapterCompletion, ExecutionAdapter +from .errors import ( + AdapterCapacityError, + AdapterClosedError, + AdapterProtocolError, + AdapterUnavailableError, + normalize_adapter_exception, +) + + +def _is_positive_int(value: object) -> bool: + return type(value) is int and value > 0 + + +class AsyncioAdapter(ExecutionAdapter): + """Run async callable factories on one persistent asyncio loop thread.""" + + def __init__( + self, + max_pending: int = 64, + thread_name: str = "smallos-asyncio-adapter", + startup_timeout: float = 5.0, + ) -> None: + if not _is_positive_int(max_pending): + raise ValueError("max_pending must be a positive integer") + if isinstance(startup_timeout, bool) or startup_timeout <= 0: + raise ValueError("startup_timeout must be positive") + super().__init__("asyncio") + self._max_pending: int = max_pending + self._channel: CompletionChannel = CompletionChannel() + self._lock: threading.Lock = threading.Lock() + self._ready: threading.Event = threading.Event() + self._closed: bool = False + self._loop: asyncio.AbstractEventLoop | None = None + self._startup_error: BaseException | None = None + self._accepted: set[int] = set() + self._cancel_requested: set[int] = set() + self._tasks: dict[int, asyncio.Future[Any]] = {} + self._shutdown_task: asyncio.Task[None] | None = None + self._shutdown_error: BaseException | None = None + self._thread: threading.Thread = threading.Thread( + target=self._run_loop, + name=thread_name, + daemon=False, + ) + try: + self._thread.start() + except BaseException as exc: + with self._lock: + self._closed = True + self._channel.close() + raise AdapterUnavailableError( + "AsyncioAdapter loop thread could not start: {}".format(exc) + ) from exc + if not self._ready.wait(startup_timeout): + with self._lock: + self._closed = True + self._channel.close() + raise AdapterUnavailableError( + "AsyncioAdapter loop did not start within {} seconds".format( + startup_timeout + ) + ) + if self._startup_error is not None: + self._thread.join() + raise AdapterUnavailableError( + "AsyncioAdapter loop failed to start: {}".format(self._startup_error) + ) + + def call( + self, + callable_obj: Callable[..., Awaitable[_ResultT]], + /, + *args: Any, + **kwargs: Any, + ) -> InstructionAwaitable[_ResultT]: + """Return an awaitable for the async callable's eventual result type.""" + if not callable(callable_obj): + raise TypeError("adapter call target must be callable") + instruction = adapter_call_instruction(self, callable_obj, args, kwargs) + return cast("InstructionAwaitable[_ResultT]", instruction) + + @property + def wait_object(self) -> socket.socket: + return self._channel.wait_object + + @property + def pending_count(self) -> int: + with self._lock: + return len(self._accepted) + + @property + def closed(self) -> bool: + with self._lock: + return self._closed + + @property + def shutdown_error(self) -> BaseException | None: + """Return an unexpected loop or teardown failure, if one occurred.""" + with self._lock: + return self._shutdown_error + + def _run_loop(self) -> None: + loop: asyncio.AbstractEventLoop | None = None + try: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + with self._lock: + self._loop = loop + should_run = not self._closed + self._ready.set() + if should_run: + loop.run_forever() + with self._lock: + if not self._closed and self._shutdown_error is None: + self._shutdown_error = AdapterUnavailableError( + "AsyncioAdapter loop stopped unexpectedly" + ) + self._closed = True + except BaseException as exc: + with self._lock: + if not self._ready.is_set(): + self._startup_error = exc + elif self._shutdown_error is None: + self._shutdown_error = exc + self._closed = True + self._ready.set() + finally: + if loop is not None: + try: + asyncio.set_event_loop(None) + except BaseException as exc: + with self._lock: + if self._shutdown_error is None: + self._shutdown_error = exc + try: + loop.close() + except BaseException as exc: + with self._lock: + if self._shutdown_error is None: + self._shutdown_error = exc + self._channel.close() + + def submit( + self, + job_id: int, + callable_obj: Callable[..., Any], + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: + if not callable(callable_obj): + raise TypeError("AsyncioAdapter target must be callable") + + with self._lock: + if self._closed: + raise AdapterClosedError("AsyncioAdapter is closed") + if len(self._accepted) >= self._max_pending: + raise AdapterCapacityError( + "AsyncioAdapter reached max_pending ({})".format( + self._max_pending + ) + ) + loop = self._loop + self._accepted.add(job_id) + if loop is None or not loop.is_running(): + self._accepted.discard(job_id) + raise AdapterUnavailableError("AsyncioAdapter loop is not running") + try: + # Queue the start callback while holding the lifecycle lock so + # shutdown cannot overtake an accepted submission. + loop.call_soon_threadsafe( + self._start_job, + job_id, + callable_obj, + args, + kwargs, + ) + except BaseException: + self._accepted.discard(job_id) + raise + + def _start_job( + self, + job_id: int, + callable_obj: Callable[..., Any], + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: + with self._lock: + if job_id not in self._accepted: + return + cancel_requested = job_id in self._cancel_requested + + if cancel_requested: + self._finish_job(job_id, AdapterCompletion.cancelled_result(job_id)) + return + + try: + awaitable = callable_obj(*args, **kwargs) + except BaseException as exc: + self._finish_job( + job_id, + AdapterCompletion.failed( + job_id, + normalize_adapter_exception(exc, self.name), + ), + ) + return + + if not inspect.isawaitable(awaitable): + self._finish_job( + job_id, + AdapterCompletion.failed( + job_id, + AdapterProtocolError( + "AsyncioAdapter callable returned a non-awaitable value" + ), + ), + ) + return + + try: + loop = asyncio.get_running_loop() + if asyncio.isfuture(awaitable): + get_loop = getattr(awaitable, "get_loop", None) + if callable(get_loop) and get_loop() is not loop: + raise AdapterProtocolError( + "AsyncioAdapter cannot accept a Future or Task " + "owned by another event loop" + ) + task = asyncio.ensure_future(awaitable, loop=loop) + except BaseException as exc: + close = getattr(awaitable, "close", None) + if callable(close): + close() + self._finish_job( + job_id, + AdapterCompletion.failed( + job_id, + normalize_adapter_exception(exc, self.name), + ), + ) + return + + self._tasks[job_id] = task + task.add_done_callback( + lambda completed, completed_job_id=job_id: self._on_task_done( + completed_job_id, + completed, + ) + ) + + def _on_task_done(self, job_id: int, task: asyncio.Future[Any]) -> None: + self._tasks.pop(job_id, None) + if task.cancelled(): + completion = AdapterCompletion.cancelled_result(job_id) + else: + try: + value = task.result() + except BaseException as exc: + completion = AdapterCompletion.failed( + job_id, + normalize_adapter_exception(exc, self.name), + ) + else: + completion = AdapterCompletion.result(job_id, value) + self._finish_job(job_id, completion) + + def _finish_job(self, job_id: int, completion: AdapterCompletion) -> None: + with self._lock: + self._accepted.discard(job_id) + self._cancel_requested.discard(job_id) + self._channel.post(completion) + + def cancel(self, job_id: int) -> bool: + with self._lock: + if job_id not in self._accepted: + return False + self._cancel_requested.add(job_id) + loop = self._loop + if loop is None or not loop.is_running(): + return False + try: + loop.call_soon_threadsafe(self._cancel_on_loop, job_id) + except (RuntimeError, OSError): + return False + return True + + def _cancel_on_loop(self, job_id: int) -> None: + task = self._tasks.get(job_id) + if task is not None: + task.cancel() + + def drain_completions(self) -> list[AdapterCompletion]: + return self._channel.drain() + + async def _shutdown_async(self, cancel_pending: bool) -> None: + loop = asyncio.get_running_loop() + try: + tasks = list(self._tasks.values()) + if cancel_pending: + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + await loop.shutdown_asyncgens() + shutdown_default_executor = getattr( + loop, + "shutdown_default_executor", + None, + ) + if shutdown_default_executor is not None: + await shutdown_default_executor() + except BaseException as exc: + # Preserve cleanup progress and avoid leaving a non-daemon loop + # thread alive if a library-owned shutdown hook fails. + with self._lock: + self._shutdown_error = exc + finally: + loop.call_soon(loop.stop) + + def _begin_shutdown(self, cancel_pending: bool) -> None: + if self._shutdown_task is not None: + return + self._shutdown_task = asyncio.create_task( + self._shutdown_async(cancel_pending) + ) + + def shutdown(self, wait: bool = True, cancel_pending: bool = False) -> None: + with self._lock: + if self._closed: + if wait and self._thread.is_alive(): + thread = self._thread + else: + return + else: + self._closed = True + thread = self._thread + loop = self._loop + if loop is not None and loop.is_running(): + try: + # Every accepted submit callback was queued under this same + # lock, so shutdown is ordered after those submissions. + loop.call_soon_threadsafe(self._begin_shutdown, cancel_pending) + except (RuntimeError, OSError): + pass + if wait and thread is not threading.current_thread(): + thread.join() diff --git a/SmallPackage/adapters/base.py b/SmallPackage/adapters/base.py new file mode 100644 index 0000000..3f63d50 --- /dev/null +++ b/SmallPackage/adapters/base.py @@ -0,0 +1,169 @@ +"""Shared execution-adapter instruction and completion contracts.""" + +from __future__ import annotations + +try: + from typing import TYPE_CHECKING as _type_checking +except ImportError: # pragma: no cover - constrained runtimes + _type_checking = False + +if _type_checking: + from collections.abc import Callable + from types import TracebackType + from typing import Any, TypeVar + + from ..SmallOS import SmallOS + from ..awaitables import InstructionAwaitable + + _AdapterT = TypeVar("_AdapterT", bound="ExecutionAdapter") + +from ..awaitables import adapter_call_instruction +from .errors import AdapterUnavailableError + + +_MISSING: object = object() + + +def _is_base_exception(value: object) -> bool: + """Check dynamically supplied completion failures without trusting hints.""" + return isinstance(value, BaseException) + + +class AdapterCompletion: + """One completed adapter job, represented by exactly one outcome.""" + + def __init__( + self, + job_id: int, + value: Any = _MISSING, + exception: BaseException | None = None, + cancelled: bool = False, + ) -> None: + if type(job_id) is not int: + raise TypeError("AdapterCompletion job_id must be an integer.") + if job_id <= 0: + raise ValueError("AdapterCompletion job_id must be positive.") + if exception is not None and not _is_base_exception(exception): + raise TypeError( + "AdapterCompletion exception must derive from BaseException." + ) + if type(cancelled) is not bool: + raise TypeError("AdapterCompletion cancelled must be a boolean.") + outcome_count = int(value is not _MISSING) + int(exception is not None) + int(cancelled) + if outcome_count != 1: + raise ValueError("AdapterCompletion requires exactly one outcome.") + self.job_id: int = job_id + self.value: Any = value + self.exception: BaseException | None = exception + self.cancelled: bool = cancelled + + @property + def has_value(self) -> bool: + """Whether this completion carries a successful value, including ``None``.""" + return self.value is not _MISSING + + @classmethod + def result(cls, job_id: int, value: Any) -> AdapterCompletion: + """Build a successful completion.""" + return cls(job_id, value=value) + + @classmethod + def failed(cls, job_id: int, exc: BaseException) -> AdapterCompletion: + """Build a failed completion.""" + return cls(job_id, exception=exc) + + @classmethod + def cancelled_result(cls, job_id: int) -> AdapterCompletion: + """Build a foreign-runtime cancellation completion.""" + return cls(job_id, cancelled=True) + + +class ExecutionAdapter: + """ + Base API for adapters that execute user callables outside SmallOS. + + Runtime integration is deliberately structural: subclasses provide the + wait object and job lifecycle methods, while this class owns only the + user-facing ``call()`` instruction and single-runtime binding rule. + """ + + def __init__(self, name: str) -> None: + if not name: + raise ValueError("adapter name must not be empty") + self._adapter_name: str = str(name) + self._runtime_binding: dict[str, SmallOS] = {} + self._bound_runtime: SmallOS | None = None + + @property + def name(self) -> str: + """Stable adapter name used in diagnostics.""" + return self._adapter_name + + def call( + self, + callable_obj: Callable[..., Any], + /, + *args: Any, + **kwargs: Any, + ) -> InstructionAwaitable[Any]: + """Build an untyped call instruction for custom adapter subclasses.""" + if not callable(callable_obj): + raise TypeError("adapter call target must be callable") + return adapter_call_instruction(self, callable_obj, args, kwargs) + + def _bind_runtime(self, runtime: SmallOS) -> None: + """Bind one live adapter to exactly one SmallOS scheduler.""" + bound_runtime = self._runtime_binding.setdefault("runtime", runtime) + if bound_runtime is not runtime: + raise AdapterUnavailableError( + "{} is already bound to another SmallOS runtime".format(self.name) + ) + self._bound_runtime = bound_runtime + + @property + def wait_object(self) -> Any: + """Return the readable completion notification object.""" + raise NotImplementedError + + @property + def pending_count(self) -> int: + """Return the number of accepted jobs not yet completed.""" + raise NotImplementedError + + @property + def closed(self) -> bool: + """Whether shutdown has begun and new work must be rejected.""" + raise NotImplementedError + + def submit( + self, + job_id: int, + callable_obj: Callable[..., Any], + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: + """Submit a runtime-assigned job without blocking the scheduler.""" + raise NotImplementedError + + def cancel(self, job_id: int) -> bool: + """Request best-effort cancellation of one job.""" + raise NotImplementedError + + def drain_completions(self) -> list[AdapterCompletion]: + """Drain every currently queued completion.""" + raise NotImplementedError + + def shutdown(self, wait: bool = True, cancel_pending: bool = False) -> None: + """Stop accepting work and release adapter-owned resources.""" + raise NotImplementedError + + def __enter__(self: _AdapterT) -> _AdapterT: + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.shutdown(wait=True, cancel_pending=exc_type is not None) diff --git a/SmallPackage/adapters/errors.py b/SmallPackage/adapters/errors.py new file mode 100644 index 0000000..44f15b0 --- /dev/null +++ b/SmallPackage/adapters/errors.py @@ -0,0 +1,56 @@ +"""Dependency-free errors shared by SmallOS execution adapters.""" + +from __future__ import annotations + +class AdapterError(Exception): + """Base class for execution-adapter failures.""" + + +class AdapterUnavailableError(AdapterError): + """Raised when an adapter cannot run with the active runtime or kernel.""" + + +class AdapterClosedError(AdapterError): + """Raised when work is submitted after adapter shutdown has started.""" + + +class AdapterCapacityError(AdapterError): + """Raised when an adapter's bounded outstanding-work limit is full.""" + + +class AdapterProtocolError(AdapterError): + """Raised when an adapter violates the scheduler completion contract.""" + + +class AdapterCancelledError(AdapterError): + """Raised when the foreign runtime cancels an adapter operation.""" + + +class AdapterExecutionError(AdapterError): + """Wrap a foreign ``BaseException`` that SmallOS must not inject directly.""" + + def __init__(self, message: str, original: BaseException | None = None) -> None: + super().__init__(message) + self.original = original + + +def normalize_adapter_exception( + exc: BaseException, + adapter_name: str = "adapter", + cancelled: bool = False, +) -> Exception: + """Return an exception safe to throw through ``SmallTask.execute()``.""" + if cancelled: + return AdapterCancelledError( + "{} operation was cancelled: {}".format(adapter_name, exc) + ) + if isinstance(exc, Exception): + return exc + return AdapterExecutionError( + "{} operation raised {}: {}".format( + adapter_name, + type(exc).__name__, + exc, + ), + original=exc, + ) diff --git a/SmallPackage/adapters/threads.py b/SmallPackage/adapters/threads.py new file mode 100644 index 0000000..300587e --- /dev/null +++ b/SmallPackage/adapters/threads.py @@ -0,0 +1,195 @@ +"""Thread-backed escape hatch for user-supplied blocking callables.""" + +from __future__ import annotations + +import concurrent.futures +import inspect +import threading + +try: + from typing import TYPE_CHECKING as _type_checking +except ImportError: # pragma: no cover - desktop-only module + _type_checking = False + +if _type_checking: + from collections.abc import Callable + import socket + from typing import Any, TypeVar + + from ..awaitables import InstructionAwaitable + + _ResultT = TypeVar("_ResultT") + +from ..awaitables import adapter_call_instruction +from ._completion import CompletionChannel +from .base import AdapterCompletion, ExecutionAdapter +from .errors import ( + AdapterCapacityError, + AdapterClosedError, + AdapterProtocolError, + normalize_adapter_exception, +) + + +def _is_positive_int(value: object) -> bool: + return type(value) is int and value > 0 + + +class ThreadAdapter(ExecutionAdapter): + """Execute synchronous user callables in a bounded desktop thread pool.""" + + def __init__( + self, + max_workers: int | None = None, + max_pending: int = 64, + thread_name_prefix: str = "smallos-adapter", + ) -> None: + if max_workers is not None and not _is_positive_int(max_workers): + raise ValueError("max_workers must be a positive integer or None") + if not _is_positive_int(max_pending): + raise ValueError("max_pending must be a positive integer") + super().__init__("threads") + self._max_pending: int = max_pending + self._channel: CompletionChannel = CompletionChannel() + self._lock: threading.Lock = threading.Lock() + self._closed: bool = False + self._futures: dict[int, concurrent.futures.Future[Any]] = {} + self._reserved: int = 0 + try: + self._executor = concurrent.futures.ThreadPoolExecutor( + max_workers=max_workers, + thread_name_prefix=thread_name_prefix, + ) + except BaseException: + self._channel.close() + raise + + def call( + self, + callable_obj: Callable[..., _ResultT], + /, + *args: Any, + **kwargs: Any, + ) -> InstructionAwaitable[_ResultT]: + """Return an awaitable whose type matches the blocking callable result.""" + if not callable(callable_obj): + raise TypeError("adapter call target must be callable") + return adapter_call_instruction(self, callable_obj, args, kwargs) + + @property + def wait_object(self) -> socket.socket: + return self._channel.wait_object + + @property + def pending_count(self) -> int: + with self._lock: + return self._reserved + + @property + def closed(self) -> bool: + with self._lock: + return self._closed + + def submit( + self, + job_id: int, + callable_obj: Callable[..., Any], + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: + if not callable(callable_obj): + raise TypeError("ThreadAdapter target must be callable") + + with self._lock: + if self._closed: + raise AdapterClosedError("ThreadAdapter is closed") + if self._reserved >= self._max_pending: + raise AdapterCapacityError( + "ThreadAdapter reached max_pending ({})".format(self._max_pending) + ) + self._reserved += 1 + + try: + future = self._executor.submit(callable_obj, *args, **kwargs) + except BaseException: + with self._lock: + self._reserved -= 1 + raise + + with self._lock: + self._futures[job_id] = future + future.add_done_callback( + lambda completed, completed_job_id=job_id: self._on_done( + completed_job_id, + completed, + ) + ) + + def _on_done( + self, + job_id: int, + future: concurrent.futures.Future[Any], + ) -> None: + with self._lock: + self._futures.pop(job_id, None) + if self._reserved > 0: + self._reserved -= 1 + + if future.cancelled(): + completion = AdapterCompletion.cancelled_result(job_id) + else: + try: + value = future.result() + except BaseException as exc: + completion = AdapterCompletion.failed( + job_id, + normalize_adapter_exception(exc, self.name), + ) + else: + if inspect.isawaitable(value): + close = getattr(value, "close", None) + if callable(close): + close() + completion = AdapterCompletion.failed( + job_id, + AdapterProtocolError( + "ThreadAdapter callable returned an awaitable; " + "use AsyncioAdapter for async callables" + ), + ) + else: + completion = AdapterCompletion.result(job_id, value) + self._channel.post(completion) + with self._lock: + close_after_completion = self._closed and self._reserved == 0 + if close_after_completion: + self._channel.close() + + def cancel(self, job_id: int) -> bool: + with self._lock: + future = self._futures.get(job_id) + if future is None: + return False + return bool(future.cancel()) + + def drain_completions(self) -> list[AdapterCompletion]: + return self._channel.drain() + + def shutdown(self, wait: bool = True, cancel_pending: bool = False) -> None: + with self._lock: + was_closed = self._closed + self._closed = True + if was_closed and not wait: + return + + self._executor.shutdown(wait=wait, cancel_futures=cancel_pending) + if wait: + self._channel.close() + return + + # Running callbacks may still need to publish. Close the channel after + # the final one instead of racing them here. + with self._lock: + should_close = self._reserved == 0 + if should_close: + self._channel.close() diff --git a/SmallPackage/awaitables.py b/SmallPackage/awaitables.py index d3cb58e..2fee93d 100644 --- a/SmallPackage/awaitables.py +++ b/SmallPackage/awaitables.py @@ -19,7 +19,7 @@ TYPE_CHECKING = False if TYPE_CHECKING: - from collections.abc import Generator, Iterable + from collections.abc import Callable, Generator, Iterable from typing import Any, Generic, TypeVar from ._types import TaskTarget @@ -47,7 +47,7 @@ def __repr__(self) -> str: ) -class _InstructionAwaitable(Generic[T] if TYPE_CHECKING else object): +class InstructionAwaitable(Generic[T] if TYPE_CHECKING else object): """ Tiny awaitable wrapper shared by the public helpers below. @@ -65,36 +65,54 @@ def __await__(self) -> Generator[TaskInstruction, Any, T]: return result -def sleep_instruction(seconds: float) -> _InstructionAwaitable[None]: +def sleep_instruction(seconds: float) -> InstructionAwaitable[None]: """Create the awaitable used for cooperative sleeping.""" - return _InstructionAwaitable(TaskInstruction("sleep", seconds=seconds)) + return InstructionAwaitable(TaskInstruction("sleep", seconds=seconds)) -def wait_signal_instruction(signal: int) -> _InstructionAwaitable[int]: +def wait_signal_instruction(signal: int) -> InstructionAwaitable[int]: """Create the awaitable used for waiting on a task signal.""" - return _InstructionAwaitable(TaskInstruction("wait_signal", signal=signal)) + return InstructionAwaitable(TaskInstruction("wait_signal", signal=signal)) -def yield_now_instruction() -> _InstructionAwaitable[None]: +def yield_now_instruction() -> InstructionAwaitable[None]: """Create the awaitable used for a voluntary scheduler yield.""" - return _InstructionAwaitable(TaskInstruction("yield_now")) + return InstructionAwaitable(TaskInstruction("yield_now")) -def join_instruction(target: TaskTarget) -> _InstructionAwaitable[Any]: +def join_instruction(target: TaskTarget) -> InstructionAwaitable[Any]: """Create the awaitable used for waiting on a single task.""" - return _InstructionAwaitable(TaskInstruction("join", target=target)) + return InstructionAwaitable(TaskInstruction("join", target=target)) -def join_all_instruction(targets: Iterable[TaskTarget]) -> _InstructionAwaitable[list[Any]]: +def join_all_instruction(targets: Iterable[TaskTarget]) -> InstructionAwaitable[list[Any]]: """Create the awaitable used for waiting on several tasks at once.""" - return _InstructionAwaitable(TaskInstruction("join_all", targets=list(targets))) + return InstructionAwaitable(TaskInstruction("join_all", targets=list(targets))) -def wait_readable_instruction(io_obj: Any) -> _InstructionAwaitable[Any]: +def wait_readable_instruction(io_obj: Any) -> InstructionAwaitable[Any]: """Create the awaitable used for waiting until an I/O object is readable.""" - return _InstructionAwaitable(TaskInstruction("wait_readable", io_obj=io_obj)) + return InstructionAwaitable(TaskInstruction("wait_readable", io_obj=io_obj)) -def wait_writable_instruction(io_obj: Any) -> _InstructionAwaitable[Any]: +def wait_writable_instruction(io_obj: Any) -> InstructionAwaitable[Any]: """Create the awaitable used for waiting until an I/O object is writable.""" - return _InstructionAwaitable(TaskInstruction("wait_writable", io_obj=io_obj)) + return InstructionAwaitable(TaskInstruction("wait_writable", io_obj=io_obj)) + + +def adapter_call_instruction( + adapter: Any, + callable_obj: Callable[..., T], + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> InstructionAwaitable[T]: + """Create an awaitable for one call through an execution adapter.""" + return InstructionAwaitable( + TaskInstruction( + "adapter_call", + adapter=adapter, + callable=callable_obj, + args=args, + kwargs=kwargs, + ) + ) diff --git a/demos/adapters_asyncio_demo.py b/demos/adapters_asyncio_demo.py new file mode 100644 index 0000000..7d1ad97 --- /dev/null +++ b/demos/adapters_asyncio_demo.py @@ -0,0 +1,107 @@ +"""Reuse standard-library asyncio resources on one persistent adapter loop.""" + +from __future__ import annotations + +import asyncio + +from common import build_runtime, task_runtime + +from SmallPackage import SmallTask, Unix +from SmallPackage.adapters.asyncio_loop import AsyncioAdapter + + +class AsyncioWorker: + """Own a queue and background task that live on the adapter loop.""" + + def __init__(self) -> None: + self.queue: asyncio.Queue[ + tuple[str, asyncio.Future[str]] | None + ] | None = None + self.worker_task: asyncio.Task[None] | None = None + self.owner_loop_id: int | None = None + + def _check_loop(self) -> int: + loop_id = id(asyncio.get_running_loop()) + if self.owner_loop_id is not None and loop_id != self.owner_loop_id: + raise RuntimeError("asyncio resource used from the wrong event loop") + return loop_id + + async def open(self) -> int: + self.owner_loop_id = self._check_loop() + self.queue = asyncio.Queue() + self.worker_task = asyncio.create_task(self._run()) + return self.owner_loop_id + + async def _run(self) -> None: + self._check_loop() + queue = self.queue + if queue is None: + raise RuntimeError("asyncio worker is not open") + while True: + item = await queue.get() + if item is None: + return + message, response = item + await asyncio.sleep(0) + response.set_result(message.upper()) + + async def process(self, message: str) -> tuple[str, int]: + loop_id = self._check_loop() + queue = self.queue + if queue is None: + raise RuntimeError("asyncio worker is not open") + response: asyncio.Future[str] = asyncio.get_running_loop().create_future() + await queue.put((message, response)) + return await response, loop_id + + async def close(self) -> None: + self._check_loop() + if self.queue is not None: + await self.queue.put(None) + if self.worker_task is not None: + await self.worker_task + self.queue = None + self.worker_task = None + self.owner_loop_id = None + + +async def asyncio_example( + task: SmallTask[tuple[str, str]], + adapter: AsyncioAdapter, + service: AsyncioWorker, +) -> tuple[str, str]: + """Create and reuse a loop-affine service across adapter calls.""" + opened_loop = await adapter.call(service.open) + try: + first, first_loop = await adapter.call(service.process, "smallos") + second, second_loop = await adapter.call(service.process, "asyncio") + finally: + await adapter.call(service.close) + + if opened_loop != first_loop or first_loop != second_loop: + raise RuntimeError("adapter calls did not reuse one asyncio loop") + + task_runtime(task).print("Asyncio replies: {}, {}\n".format(first, second)) + return first, second + + +def main() -> None: + runtime = build_runtime(Unix()) + service = AsyncioWorker() + + with AsyncioAdapter(max_pending=8) as foreign_async: + target = SmallTask( + 2, + asyncio_example, + name="asyncio_adapter_example", + args=(foreign_async, service), + ) + runtime.fork(target) + runtime.start() + + if target.exception is not None: + raise target.exception + + +if __name__ == "__main__": + main() diff --git a/demos/adapters_demo.py b/demos/adapters_demo.py new file mode 100644 index 0000000..2f07600 --- /dev/null +++ b/demos/adapters_demo.py @@ -0,0 +1,77 @@ +"""Run blocking and asyncio-owned user code alongside SmallOS tasks.""" + +from __future__ import annotations + +import asyncio +import time + +from common import build_runtime, task_runtime + +from SmallPackage import SmallTask, Unix +from SmallPackage.adapters.asyncio_loop import AsyncioAdapter +from SmallPackage.adapters.threads import ThreadAdapter + + +def blocking_library_call(label: str) -> str: + """Stand in for a user-installed synchronous driver or SDK.""" + time.sleep(0.05) + return "blocking result for {}".format(label) + + +async def asyncio_library_call(label: str) -> str: + """Stand in for a user-installed library that owns asyncio primitives.""" + await asyncio.sleep(0.05) + return "asyncio result for {}".format(label) + + +async def blocking_adapter_demo( + task: SmallTask[str], + adapter: ThreadAdapter, +) -> str: + result = await adapter.call(blocking_library_call, "SmallOS") + task_runtime(task).print(result + "\n") + return result + + +async def asyncio_adapter_demo( + task: SmallTask[str], + adapter: AsyncioAdapter, +) -> str: + result = await adapter.call(asyncio_library_call, "SmallOS") + task_runtime(task).print(result + "\n") + return result + + +async def cooperative_peer(task: SmallTask[str]) -> str: + for index in range(3): + task_runtime(task).print("SmallOS peer step {}\n".format(index)) + await task.sleep(0.5) + return "peer complete" + + +def main() -> None: + runtime = build_runtime(Unix()) + with ThreadAdapter(max_workers=2, max_pending=8) as blocking: + with AsyncioAdapter(max_pending=8) as foreign_async: + runtime.fork( + [ + SmallTask( + 2, + blocking_adapter_demo, + name="blocking_adapter", + args=(blocking,), + ), + SmallTask( + 3, + asyncio_adapter_demo, + name="asyncio_adapter", + args=(foreign_async,), + ), + SmallTask(4, cooperative_peer, name="cooperative_peer"), + ] + ) + runtime.start() + + +if __name__ == "__main__": + main() diff --git a/demos/adapters_sqlite_demo.py b/demos/adapters_sqlite_demo.py new file mode 100644 index 0000000..9fea534 --- /dev/null +++ b/demos/adapters_sqlite_demo.py @@ -0,0 +1,96 @@ +"""Use a thread adapter with a user-owned, thread-affine SQLite connection.""" + +from __future__ import annotations + +from collections.abc import Sequence +import sqlite3 +import threading + +from common import build_runtime, task_runtime + +from SmallPackage import SmallTask, Unix +from SmallPackage.adapters.threads import ThreadAdapter + + +class SQLiteStore: + """Own a SQLite connection without making it part of SmallOS.""" + + def __init__(self) -> None: + self.connection: sqlite3.Connection | None = None + self.owner_thread_id: int | None = None + + def open(self) -> None: + self.owner_thread_id = threading.get_ident() + self.connection = sqlite3.connect(":memory:") + self.connection.execute( + "CREATE TABLE messages (id INTEGER PRIMARY KEY, body TEXT NOT NULL)" + ) + + def _owned_connection(self) -> sqlite3.Connection: + if self.connection is None: + raise RuntimeError("SQLite connection is not open") + if threading.get_ident() != self.owner_thread_id: + raise RuntimeError("SQLite connection used from the wrong thread") + return self.connection + + def insert_messages(self, messages: Sequence[str]) -> None: + connection = self._owned_connection() + connection.executemany( + "INSERT INTO messages (body) VALUES (?)", + [(message,) for message in messages], + ) + connection.commit() + + def read_messages(self) -> list[tuple[int, str]]: + cursor = self._owned_connection().execute( + "SELECT id, body FROM messages ORDER BY id" + ) + return [(int(row[0]), str(row[1])) for row in cursor.fetchall()] + + def close(self) -> None: + connection = self._owned_connection() + connection.close() + self.connection = None + self.owner_thread_id = None + + +async def sqlite_example( + task: SmallTask[list[tuple[int, str]]], + adapter: ThreadAdapter, + store: SQLiteStore, +) -> list[tuple[int, str]]: + """Create, use, and close SQLite entirely on the adapter worker.""" + await adapter.call(store.open) + try: + await adapter.call( + store.insert_messages, + ["blocking libraries", "remain cooperative"], + ) + rows = await adapter.call(store.read_messages) + task_runtime(task).print("SQLite rows: {}\n".format(rows)) + return rows + finally: + await adapter.call(store.close) + + +def main() -> None: + runtime = build_runtime(Unix()) + store = SQLiteStore() + + # One worker creates a serialized execution lane for this connection. + with ThreadAdapter(max_workers=1, max_pending=8) as blocking: + target = SmallTask( + 2, + sqlite_example, + name="sqlite_adapter_example", + args=(blocking, store), + ) + runtime.fork(target) + runtime.start() + + if target.exception is not None: + raise target.exception + + +if __name__ == "__main__": + main() diff --git a/demos/common.py b/demos/common.py index 0b5063d..6a70455 100644 --- a/demos/common.py +++ b/demos/common.py @@ -6,8 +6,11 @@ handler, spawn tasks, and start the runtime. """ +from __future__ import annotations + import os import sys +from typing import Any REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) @@ -18,6 +21,7 @@ from SmallPackage.SmallConfig import SmallOSConfig from SmallPackage.SmallOS import SmallOS from SmallPackage.SmallTask import SmallTask +from SmallPackage.Kernel import Kernel CONFIG_PATH = os.path.join(REPO_ROOT, "smallos.config.json") @@ -32,12 +36,20 @@ def load_demo_config(**overrides): return config -def build_runtime(kernel, **config_overrides): +def build_runtime(kernel: Kernel, **config_overrides: Any) -> SmallOS: """Create a ``SmallOS`` instance wired to the chosen kernel.""" runtime = SmallOS(config=load_demo_config(**config_overrides)).setKernel(kernel) return install_demo_error_handler(runtime) +def task_runtime(task: SmallTask[Any]) -> SmallOS: + """Return the runtime attached before a registered task is executed.""" + runtime = task.OS + if runtime is None: + raise RuntimeError("demo task is not attached to a SmallOS runtime") + return runtime + + def _format_failure_event(event): """Return a readable multi-line summary for demo task failures.""" details = [] @@ -53,6 +65,13 @@ def _format_failure_event(event): details.append("join_target={}".format(event["join_target_id"])) if event["join_pending_ids"]: details.append("join_pending={}".format(event["join_pending_ids"])) + if event.get("adapter_name") is not None: + details.append( + "adapter={}#{}".format( + event["adapter_name"], + event.get("adapter_job_id"), + ) + ) header = "[smallOS demo] task failure" if event["task_name"]: diff --git a/pyproject.toml b/pyproject.toml index 3df95ae..5e7bb0b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,12 +48,28 @@ fallback_version = "0.0.0" [tool.pyright] pythonVersion = "3.10" typeCheckingMode = "basic" -include = ["SmallPackage", "demos"] +include = ["SmallPackage", "demos", "tests/typing"] +strict = [ + "SmallPackage/adapters", + "demos/adapters_demo.py", + "demos/adapters_sqlite_demo.py", + "demos/adapters_asyncio_demo.py", + "tests/typing", +] exclude = ["build", "dist"] ignore = [ "SmallPackage/clients", "SmallPackage/shells.py", - "demos", + "demos/esp32_demo.py", + "demos/http_demo.py", + "demos/micropython_autodetect_demo.py", + "demos/mqtt_demo.py", + "demos/pico_w_demo.py", + "demos/redis_demo.py", + "demos/runtime_demo.py", + "demos/shell_demo.py", + "demos/unix_demo.py", + "demos/web_app_demo.py", ] reportMissingTypeStubs = false reportUnknownMemberType = false diff --git a/tests/test_adapters.py b/tests/test_adapters.py new file mode 100644 index 0000000..2a0b3b1 --- /dev/null +++ b/tests/test_adapters.py @@ -0,0 +1,772 @@ +import asyncio +import sqlite3 +import subprocess +import sys +import threading +import unittest + +sys.path.append("..") + +from SmallPackage.Kernel import Kernel, Unix +from SmallPackage.SmallErrors import TaskCancelledError +from SmallPackage.SmallOS import SmallOS +from SmallPackage.SmallTask import SmallTask +from SmallPackage.adapters import ( + AdapterCancelledError, + AdapterCapacityError, + AdapterCompletion, + AdapterProtocolError, + AdapterUnavailableError, + ExecutionAdapter, +) +from SmallPackage.adapters._completion import CompletionChannel +from SmallPackage.adapters.asyncio_loop import AsyncioAdapter +from SmallPackage.adapters.threads import ThreadAdapter + + +class AdapterTestKernel(Kernel): + """Deterministic readiness kernel for the scheduler contract tests.""" + + def __init__(self): + super().__init__() + self.now = 0 + self.readable = set() + + def ticks_ms(self): + return self.now + + def sleep_ms(self, delay_ms): + self.now += max(0, int(delay_ms)) + + def io_wait(self, readables, writables, timeout_ms=None): + if timeout_ms is not None: + self.now += max(0, int(timeout_ms)) + ready = [obj for obj in readables if obj in self.readable] + for obj in ready: + self.readable.discard(obj) + return ready, [] + + def supports_external_wait_objects(self): + return True + + def mark_readable(self, obj): + self.readable.add(obj) + + +class NoExternalWaitKernel(AdapterTestKernel): + def supports_external_wait_objects(self): + return False + + +class ImmediateAdapter(ExecutionAdapter): + """No-thread adapter used to test only the scheduler integration.""" + + def __init__(self): + super().__init__("manual") + self._wait_object = object() + self._completions = [] + self._pending = set() + self._closed = False + + @property + def wait_object(self): + return self._wait_object + + @property + def pending_count(self): + return len(self._pending) + + @property + def closed(self): + return self._closed + + def submit(self, job_id, callable_obj, args, kwargs): + self._pending.add(job_id) + try: + value = callable_obj(*args, **kwargs) + except BaseException as exc: + completion = AdapterCompletion.failed(job_id, exc) + else: + completion = AdapterCompletion.result(job_id, value) + self._completions.append(completion) + self._bound_runtime.kernel.mark_readable(self.wait_object) + + def cancel(self, job_id): + was_pending = job_id in self._pending + self._pending.discard(job_id) + return was_pending + + def drain_completions(self): + completions = list(self._completions) + self._completions = [] + for completion in completions: + self._pending.discard(completion.job_id) + return completions + + def shutdown(self, wait=True, cancel_pending=False): + self._closed = True + + +class StructuralCompletion: + """Completion-shaped object used to exercise runtime validation.""" + + def __init__( + self, + *, + job_id, + value=None, + exception=None, + cancelled=False, + has_value=True, + ): + self.job_id = job_id + self.value = value + self.exception = exception + self.cancelled = cancelled + self.has_value = has_value + + +class ScriptedCompletionAdapter(ImmediateAdapter): + """Manual adapter that returns a caller-selected structural completion.""" + + def __init__(self, completion_factory): + super().__init__() + self._completion_factory = completion_factory + + def submit(self, job_id, callable_obj, args, kwargs): + self._pending.add(job_id) + self._completions.append(self._completion_factory(job_id)) + self._bound_runtime.kernel.mark_readable(self.wait_object) + + +class TestAdapterSchedulerContract(unittest.TestCase): + def run_task(self, adapter, routine, events=None): + kernel = AdapterTestKernel() + runtime = SmallOS().setKernel(kernel) + task = SmallTask(2, routine, name="adapter-test") + runtime.fork(task) + if events is not None: + runtime.setErrorHandler(events.append) + runtime.start() + return task, runtime + + def test_manual_adapter_returns_result_and_keeps_runtime_alive(self): + adapter = ImmediateAdapter() + + async def routine(task): + return await adapter.call(lambda left, right: left + right, 20, 22) + + task, runtime = self.run_task(adapter, routine) + + self.assertEqual(42, task.result) + self.assertEqual({}, runtime._adapter_jobs) + self.assertIsNone(task._adapter) + self.assertIsNone(task._adapter_job_id) + + def test_manual_adapter_preserves_library_exception_type(self): + adapter = ImmediateAdapter() + + class UserLibraryError(Exception): + pass + + def fail(): + raise UserLibraryError("library failed") + + async def routine(task): + try: + await adapter.call(fail) + except UserLibraryError as exc: + return str(exc) + return "missed" + + task, _runtime = self.run_task(adapter, routine) + self.assertEqual("library failed", task.result) + + def test_uncaught_adapter_failure_includes_resume_origin(self): + adapter = ImmediateAdapter() + events = [] + + def fail(): + raise ValueError("foreign failure") + + async def routine(task): + await adapter.call(fail) + + task, _runtime = self.run_task(adapter, routine, events=events) + + self.assertIsInstance(task.exception, ValueError) + self.assertEqual(1, len(events)) + self.assertEqual("manual", events[0]["adapter_name"]) + self.assertIsInstance(events[0]["adapter_job_id"], int) + + def test_foreign_base_exception_is_normalized(self): + adapter = ImmediateAdapter() + + def stop(): + raise SystemExit("do not stop SmallOS") + + async def routine(task): + try: + await adapter.call(stop) + except Exception as exc: + return type(exc).__name__ + return "missed" + + task, _runtime = self.run_task(adapter, routine) + self.assertEqual("AdapterExecutionError", task.result) + + def test_adapter_requires_kernel_external_wait_capability(self): + adapter = ImmediateAdapter() + runtime = SmallOS().setKernel(NoExternalWaitKernel()) + + async def routine(task): + try: + await adapter.call(lambda: 42) + except AdapterUnavailableError: + return "unsupported" + return "missed" + + target = SmallTask(2, routine, name="unsupported-kernel") + runtime.fork(target) + runtime.start() + + self.assertEqual("unsupported", target.result) + self.assertEqual(0, adapter.pending_count) + + def test_adapter_cannot_be_shared_by_live_runtimes(self): + adapter = ImmediateAdapter() + + async def first(task): + return await adapter.call(lambda: "first") + + first_task, _runtime = self.run_task(adapter, first) + self.assertEqual("first", first_task.result) + + async def second(task): + try: + await adapter.call(lambda: "second") + except AdapterUnavailableError: + return "bound" + return "missed" + + second_task, _runtime = self.run_task(adapter, second) + self.assertEqual("bound", second_task.result) + + def test_core_import_does_not_load_desktop_adapter_backends(self): + code = ( + "import sys; import SmallPackage; " + "assert 'SmallPackage.adapters.threads' not in sys.modules; " + "assert 'SmallPackage.adapters.asyncio_loop' not in sys.modules; " + "assert 'SmallPackage.adapters._completion' not in sys.modules" + ) + completed = subprocess.run( + [sys.executable, "-c", code], + check=False, + capture_output=True, + text=True, + ) + self.assertEqual(0, completed.returncode, completed.stderr) + + def test_completion_channel_closes_when_notification_fails(self): + channel = CompletionChannel() + channel._writer.close() + + posted = channel.post(AdapterCompletion.result(1, 42)) + + self.assertFalse(posted) + self.assertTrue(channel.closed) + self.assertEqual(-1, channel.wait_object.fileno()) + + def test_adapter_completion_enforces_runtime_invariants(self): + with self.assertRaises(TypeError): + AdapterCompletion.result(True, 42) + with self.assertRaises(ValueError): + AdapterCompletion.result(0, 42) + with self.assertRaises(TypeError): + AdapterCompletion(1, exception="not an exception") + with self.assertRaises(TypeError): + AdapterCompletion(1, cancelled=1) + + def test_structural_completion_fields_are_not_truthiness_coerced(self): + invalid_completions = { + "boolean job ID": lambda _job_id: StructuralCompletion( + job_id=True, + value=42, + ), + "non-boolean cancellation": lambda job_id: StructuralCompletion( + job_id=job_id, + cancelled="yes", + has_value=False, + ), + "non-boolean value state": lambda job_id: StructuralCompletion( + job_id=job_id, + value=42, + has_value=1, + ), + "non-exception failure": lambda job_id: StructuralCompletion( + job_id=job_id, + exception="failed", + has_value=False, + ), + } + + for label, completion_factory in invalid_completions.items(): + with self.subTest(label=label): + adapter = ScriptedCompletionAdapter(completion_factory) + + async def routine(task): + try: + await adapter.call(lambda: 42) + except AdapterProtocolError: + return "protocol" + return "accepted" + + task, _runtime = self.run_task(adapter, routine) + self.assertEqual("protocol", task.result) + + def test_structural_adapter_closed_state_must_be_boolean(self): + class InvalidClosedStateAdapter(ImmediateAdapter): + @property + def closed(self): + return 0 + + adapter = InvalidClosedStateAdapter() + + async def routine(task): + try: + await adapter.call(lambda: 42) + except AdapterProtocolError: + return "protocol" + return "accepted" + + task, _runtime = self.run_task(adapter, routine) + self.assertEqual("protocol", task.result) + + +class TestThreadAdapter(unittest.TestCase): + def build_runtime(self, *tasks): + runtime = SmallOS().setKernel(Unix()) + runtime.fork(list(tasks)) + runtime.start() + return runtime + + def test_thread_adapter_rejects_boolean_capacity_values(self): + with self.assertRaises(ValueError): + ThreadAdapter(max_workers=True) + with self.assertRaises(ValueError): + ThreadAdapter(max_pending=True) + + def test_thread_adapter_returns_result_and_propagates_exception(self): + class UserLibraryError(Exception): + pass + + def fail(): + raise UserLibraryError("thread failure") + + with ThreadAdapter(max_workers=2, max_pending=4) as adapter: + async def routine(task): + value = await adapter.call(lambda: 21 * 2) + try: + await adapter.call(fail) + except UserLibraryError as exc: + return value, str(exc) + return None + + target = SmallTask(2, routine, name="thread-result") + self.build_runtime(target) + + self.assertEqual((42, "thread failure"), target.result) + self.assertEqual(0, adapter.pending_count) + self.assertEqual(-1, adapter.wait_object.fileno()) + + def test_thread_adapter_rejects_awaitable_results(self): + async def async_result(): + return 42 + + with ThreadAdapter(max_workers=1, max_pending=2) as adapter: + async def routine(task): + try: + await adapter.call(async_result) + except AdapterProtocolError as exc: + return "AsyncioAdapter" in str(exc) + return False + + target = SmallTask(2, routine, name="thread-awaitable") + self.build_runtime(target) + + self.assertTrue(target.result) + + def test_blocking_thread_does_not_block_other_smallos_tasks(self): + release = threading.Event() + events = [] + + def blocked_call(): + if not release.wait(2): + raise TimeoutError("test release was not set") + return "released" + + with ThreadAdapter(max_workers=1, max_pending=2) as adapter: + async def waiting(task): + events.append("waiting-start") + result = await adapter.call(blocked_call) + events.append("waiting-end") + return result + + async def peer(task): + events.append("peer-ran") + release.set() + return "peer" + + waiting_task = SmallTask(1, waiting, name="waiting") + peer_task = SmallTask(2, peer, name="peer") + self.build_runtime(waiting_task, peer_task) + + self.assertEqual("released", waiting_task.result) + self.assertEqual("peer", peer_task.result) + self.assertEqual(["waiting-start", "peer-ran", "waiting-end"], events) + + def test_thread_adapter_capacity_rejection_is_catchable(self): + release = threading.Event() + + def blocked_call(): + if not release.wait(2): + raise TimeoutError("test release was not set") + return "first" + + with ThreadAdapter(max_workers=1, max_pending=1) as adapter: + async def first(task): + return await adapter.call(blocked_call) + + async def second(task): + try: + await adapter.call(lambda: "second") + except AdapterCapacityError: + release.set() + return "capacity" + return "missed" + + first_task = SmallTask(1, first, name="first") + second_task = SmallTask(2, second, name="second") + self.build_runtime(first_task, second_task) + + self.assertEqual("first", first_task.result) + self.assertEqual("capacity", second_task.result) + + def test_one_worker_preserves_thread_affinity_across_calls(self): + thread_ids = [] + + def record_thread(): + thread_id = threading.get_ident() + thread_ids.append(thread_id) + return thread_id + + with ThreadAdapter(max_workers=1, max_pending=4) as adapter: + async def routine(task): + first = await adapter.call(record_thread) + second = await adapter.call(record_thread) + return first, second + + target = SmallTask(2, routine, name="affinity") + self.build_runtime(target) + + self.assertEqual(2, len(thread_ids)) + self.assertEqual(thread_ids[0], thread_ids[1]) + self.assertEqual(tuple(thread_ids), target.result) + + def test_user_owned_sqlite_connection_can_stay_on_one_worker(self): + state = {} + + def open_database(): + connection = sqlite3.connect(":memory:") + connection.execute("CREATE TABLE records (value INTEGER)") + connection.execute("INSERT INTO records VALUES (42)") + state["connection"] = connection + + def read_database(): + return state["connection"].execute( + "SELECT value FROM records" + ).fetchone()[0] + + def close_database(): + state.pop("connection").close() + + with ThreadAdapter(max_workers=1, max_pending=4) as adapter: + async def routine(task): + await adapter.call(open_database) + try: + return await adapter.call(read_database) + finally: + await adapter.call(close_database) + + target = SmallTask(2, routine, name="sqlite-user-resource") + self.build_runtime(target) + + self.assertEqual(42, target.result) + self.assertEqual({}, state) + + def test_cancelling_thread_waiter_discards_late_completion(self): + started = threading.Event() + release = threading.Event() + + def blocked_call(): + started.set() + release.wait(2) + return "late" + + with ThreadAdapter(max_workers=1, max_pending=2) as adapter: + async def waiting(task): + return await adapter.call(blocked_call) + + async def killer(task, target): + while not started.is_set(): + await task.yield_now() + target.kill() + release.set() + return "killed" + + waiting_task = SmallTask(1, waiting, name="waiting") + killer_task = SmallTask(2, killer, name="killer", args=(waiting_task,)) + self.build_runtime(waiting_task, killer_task) + + self.assertIsInstance(waiting_task.exception, TaskCancelledError) + self.assertEqual("killed", killer_task.result) + self.assertEqual(0, adapter.pending_count) + + def test_thread_shutdown_can_escalate_from_nonblocking_to_waiting(self): + started = threading.Event() + release = threading.Event() + + def blocked_call(): + started.set() + release.wait(2) + + adapter = ThreadAdapter(max_workers=1, max_pending=1) + adapter.submit(1, blocked_call, (), {}) + self.assertTrue(started.wait(1)) + adapter.shutdown(wait=False) + release.set() + + adapter.shutdown(wait=True) + + self.assertEqual(0, adapter.pending_count) + self.assertEqual(-1, adapter.wait_object.fileno()) + + +class TestAsyncioAdapter(unittest.TestCase): + def build_runtime(self, *tasks): + runtime = SmallOS().setKernel(Unix()) + runtime.fork(list(tasks)) + runtime.start() + return runtime + + def test_asyncio_adapter_rejects_boolean_limits(self): + with self.assertRaises(ValueError): + AsyncioAdapter(max_pending=True) + with self.assertRaises(ValueError): + AsyncioAdapter(startup_timeout=True) + + def test_asyncio_adapter_reuses_persistent_loop_thread(self): + async def identify(): + await asyncio.sleep(0) + return id(asyncio.get_running_loop()), threading.get_ident() + + scheduler_thread = threading.get_ident() + with AsyncioAdapter(max_pending=4) as adapter: + async def routine(task): + first = await adapter.call(identify) + second = await adapter.call(identify) + return first, second + + target = SmallTask(2, routine, name="asyncio-identity") + self.build_runtime(target) + + first, second = target.result + self.assertEqual(first, second) + self.assertNotEqual(scheduler_thread, first[1]) + self.assertIsNone(adapter.shutdown_error) + self.assertEqual(-1, adapter.wait_object.fileno()) + + def test_unexpected_loop_stop_is_observable(self): + adapter = AsyncioAdapter(max_pending=1) + + async def stop_loop(): + asyncio.get_running_loop().stop() + return "stopped" + + adapter.submit(1, stop_loop, (), {}) + adapter._thread.join(1) + + self.assertFalse(adapter._thread.is_alive()) + self.assertIsInstance(adapter.shutdown_error, AdapterUnavailableError) + self.assertTrue(adapter.closed) + adapter.shutdown() + + def test_asyncio_resources_can_be_reused_on_the_adapter_loop(self): + state = {} + + async def create_resource(): + loop = asyncio.get_running_loop() + state["future"] = loop.create_future() + return id(loop) + + async def use_resource(): + loop = asyncio.get_running_loop() + future = state.pop("future") + future.set_result(42) + return id(loop), await future + + with AsyncioAdapter(max_pending=4) as adapter: + async def routine(task): + created_loop = await adapter.call(create_resource) + used_loop, value = await adapter.call(use_resource) + return created_loop, used_loop, value + + target = SmallTask(2, routine, name="asyncio-resource") + self.build_runtime(target) + + created_loop, used_loop, value = target.result + self.assertEqual(created_loop, used_loop) + self.assertEqual(42, value) + self.assertEqual({}, state) + + def test_asyncio_library_wait_does_not_block_smallos(self): + release = threading.Event() + events = [] + + async def blocked_call(): + while not release.is_set(): + await asyncio.sleep(0) + return "released" + + with AsyncioAdapter(max_pending=4) as adapter: + async def waiting(task): + events.append("waiting-start") + value = await adapter.call(blocked_call) + events.append("waiting-end") + return value + + async def peer(task): + events.append("peer-ran") + release.set() + return "peer" + + waiting_task = SmallTask(1, waiting, name="waiting") + peer_task = SmallTask(2, peer, name="peer") + self.build_runtime(waiting_task, peer_task) + + self.assertEqual("released", waiting_task.result) + self.assertEqual("peer", peer_task.result) + self.assertEqual(["waiting-start", "peer-ran", "waiting-end"], events) + + def test_asyncio_exception_protocol_and_cancellation_are_catchable(self): + class UserAsyncError(Exception): + pass + + async def fail(): + raise UserAsyncError("async failure") + + async def cancel_self(): + raise asyncio.CancelledError() + + with AsyncioAdapter(max_pending=4) as adapter: + async def routine(task): + outcomes = [] + try: + await adapter.call(fail) + except UserAsyncError as exc: + outcomes.append(str(exc)) + try: + await adapter.call(lambda: 42) + except AdapterProtocolError: + outcomes.append("protocol") + try: + await adapter.call(cancel_self) + except AdapterCancelledError: + outcomes.append("cancelled") + return outcomes + + target = SmallTask(2, routine, name="async-errors") + self.build_runtime(target) + + self.assertEqual(["async failure", "protocol", "cancelled"], target.result) + + def test_asyncio_adapter_rejects_future_from_another_loop(self): + other_loop = asyncio.new_event_loop() + foreign_future = other_loop.create_future() + try: + with AsyncioAdapter(max_pending=2) as adapter: + async def routine(task): + try: + await adapter.call(lambda: foreign_future) + except AdapterProtocolError as exc: + return "another event loop" in str(exc) + return False + + target = SmallTask(2, routine, name="foreign-future") + self.build_runtime(target) + finally: + foreign_future.cancel() + other_loop.close() + + self.assertTrue(target.result) + + def test_asyncio_capacity_rejection_is_catchable(self): + release = threading.Event() + + async def blocked_call(): + while not release.is_set(): + await asyncio.sleep(0) + return "first" + + with AsyncioAdapter(max_pending=1) as adapter: + async def first(task): + return await adapter.call(blocked_call) + + async def second(task): + try: + await adapter.call(asyncio.sleep, 0) + except AdapterCapacityError: + release.set() + return "capacity" + return "missed" + + first_task = SmallTask(1, first, name="first") + second_task = SmallTask(2, second, name="second") + self.build_runtime(first_task, second_task) + + self.assertEqual("first", first_task.result) + self.assertEqual("capacity", second_task.result) + + def test_cancelling_smallos_task_cancels_asyncio_task(self): + started = threading.Event() + cleaned_up = threading.Event() + + async def blocked_call(): + started.set() + try: + await asyncio.sleep(10) + finally: + cleaned_up.set() + + with AsyncioAdapter(max_pending=2) as adapter: + async def waiting(task): + return await adapter.call(blocked_call) + + async def killer(task, target): + while not started.is_set(): + await task.yield_now() + target.kill() + return "killed" + + waiting_task = SmallTask(1, waiting, name="waiting") + killer_task = SmallTask(2, killer, name="killer", args=(waiting_task,)) + self.build_runtime(waiting_task, killer_task) + + self.assertIsInstance(waiting_task.exception, TaskCancelledError) + self.assertEqual("killed", killer_task.result) + self.assertTrue(cleaned_up.is_set()) + self.assertFalse(adapter._thread.is_alive()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/typing/adapter_contract.py b/tests/typing/adapter_contract.py new file mode 100644 index 0000000..c828371 --- /dev/null +++ b/tests/typing/adapter_contract.py @@ -0,0 +1,33 @@ +# pyright: strict + +"""Static consumer contract for execution-adapter result inference.""" + +from SmallPackage.adapters.asyncio_loop import AsyncioAdapter +from SmallPackage.adapters.threads import ThreadAdapter +from SmallPackage._types import ExecutionAdapterLike + + +def sync_value() -> int: + return 42 + + +async def async_value() -> int: + return 42 + + +def accepts_scheduler_adapter(adapter: ExecutionAdapterLike) -> None: + """Require concrete adapters to satisfy the scheduler's protocol.""" + + +async def verify_adapter_result_types() -> None: + with ThreadAdapter() as threads: + accepts_scheduler_adapter(threads) + thread_result: int = await threads.call(sync_value) + + with AsyncioAdapter() as asyncio_adapter: + accepts_scheduler_adapter(asyncio_adapter) + asyncio_result: int = await asyncio_adapter.call(async_value) + shutdown_error: BaseException | None = asyncio_adapter.shutdown_error + + assert thread_result == asyncio_result + assert shutdown_error is None