perf(client): cut per-request overhead

Cargo.toml

@@ -84,6 +84,7 @@ btls = { git = "https://github.com/0x676e67/btls", branch = "main" }
 btls-sys = { git = "https://github.com/0x676e67/btls", branch = "main" }
 tokio-btls = { git = "https://github.com/0x676e67/btls", branch = "main" }
 netty = { git = "https://github.com/0x676e67/netty", branch = "main" }
+http2 = { git = "https://github.com/0x676e67/http2", rev = "fb5bf8bdbf3e9d9aad7073759a6f06f47373342e" }
 
 [profile.release]
 codegen-units = 1

docs/source/guide/redirect-errors.md

@@ -107,6 +107,9 @@ reading its response body. `connect_timeout` limits connection establishment;
 `read_timeout` limits the wait for a read and resets after successful reads.
 Requests can override `timeout` and `read_timeout`.
 
+Waiting before the first response body read does not use up its `read_timeout`.
+An overall `timeout` still runs during that wait.
+
 ```python
 import asyncio
 from datetime import timedelta

docs/source/guide/runtime.md

@@ -20,6 +20,10 @@ Both worker schedulers support async and blocking clients. `CURRENT_THREAD`
 supports only `wreq.blocking.Client`; passing it to `wreq.Client` raises
 `ValueError`.
 
+With `WORK_STEALING`, blocking requests run on the calling thread while
+background workers handle connection I/O. `PER_WORKER` submits requests to the
+client's selected worker. Both keep processing connections between calls.
+
 Runtime worker count and request concurrency are different settings. For
 example, one async client can have many requests in flight with `workers=1`.
 With `PER_WORKER`, increasing the worker count distributes newly created clients
@@ -74,7 +78,8 @@ or all process resources.
 
 ## Use the calling thread for blocking I/O {#current-thread}
 
-`CURRENT_THREAD` avoids dispatching each request to a background I/O worker.
+`CURRENT_THREAD` runs connection I/O on the calling thread, without background
+I/O workers.
 Create the client and runtime together and reuse them for subsequent calls:
 
 ```python

src/client.rs

@@ -1,12 +1,14 @@
 pub mod body;
 pub mod req;
 pub mod resp;
+pub mod ws;
 
 mod param;
 mod query;
 
 use std::{
     net::{IpAddr, Ipv4Addr, Ipv6Addr},
+    sync::Arc,
     time::Duration,
 };
 
