Merge pull request #6 from RustedBytes/compio

compio integration

Author: egorsmkv

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "rsloop"
-version = "0.1.4"
+version = "0.1.5"
 edition = "2021"
 description = "An event loop for asyncio written in Rust"
 license = "Apache-2.0"
@@ -19,6 +19,7 @@ profiler = ["dep:pprof"]
 
 [dependencies]
 async-std = "1"
+compio = { version = "0.18", default-features = false, features = ["runtime", "net", "io", "signal", "time", "io-uring"] }
 crossbeam-channel = "0.5"
 futures = { version = "0.3", default-features = false, features = ["std"] }
 libc = "0.2"
@@ -41,7 +42,7 @@ opt-level = 3
 lto = "fat"
 codegen-units = 1
 incremental = false
-strip = true
+# strip = true
 
 [profile.bench]
 inherits = "release"

Cargo.toml

@@ -8,9 +8,11 @@
 `rsloop` is a PyO3-based `asyncio` event loop implemented in Rust.
 
 Each `rsloop.Loop` owns a dedicated Rust runtime thread for loop coordination
-and I/O work. Python callbacks, tasks, and coroutines still run on the thread
-that calls `run_forever()` or `run_until_complete()` (usually the main Python
-thread).
+and I/O work. On Linux, low-level fd watchers plus plain TCP / Unix socket
+readiness, socket reads, and non-TLS server accepts are driven from that thread
+through `compio` with `io_uring` support enabled. Python callbacks, tasks, and
+coroutines still run on the thread that calls `run_forever()` or
+`run_until_complete()` (usually the main Python thread).
 
 The package exposes:
 
@@ -146,6 +148,19 @@ Otherwise `rsloop` falls back to the stdlib `asyncio.streams` helpers.
 The implementation lives in `src/fast_streams.rs` and is backed by the lower
 level transport code in `src/stream_transport.rs`.
 
+## Runtime Model
+
+Today the runtime is hybrid rather than fully single-threaded:
+
+- the loop coordination thread is always the central scheduler
+- on Linux, `add_reader` / `add_writer`, plain socket reads, and non-TLS socket
+  accept loops use the `compio` runtime on that thread
+- some transport paths still fall back to helper threads, especially TLS I/O,
+  TLS server accept, and parts of the legacy transport write path
+
+That means the codebase has started the move toward a single-runtime-thread I/O
+model, but has not finished eliminating every helper thread yet.
+
 ## Current Limitations
 
 These gaps are visible in the current implementation.
@@ -163,7 +178,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
-  stdlib helpers whenever `ssl` is enabled.
+  stdlib helpers whenever `ssl` is enabled. TLS transport internals also still
+  use helper-thread paths instead of the newer runtime-thread `compio` socket
+  path.
 - Subprocess support is intentionally incomplete:
   `preexec_fn` is unsupported, and text mode is rejected for
   `asyncio.create_subprocess_exec()` /
@@ -178,6 +195,10 @@ These gaps are visible in the current implementation.
   Unix socket APIs and Unix signal handlers remain Unix-only, and several
   subprocess options such as `pass_fds`, `user`, `group`, and `umask` are
   still specific to Unix process spawning.
+- The transport runtime model is still in transition:
+  plain socket reads and non-TLS accepts now run on the loop runtime thread on
+  Linux, but writes and TLS-heavy paths are not fully collapsed onto that same
+  single-threaded I/O path yet.
 
 ## Build
 

README.md

@@ -1,3 +1,4 @@
 fn main() {
+    pyo3_build_config::use_pyo3_cfgs();
     pyo3_build_config::add_extension_module_link_args();
 }

build.rs

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "rsloop"
-version = "0.1.4"
+version = "0.1.5"
 description = "An event loop for asyncio written in Rust"
 readme = "README.md"
 license = { file = "LICENSE" }

pyproject.toml

