From 694b75f9e817a4e71af72d0b681cba4735839c35 Mon Sep 17 00:00:00 2001 From: Jeremy Howard Date: Fri, 28 Aug 2026 21:43:46 +1000 Subject: [PATCH] Rebuild on stdlib loop machinery: subclass BaseEventLoop, remove Tokio, require Python 3.11+ --- .github/workflows/ci.yml | 7 +- Cargo.toml | 6 +- DEV.md | 153 ++-------- README.md | 7 +- pyproject.toml | 2 +- python/loopmini/loop.py | 558 ++++++---------------------------- python/loopmini/subproc.py | 118 +------ python/loopmini/transports.py | 377 ++++------------------- rustfmt.toml | 3 + src/lib.rs | 4 +- src/pyreactor.rs | 37 +-- src/reactor.rs | 38 ++- src/tokio_core.rs | 118 ------- tests/oracle_util.py | 90 +++--- tests/test_bench.py | 27 +- 15 files changed, 300 insertions(+), 1245 deletions(-) create mode 100644 rustfmt.toml delete mode 100644 src/tokio_core.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e4c0850..cf96825 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,13 +8,16 @@ on: jobs: test: + strategy: + matrix: + python-version: ['3.11', '3.12', '3.13', '3.14'] runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 - uses: dtolnay/rust-toolchain@stable - uses: actions/setup-python@v7 with: - python-version: '3.12' + python-version: ${{ matrix.python-version }} - run: pip install -e '.[dev]' - run: pytest -q @@ -28,7 +31,7 @@ jobs: - uses: actions/checkout@v7 - uses: PyO3/maturin-action@v1 with: - args: --release --out dist -i python3.10 -i python3.11 -i python3.12 -i python3.13 + args: --release --out dist -i python3.11 -i python3.12 -i python3.13 -i python3.14 manylinux: auto - uses: actions/upload-artifact@v7 with: diff --git a/Cargo.toml b/Cargo.toml index 0a38662..8cf4e13 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "loopmini" -version = "0.1.2" +version = "0.2.0" edition = "2021" license = "Apache-2.0" description = "Rust-backed asyncio event loop" @@ -13,10 +13,6 @@ crate-type = ["cdylib", "rlib"] [dependencies] pyo3 = { version = ">=0.28", features = ["py-clone"] } polling = "3" -tokio = { version = "1", features = ["rt", "time", "net"] } - -[dev-dependencies] -tokio = { version = "1", features = ["rt-multi-thread"] } [features] extension-module = ["pyo3/extension-module"] diff --git a/DEV.md b/DEV.md index ee15dfd..affef0f 100644 --- a/DEV.md +++ b/DEV.md @@ -2,104 +2,27 @@ ## Why this exists -Kernmini's Rust engine (see kernmini `meta/ROUGH.md`, "Separate project idea: a -Tokio-backed asyncio loop" and the 2026-08-28 review) needs Python kernels to -run arbitrary user asyncio code. loopmini tests whether a Rust reactor can host -that loop with full asyncio compatibility. The compatibility bar is the full -public asyncio surface, because solveit users run arbitrary packages; the -strategy for meeting it is maximal reuse of CPython's own asyncio machinery, -with Rust owning only what Python cannot express well. +Kernmini's Rust engine (see kernmini `meta/ROUGH.md`, "Separate project idea: a Rust-backed asyncio loop" and the 2026-08-28 review) needs Python kernels to run arbitrary user asyncio code. loopmini tests whether a Rust reactor can host that loop with full asyncio compatibility. The compatibility bar is the full public asyncio surface, because solveit users run arbitrary packages; the strategy is maximal reuse of CPython's own asyncio machinery, with Rust owning only what Python cannot express well. ## Crate/Python split -A key goal is that Rust crates and PyO3 wrappers share code: the same reactor -must be drivable by a pure-Rust kernel (as kernmini's native crate will be). -So the crate has three layers: - -- `src/reactor.rs`: `Reactor`, a PyO3-free, Tokio-free core generic over - the handle type. Ready queue, timer map, fd interest map (oneshot `polling` - sources, re-armed on delivery), thread-safe `schedule_ts` + notify, and the - turn phases: `next_timeout` / `poll` / `process` / `take_batch` / - `requeue_front`. `poll` holds no locks, so a driver may block in it with the - GIL released. The crate builds as an rlib, so a Rust consumer instantiates - `Reactor` directly. -- `src/tokio_core.rs`: the reactor hosted on a Tokio current-thread runtime. - Tokio waits on the kqueue's own fd (a kqueue is pollable) and a zero-timeout - drain then collects events, so the kqueue-native level/oneshot semantics - that asyncio's `add_reader` contract needs survive Tokio's edge-triggered - driver. Rust futures spawned on the runtime advance during every blocking - poll, with the GIL released: the shared-reactor path for kernmini's engine. - Zero-timeout turns skip the runtime entirely, because entering it rounds the - wait up to the timer driver's ~1ms tick. -- `src/pyreactor.rs`: `PyReactor`, the Python-facing pyclass: the scheduling - and readiness methods plus the canonical dispatch loop (`check_signals` each - turn, `py.detach` around `poll`, EINTR retry, and the injected-exception - requeue rule, which must exist exactly once). `loopmini._core` registers it - on an owned runtime; kernmini's Python feature compiles the same struct and constructs - it with `PyReactor::with_handle` on kernmini's runtime. Rust has no stable - dylib ABI, so runtimes and reactors never cross extension boundaries: each - extension compiles the crate in, and the Python-visible reactor methods are - the only shared surface. `Loop(reactor=...)` accepts such a foreign reactor. -- `src/lib.rs`: module registration and the public re-exports (`ReactorCore`, - `TokioCore`, `PyReactor`) for embedding crates. - -Python (`python/loopmini/loop.py`) implements `asyncio.AbstractEventLoop` by -delegating scheduling to the reactor and reusing stock `Handle`, `TimerHandle`, -`Task`, `Future`, and `wrap_future`. That reuse is a design decision, not a -shortcut: it buys exact contextvars, cancellation, and introspection semantics -and removes the most version-sensitive surface. `run_forever` on the main -thread routes signals through `signal.set_wakeup_fd` into an fd the reactor -watches, matching `BaseSelectorEventLoop`. +A key goal is that Rust crates and PyO3 wrappers share code: the same reactor must be drivable by a pure-Rust kernel (as kernmini's native crate will be). The implementation has two layers plus module registration: + +- `src/reactor.rs`: `Reactor`, a PyO3-free, Tokio-free core generic over the handle type. It owns the ready queue, timer map, fd interest map (oneshot `polling` sources, re-armed on delivery), thread-safe `schedule_ts` + notify, and the turn phases: `next_timeout` / `poll` / `process` / `take_batch` / `requeue_front`. `poll` holds no locks, so a driver may block in it with the GIL released. The crate builds as an rlib, so a Rust consumer instantiates `Reactor` directly. +- `src/pyreactor.rs`: `PyReactor`, the Python-facing pyclass: the scheduling and readiness methods plus the canonical dispatch loop (`check_signals` each turn, `py.detach` around `poll`, EINTR retry, and the injected-exception requeue rule, which must exist exactly once). The driver blocks directly in the core poller with the GIL released. Rust runtimes keep their futures on worker threads and wake the loop with its thread-safe scheduling path. +- `src/lib.rs`: module registration and the public re-exports (`ReactorCore`, `PyReactor`) for embedding crates. + +Python (`python/loopmini/loop.py`) subclasses `asyncio.BaseEventLoop`, delegating scheduling to the reactor while inheriting task/future creation, executors, exception handling, high-level networking, TLS orchestration, servers, subprocess entry points, and buffered sendfile. Reactor-neutral implementations of socket operations and accepting connections are borrowed from `BaseSelectorEventLoop`; Unix connection/server methods and pipe transports are borrowed from `_UnixSelectorEventLoop`. Loopmini supplies the hooks those implementations call. This reuse buys exact CPython contextvars, cancellation, introspection, lifecycle, and version-specific semantics. `run_forever` on the main thread routes signals through `signal.set_wakeup_fd` into an fd watched by the reactor. ## Implemented surface -Beyond the scheduling core: TCP transports (`transports.py`: `SockTransport` -speaking both `Protocol` and `BufferedProtocol`, with `TCP_NODELAY` set as the -standard loop does, and `MiniServer` with the accept loop), -`create_connection`/`create_server` and their Unix-socket twins, -`connect_accepted_socket`, UDP via `DatagramTransport` and -`create_datagram_endpoint`, `start_tls` (including the gh-142352 -buffered-StreamReader move), TLS client and server via stock `asyncio.sslproto` -over our transports, `getaddrinfo`/`getnameinfo` through the executor, and -`add_signal_handler`/`remove_signal_handler` dispatched from the -`set_wakeup_fd` socketpair (main-thread loops only, as in the standard loop), -pipe transports (`connect_read_pipe`/`connect_write_pipe`), and -`subprocess_exec`/`subprocess_shell` (`subproc.py`: Popen spawned in the -executor since fork/exec blocks, one reaper thread per child as in asyncio's -default ThreadedChildWatcher, and pipe transports over the child's fds). - -Contracts learned from the oracles, kept working by them: `remove_reader` -cancels the stored `Handle`, so an already-queued readiness callback for a -removed fd never fires (anyio's futures assume it); `connection_lost` is -scheduled exactly once however `close()`, `abort()`, and fatal errors overlap -(anyio's `aclose` does close-then-abort, and a missed schedule leaks the -socket); a cancelled connect attempt closes its socket; and a raised -`create_connection` error must not keep a referrer to the attempt-error list. -`MiniServer` mirrors 3.13 `Server` exactly (test_server pins it): -`wait_closed()` returns only once the server is closed *and* the last client -transport is gone, `close_clients()`/`abort_clients()` act on a `WeakSet` of -attached transports, a cancelled `serve_forever` closes the server and its -clients before re-raising, and a Unix socket path is unlinked at close only -when its inode still matches the one bound (`cleanup_socket=False` disables). -A cancelled timer leaves the reactor's timer map at once: `schedule_at` -returns a `(deadline µs, seq)` key, the `TimerHandle` subclass carries it, and -`_timer_handle_cancelled` removes the entry. Retention until the original -deadline would grow memory for an hour per cancelled `wait_for(..., 3600)`. -From the aiohttp suite: `loop.time()` must be anchored to `time.monotonic` -(connectors compare the two directly), so the reactor clock carries a -constant offset captured at loop creation; and a transport must not hold an -explicit `memoryview` export of its write buffer across a `send` that can -raise, because the raised exception's traceback keeps the export alive and -the buffer can then never be resized (send the bytearray itself, as the -standard loop does). +Beyond the scheduling core, loopmini implements TCP transports (`SockTransport`, speaking both `Protocol` and `BufferedProtocol`, with `TCP_NODELAY`), UDP (`DatagramTransport`), signal dispatch through the `set_wakeup_fd` socketpair, and the reactor's reader/writer registrations. CPython supplies the high-level TCP/UDP/TLS/Unix/server operations, Unix pipe transports, flow control, and buffered sendfile. `subproc.py` prepares `Popen` in the executor because fork/exec blocks, then gives it to CPython's `BaseSubprocessTransport`; one reaper thread per child reports exit through the loop's thread-safe scheduling path. + +Contracts learned from the oracles, kept working by them: `remove_reader` cancels the stored `Handle`, so an already-queued readiness callback for a removed fd never fires; `connection_lost` is scheduled exactly once however `close()`, `abort()`, and fatal errors overlap; and a Unix socket path is unlinked at close only when its inode still matches the one bound. A cancelled timer leaves the reactor's timer map at once: `schedule_at` returns a `(deadline µs, seq)` key, the `TimerHandle` subclass carries it, and `_timer_handle_cancelled` removes the entry. Retention until the original deadline would grow memory for an hour per cancelled `wait_for(..., 3600)`. The reactor clock carries a constant offset captured at loop creation because aiohttp compares `loop.time()` with `time.monotonic`. A socket transport sends its bytearray directly rather than holding an explicit `memoryview` export across a potentially raising `send`, whose traceback could otherwise keep the buffer unresizable. ## Deliberately not implemented yet -Sendfile and Windows. -`AbstractEventLoop` raises `NotImplementedError` for these, which is the -honest signal while the compatibility ladder is climbed. Next rungs and their -oracles (anyio/aiohttp/httpx/websockets test suites) are listed in kernmini -`meta/ROUGH.md`. +Native sendfile and Windows. `BaseEventLoop` provides the portable buffered sendfile fallback; the socket transport declares that capability rather than pretending to support the platform-native fast path. ## Tests @@ -107,25 +30,16 @@ The tiers, so the inner loop stays seconds: - `pytest -q` per change (~2s). - `pytest -m oracle -n auto` when a feature lands, not per edit. -- `pytest -m bench -s` and `pytest -m soak -s` only when performance or - stability is the question. +- `pytest -m bench -s` and `pytest -m soak -s` only when performance or stability is the question. - `chkstyle` once, at the final PR stage, never per edit. -`pytest -q`. Ten integration stories, deliberately few: scheduling/tasks/ -threads (timers, contextvars, gather, TaskGroup, timeout, cancellation, -cross-thread wakeup, to_thread), socket I/O through the reactor (accept/ -connect/backpressure on a 5MB payload), asyncio streams echo with drain -backpressure, a background task surviving between `run_until_complete` calls -(the kernel-persistence story), KeyboardInterrupt injection with the loop -reused afterwards, uvicorn serving an ASGI app fetched by urllib and by -httpx-over-anyio, TLS echo against a throwaway openssl cert, a subprocess -round-trip (exec and shell, streams and communicate), and the -interrupt-torture story: window-scoped `PyThreadState_SetAsyncExc` injection -under stream/timer/cross-thread load, mirroring kernmini's -`sync_execution_context` contract (0.5s by default; `LOOPMINI_TORTURE_SECONDS` -extends it), and the KI-at-`_run`-entry orphan repro that pins the -traceback-depth requeue rule. Rust-side unit tests should exist only for -reactor invariants Python stories cannot reach. +To test against another Python version locally (the workspace venv is pinned, so +plain `uv run --python` refuses): `uv run --no-project --python 3.14 --with +'.[dev]' pytest -q` from the repo root. uv builds against its managed +interpreter into its cache, so repeat runs are fast and there is no venv to +maintain. CI runs the same suite on every supported version. + +`pytest -q`. Eleven integration stories, deliberately few: scheduling/tasks/threads (timers, contextvars, gather, TaskGroup, timeout, cancellation, cross-thread wakeup, to_thread), socket I/O through the reactor (accept/connect/backpressure on a 5MB payload), asyncio streams echo with drain backpressure, a background task surviving between `run_until_complete` calls (the kernel-persistence story), KeyboardInterrupt injection with the loop reused afterwards, uvicorn serving an ASGI app fetched by urllib and by httpx-over-anyio, TLS echo against a throwaway openssl cert, a subprocess round-trip (exec and shell, streams and communicate), and the interrupt-torture story: window-scoped `PyThreadState_SetAsyncExc` injection under stream/timer/cross-thread load, mirroring kernmini's `sync_execution_context` contract (0.5s by default; `LOOPMINI_TORTURE_SECONDS` extends it), and the KI-at-`_run`-entry orphan repro that pins the traceback-depth requeue rule. Rust-side unit tests should exist only for reactor invariants Python stories cannot reach. Two CPython 3.13 facts the torture test depends on, discovered the hard way: an async-injected exception raised at an eval-breaker check escapes the @@ -145,8 +59,7 @@ pytest -m oracle -n auto # all four suites, one worker per module pytest -m oracle -k uvloop # one suite; add "and test_tcp" etc. for one module ``` -`tests/oracle_util.py` fetches and caches what each suite needs under -`~/.cache/loopmini-oracle` (override with `LOOPMINI_ORACLE_CACHE`): +`tests/oracle_util.py` fetches and caches what each suite needs under `~/.cache/loopmini-oracle` (override with `LOOPMINI_ORACLE_CACHE`). Modified external source trees are content-addressed by the source of their adapter function, so changing an adapter creates a fresh fixture while unchanged fixtures retain their cache: - CPython's own test_asyncio (uvloop's strategy). This uv-managed Python ships without the stdlib `test` package, so the matching source tarball is fetched @@ -165,20 +78,9 @@ pytest -m oracle -k uvloop # one suite; add "and test_tcp" etc. for one m onto loopmini with `implementation='asyncio'`, so branchy tests take the standard-loop expectation paths. `test_tcp` needs pyOpenSSL and is skipped without it. -- aiohttp's test suite: the sdist is unpacked, uvloop is stubbed to loopmini in - its conftest so `--aiohttp-loop=uvloop` selects it, and its blockbuster - fixture gains an allowlist entry for `MiniServer.close` (the same - unlink-if-unchanged stat asyncio's `_stop_serving` is allowlisted for). - Runs the pure-Python aiohttp (`AIOHTTP_NO_EXTENSIONS=1`), with a private - `--basetemp` because its permission tests leave chmod-000 dirs that - pytest's numbered-dir sweeper cannot remove, fatal under aiohttp's - `filterwarnings = error`. Deselected, with reasons in `test_oracle.py`: - tests that fail identically on the standard loop in this venv, and one - test that mock-patches `loop.time()`, which only steers loops whose timer - arithmetic reads Python-level time each turn (real uvloop fails it too). - -Status on 2026-08-28: all green; 25 oracle tests, the first three suites in -~17s under xdist and aiohttp's ~4,200 tests in a further ~2 minutes. +- aiohttp's test suite: the sdist is unpacked, uvloop is stubbed to loopmini in its conftest so `--aiohttp-loop=uvloop` selects it, and its blockbuster fixture gains an allowlist entry for the same unlink-if-unchanged stat used by asyncio's Unix loop. It runs pure-Python aiohttp (`AIOHTTP_NO_EXTENSIONS=1`) with a private `--basetemp`, because permission tests leave chmod-000 directories that pytest's numbered-directory sweeper cannot remove under aiohttp's `filterwarnings = error`. Deselections and their reasons live in `test_oracle.py`. + +Status on 2026-08-28: all green; 26 oracle stories, including about 4,200 aiohttp tests, complete in about 26 seconds under xdist on the development Mac. ## The soak gate @@ -191,12 +93,7 @@ component made progress. It watches stability, not speed. ## Benchmarks -`pytest -m bench -s` prints loopmini vs the standard loop on loop-bound -microbenchmarks; informational, never gating. Numbers on 2026-08-28 (M-series -macOS): call_soon 1.10x, sleep0 1.12x, tcp_echo 1.10x, task spawn 2.48x -(stdlib=1.0). The spawn gap is per-task PyO3 boundary crossings (two to three -`schedule` calls per task lifecycle at ~0.5µs each); loop startup is at parity. -Closing it would need handles represented Rust-side, which is later-rung work. +`pytest -m bench -s` prints loopmini vs the standard loop on loop-bound microbenchmarks; informational, never gating. Numbers on 2026-08-28 (M-series macOS): call_soon 1.04x, sleep0 1.03x, tcp_echo 1.02x, task spawn 1.17x (stdlib=1.0); the corresponding extra cost is approximately 0.5µs per scheduled callback, 0.5µs per `sleep(0)` suspend/resume, 1.1µs per TCP echo, and 0.3µs per spawned task. Median 1ms-timer overshoot is 0.188ms versus 0.156ms. ## Style and releases diff --git a/README.md b/README.md index 2d485b1..af97fc4 100644 --- a/README.md +++ b/README.md @@ -20,9 +20,9 @@ pip install loopmini ## Design -The Python side subclasses `asyncio.AbstractEventLoop` and reuses the stock `Task`, `Future`, `Handle`, and `sslproto` machinery. Contextvars, cancellation, and task introspection therefore behave exactly as in the standard loop. CPython releases change little that loopmini must track, because the version-sensitive objects are CPython's own. +The Python side subclasses `asyncio.BaseEventLoop` and reuses its task, future, executor, error-handling, networking, server, sendfile, and subprocess machinery. Reactor-neutral socket, Unix, pipe, and accept implementations come directly from CPython's selector loop. Loopmini supplies the scheduling and fd-readiness hooks plus socket and datagram transports, so version-sensitive asyncio behavior remains CPython's own. -The Rust side owns fd readiness, timers, and cross-thread wakeup. A small level-triggered reactor core (the `polling` crate, kqueue/epoll) is hosted on a Tokio current-thread runtime, which waits on the reactor's own pollable fd. This split preserves the level-triggered `add_reader` contract that asyncio requires and Tokio's edge-triggered driver cannot express. Rust futures spawned on the runtime advance during every blocking poll, with the GIL released, on the same thread and reactor as the Python loop. +The Rust side owns fd readiness, timers, and cross-thread wakeup through a small level-triggered reactor core (`polling`, using kqueue/epoll). The Python driver blocks directly in the reactor with the GIL released. The same PyO3-free core is available to Rust consumers; embedding runtimes can run futures on their own workers and wake the Python loop through its thread-safe scheduling path. ## Compatibility @@ -34,5 +34,4 @@ A KeyboardInterrupt injected while the loop runs (the kernel interrupt mechanism ## Performance -Throughput matches the standard loop on real I/O workloads. A 30-second soak serving a fasthtml app under concurrent HTTP and websocket load holds a 2.6ms median response with no fd or memory growth. Microbenchmarks run 5 to 15% slower than the standard loop, and creating a task costs about twice as much, because each schedule crosses the Python/Rust boundary. uvloop is faster where speed is the requirement. - +Throughput matches the standard loop on real I/O workloads. A 30-second soak serving a fasthtml app under concurrent HTTP and websocket load holds a 2.6ms median response with no fd or memory growth. There is a small overhead involved in getting the better interrupt semantics, due to having to cross the Rust boundary. It applies only when something is scheduled onto the loop, and is then under 1µs per operation: `create_task`, an `await` that suspends, `asyncio.sleep`, or `call_soon`. Code that stays inside Python, including an `await` that does not suspend, pays nothing. The median overshoot of a 1ms timer is ~0.2ms, similar to Python's standard loop. diff --git a/pyproject.toml b/pyproject.toml index a8d1e65..8a15e44 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ name = "loopmini" dynamic = ["version"] description = "Rust-backed asyncio event loop" license = {text = "Apache-2.0"} -requires-python = ">=3.10" +requires-python = ">=3.11" readme = "README.md" authors = [{name = "Jeremy Howard", email = "github@jhoward.fastmail.fm"}] classifiers = [ diff --git a/python/loopmini/loop.py b/python/loopmini/loop.py index a74d046..e1607b8 100644 --- a/python/loopmini/loop.py +++ b/python/loopmini/loop.py @@ -1,19 +1,16 @@ "An asyncio event loop whose reactor (fd readiness, timers, cross-thread wakeup) runs in Rust." -import asyncio, concurrent.futures, errno, functools, itertools, os, signal, socket, ssl as ssl_mod, stat as stat_mod -import subprocess as subprocess_mod, sys, threading, time, traceback, warnings -from asyncio import base_events, events, futures, sslproto, tasks -import weakref +import errno, functools, os, signal, socket, subprocess, sys, threading, time, weakref +from asyncio import base_events, events, selector_events, sslproto, unix_events from ._core import Reactor -from .transports import SockTransport, MiniServer, DatagramTransport, ReadPipeTransport, WritePipeTransport +from .transports import SockTransport, DatagramTransport from .subproc import SubprocessTransport __all__ = ["Loop", "new_event_loop"] -def _fileno(fd): return fd if isinstance(fd, int) else fd.fileno() +_SelLoop = selector_events.BaseSelectorEventLoop +_UnixLoop = unix_events._UnixSelectorEventLoop -def _check_callback(callback, name): - if asyncio.iscoroutine(callback) or asyncio.iscoroutinefunction(callback): raise TypeError(f'coroutines cannot be used with {name}()') - if not callable(callback): raise TypeError(f'a callable object was expected by {name}(), got {callback!r}') +def _fileno(fd): return fd if isinstance(fd, int) else fd.fileno() def _sighandler_noop(signum, frame): pass @@ -39,33 +36,23 @@ def _run(self): _timed_run(self, lambda: events.TimerHandle._run(self)) _TimerHandle.__name__ = _TimerHandle.__qualname__ = 'TimerHandle' _TimedTimerHandle.__name__ = _TimedTimerHandle.__qualname__ = 'TimerHandle' -class Loop(asyncio.AbstractEventLoop): +class Loop(base_events.BaseEventLoop): def __init__(self, reactor=None, # A `_core.Reactor`-shaped object; an embedding host passes one on its own runtime ): - self._r,self._running,self._closed,self._debug = reactor if reactor is not None else Reactor(),False,False,False + super().__init__() + self._r = reactor if reactor is not None else Reactor() # Anchor the reactor clock to time.monotonic (same underlying clock, so the # offset is constant): libraries compare loop.time() against monotonic directly self._time_offset = time.monotonic() - self._r.time() - self._thread_id = None - self.slow_callback_duration = 0.1 - self._asyncgens = weakref.WeakSet() - self._asyncgens_shutdown_called = False - self._default_executor = self._exception_handler = self._task_factory = None self._signal_handlers = {} + self._unix_server_sockets = {} + self._transports = weakref.WeakValueDictionary() self._readers,self._writers = {},{} def time(self): return self._r.time() + self._time_offset - def _check_thread(self): - # Debug-mode guard, as in the standard loop: catches non-threadsafe cross-thread calls - if self._thread_id is None or self._thread_id == threading.get_ident(): return - raise RuntimeError('Non-thread-safe operation invoked on an event loop other than the current one') - - def call_soon(self, callback, *args, context=None): - self._check_closed() - _check_callback(callback, 'call_soon') - if self._debug: self._check_thread() + def _call_soon(self, callback, args, context): h = (_TimedHandle if self._debug else events.Handle)(callback, args, self, context) if h._source_traceback: del h._source_traceback[-1] self._r.schedule(h) @@ -73,22 +60,18 @@ def call_soon(self, callback, *args, context=None): def call_soon_threadsafe(self, callback, *args, context=None): self._check_closed() - _check_callback(callback, 'call_soon_threadsafe') + self._check_callback(callback, 'call_soon_threadsafe') h = (_TimedHandle if self._debug else events.Handle)(callback, args, self, context) if h._source_traceback: del h._source_traceback[-1] self._r.schedule_ts(h) return h - def call_later(self, delay, callback, *args, context=None): - _check_callback(callback, 'call_later') - h = self.call_at(self.time()+delay, callback, *args, context=context) - if h._source_traceback: del h._source_traceback[-1] - return h - def call_at(self, when, callback, *args, context=None): + if when is None: raise TypeError('when cannot be None') self._check_closed() - _check_callback(callback, 'call_at') - if self._debug: self._check_thread() + if self._debug: + self._check_thread() + self._check_callback(callback, 'call_at') h = (_TimedTimerHandle if self._debug else _TimerHandle)(when, callback, args, self, context) if h._source_traceback: del h._source_traceback[-1] h._reactor_key = self._r.schedule_at(when - self._time_offset, h) @@ -98,38 +81,15 @@ def _timer_handle_cancelled(self, handle): # False (already fired) is fine: dispatch skips the promoted handle's cancelled flag self._r.cancel_timer(handle._reactor_key) - def create_future(self): return futures.Future(loop=self) - - def create_task(self, coro, *, name=None, context=None, eager_start=False): - self._check_closed() - if self._task_factory is None: return tasks.Task(coro, loop=self, name=name, context=context, eager_start=eager_start) - task = self._task_factory(self, coro) if context is None else self._task_factory(self, coro, context=context) - if name is not None: task.set_name(name) - return task - - def set_task_factory(self, factory): - if factory is not None and not callable(factory): raise TypeError('task factory must be a callable or None') - self._task_factory = factory - def get_task_factory(self): return self._task_factory - def run_forever(self): - self._check_closed() - if self._running: raise RuntimeError('This event loop is already running') - if events._get_running_loop() is not None: raise RuntimeError('Cannot run the event loop while another loop is running') + self._run_forever_setup() main = threading.current_thread() is threading.main_thread() - if main: old_wakeup = self._setup_signal_wakeup() - old_agen_hooks = sys.get_asyncgen_hooks() - sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook, finalizer=self._asyncgen_finalizer_hook) - self._running = True - self._thread_id = threading.get_ident() - events._set_running_loop(self) - try: self._r.run() - finally: - events._set_running_loop(None) - self._running = False - self._thread_id = None - sys.set_asyncgen_hooks(*old_agen_hooks) - if main: self._teardown_signal_wakeup(old_wakeup) + try: + if main: old_wakeup = self._setup_signal_wakeup() + try: self._r.run() + finally: + if main: self._teardown_signal_wakeup(old_wakeup) + finally: self._run_forever_cleanup() def _setup_signal_wakeup(self): self._ssock,self._csock = socket.socketpair() @@ -146,7 +106,7 @@ def _drain_signal_sock(self): if h is not None and not h.cancelled(): self._r.schedule(h) def add_signal_handler(self, sig, callback, *args): - _check_callback(callback, 'add_signal_handler') + self._check_callback(callback, 'add_signal_handler') h = events.Handle(callback, args, self, None) self._signal_handlers[sig] = h try: signal.signal(sig, _sighandler_noop) @@ -173,76 +133,23 @@ def _teardown_signal_wakeup(self, old): self._ssock.close() self._csock.close() - def _run_until_complete_cb(self, fut): - # KeyboardInterrupt/SystemExit propagate via run_forever; retrieving the exception - # here (and not stopping) is CPython's gh-issue-336 behaviour - if not fut.cancelled(): - exc = fut.exception() - if isinstance(exc, (SystemExit, KeyboardInterrupt)): return - self.stop() - - def run_until_complete(self, future): - fut = tasks.ensure_future(future, loop=self) - fut.add_done_callback(self._run_until_complete_cb) - try: self.run_forever() - finally: fut.remove_done_callback(self._run_until_complete_cb) - if not fut.done(): raise RuntimeError('Event loop stopped before Future completed.') - return fut.result() - def stop(self): self._r.stop() - def is_running(self): return self._running - def is_closed(self): return self._closed def close(self): - if self._running: raise RuntimeError('Cannot close a running event loop') - if self._closed: return - self._closed = True + if self.is_running(): raise RuntimeError('Cannot close a running event loop') + if self.is_closed(): return for sig in list(self._signal_handlers): self.remove_signal_handler(sig) self._r.close() - if self._default_executor is not None: self._default_executor.shutdown(wait=False) - - def _check_closed(self): - if self._closed: raise RuntimeError('Event loop is closed') - - def _asyncgen_firstiter_hook(self, agen): - if self._asyncgens_shutdown_called: - self.call_exception_handler(dict(message=f'asynchronous generator {agen!r} was scheduled after loop.shutdown_asyncgens() call', asyncgen=agen)) - self._asyncgens.add(agen) - - def _asyncgen_finalizer_hook(self, agen): - self._asyncgens.discard(agen) - if not self.is_closed(): self.call_soon_threadsafe(self.create_task, agen.aclose()) - - async def shutdown_asyncgens(self): - self._asyncgens_shutdown_called = True - if not len(self._asyncgens): return - closing = list(self._asyncgens) - self._asyncgens.clear() - results = await tasks.gather(*[ag.aclose() for ag in closing], return_exceptions=True) - for result, agen in zip(results, closing): - if isinstance(result, Exception): - self.call_exception_handler(dict(message=f'an error occurred during closing of asynchronous generator {agen!r}', - exception=result, asyncgen=agen)) - - async def shutdown_default_executor(self, timeout=None): - ex,self._default_executor = self._default_executor,None - if ex is None: return - fut = self.create_future() - def _shut(): - ex.shutdown(wait=True) - try: self.call_soon_threadsafe(futures._set_result_unless_cancelled, fut, None) - except RuntimeError: pass # loop closed after a timed-out join, as in the standard loop - threading.Thread(target=_shut).start() - try: await (fut if timeout is None else tasks.wait_for(fut, timeout)) - except TimeoutError: warnings.warn(f'executor did not finish joining its threads within {timeout} seconds', RuntimeWarning) - - def _add_io(self, fd, callback, args, handles, add): + super().close() + + def _add_io(self, fd, callback, args, handles, add, context=None): fd = _fileno(fd) - h = events.Handle(callback, args, self, None) + h = events.Handle(callback, args, self, context) old = handles.get(fd) if old is not None: old.cancel() handles[fd] = h add(fd, h) + return h def _remove_io(self, fd, handles, remove): # A queued readiness callback may still fire this turn; cancelling the stored @@ -253,350 +160,85 @@ def _remove_io(self, fd, handles, remove): if fd < 0: return False return remove(fd) - def add_reader(self, fd, callback, *args): self._add_io(fd, callback, args, self._readers, self._r.add_reader) + def add_reader(self, fd, callback, *args): return self._add_io(fd, callback, args, self._readers, self._r.add_reader) def remove_reader(self, fd): return self._remove_io(fd, self._readers, self._r.remove_reader) - def add_writer(self, fd, callback, *args): self._add_io(fd, callback, args, self._writers, self._r.add_writer) + def add_writer(self, fd, callback, *args): return self._add_io(fd, callback, args, self._writers, self._r.add_writer) def remove_writer(self, fd): return self._remove_io(fd, self._writers, self._r.remove_writer) - - def _wait_io(self, fd, handles, add, rm): - fut = self.create_future() - def cb(): - if not fut.done(): fut.set_result(None) - add(fd, cb) - h = handles[fd] - # Remove only our own registration: a cancelled wait must not tear down a newer one - def cleanup(f): - if handles.get(fd) is h: rm(fd) - fut.add_done_callback(cleanup) - return fut - - def _readable(self, fd): return self._wait_io(fd, self._readers, self.add_reader, self.remove_reader) - def _writable(self, fd): return self._wait_io(fd, self._writers, self.add_writer, self.remove_writer) - - @staticmethod - def _check_nonblocking(sock): - if sock.getblocking(): raise ValueError('the socket must be non-blocking') - - async def sock_recv(self, sock, nbytes): - self._check_nonblocking(sock) - while True: - try: return sock.recv(nbytes) - except (BlockingIOError, InterruptedError): await self._readable(sock.fileno()) - - async def sock_sendall(self, sock, data): - self._check_nonblocking(sock) - view = memoryview(data) - while view: - try: view = view[sock.send(view):] - except (BlockingIOError, InterruptedError): await self._writable(sock.fileno()) - - async def sock_accept(self, sock): - self._check_nonblocking(sock) - while True: - try: - conn,addr = sock.accept() - conn.setblocking(False) - return conn,addr - except (BlockingIOError, InterruptedError): await self._readable(sock.fileno()) - - async def sock_connect(self, sock, address): - self._check_nonblocking(sock) - try: return sock.connect(address) - except (BlockingIOError, InterruptedError): pass - await self._writable(sock.fileno()) - err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) - if err: raise OSError(err, f'Connect call failed {address}') - - async def getaddrinfo(self, host, port, *, family=0, type=0, proto=0, flags=0): - return await self.run_in_executor(None, functools.partial(socket.getaddrinfo, host, port, family=family, type=type, proto=proto, flags=flags)) - - async def getnameinfo(self, sockaddr, flags=0): - return await self.run_in_executor(None, functools.partial(socket.getnameinfo, sockaddr, flags)) - - async def _wrap_socket(self, sock, protocol_factory, sslcontext, server_hostname, server_side=False, server=None): - protocol = protocol_factory() + def _add_reader(self, fd, callback, *args): return self.add_reader(fd, callback, *args) + def _remove_reader(self, fd): return self.remove_reader(fd) + def _add_writer(self, fd, callback, *args): return self.add_writer(fd, callback, *args) + def _remove_writer(self, fd): return self.remove_writer(fd) + + _ensure_fd_no_transport,_sock_read_done,_sock_write_done = _SelLoop._ensure_fd_no_transport,_SelLoop._sock_read_done,_SelLoop._sock_write_done + sock_recv,_sock_recv = _SelLoop.sock_recv,_SelLoop._sock_recv + sock_recv_into,_sock_recv_into = _SelLoop.sock_recv_into,_SelLoop._sock_recv_into + sock_recvfrom,_sock_recvfrom = _SelLoop.sock_recvfrom,_SelLoop._sock_recvfrom + sock_recvfrom_into,_sock_recvfrom_into = _SelLoop.sock_recvfrom_into,_SelLoop._sock_recvfrom_into + sock_sendall,_sock_sendall = _SelLoop.sock_sendall,_SelLoop._sock_sendall + sock_sendto,_sock_sendto = _SelLoop.sock_sendto,_SelLoop._sock_sendto + sock_accept,_sock_accept = _SelLoop.sock_accept,_SelLoop._sock_accept + sock_connect,_sock_connect,_sock_connect_cb = _SelLoop.sock_connect,_SelLoop._sock_connect,_SelLoop._sock_connect_cb + + def _make_socket_transport(self, sock, protocol, waiter=None, *, extra=None, server=None, context=None): + return SockTransport(self, sock, protocol, waiter, extra, server, context) + + def _make_ssl_transport(self, sock, protocol, sslcontext, waiter=None, *, server_side=False, server_hostname=None, extra=None, server=None, + ssl_handshake_timeout=None, ssl_shutdown_timeout=None, call_connection_made=True, context=None): + ssl_protocol = sslproto.SSLProtocol(self, protocol, sslcontext, waiter, server_side, server_hostname, + call_connection_made=call_connection_made, ssl_handshake_timeout=ssl_handshake_timeout, ssl_shutdown_timeout=ssl_shutdown_timeout) + SockTransport(self, sock, ssl_protocol, extra=extra, server=server, context=context) + return ssl_protocol._app_transport + + def _make_datagram_transport(self, sock, protocol, address=None, waiter=None, extra=None): + return DatagramTransport(self, sock, protocol, address, waiter, extra) + def _make_read_pipe_transport(self, pipe, protocol, waiter=None, extra=None): return unix_events._UnixReadPipeTransport(self, pipe, protocol, waiter, extra) + def _make_write_pipe_transport(self, pipe, protocol, waiter=None, extra=None): + return unix_events._UnixWritePipeTransport(self, pipe, protocol, waiter, extra) + + async def _make_subprocess_transport(self, protocol, args, shell, stdin, stdout, stderr, bufsize, extra=None, **kwargs): + popen = functools.partial(subprocess.Popen, args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr, bufsize=bufsize, **kwargs) + proc = await self.run_in_executor(None, popen) waiter = self.create_future() - if sslcontext is not None: - sslp = sslproto.SSLProtocol(self, protocol, sslcontext, waiter, server_side=server_side, server_hostname=server_hostname) - SockTransport(self, sock, sslp, server=server) - transport = sslp._app_transport - else: transport = SockTransport(self, sock, protocol, waiter=waiter, server=server) + transport = SubprocessTransport(self, protocol, args, shell, stdin, stdout, stderr, bufsize, proc, waiter, extra, **kwargs) try: await waiter except BaseException: transport.close() + await transport._wait() raise - return transport, protocol - - @staticmethod - def _check_ssl_args(ssl, ssl_handshake_timeout): - if ssl_handshake_timeout is not None and not ssl: raise ValueError('ssl_handshake_timeout is only meaningful with ssl') - - async def create_connection(self, protocol_factory, host=None, port=None, *, ssl=None, sock=None, family=0, - proto=0, flags=0, local_addr=None, server_hostname=None, ssl_handshake_timeout=None, **kw): - if server_hostname is not None and not ssl: raise ValueError('server_hostname is only meaningful with ssl') - self._check_ssl_args(ssl, ssl_handshake_timeout) - if sock is None: - infos = await self.getaddrinfo(host, port, family=family, type=socket.SOCK_STREAM, proto=proto, flags=flags) - if not infos: raise OSError(f'getaddrinfo({host!r}) returned empty list') - errors = [] - for af,st,pr,_,addr in infos: - sock = socket.socket(af, st, pr) - sock.setblocking(False) - try: - if local_addr is not None: sock.bind(local_addr) - await self.sock_connect(sock, addr) - break - except OSError as e: - sock.close() - sock = None - errors.append(e) - except BaseException: - sock.close() - raise - if sock is None: - try: - if len(errors) == 1: raise errors[0] - raise OSError(f'could not connect to {host}:{port}', errors) - finally: errors = None # a kept list would give the exception a referrer - else: sock.setblocking(False) - sslcontext = None - if ssl: - sslcontext = ssl_mod.create_default_context() if ssl is True else ssl - if server_hostname is None: server_hostname = host - return await self._wrap_socket(sock, protocol_factory, sslcontext, server_hostname) - - async def create_server(self, protocol_factory, host=None, port=None, *, sock=None, backlog=100, ssl=None, - family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, reuse_address=None, reuse_port=None, - start_serving=True, ssl_handshake_timeout=None, **kw): - if ssl is True: raise ValueError('ssl=True needs an SSLContext holding the server certificate') - self._check_ssl_args(ssl, ssl_handshake_timeout) - if host is not None or port is not None: - if sock is not None: raise ValueError('host/port and sock can not be specified at the same time') - if host == '': hosts = [None] - elif isinstance(host, str) or not hasattr(host, '__iter__'): hosts = [host] - else: hosts = host - all_infos = await tasks.gather(*[self.getaddrinfo(h, port, family=family, type=socket.SOCK_STREAM, flags=flags) for h in hosts]) - infos = set(itertools.chain.from_iterable(all_infos)) - socks,completed = [],False - try: - for af,st,pr,_,addr in infos: - try: s = socket.socket(af, st, pr) - except OSError: continue # bad family/type/protocol combination - socks.append(s) - if reuse_address or reuse_address is None: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - if reuse_port and af in (socket.AF_INET, socket.AF_INET6): s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) - # Disable dual-stack on the ipv6 socket, or its bind of :: takes the port over for v4 too - if socket.has_ipv6 and af == socket.AF_INET6 and hasattr(socket, 'IPPROTO_IPV6'): - s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, True) - try: s.bind(addr) - except OSError as err: - msg = f'error while attempting to bind on address {addr!r}: {str(err).lower()}' - if err.errno == errno.EADDRNOTAVAIL: # assume the family is not enabled (bpo-30945) - socks.pop() - s.close() - continue - raise OSError(err.errno, msg) from None - if not socks: raise OSError(f'could not bind on any address out of {[i[4] for i in infos]!r}') - completed = True - finally: - if not completed: - for s in socks: s.close() - else: - if sock is None: raise ValueError('Neither host/port nor sock were specified') - if sock.type != socket.SOCK_STREAM: raise ValueError(f'A Stream Socket was expected, got {sock!r}') - socks = [sock] - for s in socks: - s.listen(backlog) - s.setblocking(False) - server = MiniServer(self, socks, protocol_factory, ssl) - if start_serving: server._start_serving() - return server - - async def create_unix_connection(self, protocol_factory, path=None, *, ssl=None, sock=None, - server_hostname=None, ssl_handshake_timeout=None, **kw): - if ssl and server_hostname is None: raise ValueError('you have to pass server_hostname when using ssl') - if server_hostname is not None and not ssl: raise ValueError('server_hostname is only meaningful with ssl') - self._check_ssl_args(ssl, ssl_handshake_timeout) - if sock is None: - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.setblocking(False) - try: await self.sock_connect(sock, path) - except BaseException: - sock.close() - raise - else: sock.setblocking(False) - sslcontext = (ssl_mod.create_default_context() if ssl is True else ssl) if ssl else None - return await self._wrap_socket(sock, protocol_factory, sslcontext, server_hostname) - - async def create_unix_server(self, protocol_factory, path=None, *, sock=None, backlog=100, ssl=None, - start_serving=True, ssl_handshake_timeout=None, cleanup_socket=True, **kw): - if ssl is True: raise ValueError('ssl=True needs an SSLContext holding the server certificate') - self._check_ssl_args(ssl, ssl_handshake_timeout) - if sock is None: - def _unlink_stale(): - try: - if stat_mod.S_ISSOCK(os.stat(path).st_mode): os.remove(path) - except (FileNotFoundError, OSError): pass - await self.run_in_executor(None, _unlink_stale) - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - try: sock.bind(path) - except OSError as e: - sock.close() - if e.errno == errno.EADDRINUSE: raise OSError(errno.EADDRINUSE, f'Address {path!r} is already in use') from None - raise - else: path = sock.getsockname() - sock.listen(backlog) - sock.setblocking(False) - unlink = () - if cleanup_socket and path and path[0] not in (0, '\x00'): - try: unlink = ((path, (await self.run_in_executor(None, os.stat, path)).st_ino),) - except FileNotFoundError: pass - server = MiniServer(self, [sock], protocol_factory, ssl, unlink_paths=unlink) - if start_serving: server._start_serving() - return server - - async def connect_accepted_socket(self, protocol_factory, sock, *, ssl=None, ssl_handshake_timeout=None, **kw): - sock.setblocking(False) - return await self._wrap_socket(sock, protocol_factory, ssl, None, server_side=True) - - async def create_datagram_endpoint(self, protocol_factory, local_addr=None, remote_addr=None, *, family=0, - proto=0, flags=0, reuse_port=None, allow_broadcast=None, sock=None): - if sock is None: - if remote_addr or local_addr: - infos = await self.getaddrinfo(*(remote_addr or local_addr), family=family, - type=socket.SOCK_DGRAM, proto=proto, flags=flags) - if not infos: raise OSError('getaddrinfo returned empty list') - family = infos[0][0] - sock = socket.socket(family or socket.AF_INET, socket.SOCK_DGRAM, proto) - try: - if reuse_port: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) - if allow_broadcast: sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) - sock.setblocking(False) - if local_addr: sock.bind(local_addr) - if remote_addr: await self.sock_connect(sock, tuple(remote_addr)) - except BaseException: - sock.close() - raise - else: sock.setblocking(False) - # The canonical peer address (a 4-tuple for IPv6), whatever form the caller used - remote_addr = None - try: remote_addr = sock.getpeername() + return transport + + _start_serving = _SelLoop._start_serving + _accept_connection = _SelLoop._accept_connection + _accept_connection2 = _SelLoop._accept_connection2 + + def _stop_serving(self, sock): + path = sock.getsockname() if sock in self._unix_server_sockets else None + self.remove_reader(sock.fileno()) + sock.close() + if path is None: return + inode = self._unix_server_sockets.pop(sock) + try: + if os.stat(path).st_ino == inode: os.unlink(path) except OSError: pass - protocol = protocol_factory() - waiter = self.create_future() - transport = DatagramTransport(self, sock, protocol, address=remote_addr or None, waiter=waiter) - try: await waiter - except BaseException: - transport.close() - raise - return transport, protocol - - async def connect_read_pipe(self, protocol_factory, pipe): - protocol = protocol_factory() - waiter = self.create_future() - transport = ReadPipeTransport(self, pipe, protocol, waiter) - try: await waiter - except BaseException: - transport.close() - raise - return transport, protocol - async def connect_write_pipe(self, protocol_factory, pipe): - protocol = protocol_factory() - waiter = self.create_future() - transport = WritePipeTransport(self, pipe, protocol, waiter) - try: await waiter - except BaseException: - transport.close() - raise - return transport, protocol - - async def _subprocess(self, protocol_factory, args, shell, stdin, stdout, stderr, bufsize, - universal_newlines, encoding, errors, text, **kwargs): - if universal_newlines or encoding or errors or text: raise ValueError('text mode not supported by the event loop') - protocol = protocol_factory() - transport = SubprocessTransport(self, protocol, args, shell, stdin, stdout, stderr, bufsize, **kwargs) - await transport._setup() - return transport, protocol - - async def subprocess_exec(self, protocol_factory, program, *args, stdin=subprocess_mod.PIPE, - stdout=subprocess_mod.PIPE, stderr=subprocess_mod.PIPE, universal_newlines=False, shell=False, - bufsize=0, encoding=None, errors=None, text=None, **kwargs): - if shell: raise ValueError('subprocess_exec() does not take a shell argument of True') - return await self._subprocess(protocol_factory, (program,)+args, False, stdin, stdout, stderr, - bufsize, universal_newlines, encoding, errors, text, **kwargs) - - async def subprocess_shell(self, protocol_factory, cmd, *, stdin=subprocess_mod.PIPE, - stdout=subprocess_mod.PIPE, stderr=subprocess_mod.PIPE, universal_newlines=False, shell=True, - bufsize=0, encoding=None, errors=None, text=None, **kwargs): - if not isinstance(cmd, (bytes, str)): raise ValueError('cmd must be a string') - if not shell: raise ValueError('subprocess_shell() requires shell=True') - return await self._subprocess(protocol_factory, cmd, True, stdin, stdout, stderr, - bufsize, universal_newlines, encoding, errors, text, **kwargs) - - async def start_tls(self, transport, protocol, sslcontext, *, server_side=False, server_hostname=None, - ssl_handshake_timeout=None, ssl_shutdown_timeout=None): - waiter = self.create_future() - sslp = sslproto.SSLProtocol(self, protocol, sslcontext, waiter, server_side=server_side, - server_hostname=server_hostname, call_connection_made=False) - transport.pause_reading() - # gh-142352: TLS bytes may already sit in a server-side StreamReader; move them - # into the SSL protocol's incoming BIO or the handshake never sees them - if server_side and isinstance(protocol, asyncio.streams.StreamReaderProtocol): - reader = getattr(protocol, '_stream_reader', None) - if reader is not None and reader._buffer: - sslp._incoming.write(reader._buffer) - reader._buffer.clear() - transport.set_protocol(sslp) - self.call_soon(sslp.connection_made, transport) - self.call_soon(transport.resume_reading) - try: await waiter - except BaseException: - transport.close() - raise - return sslp._app_transport + create_unix_connection = _UnixLoop.create_unix_connection + create_unix_server = _UnixLoop.create_unix_server - def run_in_executor(self, executor, func, *args): +if not hasattr(base_events.BaseEventLoop, '_run_forever_setup'): # the helpers first appear in 3.12; transcribed from 3.11's inline run_forever, which is frozen + def _run_forever_setup(self): self._check_closed() - if executor is None: - if self._default_executor is None: self._default_executor = concurrent.futures.ThreadPoolExecutor(thread_name_prefix='asyncio') - executor = self._default_executor - return futures.wrap_future(executor.submit(func, *args), loop=self) - - def set_default_executor(self, executor): self._default_executor = executor - - def get_debug(self): return self._debug - def set_debug(self, enabled): self._debug = bool(enabled) - - def set_exception_handler(self, handler): self._exception_handler = handler - def get_exception_handler(self): return self._exception_handler - - def call_exception_handler(self, context): - if self._exception_handler is None: return self.default_exception_handler(context) - try: self._exception_handler(self, context) - except (SystemExit, KeyboardInterrupt): raise - except BaseException as exc: - # A broken handler must not take down the loop: report it via the default one - try: self.default_exception_handler(dict(message='Unhandled error in exception handler', - exception=exc, context=context)) - except (SystemExit, KeyboardInterrupt): raise - except BaseException: base_events.logger.error('Exception in default exception handler', exc_info=True) - - def default_exception_handler(self, context): - message = context.get('message') or 'Unhandled exception in event loop' - exc = context.get('exception') - exc_info = (type(exc), exc, exc.__traceback__) if exc is not None else False - log_lines = [message] - for key in sorted(context): - if key in ('message', 'exception'): continue - value = context[key] - if key == 'source_traceback': - value = 'Object created at (most recent call last):\n' + ''.join(traceback.format_list(value)).rstrip() - elif key == 'handle_traceback': - value = 'Handle created at (most recent call last):\n' + ''.join(traceback.format_list(value)).rstrip() - else: value = repr(value) - log_lines.append(f'{key}: {value}') - base_events.logger.error('\n'.join(log_lines), exc_info=exc_info) + self._check_running() + self._set_coroutine_origin_tracking(self._debug) + self._thread_id = threading.get_ident() + self._old_agen_hooks = sys.get_asyncgen_hooks() + sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook, finalizer=self._asyncgen_finalizer_hook) + events._set_running_loop(self) + def _run_forever_cleanup(self): + self._thread_id = None + events._set_running_loop(None) + self._set_coroutine_origin_tracking(False) + sys.set_asyncgen_hooks(*self._old_agen_hooks) + Loop._run_forever_setup, Loop._run_forever_cleanup = _run_forever_setup, _run_forever_cleanup def new_event_loop(reactor=None): "An event loop on a Rust reactor, for `asyncio.run(..., loop_factory=new_event_loop)`." diff --git a/python/loopmini/subproc.py b/python/loopmini/subproc.py index 77ec9ad..caf1e96 100644 --- a/python/loopmini/subproc.py +++ b/python/loopmini/subproc.py @@ -1,114 +1,20 @@ -"Subprocess transport: Popen plus pipe transports, with a reaper thread per child." -import asyncio, collections, functools, subprocess, threading +"A nonblocking process spawn feeding asyncio's standard subprocess transport." +import threading +from asyncio import base_subprocess __all__ = ["SubprocessTransport"] -class _PipeReadProto(asyncio.Protocol): - def __init__(self, subtr, fd): self.subtr,self.fd,self.pipe = subtr,fd,None - def connection_made(self, transport): self.pipe = transport - def data_received(self, data): self.subtr._call(self.subtr._protocol.pipe_data_received, self.fd, data) - def connection_lost(self, exc): self.subtr._pipe_connection_lost(self.fd, exc) - -class _PipeWriteProto(asyncio.BaseProtocol): - def __init__(self, subtr, fd): self.subtr,self.fd,self.pipe = subtr,fd,None - def connection_made(self, transport): self.pipe = transport - def connection_lost(self, exc): self.subtr._pipe_connection_lost(self.fd, exc) - def pause_writing(self): self.subtr._protocol.pause_writing() - def resume_writing(self): self.subtr._protocol.resume_writing() - -class SubprocessTransport(asyncio.SubprocessTransport): - def __init__(self, loop, protocol, args, shell, stdin, stdout, stderr, bufsize, **kwargs): - super().__init__() - self._loop,self._protocol = loop,protocol - self._closed = self._finished = False - # Buffers protocol calls until connection_made wires the protocol: a fast child - # can produce output (even exit) while later pipes are still being connected - self._pending_calls = collections.deque() - self._returncode = None - self._exit_waiters = [] - self._pipes = {} - self._disconnected = set() - self._popen = functools.partial(subprocess.Popen, args, shell=shell, stdin=stdin, stdout=stdout, - stderr=stderr, bufsize=bufsize, **kwargs) - - async def _setup(self): - # In the executor because fork/exec blocks (the standard loop blocks its loop thread here) - proc = self._proc = await self._loop.run_in_executor(None, self._popen) - self._extra['subprocess'] = proc - del self._popen - if proc.stdin is not None: - tr,_ = await self._loop.connect_write_pipe(lambda: _PipeWriteProto(self, 0), proc.stdin) - self._pipes[0] = tr - if proc.stdout is not None: - tr,_ = await self._loop.connect_read_pipe(lambda: _PipeReadProto(self, 1), proc.stdout) - self._pipes[1] = tr - if proc.stderr is not None: - tr,_ = await self._loop.connect_read_pipe(lambda: _PipeReadProto(self, 2), proc.stderr) - self._pipes[2] = tr - # Called directly so the protocol is fully wired before subprocess_exec returns - self._protocol.connection_made(self) - for cb,data in self._pending_calls: self._loop.call_soon(cb, *data) - self._pending_calls = None - # Same model as asyncio's default ThreadedChildWatcher: one waiting thread per child +class SubprocessTransport(base_subprocess.BaseSubprocessTransport): + def __init__(self, loop, protocol, args, shell, stdin, stdout, stderr, bufsize, proc, waiter=None, extra=None, **kwargs): + self._prepared_proc = proc + super().__init__(loop, protocol, args, shell, stdin, stdout, stderr, bufsize, waiter, extra, **kwargs) threading.Thread(target=self._reap, daemon=True).start() + def _start(self, **kwargs): + self._proc = self._prepared_proc + del self._prepared_proc + def _reap(self): returncode = self._proc.wait() try: self._loop.call_soon_threadsafe(self._process_exited, returncode) - except RuntimeError: pass # loop closed before the child exited - - def _call(self, cb, *data): - if self._pending_calls is not None: self._pending_calls.append((cb, data)) - else: cb(*data) - - def _process_exited(self, returncode): - self._returncode = returncode - self._call(self._protocol.process_exited) - for w in self._exit_waiters: - if not w.cancelled(): w.set_result(returncode) - self._exit_waiters = [] - self._try_finish() - - def _pipe_connection_lost(self, fd, exc): - self._call(self._protocol.pipe_connection_lost, fd, exc) - self._disconnected.add(fd) - self._try_finish() - - def _try_finish(self): - # The subprocess protocol's connection_lost fires once: process exited, all pipes gone - if self._finished or self._returncode is None or self._disconnected < set(self._pipes): return - self._finished = True - self._loop.call_soon(self._protocol.connection_lost, None) - - async def _wait(self): - if self._returncode is not None: return self._returncode - fut = self._loop.create_future() - self._exit_waiters.append(fut) - return await fut - - def get_pid(self): return self._proc.pid - def get_returncode(self): return self._returncode - def get_pipe_transport(self, fd): return self._pipes.get(fd) - def is_closing(self): return self._closed - - def _check_running(self): - if self._returncode is not None: raise ProcessLookupError() - - def send_signal(self, signal): - self._check_running() - self._proc.send_signal(signal) - - def terminate(self): - self._check_running() - self._proc.terminate() - - def kill(self): - self._check_running() - self._proc.kill() - - def close(self): - if self._closed: return - self._closed = True - for tr in self._pipes.values(): - if tr is not None: tr.close() - if self._returncode is None and self._proc.poll() is None: self._proc.kill() + except RuntimeError: pass diff --git a/python/loopmini/transports.py b/python/loopmini/transports.py index 2e84491..0234a40 100644 --- a/python/loopmini/transports.py +++ b/python/loopmini/transports.py @@ -1,49 +1,39 @@ -"TCP transports and server for the Rust-reactor loop, feeding stock asyncio protocols." -import asyncio, os, socket, weakref, stat, sys -from asyncio import futures +"Socket transports for the Rust-reactor loop, feeding stock asyncio protocols." +import asyncio, socket +from asyncio import constants, futures, transports -class _FlowControl: - "Write-buffer flow control shared by buffering transports, with CPython's _FlowControlMixin semantics." - _protocol_paused = False - _high,_low = 65536,16384 - - def get_write_buffer_size(self): return len(self._buffer) - def get_write_buffer_limits(self): return (self._low, self._high) - - def set_write_buffer_limits(self, high=None, low=None): - if high is None: high = 65536 if low is None else 4*low - if low is None: low = high//4 - if not high >= low >= 0: raise ValueError(f'high ({high}) must be >= low ({low}) must be >= 0') - self._high,self._low = high,low - self._maybe_pause_protocol() +class _TransportLifecycle: + "Protocol binding and exactly-once connection_lost delivery shared by fd transports." + def get_protocol(self): return self._protocol + def set_protocol(self, protocol): self._protocol = protocol + def is_closing(self): return self._closing - def _maybe_pause_protocol(self): - if not self._protocol_paused and len(self._buffer) > self._high: - self._protocol_paused = True - try: self._protocol.pause_writing() - except Exception as e: self._loop.call_exception_handler(dict(message='protocol.pause_writing() failed', exception=e, transport=self, protocol=self._protocol)) + def _schedule_lost(self, exc): + if self._lost: return + self._lost = True + self._loop.call_soon(self._call_connection_lost, exc) - def _maybe_resume_protocol(self): - if self._protocol_paused and len(self._buffer) <= self._low: - self._protocol_paused = False - try: self._protocol.resume_writing() - except Exception as e: self._loop.call_exception_handler(dict(message='protocol.resume_writing() failed', exception=e, transport=self, protocol=self._protocol)) + def _call_connection_lost(self, exc): + try: self._protocol.connection_lost(exc) + finally: self._close_resource() -__all__ = ["SockTransport", "MiniServer"] +__all__ = ["SockTransport", "DatagramTransport"] -class SockTransport(_FlowControl, asyncio.Transport): +class SockTransport(_TransportLifecycle, transports._FlowControlMixin): "Bidirectional TCP transport over the loop's fd-readiness primitives." max_size = 262144 _start_tls_compatible = True + _sendfile_compatible = constants._SendfileMode.FALLBACK - def __init__(self, loop, sock, protocol, waiter=None, extra=None, server=None): - super().__init__(extra) + def __init__(self, loop, sock, protocol, waiter=None, extra=None, server=None, context=None): + super().__init__(extra, loop) self._extra['socket'] = sock try: self._extra['sockname'] = sock.getsockname() except OSError: pass try: self._extra['peername'] = sock.getpeername() except OSError: pass - self._loop,self._sock,self._protocol,self._server = loop,sock,protocol,server + self._loop,self._sock,self._protocol,self._server,self._fd = loop,sock,protocol,server,sock.fileno() + loop._transports[self._fd] = self self._buffered = isinstance(protocol, asyncio.BufferedProtocol) if sock.family in (socket.AF_INET, socket.AF_INET6): try: sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) @@ -51,24 +41,28 @@ def __init__(self, loop, sock, protocol, waiter=None, extra=None, server=None): self._buffer = bytearray() self._closing = self._reading = self._eof = self._lost = False if server is not None: server._attach(self) - loop.call_soon(protocol.connection_made, self) - loop.call_soon(self.resume_reading) + loop.call_soon(protocol.connection_made, self, context=context) + loop.call_soon(self.resume_reading, context=context) if waiter is not None: loop.call_soon(futures._set_result_unless_cancelled, waiter, None) - def get_protocol(self): return self._protocol def set_protocol(self, protocol): self._protocol,self._buffered = protocol,isinstance(protocol, asyncio.BufferedProtocol) - def is_closing(self): return self._closing + + def get_write_buffer_size(self): return len(self._buffer) + def is_reading(self): return self._reading - def pause_reading(self): - if self._closing or not self._reading: return + def _stop_reading(self): + if not self._reading: return self._reading = False - self._loop.remove_reader(self._sock.fileno()) + self._loop.remove_reader(self._fd) + + def pause_reading(self): + if not self._closing: self._stop_reading() def resume_reading(self): if self._closing or self._reading: return self._reading = True - self._loop.add_reader(self._sock.fileno(), self._read_ready) + self._loop.add_reader(self._fd, self._read_ready) def _read_ready(self): if not self._reading: return @@ -83,17 +77,17 @@ def _read_ready(self): except OSError as e: return self._fatal_error(e, 'Fatal read error on socket transport') if self._protocol.eof_received(): self._reading = False - self._loop.remove_reader(self._sock.fileno()) + self._loop.remove_reader(self._fd) else: self.close() def write(self, data): - if not data or self._closing: return + if not data or self._closing or self._lost: return if not self._buffer: try: data = data[self._sock.send(data):] except (BlockingIOError, InterruptedError): pass except OSError as e: return self._fatal_error(e, 'Fatal write error on socket transport') if not data: return - self._loop.add_writer(self._sock.fileno(), self._write_ready) + self._loop.add_writer(self._fd, self._write_ready) self._buffer.extend(data) self._maybe_pause_protocol() @@ -101,13 +95,13 @@ def writelines(self, list_of_data): self.write(b''.join(list_of_data)) def _write_ready(self): if not self._buffer: return - try: n = self._sock.send(self._buffer) # no explicit memoryview: a raising send would leave an export alive in the traceback, and clear() then fails + try: n = self._sock.send(self._buffer) except (BlockingIOError, InterruptedError): return except OSError as e: return self._fatal_error(e, 'Fatal write error on socket transport') del self._buffer[:n] self._maybe_resume_protocol() if self._buffer: return - self._loop.remove_writer(self._sock.fileno()) + self._loop.remove_writer(self._fd) if self._eof: self._sock.shutdown(socket.SHUT_WR) if self._closing: self._schedule_lost(None) @@ -118,18 +112,10 @@ def write_eof(self): self._eof = True if not self._buffer: self._sock.shutdown(socket.SHUT_WR) - def _schedule_lost(self, exc): - # close(), abort(), and fatal errors may overlap; connection_lost runs exactly once - if self._lost: return - self._lost = True - self._loop.call_soon(self._call_connection_lost, exc) - def close(self): if self._closing: return self._closing = True - if self._reading: - self._reading = False - self._loop.remove_reader(self._sock.fileno()) + self._stop_reading() if not self._buffer: self._schedule_lost(None) def abort(self): self._force_close(None) @@ -143,136 +129,35 @@ def _force_close(self, exc): if self._lost: return if self._buffer: self._buffer.clear() - self._loop.remove_writer(self._sock.fileno()) - if self._reading: - self._reading = False - self._loop.remove_reader(self._sock.fileno()) + self._loop.remove_writer(self._fd) + self._stop_reading() self._closing = True self._schedule_lost(exc) - def _call_connection_lost(self, exc): - try: self._protocol.connection_lost(exc) - finally: - self._sock.close() - if self._server is not None: - self._server._detach(self) - self._server = None - -class MiniServer(asyncio.AbstractServer): - "Listening sockets plus the accept loop; hands accepted connections to the loop." - def __init__(self, loop, sockets, protocol_factory, sslcontext=None, unlink_paths=()): - self._loop,self._sockets,self._factory,self._ssl = loop,list(sockets),protocol_factory,sslcontext - self._unlink_paths = unlink_paths # (path, inode) pairs: unlinked at close only if the inode still matches - self._serving = False - self._serving_forever_fut = None - self._transports = weakref.WeakSet() - self._waiters = [] # becomes None once closed with no remaining client transports - - @property - def sockets(self): return tuple(self._sockets or ()) - def is_serving(self): return self._serving - def get_loop(self): return self._loop - def _attach(self, transport): self._transports.add(transport) + def _close_resource(self): + self._sock.close() + if self._server is not None: + self._server._detach(self) + self._server = None - def _detach(self, transport): - self._transports.discard(transport) - if not self._transports and self._sockets is None: self._wakeup() - - def _wakeup(self): - waiters,self._waiters = self._waiters,None - for w in waiters: - if not w.done(): w.set_result(None) - - def _start_serving(self): - if self._serving: return - self._serving = True - for s in self._sockets: self._loop.add_reader(s.fileno(), self._accept_ready, s) - - async def start_serving(self): self._start_serving() - - def _accept_ready(self, s): - for _ in range(16): - try: conn,addr = s.accept() - except (BlockingIOError, InterruptedError): return - except OSError: return - conn.setblocking(False) - self._loop.create_task(self._accept_conn(conn)) - - async def _accept_conn(self, conn): - try: await self._loop._wrap_socket(conn, self._factory, self._ssl, None, server_side=True, server=self) - except Exception as e: - conn.close() - self._loop.call_exception_handler(dict(message='Error on accepted connection', exception=e)) - - def close(self): - socks,self._sockets = self._sockets,None - if socks is None: return - for s in socks: - self._loop.remove_reader(s.fileno()) - s.close() - for p,ino in self._unlink_paths: - try: - if os.stat(p).st_ino == ino: os.unlink(p) - except OSError: pass - self._unlink_paths = () - self._serving = False - if self._serving_forever_fut is not None and not self._serving_forever_fut.done(): - self._serving_forever_fut.cancel() - self._serving_forever_fut = None - if not self._transports: self._wakeup() - - def close_clients(self): - for t in list(self._transports): t.close() - - def abort_clients(self): - for t in list(self._transports): t.abort() - - async def wait_closed(self): - if self._waiters is None: return - w = self._loop.create_future() - self._waiters.append(w) - await w - - async def serve_forever(self): - if self._serving_forever_fut is not None: raise RuntimeError(f'server {self!r} is already being awaited on serve_forever()') - if self._sockets is None: raise RuntimeError(f'server {self!r} is closed') - self._start_serving() - self._serving_forever_fut = self._loop.create_future() - try: await self._serving_forever_fut - except asyncio.CancelledError: - try: - self.close() - self.close_clients() - await self.wait_closed() - finally: raise - finally: self._serving_forever_fut = None - - async def __aenter__(self): return self - async def __aexit__(self, *exc): - self.close() - await self.wait_closed() - -class DatagramTransport(asyncio.DatagramTransport): +class DatagramTransport(_TransportLifecycle, asyncio.DatagramTransport): "UDP transport: sendto with backpressure buffering, datagram_received/error_received dispatch." max_size = 65536 - def __init__(self, loop, sock, protocol, address=None, waiter=None): - super().__init__() + def __init__(self, loop, sock, protocol, address=None, waiter=None, extra=None): + super().__init__(extra) self._extra['socket'] = sock try: self._extra['sockname'] = sock.getsockname() except OSError: pass if address is not None: self._extra['peername'] = address self._loop,self._sock,self._protocol,self._address = loop,sock,protocol,address + loop._transports[sock.fileno()] = self self._buffer = [] self._closing = self._lost = False loop.call_soon(protocol.connection_made, self) loop.call_soon(loop.add_reader, sock.fileno(), self._read_ready) if waiter is not None: loop.call_soon(futures._set_result_unless_cancelled, waiter, None) - def get_protocol(self): return self._protocol - def set_protocol(self, protocol): self._protocol = protocol - def is_closing(self): return self._closing - def _read_ready(self): try: data,addr = self._sock.recvfrom(self.max_size) except (BlockingIOError, InterruptedError): return @@ -320,162 +205,4 @@ def abort(self): self._loop.remove_reader(self._sock.fileno()) self._schedule_lost(None) - def _schedule_lost(self, exc): - if self._lost: return - self._lost = True - self._loop.call_soon(self._call_connection_lost, exc) - - def _call_connection_lost(self, exc): - try: self._protocol.connection_lost(exc) - finally: self._sock.close() - -class ReadPipeTransport(asyncio.ReadTransport): - "Read side of a pipe: os.read on readiness, eof_received then connection_lost at EOF." - max_size = 262144 - - def __init__(self, loop, pipe, protocol, waiter=None): - super().__init__() - self._extra['pipe'] = pipe - self._loop,self._pipe,self._protocol,self._fd = loop,pipe,protocol,pipe.fileno() - self._closing = self._reading = self._lost = False - os.set_blocking(self._fd, False) - loop.call_soon(protocol.connection_made, self) - loop.call_soon(self.resume_reading) - if waiter is not None: loop.call_soon(futures._set_result_unless_cancelled, waiter, None) - - def get_protocol(self): return self._protocol - def set_protocol(self, protocol): self._protocol = protocol - def is_closing(self): return self._closing - def is_reading(self): return self._reading - - def pause_reading(self): - if self._closing or not self._reading: return - self._reading = False - self._loop.remove_reader(self._fd) - - def resume_reading(self): - if self._closing or self._reading: return - self._reading = True - self._loop.add_reader(self._fd, self._read_ready) - - def _read_ready(self): - if not self._reading: return - try: data = os.read(self._fd, self.max_size) - except (BlockingIOError, InterruptedError): return - except OSError as e: return self._force_close(e) - if data: return self._protocol.data_received(data) - self._closing = True - self._reading = False - self._loop.remove_reader(self._fd) - self._loop.call_soon(self._protocol.eof_received) - self._schedule_lost(None) - - def close(self): - if self._closing: return - self._closing = True - if self._reading: - self._reading = False - self._loop.remove_reader(self._fd) - self._schedule_lost(None) - - def _force_close(self, exc): - if self._lost: return - if self._reading: - self._reading = False - self._loop.remove_reader(self._fd) - self._closing = True - self._schedule_lost(exc) - - def _schedule_lost(self, exc): - if self._lost: return - self._lost = True - self._loop.call_soon(self._call_connection_lost, exc) - - def _call_connection_lost(self, exc): - try: self._protocol.connection_lost(exc) - finally: self._pipe.close() - -class WritePipeTransport(_FlowControl, asyncio.Transport): - """Write side of a pipe: buffered os.write with the same flow control as SockTransport. - - The asyncio.Transport base supplies read-side methods raising NotImplementedError, - matching the standard write-pipe transport's contract. A reader watches the fd where - the platform reports peer close that way (sockets, and unnamed pipes off macOS). - """ - - def __init__(self, loop, pipe, protocol, waiter=None): - super().__init__() - self._extra['pipe'] = pipe - self._loop,self._pipe,self._protocol,self._fd = loop,pipe,protocol,pipe.fileno() - mode = os.fstat(self._fd).st_mode - if not (stat.S_ISFIFO(mode) or stat.S_ISSOCK(mode) or stat.S_ISCHR(mode)): - raise ValueError('Pipe transport is only for pipes, sockets and character devices') - self._buffer = bytearray() - self._closing = self._lost = self._has_reader = False - os.set_blocking(self._fd, False) - loop.call_soon(protocol.connection_made, self) - named_fifo = sys.platform == 'darwin' and os.fstat(self._fd).st_nlink > 0 - if stat.S_ISSOCK(mode) or (stat.S_ISFIFO(mode) and not named_fifo): - self._has_reader = True - loop.call_soon(loop.add_reader, self._fd, self._hangup_ready) - if waiter is not None: loop.call_soon(futures._set_result_unless_cancelled, waiter, None) - - def _hangup_ready(self): self._force_close(BrokenPipeError() if self._buffer else None) - - def get_protocol(self): return self._protocol - def set_protocol(self, protocol): self._protocol = protocol - def is_closing(self): return self._closing - def write(self, data): - if not data or self._closing or self._lost: return - if not self._buffer: - try: data = data[os.write(self._fd, data):] - except (BlockingIOError, InterruptedError): pass - except OSError as e: return self._force_close(e) - if not data: return - self._loop.add_writer(self._fd, self._write_ready) - self._buffer.extend(data) - self._maybe_pause_protocol() - - def writelines(self, list_of_data): self.write(b''.join(list_of_data)) - - def _write_ready(self): - if not self._buffer: return - try: n = os.write(self._fd, self._buffer) - except (BlockingIOError, InterruptedError): return - except OSError as e: return self._force_close(e) - del self._buffer[:n] - self._maybe_resume_protocol() - if self._buffer: return - self._loop.remove_writer(self._fd) - if self._closing: self._schedule_lost(None) - - def can_write_eof(self): return True - - def write_eof(self): self.close() - - def close(self): - if self._closing: return - self._closing = True - if not self._buffer: self._schedule_lost(None) - - def abort(self): self._force_close(None) - - def _force_close(self, exc): - if self._lost: return - if self._buffer: - self._buffer.clear() - self._loop.remove_writer(self._fd) - self._closing = True - self._schedule_lost(exc) - - def _schedule_lost(self, exc): - if self._lost: return - self._lost = True - if self._has_reader: - self._has_reader = False - self._loop.remove_reader(self._fd) - self._loop.call_soon(self._call_connection_lost, exc) - - def _call_connection_lost(self, exc): - try: self._protocol.connection_lost(exc) - finally: self._pipe.close() + def _close_resource(self): self._sock.close() diff --git a/rustfmt.toml b/rustfmt.toml new file mode 100644 index 0000000..59fc332 --- /dev/null +++ b/rustfmt.toml @@ -0,0 +1,3 @@ +max_width = 160 +use_small_heuristics = "Max" +use_field_init_shorthand = true diff --git a/src/lib.rs b/src/lib.rs index 338088f..d0f97dd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,12 +1,10 @@ -pub mod reactor; -pub mod tokio_core; mod pyreactor; +pub mod reactor; use pyo3::prelude::*; pub use pyreactor::PyReactor; pub use reactor::Reactor as ReactorCore; -pub use tokio_core::TokioCore; #[pymodule] fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> { diff --git a/src/pyreactor.rs b/src/pyreactor.rs index 87c7a01..9d87953 100644 --- a/src/pyreactor.rs +++ b/src/pyreactor.rs @@ -1,31 +1,19 @@ //! The Python-facing reactor: scheduling and readiness methods plus the canonical -//! dispatch loop, as one pyclass both extensions compile in. `loopmini._core` -//! registers it with an owned runtime; an embedding extension (`kernmini._native`) -//! constructs it with `with_handle` on its own runtime. The dispatch loop carries -//! the injected-exception requeue rule, which must exist exactly once. -use crate::tokio_core::TokioCore; +//! dispatch loop. The dispatch loop carries the injected-exception requeue rule, +//! which must exist exactly once. +use crate::reactor::Reactor as ReactorCore; use polling::Events; use pyo3::intern; use pyo3::prelude::*; use std::ops::ControlFlow; -use tokio::runtime::Handle; #[pyclass(name = "Reactor")] -pub struct PyReactor { core: TokioCore> } - -impl PyReactor { - /// A reactor whose blocking waits run on `handle`'s runtime. Call `run` only - /// from a thread outside that runtime (a Python main or session thread): - /// Tokio panics on `block_on` from one of the runtime's own workers. - pub fn with_handle(handle: Handle) -> PyResult { - Ok(Self { core: TokioCore::with_handle(handle)? }) - } -} +pub struct PyReactor { core: ReactorCore> } #[pymethods] impl PyReactor { #[new] - fn new() -> PyResult { Ok(Self { core: TokioCore::new()? }) } + fn new() -> PyResult { Ok(Self { core: ReactorCore::new()? }) } fn time(&self) -> f64 { self.core.time() } fn schedule(&self, h: Py) { self.core.schedule(h) } @@ -56,23 +44,14 @@ impl PyReactor { // An injected async exception (e.g. KeyboardInterrupt) can surface at either // Python call; only a handle whose _run began may be dropped without requeue. let cancelled = h.bind(py).getattr(intern!(py, "_cancelled")).and_then(|v| v.extract::()); - match cancelled { - Err(e) => { - self.core.requeue_front(std::iter::once(h).chain(it)); - return Err(e); - } - Ok(true) => continue, - Ok(false) => {} - } + match cancelled { Err(e) => { self.core.requeue_front(std::iter::once(h).chain(it)); return Err(e); } Ok(true) => continue, Ok(false) => {} } if let Err(e) = h.bind(py).call_method0(intern!(py, "_run")) { // An injected exception surfacing at `_run`'s entry leaves a single-frame // traceback: the callback never ran, so requeue the handle - dropping it // would lose e.g. a task wakeup and orphan the task. A deeper traceback // means the callback began, so the handle is consumed (CPython semantics). - let entry_only = e.traceback(py) - .map_or(true, |tb| tb.getattr("tb_next").ok().is_none_or(|n| n.is_none())); - if entry_only { self.core.requeue_front(std::iter::once(h).chain(it)) } - else { self.core.requeue_front(it) } + let entry_only = e.traceback(py).map_or(true, |tb| tb.getattr("tb_next").ok().is_none_or(|n| n.is_none())); + if entry_only { self.core.requeue_front(std::iter::once(h).chain(it)) } else { self.core.requeue_front(it) } return Err(e); } } diff --git a/src/reactor.rs b/src/reactor.rs index e0b766f..cef2eb0 100644 --- a/src/reactor.rs +++ b/src/reactor.rs @@ -51,9 +51,6 @@ impl Reactor { pub fn time(&self) -> f64 { self.start.elapsed().as_secs_f64() } - /// The poller's own fd (a kqueue/epoll fd is itself pollable), for embedding in another reactor. - pub fn poller_fd(&self) -> std::os::fd::RawFd { std::os::fd::AsRawFd::as_raw_fd(&self.poller) } - pub fn schedule(&self, h: H) { self.inner.lock().unwrap().ready.push_back(h) } /// Safe from any thread: queues the handle and wakes a blocked `poll`. @@ -74,9 +71,7 @@ impl Reactor { /// Drop a scheduled timer. False when it already fired (or was already removed): /// the promoted handle then carries its own cancelled flag, which dispatch skips. - pub fn cancel_timer(&self, key: (u64, u64)) -> bool { - self.inner.lock().unwrap().timers.remove(&key).is_some() - } + pub fn cancel_timer(&self, key: (u64, u64)) -> bool { self.inner.lock().unwrap().timers.remove(&key).is_some() } pub fn timer_count(&self) -> usize { self.inner.lock().unwrap().timers.len() } @@ -95,10 +90,7 @@ impl Reactor { let had = if write { e.writer.take().is_some() } else { e.reader.take().is_some() }; let empty = e.reader.is_none() && e.writer.is_none(); let ev = interest(fd, e); - if empty { - inner.fds.remove(&fd); - let _ = self.poller.delete(borrowed(fd)); - } else { self.poller.modify(borrowed(fd), ev)? } + if empty { inner.fds.remove(&fd); let _ = self.poller.delete(borrowed(fd)); } else { self.poller.modify(borrowed(fd), ev)? } Ok(had) } @@ -134,16 +126,12 @@ impl Reactor { let Inner { ready, timers, .. } = &mut *inner; self.drain_tsq(ready); let now = self.now_us(); - while timers.first_key_value().is_some_and(|((when, _), _)| *when <= now) { - let (_, h) = timers.pop_first().unwrap(); - ready.push_back(h); - } + while timers.first_key_value().is_some_and(|((when, _), _)| *when <= now) { let (_, h) = timers.pop_first().unwrap(); ready.push_back(h); } // Clamped like CPython's MAXIMUM_SELECT_TIMEOUT: a sleep(inf) timer saturates // to u64::MAX, and a pure-Rust consumer blocking in `poll` would hand that // timespec to kqueue, which rejects it with EINVAL const MAX_POLL_US: u64 = 86_400_000_000; - ControlFlow::Continue(if !ready.is_empty() { Some(Duration::ZERO) } - else { timers.first_key_value().map(|((when, _), _)| Duration::from_micros((when - now + 1).min(MAX_POLL_US))) }) + ControlFlow::Continue(if !ready.is_empty() { Some(Duration::ZERO) } else { timers.first_key_value().map(|((when, _), _)| Duration::from_micros((when - now + 1).min(MAX_POLL_US))) }) } /// Blocking wait; holds no locks, so drivers may run it with the GIL released. @@ -176,3 +164,21 @@ impl Reactor { for h in items.rev() { inner.ready.push_front(h) } } } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::thread; + + #[test] + fn cross_thread_schedule_wakes_poll() { + let reactor = Arc::new(Reactor::new().unwrap()); + let sender = reactor.clone(); + let thread = thread::spawn(move || { thread::sleep(Duration::from_millis(10)); sender.schedule_ts("ready").unwrap(); }); + let events = reactor.poll(None).unwrap(); + reactor.process(&events); + thread.join().unwrap(); + assert_eq!(reactor.take_batch().into_iter().collect::>(), vec!["ready"]); + } +} diff --git a/src/tokio_core.rs b/src/tokio_core.rs deleted file mode 100644 index dede98b..0000000 --- a/src/tokio_core.rs +++ /dev/null @@ -1,118 +0,0 @@ -//! The same reactor hosted on a Tokio current-thread runtime: Tokio waits on the -//! kqueue's own fd (a kqueue is pollable), then a zero-timeout drain preserves the -//! kqueue-native level/oneshot semantics that asyncio's add_reader contract needs. -//! Rust futures spawned on the runtime advance during every poll phase, with the -//! GIL released, which is the shared-reactor story in one file. -use crate::reactor::Reactor as ReactorCore; -use polling::Events; -use std::io; -use std::os::fd::RawFd; -use std::time::Duration; -use tokio::io::unix::AsyncFd; -use tokio::io::Interest; -use tokio::runtime::{Builder, Handle, Runtime}; -use std::ops::Deref; - -enum Rt { Owned(Runtime), Borrowed(Handle) } - -impl Rt { - fn handle(&self) -> Handle { - match self { Rt::Owned(r) => r.handle().clone(), Rt::Borrowed(h) => h.clone() } - } - fn block_on(&self, f: F) -> F::Output { - match self { Rt::Owned(r) => r.block_on(f), Rt::Borrowed(h) => h.block_on(f) } - } -} - -pub struct TokioCore { - afd: AsyncFd, // declared before rt: must deregister while an owned driver lives - rt: Rt, - core: ReactorCore, -} - -impl Deref for TokioCore { - type Target = ReactorCore; - fn deref(&self) -> &ReactorCore { &self.core } -} - -impl TokioCore { - /// A reactor on its own current-thread runtime, for standalone use. - pub fn new() -> io::Result { - Self::build(Rt::Owned(Builder::new_current_thread().enable_io().enable_time().build()?)) - } - - /// A reactor on an existing runtime. Call `poll` only from a thread outside that - /// runtime: Tokio panics on `block_on` from one of its own workers. The runtime - /// must outlive this reactor. - pub fn with_handle(handle: Handle) -> io::Result { Self::build(Rt::Borrowed(handle)) } - - fn build(rt: Rt) -> io::Result { - let core = ReactorCore::new()?; - let handle = rt.handle(); - let afd = { - let _g = handle.enter(); - AsyncFd::with_interest(core.poller_fd(), Interest::READABLE)? - }; - Ok(Self { afd, rt, core }) - } - - - /// Wait in Tokio, then drain without blocking. Readiness is cleared *before* - /// the drain: an event arriving after the clear is either picked up by this - /// drain or arrives as a fresh edge. A drain capped by Events capacity leaves - /// the ready queue non-empty, so the next turn polls with a zero timeout and - /// drains again: no burst is lost. - pub fn poll(&self, timeout: Option) -> io::Result { - // A zero timeout means "drain, don't wait": entering the runtime would round - // the wait up to the timer driver's ~1ms tick, at 1ms per busy loop turn - if timeout == Some(Duration::ZERO) { return self.core.poll(timeout) } - self.rt.block_on(async { - let ready = self.afd.readable(); - match timeout { - Some(t) => { - if let Ok(Ok(mut guard)) = tokio::time::timeout(t, ready).await { guard.clear_ready() } - } - None => { - if let Ok(mut guard) = ready.await { guard.clear_ready() } - } - } - }); - self.core.poll(Some(Duration::ZERO)) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::ops::ControlFlow; - use std::time::Instant; - - #[test] - fn borrowed_handle_drives_waits() { - let rt = Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap(); - let core: TokioCore = TokioCore::with_handle(rt.handle().clone()).unwrap(); - core.schedule_at(core.time() + 0.01, 7); - let deadline = Instant::now() + Duration::from_secs(5); - loop { - let ControlFlow::Continue(t) = core.next_timeout() else { panic!("unexpected stop") }; - let ev = core.poll(t).unwrap(); - core.process(&ev); - if core.take_batch().into_iter().any(|h| h == 7) { break } - assert!(Instant::now() < deadline, "timer never fired through the borrowed runtime"); - } - } - - #[test] - fn borrowed_handle_wakes_from_another_thread() { - let rt = Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap(); - let core = std::sync::Arc::new(TokioCore::with_handle(rt.handle().clone()).unwrap()); - let sender = core.clone(); - std::thread::spawn(move || { - std::thread::sleep(Duration::from_millis(10)); - sender.schedule_ts(7).unwrap(); - }); - let ev = core.poll(Some(Duration::from_secs(5))).unwrap(); - core.process(&ev); - assert_eq!(core.take_batch().into_iter().collect::>(), [7]); - } -} diff --git a/tests/oracle_util.py b/tests/oracle_util.py index 4b545b3..b47d6db 100644 --- a/tests/oracle_util.py +++ b/tests/oracle_util.py @@ -1,21 +1,10 @@ "Fetch, cache, and adapt external event-loop conformance suites (CPython, anyio, uvloop)." -import asyncio, importlib.metadata, importlib.util, os, platform, shutil, sys, tarfile, types, unittest, urllib.request +import asyncio, hashlib, importlib.metadata, importlib.util, inspect, os, platform, shutil, sys, tarfile, types, unittest, urllib.request from pathlib import Path import loopmini CACHE = Path(os.environ.get('LOOPMINI_ORACLE_CACHE', Path.home()/'.cache'/'loopmini-oracle')) -ANYIO_PARAM = '''import loopmini - -asyncio_params.append( - pytest.param( - ("asyncio", {"debug": True, "loop_factory": loopmini.new_event_loop}), - id="asyncio+loopmini", - ), -) - -''' - class LoopminiPolicy(asyncio.DefaultEventLoopPolicy): def new_event_loop(self): return loopmini.new_event_loop() @@ -41,49 +30,64 @@ def cpython_test_dir(): if not (lib/'test').exists(): _fetch_tar(f'https://www.python.org/ftp/python/{ver}/Python-{ver}.tgz', CACHE/f'cpython-{ver}') return lib +def _adapted_sdist(name, ver, adapt): + "Fetch an sdist into a cache keyed by the source of its local adapter." + key = hashlib.sha256(inspect.getsource(adapt).encode()).hexdigest()[:12] + cache = CACHE/f'{name}-{ver}-{key}' + root,ready = cache/f'{name}-{ver}',cache/'.ready' + if not ready.exists(): + _fetch_tar(f'https://files.pythonhosted.org/packages/source/{name[0]}/{name}/{name}-{ver}.tar.gz', cache) + adapt(root) + ready.touch() + return root + +def _adapt_anyio(root): + conftest = root/'tests'/'conftest.py' + src = conftest.read_text() + anchor = 'backend_params = asyncio_params.copy()' + assert anchor in src, 'anyio conftest layout changed; adjust oracle_util.anyio_test_dir' + param = '''import loopmini + +asyncio_params.append( + pytest.param( + ("asyncio", {"debug": True, "loop_factory": loopmini.new_event_loop}), + id="asyncio+loopmini", + ), +) + +''' + conftest.write_text(src.replace(anchor, param + anchor)) + def anyio_test_dir(): "The sdist of the installed anyio version, with a loopmini entry added to its backend params." ver = importlib.metadata.version('anyio') - root = CACHE/f'anyio-{ver}'/f'anyio-{ver}' - conftest = root/'tests'/'conftest.py' - if not conftest.exists(): - _fetch_tar(f'https://files.pythonhosted.org/packages/source/a/anyio/anyio-{ver}.tar.gz', CACHE/f'anyio-{ver}') - src = conftest.read_text() - anchor = 'backend_params = asyncio_params.copy()' - assert anchor in src, 'anyio conftest layout changed; adjust oracle_util.anyio_test_dir' - conftest.write_text(src.replace(anchor, ANYIO_PARAM + anchor)) - return root + return _adapted_sdist('anyio', ver, _adapt_anyio) -AIOHTTP_STUB = """import sys, types, loopmini +def _adapt_aiohttp(root): + conftest = root/'tests'/'conftest.py' + src = conftest.read_text() + anchor = 'try:\n if sys.platform == "win32":' + assert anchor in src, 'aiohttp conftest layout changed; adjust oracle_util.aiohttp_test_dir' + stub = """import sys, types, loopmini uvloop = types.ModuleType("uvloop") uvloop.new_event_loop = loopmini.new_event_loop sys.modules["uvloop"] = uvloop """ + src = src.replace(anchor, stub + anchor) + anchor = 'bb.functions["threading.Lock.acquire"].deactivate()' + assert anchor in src, 'aiohttp blockbuster fixture changed; adjust oracle_util.aiohttp_test_dir' + patch = '''# loopmini does the same unlink-if-unchanged close as asyncio's unix loop + for func in ("os.stat", "os.unlink"): + bb.functions[func].can_block_in("loopmini/loop.py", "_stop_serving") + ''' + conftest.write_text(src.replace(anchor, patch + anchor)) + cfg = root/'setup.cfg' + cfg.write_text('\n'.join(l for l in cfg.read_text().splitlines() if 'CoverageWarning' not in l) + '\n') def aiohttp_test_dir(): "The sdist of the installed aiohttp version, with uvloop stubbed to loopmini so `--aiohttp-loop=uvloop` selects it." ver = importlib.metadata.version('aiohttp') - root = CACHE/f'aiohttp-{ver}'/f'aiohttp-{ver}' - conftest = root/'tests'/'conftest.py' - if not conftest.exists(): - _fetch_tar(f'https://files.pythonhosted.org/packages/source/a/aiohttp/aiohttp-{ver}.tar.gz', CACHE/f'aiohttp-{ver}') - src = conftest.read_text() - anchor = 'try:\n if sys.platform == "win32":' - assert anchor in src, 'aiohttp conftest layout changed; adjust oracle_util.aiohttp_test_dir' - conftest.write_text(src.replace(anchor, AIOHTTP_STUB + anchor)) - src = conftest.read_text() - anchor = 'bb.functions["threading.Lock.acquire"].deactivate()' - assert anchor in src, 'aiohttp blockbuster fixture changed; adjust oracle_util.aiohttp_test_dir' - patch = '''# loopmini does the same unlink-if-unchanged close as asyncio's - # unix_events._stop_serving, which blockbuster allowlists by filename - for func in ("os.stat", "os.unlink"): - bb.functions[func].can_block_in("loopmini/transports.py", "close") - ''' - conftest.write_text(src.replace(anchor, patch + anchor)) - cfg = root/'setup.cfg' - # Its warning filter needs the coverage package; the tests do not - cfg.write_text('\n'.join(l for l in cfg.read_text().splitlines() if 'CoverageWarning' not in l) + '\n') - return root + return _adapted_sdist('aiohttp', ver, _adapt_aiohttp) def uvloop_repo(): "A uvloop source clone (its test suite is loop-parameterized by design); None if not present." diff --git a/tests/test_bench.py b/tests/test_bench.py index 8d59948..65dffbc 100644 --- a/tests/test_bench.py +++ b/tests/test_bench.py @@ -3,7 +3,7 @@ Deselected by default; run with `pytest -m bench -s`. Informational, no assertions on timings: the numbers are for watching drift, not gating. """ -import asyncio, time +import asyncio, statistics, time import pytest, loopmini pytestmark = pytest.mark.bench @@ -13,6 +13,9 @@ def _timed(loop_factory, coro_fn): asyncio.run(coro_fn(), loop_factory=loop_factory) return time.perf_counter() - t0 +def _run(loop_factory, coro_fn): return asyncio.run(coro_fn(), loop_factory=loop_factory) +def _measure(loop_factory, coro_fn): return statistics.median(_timed(loop_factory, coro_fn) for _ in range(3)) + async def bench_call_soon(): loop = asyncio.get_running_loop() fut = loop.create_future() @@ -25,6 +28,14 @@ def cb(i): async def bench_sleep0(): for _ in range(20_000): await asyncio.sleep(0) +async def bench_timer(): + delays = [] + for _ in range(100): + start = time.perf_counter() + await asyncio.sleep(0.001) + delays.append(time.perf_counter() - start - 0.001) + return statistics.median(delays) + async def bench_spawn(): async def noop(): pass await asyncio.gather(*(noop() for _ in range(20_000))) @@ -48,12 +59,14 @@ async def handler(reader, writer): server.close() await server.wait_closed() -BENCHES = [bench_call_soon, bench_sleep0, bench_spawn, bench_tcp_echo] +BENCHES = [(bench_call_soon, 200_000), (bench_sleep0, 20_000), (bench_spawn, 20_000), (bench_tcp_echo, 5_000)] def test_bench(): print() - print(f'{"bench":<16} {"stdlib":>8} {"loopmini":>9} ratio') - for fn in BENCHES: - std = _timed(None, fn) - lm = _timed(loopmini.new_event_loop, fn) - print(f'{fn.__name__[6:]:<16} {std:>7.3f}s {lm:>8.3f}s {lm/std:>5.2f}x') + print(f'{"bench":<16} {"stdlib":>8} {"loopmini":>9} {"ratio":>7} {"extra/op":>10}') + for fn,n in BENCHES: + std = _measure(None, fn) + lm = _measure(loopmini.new_event_loop, fn) + print(f'{fn.__name__[6:]:<16} {std:>7.3f}s {lm:>8.3f}s {lm/std:>6.2f}x {(lm-std)*1e6/n:>8.2f}µs') + std, lm = _run(None, bench_timer), _run(loopmini.new_event_loop, bench_timer) + print(f'{"timer overshoot":<16} {std*1e3:>7.3f}ms {lm*1e3:>7.3f}ms')