From e4e8a433a6ca9215b290e78b08fc9fe9a52f564c Mon Sep 17 00:00:00 2001 From: Yehor Smoliakov Date: Sun, 4 Oct 2026 11:12:12 +0000 Subject: [PATCH] Make native streams mandatory for TCP and TLS --- benches/compare_event_loops.py | 23 +---- benches/workload_matrix.py | 2 - docs/benchmarks.md | 5 +- docs/fast-streams.md | 35 ++++--- docs/getting-started.md | 12 +-- docs/hotpath-lab.md | 2 +- docs/how-it-works.md | 7 +- docs/pypi-0.1.54-benchmark.md | 132 ++++++++++++++++++++++++++ examples/fastapi_service.py | 13 +-- python/rsloop/_loop_compat.py | 30 +++++- scripts/hotpath_lab.py | 1 - src/transport/stream/fast.rs | 87 +++++------------- src/transport/stream/protocol.rs | 5 +- src/transport/stream/tls_session.rs | 25 ++++- src/transport/stream/writer.rs | 10 +- tests/test_stream_reader.py | 8 -- tests/test_stream_routing.py | 138 ++++++++++++++++++++++++++++ tests/test_tls.py | 68 +++++++++++++- tests/tooling/test_hotpath_lab.py | 2 +- 19 files changed, 444 insertions(+), 161 deletions(-) create mode 100644 docs/pypi-0.1.54-benchmark.md create mode 100644 tests/test_stream_routing.py diff --git a/benches/compare_event_loops.py b/benches/compare_event_loops.py index 46fd43c..487c602 100644 --- a/benches/compare_event_loops.py +++ b/benches/compare_event_loops.py @@ -130,20 +130,6 @@ def parse_args() -> argparse.Namespace: default=1024, help="Payload size in bytes for the tcp_streams workload", ) - stream_mode = parser.add_mutually_exclusive_group() - stream_mode.add_argument( - "--rsloop-fast-streams", - dest="rsloop_fast_streams", - action="store_true", - help="Use rsloop's native stream wrappers for tcp_streams (default).", - ) - stream_mode.add_argument( - "--no-rsloop-fast-streams", - dest="rsloop_fast_streams", - action="store_false", - help="Use the stdlib asyncio streams layer for tcp_streams on rsloop.", - ) - parser.set_defaults(rsloop_fast_streams=True) parser.add_argument( "--json-output", type=str, @@ -544,8 +530,6 @@ def run_child( env["PYTHONPATH"] = os.pathsep.join( filter(None, (os.path.dirname(script_path), env.get("PYTHONPATH"))) ) - if loop_name == "rsloop": - env["RSLOOP_USE_FAST_STREAMS"] = "1" if args.rsloop_fast_streams else "0" proc = subprocess.run( cmd, check=False, @@ -687,12 +671,7 @@ def parent_main(args: argparse.Namespace) -> int: for workload in selected_workloads: workload_runs: dict[str, list[ChildResult]] = {} if workload == "tcp_streams": - stream_mode = ( - "rsloop native fast streams" - if args.rsloop_fast_streams - else "stdlib asyncio streams" - ) - print(f"tcp_streams mode: {stream_mode}") + print("tcp_streams mode: rsloop native fast streams") for loop_name in available_loops: print(f"Running {workload} on {loop_name}...") if profile_rsloop_dir and loop_name == "rsloop": diff --git a/benches/workload_matrix.py b/benches/workload_matrix.py index b3d7660..510952e 100644 --- a/benches/workload_matrix.py +++ b/benches/workload_matrix.py @@ -1176,8 +1176,6 @@ def run_child_batch( env["PYTHONPATH"] = os.pathsep.join( filter(None, (str(Path(__file__).resolve().parent), env.get("PYTHONPATH"))) ) - if loop_name == "rsloop": - env["RSLOOP_USE_FAST_STREAMS"] = "1" cmd = child_command(args, loop_name, scenario, child_runs) if profile_output is not None: cmd = sampling_profiler_command(cmd, profile_output) diff --git a/docs/benchmarks.md b/docs/benchmarks.md index e32f9aa..f7aca29 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -22,8 +22,9 @@ Times are milliseconds; lower is better. | 5,000 TCP roundtrips | 151.77 | 127.87 | 104.26 | **83.30** | The TCP workload uses 1,024-byte payloads and rsloop's native fast streams; -the other loops use stdlib asyncio streams. Use `--no-rsloop-fast-streams` -to compare all loops through the stdlib streams layer. Zuvloop led callbacks +the other loops use stdlib asyncio streams. Current rsloop always uses native +streams, and the former stream-mode benchmark switches have been removed. +Zuvloop led callbacks and tasks in this run, while rsloop led TCP roundtrips. These are local microbenchmarks, not isolated-lab measurements or general application performance claims; do not compare them directly with the historical macOS results below. diff --git a/docs/fast-streams.md b/docs/fast-streams.md index d01be00..ce4fa6c 100644 --- a/docs/fast-streams.md +++ b/docs/fast-streams.md @@ -1,24 +1,18 @@ # Fast Streams Importing `rsloop` patches `asyncio.open_connection()` and -`asyncio.start_server()` by default. - -That import-time behavior is controlled by `RSLOOP_USE_FAST_STREAMS` and can be -disabled with: - -```bash -export RSLOOP_USE_FAST_STREAMS=0 -``` - -The native fast-stream path is used only when: - -- the running loop is an `rsloop.Loop` -- `ssl` is unset or `None` - -Otherwise `rsloop` falls back to the standard-library `asyncio.streams` -helpers. - -On that path the reader handed to your code is the native +`asyncio.start_server()`, including their `asyncio.streams` entry points. +On an `rsloop.Loop`, these helpers always use native fast streams for both +plaintext TCP and TLS. There is no stream-mode option or automatic stdlib +fallback on rsloop; the former `RSLOOP_USE_FAST_STREAMS` environment variable +has no effect. Connection and TLS errors propagate to the caller. + +The helpers remain coroutine functions and select the running loop when +awaited, so they can be created before `run()` and passed to `create_task()`. +Calls running on other event loops continue to use those loops' standard +asyncio stream helpers. + +The reader handed to your rsloop code is the native `PyFastStreamReader` rather than `asyncio.StreamReader`. It implements the reading surface protocols actually use: @@ -34,6 +28,11 @@ left in the buffer afterwards. `tests/test_stream_reader.py` pins that behavior by driving the same feed scripts through both readers and comparing the results. +Native writers support `write`, `writelines`, `drain`, `close`, `wait_closed`, +`is_closing`, `get_extra_info`, `write_eof`, and `can_write_eof`. TLS transports +do not support half-close; TLS peer EOF closes the stream instead of leaving +its write side open. TLS negotiation continues to use rsloop's TLS transport. + The implementation lives in `src/transport/stream/fast.rs` and is backed by the lower-level transport code in `src/transport/stream/mod.rs`. diff --git a/docs/getting-started.md b/docs/getting-started.md index 4ad8560..d1b94ec 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -132,15 +132,11 @@ Importing `rsloop` does a little setup work: - it boots the native extension - it patches `asyncio.set_event_loop(...)` for compatibility, especially on older Python versions -- it can patch `asyncio.open_connection(...)` and `asyncio.start_server(...)` to use `rsloop`'s fast stream path +- it patches `asyncio.open_connection(...)` and `asyncio.start_server(...)` to use native fast streams on rsloop, including TLS -That fast stream behavior is controlled by `RSLOOP_USE_FAST_STREAMS`. - -Disable it like this: - -```bash -export RSLOOP_USE_FAST_STREAMS=0 -``` +Native fast streams are automatic on rsloop, with no mode switch or stdlib +fallback. Other event loops retain their standard stream helpers. See +[Fast Streams](fast-streams.md) for the supported stream interface. ## Useful examples diff --git a/docs/hotpath-lab.md b/docs/hotpath-lab.md index 8747c7f..27d115c 100644 --- a/docs/hotpath-lab.md +++ b/docs/hotpath-lab.md @@ -60,7 +60,7 @@ This harness does not apply model output or create commits automatically. emitted. SHA-256 inventories detect changed source/package/compiler outputs. - Every benchmark child imports the selected artifact first and checks the actual native extension path and hash. The development installation is not - used. `PYTHONHASHSEED=0`, fast streams enabled, and GC disabled during both + used. `PYTHONHASHSEED=0`, native streams, and GC disabled during both measured and warmup workloads are explicit harness choices. - PGO builds use an explicit native Cargo `--target` so build scripts and proc macros are not instrumented. Training sets `LLVM_PROFILE_FILE` only in workload diff --git a/docs/how-it-works.md b/docs/how-it-works.md index 83f1fcd..982ee18 100644 --- a/docs/how-it-works.md +++ b/docs/how-it-works.md @@ -109,10 +109,9 @@ These gaps are visible in the current implementation: - TLS uses a `rustls` backend with a narrower compatibility surface than CPython's OpenSSL-backed `ssl` module. In particular, encrypted private keys - are not supported yet, and the fast-stream monkeypatch still falls back to - standard-library helpers whenever `ssl` is enabled. TLS transport internals - also still use helper-thread paths instead of the runtime-thread `vibeio` - socket path. + are not supported yet. TLS uses native fast streams, but its transport + internals still use helper-thread paths instead of the runtime-thread + `vibeio` socket path. - Subprocess support still has one notable gap: `preexec_fn` remains unsupported because running arbitrary Python between `fork()` and `exec()` is unsafe in this runtime model. diff --git a/docs/pypi-0.1.54-benchmark.md b/docs/pypi-0.1.54-benchmark.md new file mode 100644 index 0000000..9e01337 --- /dev/null +++ b/docs/pypi-0.1.54-benchmark.md @@ -0,0 +1,132 @@ +# PyPI rsloop 0.1.54 benchmark + +On October 4, 2026, the live [PyPI JSON API](https://pypi.org/pypi/rsloop/json) +reported [0.1.54](https://pypi.org/project/rsloop/0.1.54/) as the latest release. +The CPython 3.14 Linux wheel was uploaded at `2026-10-04T09:19:49.632695Z`. + +Downloaded and benchmarked: + +```text +rsloop-0.1.54-cp314-cp314-manylinux_2_39_x86_64.whl +SHA-256: aabab6f48b949810ae82716e342398484b111ac314172722751ac061575cc443 +``` + +The checksum matches PyPI metadata. The wheel was installed with `uv pip install +--target` under `target/pypi-speed-check/pypi/package`, and its import path and +distribution version were verified. The development installation was preserved. +The wheel's `Handle` and `TimerHandle` object sizes are 80 and 56 bytes, +respectively, matching the optimized layouts on the local branch. Matching +layouts do not establish identical source or build settings. + +## Standard loop comparison + +Linux x86-64, Intel Core i9-9900K, GIL-enabled CPython 3.14.7, uvloop 0.22.1. +The repository's `benches/compare_event_loops.py` ran seven measured fresh +processes after two warmups per loop/workload. Times are medians; lower is better. + +| Workload | asyncio | uvloop | PyPI rsloop 0.1.54 | +| --- | ---: | ---: | ---: | +| 200,000 callbacks | 106.30 ms | 51.06 ms | 43.87 ms | +| 50,000 tasks, batches of 5,000 | 141.76 ms | 87.40 ms | 85.42 ms | +| 5,000 TCP roundtrips, 1 KiB | 142.74 ms | 117.73 ms | 77.31 ms | + +The default TCP comparison uses rsloop's native fast streams and stdlib streams +for the other loops. These are short, sequential microbenchmarks, not paired +application-level comparisons. In particular, the small task difference from +uvloop should not be generalized into an overall performance advantage. + +A separate run disabled rsloop's native fast streams, using the stdlib streams +layer for all loops (same 5,000 roundtrips, 1 KiB, seven runs/two warmups): + +| asyncio | uvloop | PyPI rsloop 0.1.54 | +| ---: | ---: | ---: | +| 147.30 ms | 117.68 ms | 125.93 ms | + +Here uvloop had the lowest median; rsloop took 7.0% longer. The default native +stream result therefore should not be presented as a win with every streams +configuration. These separate short runs remain descriptive measurements. + +These commands used the benchmark scripts at `a570b9f`. The stream-mode switches +were subsequently removed when native streams became mandatory on rsloop. +Use that revision or the archived harness to reproduce this historical check. + +```bash +PYTHONPATH="$PWD/target/pypi-speed-check/pypi/package" \ + .venv/bin/python benches/compare_event_loops.py \ + --loops asyncio,uvloop,rsloop --workloads callbacks,tasks,tcp_streams \ + --warmups 2 --repeat 7 \ + --json-output target/pypi-speed-check/event-loops.json +PYTHONPATH="$PWD/target/pypi-speed-check/pypi/package" \ + .venv/bin/python benches/compare_event_loops.py \ + --loops asyncio,uvloop,rsloop --workloads tcp_streams \ + --no-rsloop-fast-streams --warmups 2 --repeat 7 \ + --json-output target/pypi-speed-check/stdlib-tcp.json +``` + +## Published wheel versus local branch + +The local binary is a normal `maturin develop --release` build of `a570b9f` +on `perf/hotpath-rs-profiling` (project version 0.1.53). It was copied into a +second isolated package directory. Both ELF `.comment` sections report Rust +1.100.0-nightly (`f7575a9da`, September 24), GCC 13.3.0, and LLD 23.1.1. All four +packaged Python `.py` files have identical hashes. The complete published build +settings and source revision are unavailable, so this compares two distributable +binaries rather than isolating the effect of source edits. + +The comparison reuses `scripts/hotpath_lab.py` workload configurations, +`run_sample`, balanced AB/BA orders, and paired bootstrap statistics. An archived +driver runs 12 shuffled paired blocks across eight workloads, with one full +warmup per fresh process. Every child verifies the selected native extension's +path and package inventory. Holdout workload sizes keep measured runs longer +than the short standard comparison above. GC is disabled during workloads. + +The ordinary lab comparison intentionally requires matching compiler metadata +and flags. The published manifest records build provenance as unknown; embedded +ELF compiler labels are preserved separately. This distribution comparison uses +a separate driver, without applying the source-change promotion gate. +Both manifests and all benchmark sources are archived. No compilation or tests +run concurrently with timings; CPU frequency and unrelated host load are not +controlled. + +All 12 blocks completed. The table shows median elapsed time and the paired +change from the published wheel to the local branch; negative means the branch +is faster. The paired change uses log-time ratios, not the ratio of medians. +Bootstrap intervals adjust across eight workloads and two metrics with a 95% +family confidence target. Every measured sample lasted at least 0.38 seconds. + +| Workload | PyPI 0.1.54 | Branch `a570b9f` | Branch timing change (adjusted interval) | +| --- | ---: | ---: | ---: | +| 3,000,000 callbacks | 655.43 ms | 650.93 ms | -0.7% [-2.4%, +0.8%] | +| 350,000 tasks, batches of 777 | 584.30 ms | 583.41 ms | -1.0% [-2.8%, +0.3%] | +| 25,000 native TCP roundtrips, 8 KiB | 485.43 ms | 490.18 ms | +0.2% [-1.1%, +1.6%] | +| HTTP, 7 connections, 4,000 requests each | 495.83 ms | 498.93 ms | +1.1% [-1.4%, +4.0%] | +| 1,200,000 retained timers | 414.53 ms | 418.46 ms | +1.0% [-0.05%, +2.3%] | +| 1,200,000 discarded-handle timers | 388.67 ms | 386.12 ms | -1.2% [-3.2%, +0.05%] | +| 1,200,000 cancelled timers | 404.27 ms | 405.51 ms | +0.3% [-1.1%, +1.8%] | +| 4,000 generic TCP connections, 8 KiB | 693.48 ms | 691.55 ms | +0.1% [-2.0%, +2.0%] | + +Timer batch size is 777. All timing intervals include zero: this experiment +does not establish a speed difference between the published wheel and the +local branch. This is not proof of exact equivalence or of identical source. + +The published wheel's median peak RSS was about 0.6–0.9 MiB lower in most +workloads. For example, tasks used 33.42 versus 34.30 MiB and callbacks used +515.84 versus 516.50 MiB. Connection-churn peak RSS was essentially unchanged. +RSS includes the warmup and reflects the GC-disabled benchmark configuration; +these are whole-process measurements, not Python object-size measurements. + +## Local artifacts + +`target/pypi-speed-check/` contains the original wheel, full PyPI metadata, +selected-wheel metadata and digest, isolated packages, standard benchmark JSON +and log, comparison driver, plans, raw samples, and archived harness sources. +The driver is `target/pypi-speed-check/compare_distributions.py`; it refuses to +overwrite its `paired/` output directory. + +Install the verified wheel into its isolated directory with: + +```bash +uv pip install --python .venv/bin/python --no-deps \ + --target target/pypi-speed-check/pypi/package \ + target/pypi-speed-check/rsloop-0.1.54-cp314-cp314-manylinux_2_39_x86_64.whl +``` diff --git a/examples/fastapi_service.py b/examples/fastapi_service.py index fa4cfc8..05a1da9 100644 --- a/examples/fastapi_service.py +++ b/examples/fastapi_service.py @@ -4,7 +4,6 @@ import argparse import asyncio import importlib -import os import signal import sys import time @@ -31,11 +30,6 @@ def normalize_event_loop(value: str) -> str: return value -def configure_rsloop_fast_streams(loop_name: str) -> None: - if normalize_event_loop(loop_name) == "rsloop": - os.environ["RSLOOP_USE_FAST_STREAMS"] = "1" - - def loop_factory_for(loop_name: str) -> EventLoopFactory: normalized = normalize_event_loop(loop_name) if normalized == "asyncio": @@ -103,11 +97,7 @@ def current_loop_payload() -> dict[str, str]: loop = asyncio.get_running_loop() stream_mode = "stdlib asyncio streams" if app.state.selected_event_loop == "rsloop": - stream_mode = ( - "rsloop fast streams" - if os.environ.get("RSLOOP_USE_FAST_STREAMS", "1") != "0" - else "stdlib asyncio streams" - ) + stream_mode = "rsloop fast streams" return { "selected": app.state.selected_event_loop, "module": type(loop).__module__, @@ -267,7 +257,6 @@ def parse_args() -> argparse.Namespace: def main() -> int: args = parse_args() - configure_rsloop_fast_streams(args.event_loop) app.state.selected_event_loop = normalize_event_loop(args.event_loop) previous_sigint_handler = signal.getsignal(signal.SIGINT) diff --git a/python/rsloop/_loop_compat.py b/python/rsloop/_loop_compat.py index e1bdbac..ed24626 100644 --- a/python/rsloop/_loop_compat.py +++ b/python/rsloop/_loop_compat.py @@ -11,9 +11,9 @@ from ._loop import PyLoop as Loop from ._loop import __version__ as __version__ from ._loop import build_info as build_info -from ._loop import open_connection as __open_connection +from ._loop import open_connection as __native_open_connection from ._loop import reset_transport_stats as reset_transport_stats -from ._loop import start_server as __start_server +from ._loop import start_server as __native_start_server from ._loop import transport_stats as transport_stats _T = __typing.TypeVar("_T") @@ -66,14 +66,34 @@ def __set_event_loop(loop: Loop | None) -> None: __ORIG_SHUTDOWN_ASYNCGENS = Loop.shutdown_asyncgens __ORIG_CLOSE: __typing.Any = Loop.close __ORIG_CREATE_TASK = Loop.create_task -__USE_FAST_STREAMS = __os.environ.get("RSLOOP_USE_FAST_STREAMS", "1") != "0" __ASYNCGEN_STATE: dict[Loop, dict[str, __typing.Any]] = {} __LOOP_CONFIG: dict[Loop, dict[str, __typing.Any]] = {} -if __USE_FAST_STREAMS and __asyncio.open_connection is __ORIG_OPEN_CONNECTION: + +async def __open_connection(host=None, port=None, *, limit=2**16, **kwargs): + # Select the implementation when awaited, so calls created before run() + # still use the implementation belonging to the running event loop. + if isinstance(__asyncio.get_running_loop(), Loop): + return await __native_open_connection(host, port, limit=limit, **kwargs) + return await __ORIG_OPEN_CONNECTION(host, port, limit=limit, **kwargs) + + +async def __start_server(client_connected_cb, host=None, port=None, *, limit=2**16, **kwargs): + if isinstance(__asyncio.get_running_loop(), Loop): + return await __native_start_server( + client_connected_cb, host, port, limit=limit, **kwargs + ) + return await __ORIG_START_SERVER(client_connected_cb, host, port, limit=limit, **kwargs) + + +if __asyncio.open_connection is __ORIG_OPEN_CONNECTION: __asyncio.open_connection = __open_connection -if __USE_FAST_STREAMS and __asyncio.start_server is __ORIG_START_SERVER: +if __asyncio.streams.open_connection is __ORIG_OPEN_CONNECTION: + __asyncio.streams.open_connection = __open_connection +if __asyncio.start_server is __ORIG_START_SERVER: __asyncio.start_server = __start_server +if __asyncio.streams.start_server is __ORIG_START_SERVER: + __asyncio.streams.start_server = __start_server _asyncio = __asyncio _io = __io diff --git a/scripts/hotpath_lab.py b/scripts/hotpath_lab.py index 072c063..cc47934 100644 --- a/scripts/hotpath_lab.py +++ b/scripts/hotpath_lab.py @@ -431,7 +431,6 @@ def run_sample( profile_pattern: Path | None = None, ) -> dict: env = clean_environment() - env["RSLOOP_USE_FAST_STREAMS"] = "1" if profile_pattern: env["LLVM_PROFILE_FILE"] = str(profile_pattern) command = [ diff --git a/src/transport/stream/fast.rs b/src/transport/stream/fast.rs index 712ebec..909f6be 100644 --- a/src/transport/stream/fast.rs +++ b/src/transport/stream/fast.rs @@ -1824,6 +1824,7 @@ pub struct PyFastStreamProtocol { paused: bool, drain_waiters: Vec>, connection_lost: bool, + over_ssl: bool, } impl PyFastStreamProtocol { @@ -1852,6 +1853,7 @@ impl PyFastStreamProtocol { paused: false, drain_waiters: Vec::new(), connection_lost: false, + over_ssl: false, }) } @@ -1942,8 +1944,13 @@ impl PyFastStreamProtocol { py: Python<'_>, transport: Py, ) -> PyResult<()> { + let over_ssl = !transport + .call_method1(py, "get_extra_info", ("sslcontext",))? + .bind(py) + .is_none(); { let mut protocol = slf.borrow_mut(py); + protocol.over_ssl = over_ssl; protocol.transport = transport.clone_ref(py); protocol .reader @@ -2097,9 +2104,10 @@ impl PyFastStreamProtocol { self.reader.borrow_mut(py).feed_data_internal(py, data) } - fn eof_received(&mut self, py: Python<'_>) -> PyResult { + pub(super) fn eof_received(&mut self, py: Python<'_>) -> PyResult { self.reader.borrow_mut(py).feed_eof_internal(py)?; - Ok(true) + // TLS does not support keeping a write half open after peer EOF. + Ok(!self.over_ssl) } fn connection_lost(&mut self, py: Python<'_>, exc: Option>) -> PyResult<()> { @@ -2255,31 +2263,6 @@ fn running_loop(py: Python<'_>) -> PyResult> { .unbind()) } -fn call_asyncio_streams_function( - py: Python<'_>, - name: &str, - args: &Bound<'_, PyTuple>, - kwargs: Option<&Bound<'_, PyDict>>, -) -> PyResult> { - let module = py.import("asyncio.streams")?; - Ok(module.getattr(name)?.call(args, kwargs)?.unbind()) -} - -fn kwargs_with_limit<'py>( - py: Python<'py>, - kwargs: Option<&Bound<'py, PyDict>>, - limit: usize, -) -> PyResult> { - let dict = PyDict::new(py); - if let Some(kwargs) = kwargs { - for (key, value) in kwargs.iter() { - dict.set_item(key, value)?; - } - } - dict.set_item("limit", limit)?; - Ok(dict) -} - fn copy_kwargs<'py>( py: Python<'py>, kwargs: Option<&Bound<'py, PyDict>>, @@ -2295,21 +2278,14 @@ fn copy_kwargs<'py>( Ok(Some(copied)) } -fn native_stream_loop( - py: Python<'_>, - kwargs: Option<&Bound<'_, PyDict>>, -) -> PyResult>> { +fn native_stream_loop(py: Python<'_>) -> PyResult> { let loop_obj = running_loop(py)?; if !loop_obj.bind(py).is_instance_of::() { - return Ok(None); - } - if let Some(kwargs) = kwargs - && let Some(ssl) = kwargs.get_item("ssl")? - && !ssl.is_none() - { - return Ok(None); + return Err(pyo3::exceptions::PyRuntimeError::new_err( + "native streams require an rsloop event loop", + )); } - Ok(Some(loop_obj)) + Ok(loop_obj) } fn host_port_objects( @@ -2370,10 +2346,10 @@ fn fast_open_connection_result(py: Python<'_>, created: Py) -> PyResult

>, ) -> PyResult> { let (host_obj, port_obj) = host_port_objects(py, host, port); - let args = PyTuple::new(py, [host_obj.clone_ref(py), port_obj.clone_ref(py)])?; - - let Some(loop_obj) = native_stream_loop(py, kwargs)? else { - let kwargs = kwargs_with_limit(py, kwargs, limit)?; - return call_asyncio_streams_function(py, "open_connection", &args, Some(&kwargs)); - }; + let loop_obj = native_stream_loop(py)?; let (locals, awaitable) = fast_open_connection_awaitable(py, &loop_obj, host_obj, port_obj, limit, kwargs)?; @@ -2412,10 +2383,10 @@ pub fn open_connection( /// Returns an awaitable that starts a stream server. /// -/// With a running [`PyLoop`](crate::PyLoop) and no TLS argument, accepted +/// With a running [`PyLoop`](crate::PyLoop), accepted /// connections use native fast readers and writers before invoking -/// `client_connected_cb`. Unsupported configurations delegate to -/// `asyncio.start_server`. +/// `client_connected_cb`, including over TLS. A running rsloop event loop is +/// required. /// /// `limit` controls each reader's buffer limit; extra keyword arguments are /// forwarded to the loop server factory. @@ -2436,19 +2407,7 @@ pub fn start_server( .as_ref() .map(|value| value.clone_ref(py)) .unwrap_or_else(|| py.None()); - let args = PyTuple::new( - py, - [ - client_connected_cb.clone_ref(py), - host_obj.clone_ref(py), - port_obj.clone_ref(py), - ], - )?; - - let Some(loop_obj) = native_stream_loop(py, kwargs)? else { - let kwargs = kwargs_with_limit(py, kwargs, limit)?; - return call_asyncio_streams_function(py, "start_server", &args, Some(&kwargs)); - }; + let loop_obj = native_stream_loop(py)?; let locals = task_locals_for_loop(py, &loop_obj)?; let factory = Py::new( diff --git a/src/transport/stream/protocol.rs b/src/transport/stream/protocol.rs index 838b259..7d0dec0 100644 --- a/src/transport/stream/protocol.rs +++ b/src/transport/stream/protocol.rs @@ -286,10 +286,7 @@ impl StreamReaderFastPath { pub(super) fn eof_received(&self, py: Python<'_>) -> PyResult { match self { - Self::Native { reader, .. } => { - reader.borrow_mut(py).feed_eof_internal(py)?; - Ok(true) - } + Self::Native { protocol, .. } => protocol.borrow_mut(py).eof_received(py), Self::Generic { .. } => { self.feed_eof(py)?; Ok(true) diff --git a/src/transport/stream/tls_session.rs b/src/transport/stream/tls_session.rs index abc0511..fbcf4db 100644 --- a/src/transport/stream/tls_session.rs +++ b/src/transport/stream/tls_session.rs @@ -74,10 +74,10 @@ impl TlsConnectionKind { } } - pub(super) fn writer_write_all(&mut self, data: &[u8]) -> io::Result<()> { + fn writer_write(&mut self, data: &[u8]) -> io::Result { match self { - Self::Client(conn) => conn.writer().write_all(data), - Self::Server(conn) => conn.writer().write_all(data), + Self::Client(conn) => conn.writer().write(data), + Self::Server(conn) => conn.writer().write(data), } } @@ -98,6 +98,25 @@ pub(super) struct TlsIoState { pub(super) type SharedTlsIoState = Arc>; impl TlsIoState { + pub(super) fn write_plaintext_all(&mut self, mut data: &[u8]) -> io::Result<()> { + while !data.is_empty() { + let written = self.connection.writer_write(data)?; + if written == 0 && !self.connection.wants_write() { + return Err(io::Error::new( + io::ErrorKind::WriteZero, + "failed to buffer TLS plaintext", + )); + } + data = &data[written..]; + if !data.is_empty() { + // rustls accepts only as much plaintext as its record buffer + // can hold. Flush those records before submitting the rest. + flush_tls_io_locked(self)?; + } + } + Ok(()) + } + #[inline] pub(super) fn fd(&self) -> fd_ops::RawFd { self.stream.fd() diff --git a/src/transport/stream/writer.rs b/src/transport/stream/writer.rs index e87da4b..16e6af8 100644 --- a/src/transport/stream/writer.rs +++ b/src/transport/stream/writer.rs @@ -262,7 +262,7 @@ pub(super) fn write_tls_data_batch( ) -> bool { let mut buffered_len = 0; let mut state = tls_state.lock().expect("poisoned tls state"); - if let Err(err) = state.connection.writer_write_all(data.remaining()) { + if let Err(err) = state.write_plaintext_all(data.remaining()) { drop(state); report_writer_io_error(core, err); return false; @@ -271,7 +271,7 @@ pub(super) fn write_tls_data_batch( loop { match writer_rx.try_recv() { Ok(WriterCommand::Data(next)) => { - if let Err(err) = state.connection.writer_write_all(next.remaining()) { + if let Err(err) = state.write_plaintext_all(next.remaining()) { drop(state); report_writer_io_error(core, err); return false; @@ -294,9 +294,9 @@ pub(super) fn write_tls_data_batch( } } - // Feed all immediately available plaintext into rustls before flushing - // encrypted records. This turns common header/body write pairs into one - // socket-flush pass and one TLS-state lock acquisition. + // Flush the batch's remaining records. Writes flush earlier when rustls's + // buffer fills; small header/body pairs still share one socket-flush pass + // and one TLS-state lock acquisition. if let Err(err) = flush_tls_io_locked(&mut state) { drop(state); report_writer_io_error(core, err); diff --git a/tests/test_stream_reader.py b/tests/test_stream_reader.py index 9da1eb9..9e41496 100644 --- a/tests/test_stream_reader.py +++ b/tests/test_stream_reader.py @@ -25,10 +25,6 @@ EOF = "eof" EXC = "exc" -FAST_STREAMS_ACTIVE = getattr(asyncio.open_connection, "__module__", "").startswith( - "rsloop" -) - # The native reader accepts a tuple of separators on every supported version, # but `asyncio.StreamReader` only learned the form in 3.13 — below that it # raises TypeError out of `bytearray.find`. These cases therefore only have a @@ -457,10 +453,6 @@ async def send(reader, writer): assert self._round_trip(send) == (b"ONLY\n", b"") - @pytest.mark.skipif( - not (FAST_STREAMS_ACTIVE), - reason="requires the fast-stream patch (RSLOOP_USE_FAST_STREAMS)", - ) def test_open_connection_really_uses_the_native_reader(self) -> None: # Without this the round-trip tests above would still pass against the # stdlib reader and quietly stop covering the native one. diff --git a/tests/test_stream_routing.py b/tests/test_stream_routing.py new file mode 100644 index 0000000..1947b3f --- /dev/null +++ b/tests/test_stream_routing.py @@ -0,0 +1,138 @@ +"""Native stream selection is mandatory on rsloop, including deferred calls.""" + +import asyncio +import os +import subprocess +import sys +from pathlib import Path +from typing import Any, cast + +import pytest +import rsloop +import rsloop._loop as loop_module + +PyFastStreamReader = cast(Any, loop_module).PyFastStreamReader +PyFastStreamWriter = cast(Any, loop_module).PyFastStreamWriter + + +def test_rsloop_streams_do_not_call_stdlib_helpers(monkeypatch): + def forbidden(*args, **kwargs): + raise AssertionError("rsloop fell back to a stdlib stream helper") + + monkeypatch.setattr(asyncio.streams, "open_connection", forbidden) + monkeypatch.setattr(asyncio.streams, "start_server", forbidden) + loop = rsloop.new_event_loop() + accepted = [] + + async def echo(reader, writer): + accepted.append((type(reader), type(writer))) + try: + writer.write(await reader.readexactly(4)) + await writer.drain() + finally: + writer.close() + await writer.wait_closed() + + # Constructing either coroutine without a running loop must not select + # stdlib readers, and both helpers must remain usable with create_task(). + startup = asyncio.start_server(echo, "127.0.0.1", 0) + try: + server = loop.run_until_complete(startup) + pending = asyncio.open_connection("127.0.0.1", server.sockets[0].getsockname()[1]) + + async def exchange(): + reader, writer = await asyncio.create_task(pending) + assert type(reader) is PyFastStreamReader + assert type(writer) is PyFastStreamWriter + try: + writer.write(b"test") + await writer.drain() + assert await reader.readexactly(4) == b"test" + finally: + writer.close() + await writer.wait_closed() + + try: + loop.run_until_complete(asyncio.wait_for(exchange(), 5)) + assert accepted == [(PyFastStreamReader, PyFastStreamWriter)] + finally: + server.close() + loop.run_until_complete(server.wait_closed()) + finally: + loop.close() + + +def test_legacy_environment_switch_cannot_disable_native_streams(): + env = os.environ.copy() + env["RSLOOP_USE_FAST_STREAMS"] = "0" + subprocess.run( + [ + sys.executable, + "-c", + ( + "import runpy, sys\n" + "from pytest import MonkeyPatch\n" + "with MonkeyPatch.context() as patch:\n" + " runpy.run_path(sys.argv[1])['test_rsloop_streams_do_not_call_stdlib_helpers'](patch)\n" + ), + str(Path(__file__).resolve()), + ], + env=env, + check=True, + timeout=15, + ) + + +def test_asyncio_streams_namespace_also_uses_native_helpers(): + assert asyncio.streams.open_connection is asyncio.open_connection + assert asyncio.streams.start_server is asyncio.start_server + + +def test_native_entry_points_reject_other_loops(): + from rsloop import _loop + + async def main(): + with pytest.raises(RuntimeError, match="require an rsloop event loop"): + _ = _loop.open_connection("127.0.0.1", 1) + with pytest.raises(RuntimeError, match="require an rsloop event loop"): + _ = _loop.start_server(lambda *_: None, "127.0.0.1", 0) + + asyncio.run(main()) + + +@pytest.mark.parametrize("loop_name", ["asyncio", "uvloop"]) +def test_other_loops_keep_their_stdlib_streams(loop_name): + if loop_name == "uvloop": + factory = pytest.importorskip("uvloop").new_event_loop + else: + factory = asyncio.new_event_loop + loop = factory() + accepted = [] + + async def echo(reader, writer): + accepted.append(type(reader)) + writer.close() + await writer.wait_closed() + + async def main(): + server = await asyncio.start_server(echo, "127.0.0.1", 0) + try: + reader, writer = await asyncio.open_connection( + "127.0.0.1", server.sockets[0].getsockname()[1] + ) + try: + assert type(reader) is asyncio.StreamReader + assert type(writer) is asyncio.StreamWriter + await reader.read() + assert accepted == [asyncio.StreamReader] + finally: + writer.close() + await writer.wait_closed() + finally: + server.close() + await server.wait_closed() + + try: + loop.run_until_complete(asyncio.wait_for(main(), 5)) + finally: + loop.close() diff --git a/tests/test_tls.py b/tests/test_tls.py index 9452424..8a7d42f 100644 --- a/tests/test_tls.py +++ b/tests/test_tls.py @@ -10,7 +10,7 @@ import subprocess import sys import tempfile -from typing import cast +from typing import Any, cast import pytest import rsloop @@ -81,6 +81,72 @@ def make_ssl_contexts(tmpdir: str): class TestTls: + @pytest.mark.parametrize("chunk_size", [1024, 147456]) + def test_tls_streams_use_native_objects_without_stdlib_fallback( + self, monkeypatch, chunk_size + ): + from rsloop import _loop + + native = cast(Any, _loop) + + def forbidden(*args, **kwargs): + raise AssertionError("TLS streams fell back to stdlib") + + monkeypatch.setattr(asyncio.streams, "open_connection", forbidden) + monkeypatch.setattr(asyncio.streams, "start_server", forbidden) + + async def main(): + accepted = [] + payload = b"native TLS streams" * 8192 + + async def echo(reader, writer): + accepted.append((type(reader), type(writer))) + try: + writer.write(await reader.readexactly(len(payload))) + await writer.drain() + finally: + writer.close() + await writer.wait_closed() + + with tempfile.TemporaryDirectory() as tmpdir: + server_ctx, client_ctx = make_ssl_contexts(tmpdir) + server = await asyncio.create_task( + asyncio.start_server(echo, "127.0.0.1", 0, ssl=server_ctx, limit=32) + ) + try: + reader, writer = await asyncio.create_task( + asyncio.open_connection( + "127.0.0.1", + server.sockets[0].getsockname()[1], + ssl=client_ctx, + server_hostname="localhost", + limit=32, + ) + ) + try: + assert type(reader) is native.PyFastStreamReader + assert type(writer) is native.PyFastStreamWriter + assert cast(Any, reader)._limit == 32 + assert writer.get_extra_info("sslcontext") is client_ctx + assert not writer.can_write_eof() + for offset in range(0, len(payload), chunk_size): + writer.write(payload[offset : offset + chunk_size]) + await writer.drain() + assert await reader.readexactly(len(payload)) == payload + assert await reader.read() == b"" + await writer.wait_closed() + assert accepted == [ + (native.PyFastStreamReader, native.PyFastStreamWriter) + ] + finally: + writer.close() + await writer.wait_closed() + finally: + server.close() + await server.wait_closed() + + rsloop.run(asyncio.wait_for(main(), 10)) + def test_create_default_context_marks_default_verify_paths(self) -> None: context = ssl.create_default_context() assert context.__dict__.get("_rsloop_use_default_verify_paths") diff --git a/tests/tooling/test_hotpath_lab.py b/tests/tooling/test_hotpath_lab.py index ea9e05d..55dfc71 100644 --- a/tests/tooling/test_hotpath_lab.py +++ b/tests/tooling/test_hotpath_lab.py @@ -223,7 +223,7 @@ def run(command, **kwargs): pattern = tmp_path / "%p-%m.profraw" lab.run_sample(tmp_path, {"workload": "tasks"}, 0, 30, pattern) assert calls[0]["env"]["LLVM_PROFILE_FILE"] == str(pattern) - assert calls[0]["env"]["RSLOOP_USE_FAST_STREAMS"] == "1" + assert "RSLOOP_USE_FAST_STREAMS" not in calls[0]["env"] def test_test_fixtures_do_not_modify_artifact_source(tmp_path, monkeypatch):