@@ -65,7 +65,6 @@ pub fn run_in_context(
     needs_run: bool,
     callback: &Py<PyAny>,
     args: &Py<PyTuple>,
-    _context_args: &Py<PyTuple>,
 ) -> PyResult<Py<PyAny>> {
     if !needs_run {
         return callback.call1(py, args.clone_ref(py));
@@ -89,18 +88,6 @@ pub fn run_in_context(
     }
 }
 
-pub fn build_context_args(
-    py: Python<'_>,
-    callback: &Py<PyAny>,
-    args: &Py<PyTuple>,
-) -> PyResult<Py<PyTuple>> {
-    let args_bound = args.bind(py);
-    let mut run_args = Vec::with_capacity(args_bound.len() + 1);
-    run_args.push(callback.clone_ref(py));
-    run_args.extend(args_bound.iter().map(|item| item.unbind()));
-    Ok(PyTuple::new(py, run_args)?.unbind())
-}
-
 #[inline]
 pub fn ensure_running_loop(py: Python<'_>, loop_obj: &Py<PyAny>) -> PyResult<()> {
     set_running_loop_fn(py)?.call1(py, (loop_obj.clone_ref(py),))?;

src/context.rs

@@ -106,11 +106,11 @@ impl PyFastStreamReader {
         value: Py<PyAny>,
     ) -> PyResult<()> {
         let future = future.bind(py);
-        match future.call_method1(python_names::set_result(py), (value,)) {
+        match python_names::call_method1(py, future, python_names::set_result(py), value.bind(py)) {
             Ok(_) => Ok(()),
             Err(err) => {
-                if future
-                    .call_method0(python_names::cancelled(py))?
+                if python_names::call_method0(py, future, python_names::cancelled(py))?
+                    .bind(py)
                     .extract::<bool>()?
                 {
                     Ok(())
@@ -127,11 +127,12 @@ impl PyFastStreamReader {
         exc: Py<PyAny>,
     ) -> PyResult<()> {
         let future = future.bind(py);
-        match future.call_method1(python_names::set_exception(py), (exc,)) {
+        match python_names::call_method1(py, future, python_names::set_exception(py), exc.bind(py))
+        {
             Ok(_) => Ok(()),
             Err(err) => {
-                if future
-                    .call_method0(python_names::cancelled(py))?
+                if python_names::call_method0(py, future, python_names::cancelled(py))?
+                    .bind(py)
                     .extract::<bool>()?
                 {
                     Ok(())
@@ -160,25 +161,28 @@ impl PyFastStreamReader {
     }
 
     fn create_future(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
-        self.loop_obj
-            .bind(py)
-            .call_method0(python_names::create_future(py))
-            .map(Bound::unbind)
+        python_names::call_method0(py, self.loop_obj.bind(py), python_names::create_future(py))
     }
 
     fn ready_result_future(&self, py: Python<'_>, value: Py<PyAny>) -> PyResult<Py<PyAny>> {
         let future = self.create_future(py)?;
-        future
-            .bind(py)
-            .call_method1(python_names::set_result(py), (value,))?;
+        python_names::call_method1(
+            py,
+            future.bind(py),
+            python_names::set_result(py),
+            value.bind(py),
+        )?;
         Ok(future)
     }
 
     fn ready_exception_future(&self, py: Python<'_>, exc: Py<PyAny>) -> PyResult<Py<PyAny>> {
         let future = self.create_future(py)?;
-        future
-            .bind(py)
-            .call_method1(python_names::set_exception(py), (exc,))?;
+        python_names::call_method1(
+            py,
+            future.bind(py),
+            python_names::set_exception(py),
+            exc.bind(py),
+        )?;
         Ok(future)
     }
 
@@ -214,21 +218,23 @@ impl PyFastStreamReader {
     fn maybe_resume_transport(&mut self, py: Python<'_>) -> PyResult<()> {
         if self.paused && self.buffer.len() <= self.limit && !self.transport.bind(py).is_none() {
             self.paused = false;
-            self.transport
-                .bind(py)
-                .call_method0(python_names::resume_reading(py))?;
+            python_names::call_method0(
+                py,
+                self.transport.bind(py),
+                python_names::resume_reading(py),
+            )?;
         }
         Ok(())
     }
 
     fn maybe_pause_transport(&mut self, py: Python<'_>) -> PyResult<()> {
         if !self.transport.bind(py).is_none() && !self.paused && self.buffer.len() > 2 * self.limit
         {
-            match self
-                .transport
-                .bind(py)
-                .call_method0(python_names::pause_reading(py))
-            {
+            match python_names::call_method0(
+                py,
+                self.transport.bind(py),
+                python_names::pause_reading(py),
+            ) {
                 Ok(_) => {
                     self.paused = true;
                 }
@@ -308,9 +314,11 @@ impl PyFastStreamReader {
         }
         if self.paused && !self.transport.bind(py).is_none() {
             self.paused = false;
-            self.transport
-                .bind(py)
-                .call_method0(python_names::resume_reading(py))?;
+            python_names::call_method0(
+                py,
+                self.transport.bind(py),
+                python_names::resume_reading(py),
+            )?;
         }
         let future = self.create_future(py)?;
         self.waiter = Some(ReadWaiter {
@@ -525,14 +533,16 @@ impl PyFastStreamProtocol {
         reader: Py<PyFastStreamReader>,
         client_connected_cb: Py<PyAny>,
     ) -> PyResult<Self> {
-        let closed = loop_obj.call_method0(py, "create_future")?;
-        let ready_none = loop_obj
-            .bind(py)
-            .call_method0(python_names::create_future(py))?
-            .unbind();
-        ready_none
-            .bind(py)
-            .call_method1(python_names::set_result(py), (py.None(),))?;
+        let closed =
+            python_names::call_method0(py, loop_obj.bind(py), python_names::create_future(py))?;
+        let ready_none =
+            python_names::call_method0(py, loop_obj.bind(py), python_names::create_future(py))?;
+        python_names::call_method1(
+            py,
+            ready_none.bind(py),
+            python_names::set_result(py),
+            py.None().bind(py),
+        )?;
         Ok(Self {
             closed,
             ready_none,
@@ -560,42 +570,55 @@ impl PyFastStreamProtocol {
     }
 
     fn ready_exception_future(&self, py: Python<'_>, exc: Py<PyAny>) -> PyResult<Py<PyAny>> {
-        let future = self
-            .loop_obj
-            .bind(py)
-            .call_method0(python_names::create_future(py))?
-            .unbind();
-        future
-            .bind(py)
-            .call_method1(python_names::set_exception(py), (exc,))?;
+        let future = python_names::call_method0(
+            py,
+            self.loop_obj.bind(py),
+            python_names::create_future(py),
+        )?;
+        python_names::call_method1(
+            py,
+            future.bind(py),
+            python_names::set_exception(py),
+            exc.bind(py),
+        )?;
         Ok(future)
     }
 
     fn push_drain_waiter(&mut self, py: Python<'_>) -> PyResult<Py<PyAny>> {
-        let future = self
-            .loop_obj
-            .bind(py)
-            .call_method0(python_names::create_future(py))?
-            .unbind();
+        let future = python_names::call_method0(
+            py,
+            self.loop_obj.bind(py),
+            python_names::create_future(py),
+        )?;
         self.drain_waiters.push(future.clone_ref(py));
         Ok(future)
     }
 
     fn resolve_drain_waiters(&mut self, py: Python<'_>, exc: Option<Py<PyAny>>) -> PyResult<()> {
         for future in self.drain_waiters.drain(..) {
             let future = future.bind(py);
-            if future
-                .call_method0(python_names::done(py))?
+            if python_names::call_method0(py, future, python_names::done(py))?
+                .bind(py)
                 .extract::<bool>()?
             {
                 continue;
             }
             match exc.as_ref() {
                 Some(exc) => {
-                    future.call_method1(python_names::set_exception(py), (exc.clone_ref(py),))?;
+                    python_names::call_method1(
+                        py,
+                        future,
+                        python_names::set_exception(py),
+                        exc.bind(py),
+                    )?;
                 }
                 None => {
-                    future.call_method1(python_names::set_result(py), (py.None(),))?;
+                    python_names::call_method1(
+                        py,
+                        future,
+                        python_names::set_result(py),
+                        py.None().bind(py),
+                    )?;
                 }
             }
         }
@@ -690,16 +713,31 @@ impl PyFastStreamProtocol {
                 self.reader
                     .borrow_mut(py)
                     .set_exception_internal(py, exc.clone_ref(py))?;
-                if !self.closed.call_method0(py, "done")?.extract::<bool>(py)? {
-                    self.closed
-                        .call_method1(py, "set_exception", (exc.clone_ref(py),))?;
+                if !python_names::call_method0(py, self.closed.bind(py), python_names::done(py))?
+                    .bind(py)
+                    .extract::<bool>()?
+                {
+                    python_names::call_method1(
+                        py,
+                        self.closed.bind(py),
+                        python_names::set_exception(py),
+                        exc.bind(py),
+                    )?;
                 }
                 self.resolve_drain_waiters(py, Some(exc))?;
             }
             None => {
                 self.reader.borrow_mut(py).feed_eof_internal(py)?;
-                if !self.closed.call_method0(py, "done")?.extract::<bool>(py)? {
-                    self.closed.call_method1(py, "set_result", (py.None(),))?;
+                if !python_names::call_method0(py, self.closed.bind(py), python_names::done(py))?
+                    .bind(py)
+                    .extract::<bool>()?
+                {
+                    python_names::call_method1(
+                        py,
+                        self.closed.bind(py),
+                        python_names::set_result(py),
+                        py.None().bind(py),
+                    )?;
                 }
                 self.resolve_drain_waiters(py, None)?;
             }