@@ -20,7 +22,8 @@ use wreq::tls::trust::CertStore;
 
 use self::{
     req::{execute_request, execute_websocket_request},
-    resp::{BlockingResponse, BlockingWebSocket, Response, WebSocket},
+    resp::{BlockingResponse, Response},
+    ws::{BlockingWebSocket, WebSocket},
 };
 use crate::{
     cookie::Jar,
@@ -211,17 +214,20 @@ struct Builder {
 /// Unknown keyword arguments raise `TypeError` at call time.
 #[derive(Clone)]
 #[pyclass(subclass, frozen)]
-pub struct Client {
+pub struct Client(Arc<ClientInner>);
+
+/// The state a [`Client`] shares with its requests. Each request task owns one clone, which
+/// keeps the runtime alive, so starting a request clones nothing else.
+pub struct ClientInner {
     inner: wreq::Client,
     /// Runs this client's requests and response reads, keeping the runtime alive.
     runtime: runtime::Runtime,
     /// Cancelled by `close()`; pending and new requests then fail with `CancelledError`.
     cancel: CancellationToken,
     /// Turn error statuses into exceptions for every request.
     raise_for_status: bool,
-
-    /// Get the cookie jar of the client.
-    #[pyo3(get)]
+    /// Prefetching must not start a configured read timeout before the caller reads.
+    has_read_timeout: bool,
     cookie_jar: Option<Jar>,
 }
 
@@ -256,10 +262,13 @@ impl Client {
         kwds: Option<Py<PyDict>>,
     ) -> PyResult<Response> {
         let kwds = Python::attach(|py| kwds.map(|kwds| kwds.bind(py).extract()).transpose())?;
-        self.runtime
-            .clone()
-            .run_task(execute_request(self, method, url, kwds))
-            .await
+        let Self(client) = self;
+        let task = client
+            .runtime
+            .spawn(execute_request(client.clone(), method, url, kwds));
+        // Release ours before waiting, leaving the task's clone as the only extra count.
+        drop(client);
+        runtime::Runtime::join(task).await
     }
 
     /// Open a WebSocket on the client's runtime, extracting options on first await.
@@ -269,23 +278,27 @@ impl Client {
         kwds: Option<Py<PyDict>>,
     ) -> PyResult<WebSocket> {
         let kwds = Python::attach(|py| kwds.map(|kwds| kwds.bind(py).extract()).transpose())?;
-        self.runtime
-            .clone()
-            .run_task(execute_websocket_request(self, url, kwds))
-            .await
+        let Self(client) = self;
+        let task = client
+            .runtime
+            .spawn(execute_websocket_request(client.clone(), url, kwds));
+        // Released before waiting, as in `execute`.
+        drop(client);
+        runtime::Runtime::join(task).await
     }
 }
 
 impl Default for Client {
     fn default() -> Self {
         let runtime = runtime::Runtime::shared();
-        Self {
+        Self(Arc::new(ClientInner {
             inner: wreq::Client::default(),
             runtime: runtime.clone(),
             cancel: CancellationToken::new(),
             raise_for_status: false,
+            has_read_timeout: false,
             cookie_jar: None,
-        }
+        }))
     }
 }
 
@@ -297,6 +310,9 @@ impl Client {
                 Some(runtime) => runtime.select()?,
                 None => runtime::Runtime::shared().clone(),
             };
+            let has_read_timeout = kwds
+                .as_ref()
+                .is_some_and(|config| config.read_timeout.is_some());
             // Create the client builder.
             let mut builder = wreq::Client::builder();
             let mut cookie_jar: Option<Jar> = None;
@@ -504,12 +520,15 @@ impl Client {
 
             builder
                 .build()
-                .map(|inner| Client {
-                    inner,
-                    runtime,
-                    cancel: CancellationToken::new(),
-                    cookie_jar,
-                    raise_for_status,
+                .map(|inner| {
+                    Client(Arc::new(ClientInner {
+                        inner,
+                        runtime,
+                        cancel: CancellationToken::new(),
+                        cookie_jar,
+                        raise_for_status,
+                        has_read_timeout,
+                    }))
                 })
                 .map_err(Error::Library)
                 .map_err(Into::into)
@@ -539,13 +558,19 @@ impl Client {
     /// Cancel pending requests and reject new ones with asyncio.CancelledError.
     /// Existing responses, WebSockets and the shared runtime remain usable.
     pub fn close(&self) {
-        self.cancel.cancel();
+        self.0.cancel.cancel();
     }
 
     /// The runtime used by this client and its responses.
     #[getter]
     pub fn runtime(&self) -> runtime::Runtime {
-        self.runtime.clone()
+        self.0.runtime.clone()
+    }
+
+    /// Get the cookie jar of the client.
+    #[getter]
+    pub fn cookie_jar(&self) -> Option<Jar> {
+        self.0.cookie_jar.clone()
     }
 
     /// Make a GET request to the given URL.
@@ -677,7 +702,7 @@ impl Client {
         _exc_val: Py<PyAny>,
         _traceback: Py<PyAny>,
     ) -> PyResult<Bound<'py, Coroutine>> {
-        let cancel = self.cancel.clone();
+        let cancel = self.0.cancel.clone();
         coroutine::local(py, "Client.__aexit__", async move {
             cancel.cancel();
             Ok(())
@@ -705,7 +730,7 @@ impl BlockingClient {
     /// Get the cookie jar of the client.
     #[getter]
     pub fn cookie_jar(&self) -> Option<Jar> {
-        self.0.cookie_jar.clone()
+        self.0.cookie_jar()
     }
 
     /// Cancel pending requests and reject new ones with asyncio.CancelledError.
@@ -812,9 +837,10 @@ impl BlockingClient {
         url: PyBackedStr,
         kwds: Option<Request>,
     ) -> PyResult<BlockingResponse> {
-        self.0
+        let Client(client) = &self.0;
+        client
             .runtime
-            .block_on_task(py, execute_request(self.0.clone(), method, url, kwds))
+            .block_on_task(py, execute_request(client.clone(), method, url, kwds))
             .map(Into::into)
     }
 
@@ -827,9 +853,10 @@ impl BlockingClient {
         url: PyBackedStr,
         kwds: Option<WebSocketRequest>,
     ) -> PyResult<BlockingWebSocket> {
-        self.0
+        let Client(client) = &self.0;
+        client
             .runtime
-            .block_on_task(py, execute_websocket_request(self.0.clone(), url, kwds))
+            .block_on_task(py, execute_websocket_request(client.clone(), url, kwds))
             .map(Into::into)
     }
 }

src/client/body/stream/async_gen.rs

@@ -3,12 +3,17 @@
 use std::{
     ffi::CStr,
     pin::Pin,
-    sync::Mutex,
+    sync::{Mutex, OnceLock},
     task::{Context, Poll, ready},
 };
 
 use futures_util::{Stream, StreamExt};
-use pyo3::{IntoPyObjectExt, intern, prelude::*, sync::PyOnceLock};
+use pyo3::{
+    IntoPyObjectExt, intern,
+    prelude::*,
+    sync::PyOnceLock,
+    types::{PyCode, PyDict, PyType},
+};
 use tokio::sync::TryAcquireError;
 
 use super::{Item, PyBytesLike, lock, queue};
@@ -53,12 +58,9 @@ impl AsyncStream {
             tx: Mutex::new(Some(tx)),
         };
         let coroutine = forward.bind(py).call1((generator, sender))?;
-        // create_task captures the caller's contextvars on the running loop.
-        let task = event_loop
-            .call_method1(intern!(py, "create_task"), (&coroutine,))
-            .inspect_err(|_| {
-                let _ = coroutine.call_method0(intern!(py, "close"));
-            })?;
+        let task = start(&event_loop, &coroutine).inspect_err(|_| {
+            let _ = coroutine.call_method0(intern!(py, "close"));
+        })?;
         Ok(Self {
             rx,
             task: Some((task.unbind(), event_loop.unbind())),
@@ -153,6 +155,81 @@ impl Sender {
     }
 }
 
+/// Run `coroutine` as a task on `event_loop`, capturing the caller's contextvars.
+///
+/// A standard loop whose `create_task` passes extra keywords to `Task`, as on Python 3.14 and
+/// later 3.13 releases, receives an eager-start hint so chunks can be queued before the
+/// request is sent. Task creation always goes through the loop, including wrappers that
+/// capture context or instrument tasks.
+fn start<'py>(
+    event_loop: &Bound<'py, PyAny>,
+    coroutine: &Bound<'py, PyAny>,
+) -> PyResult<Bound<'py, PyAny>> {
+    let py = event_loop.py();
+    let create_task = event_loop.getattr(intern!(py, "create_task"))?;
+    if eager(event_loop, &create_task) {
+        let kwargs = PyDict::new(py);
+        kwargs.set_item(intern!(py, "eager_start"), true)?;
+        return create_task.call((coroutine,), Some(&kwargs));
+    }
+    create_task.call1((coroutine,))
+}
+
+/// Only pass the hint where the standard loop API supports it. Older wrappers and custom
+/// factories keep their usual arguments; the actual loop method still controls scheduling.
+fn eager(event_loop: &Bound<'_, PyAny>, create_task: &Bound<'_, PyAny>) -> bool {
+    static BASE: PyOnceLock<Py<PyType>> = PyOnceLock::new();
+    static VERSION: OnceLock<(bool, bool)> = OnceLock::new();
+    let py = event_loop.py();
+    let (supported, forwards_keywords) = *VERSION.get_or_init(|| {
+        let version = py.version_info();
+        (version >= (3, 12), version >= (3, 14))
+    });
+    if !supported {
+        return false;
+    }
+    let standard = || -> PyResult<bool> {
+        let base = BASE.import(py, "asyncio", "BaseEventLoop")?;
+        let name = intern!(py, "create_task");
+        let method = base.getattr(name)?;
+        // Python 3.14 always passes extra keywords to `Task`, 3.12 never does, and 3.13 only
+        // in later releases; there only the unwrapped standard method shows whether it does.
+        let forwards = forwards_keywords
+            || (method
+                .getattr(intern!(py, "__module__"))?
+                .eq("asyncio.base_events")?
+                && takes_keywords(&method)?);
+        Ok(forwards
+            && event_loop.is_instance(base)?
+            && event_loop.get_type().getattr(name)?.is(&method)
+            && event_loop
+                .call_method0(intern!(py, "get_task_factory"))?
+                .is_none()
+            // Don't send a new keyword to an old-signature wrapper.
+            && takes_keywords(create_task)?)
+    };
+    standard().unwrap_or(false)
+}
+
+/// Whether the Python function `func` accepts `**kwargs` (`inspect.CO_VARKEYWORDS`).
+fn takes_keywords(func: &Bound<'_, PyAny>) -> PyResult<bool> {
+    static CODE: PyOnceLock<(Py<PyAny>, bool)> = PyOnceLock::new();
+    let py = func.py();
+    let code = func.getattr(intern!(py, "__code__"))?;
+    if let Some((cached, keywords)) = CODE.get(py)
+        && code.is(cached)
+    {
+        return Ok(*keywords);
+    }
+    let keywords = code.getattr(intern!(py, "co_flags"))?.extract::<u32>()? & 0x08 != 0;
+    // Code objects are immutable; replacing a function or its code misses this cache.
+    // A custom callable can expose a mutable __code__ proxy, so never cache that.
+    if code.is_exact_instance_of::<PyCode>() {
+        let _ = CODE.set(py, (code.unbind(), keywords));
+    }
+    Ok(keywords)
+}
+
 /// The coroutine that forwards an async generator into its [`Sender`].
 const FORWARD_SOURCE: &CStr = c"import asyncio