feat(runtime)!: add Scheduler with CURRENT_THREAD and tune low-concurrency I/O

Cargo.toml

@@ -29,15 +29,15 @@ abi3-py314 = ["pyo3/abi3-py314"]
 
 [dependencies]
 pingora-runtime = "0.9.0"
-tokio = { version = "1.53.2", features = ["rt-multi-thread", "sync", "time", "net"] }
+tokio = { version = "1.53.2", features = ["rt-multi-thread", "sync", "time", "net", "macros"] }
 tokio-util = { version = "0.7.18", features = ["rt"] }
 pyo3 = { version = "0.29.3", features = [
     "indexmap",
     "multiple-pymethods",
     "extension-module",
     "generate-import-lib",
 ] }
-futures-util = { version = "0.3.33", default-features = false, features = ["std"] }
+futures-util = { version = "0.3.33", default-features = false, features = ["std", "sink"] }
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1"
 serde_urlencoded = "0.7.1"

bench/README.md

@@ -15,7 +15,7 @@ are reported separately.
 - Async clients: wreq default runtime, wreq single worker, pyreqwest
   single-threaded and multithreaded runtimes, ry, httpx, aiohttp, niquests, and
   curl_cffi.
-- Blocking variants: wreq MT/ST, ry, requests, httpx, niquests, curl_cffi, and pycurl.
+- Blocking variants: wreq MT/ST/CT, ry, requests, httpx, niquests, curl_cffi, and pycurl.
 - Three rounds, each with one warmup and one timed batch of 300 requests.
   Case order is shuffled with a recorded seed.
 - Status, negotiated HTTP version, and total response length are checked for
@@ -42,7 +42,7 @@ worker clients have separate connection pools.
 | 4 MiB | 256 KiB |
 
 Both Full and Stream uploads are tested for every payload and concurrency.
-The complete matrix contains 1,680 supported cells across 16 client variants;
+The complete matrix contains 1,904 supported cells across 18 client variants;
 each cell has three measured rounds. The Rust benchmark uses Criterion and
 600 requests per iteration; this suite uses fixed batches of 300 requests.
 Custom payload sizes use upload chunks of at most 64 KiB.
@@ -70,7 +70,7 @@ Cargo uses a custom target directory, pass the actual executable path to `--serv
 starts, it prints the number of cases and batches. It saves stdout/stderr logs
 beside the raw JSON and generates an English `.report.md` once validation passes.
 
-The default suite has 1,680 supported cases, 5,040 timed batches and the same
+The default suite has 1,904 supported cases, 5,712 timed batches and the same
 number of warm-up batches. Allow several hours for a full run. The setup commands
 above prepare the dependencies and native binaries; the runner won't install or
 build them for you. Publishing the results to the docs is optional.
@@ -163,7 +163,7 @@ To update only the blocking results without rerunning the async clients:
 uv run --no-sync python bench/run.py --input bench/data/BLOCKING.json --publish-blocking
 ```
 
-This requires all eight blocking variants and the same complete workload and
+This requires all nine blocking variants and the same complete workload and
 batch minimums. It freezes the candidate and existing `latest.json` for the docs
 build, then selects `bench/data/latest-blocking.json` only if the build succeeds.
 The async source and all historical files stay unchanged. The two publication
@@ -205,13 +205,17 @@ shows each run's revision and collection time separately.
 ## Blocking wreq runtimes
 
 `wreq_blocking` uses the default shared multi-thread network runtime (MT).
-`wreq_blocking_st` shares one `Runtime(workers=1, work_steal=False)` across
-all logical workers in each case (ST). Both retain one client per logical
-worker and the same Python thread pool for concurrent blocking calls.
-
-The default full run includes both async and blocking wreq MT/ST, for 17
-variants and 1,792 supported cells. No `--clients` option is needed. Historical
-published snapshots remain readable; new full/blocking publication requires ST.
+`wreq_blocking_st` shares one
+`Runtime(scheduler=Scheduler.PER_WORKER, workers=1)` across all logical workers
+in each case (ST). `wreq_blocking_ct` gives each logical worker its own
+`Runtime(scheduler=Scheduler.CURRENT_THREAD)`, driven by the calling thread
+(CT). All three retain one client per logical worker and the same Python thread
+pool for concurrent blocking calls.
+
+The default full run includes async wreq MT/ST and blocking wreq MT/ST/CT, for
+18 variants and 1,904 supported cells. No `--clients` option is needed.
+Historical published snapshots remain readable; new full/blocking publication
+requires ST and CT. CT needs a wreq release with `Scheduler`.
 
 ## Maintaining the suite
 

bench/blocking_clients.py

@@ -100,8 +100,18 @@ def operations(client_id, protocol, url, body, body_kind, *, chunks=None, runtim
         from wreq.blocking import Client
         from wreq.tls import TlsVersion
 
-        if SPECS[client_id].runtime["kind"] == "custom" and runtime is None:
+        kind = SPECS[client_id].runtime["kind"]
+        if kind == "custom" and runtime is None:
             raise ValueError("Blocking ST requires one runtime shared by the case")
+        if kind == "current_thread":
+            from wreq.runtime import Runtime
+
+            try:
+                from wreq.runtime import Scheduler
+            except ImportError as error:
+                raise ValueError("Blocking CT requires wreq with Scheduler") from error
+            # Each logical worker drives its own runtime, like one curl handle each.
+            runtime = Runtime(scheduler=Scheduler.CURRENT_THREAD)
         with Client(
             runtime=runtime,
             tls_verify=False,
@@ -117,8 +127,9 @@ def post():
                 response = client.post(
                     url, body=iter(chunks) if body_kind == "stream" else body
                 )
-                status, version = response.status.as_int(), native_protocol(
-                    response.version
+                status, version = (
+                    response.status.as_int(),
+                    native_protocol(response.version),
                 )
                 total = 0
                 with response.stream() as stream:

bench/charts.py

@@ -24,6 +24,7 @@
         "track": "#edf1f8",
         "wreq": "#4657d9",
         "wreq_st": "#0f8a76",
+        "wreq_ct": "#b45309",
         "peer": "#8393aa",
     },
     "dark": {
@@ -34,6 +35,7 @@
         "track": "#223047",
         "wreq": "#a0aaff",
         "wreq_st": "#55ccb3",
+        "wreq_ct": "#f6b14b",
         "peer": "#8b9db8",
     },
 }

bench/clients.py

@@ -112,9 +112,13 @@ def network_runtime(client):
         return None
     from wreq.runtime import Runtime
 
-    return Runtime(
-        workers=spec.runtime["workers"], work_steal=spec.runtime["work_steal"]
-    )
+    workers, scheduler = spec.runtime["workers"], spec.runtime["scheduler"]
+    try:
+        from wreq.runtime import Scheduler
+    except ImportError:
+        # Releases before Scheduler choose it with `work_steal`.
+        return Runtime(workers=workers, work_steal=scheduler == "WORK_STEALING")
+    return Runtime(scheduler=getattr(Scheduler, scheduler), workers=workers)
 
 
 @asynccontextmanager

bench/registry.py

@@ -59,7 +59,7 @@ def settings(self):
     "wreq_st": ClientSpec(
         "wreq",
         "wreq (ST)",
-        runtime={"kind": "custom", "workers": 1, "work_steal": False},
+        runtime={"kind": "custom", "workers": 1, "scheduler": "PER_WORKER"},
         color="wreq_st",
     ),
     "pyreqwest_st": ClientSpec(
@@ -117,11 +117,19 @@ def settings(self):
         runtime={
             "kind": "custom",
             "workers": 1,
-            "work_steal": False,
+            "scheduler": "PER_WORKER",
             "scope": "shared per case",
         },
         color="wreq_st",
     ),
+    "wreq_blocking_ct": ClientSpec(
+        "wreq",
+        "wreq (blocking CT)",
+        "blocking",
+        response_read="stream(): native transport chunks",
+        runtime={"kind": "current_thread", "scope": "one per client"},
+        color="wreq_ct",
+    ),
     "ry_blocking": ClientSpec(
         "ry",
         "ry (blocking)",

bench/report.py

@@ -42,8 +42,13 @@ def runtime_label(runtime: dict) -> str:
         return "Single-thread runtime"
     if kind == "multi_thread":
         return "Multi-thread runtime"
+    if kind == "current_thread":
+        return "Current-thread runtime per client"
     if kind == "custom":
         workers = runtime.get("workers", "not recorded")
+        if "scheduler" in runtime:
+            return f"Custom: workers={workers}, scheduler={runtime['scheduler']}"
+        # Snapshots recorded before Scheduler.
         steal = runtime.get("work_steal", "not recorded")
         return f"Custom: workers={workers}, work_steal={steal}"
     if kind == "thread_pool":

bench/results.py

@@ -44,11 +44,14 @@ def require_publish(config, *, blocking=False, allow_legacy_snapshot=False):
     """Require the complete default suite or its independent blocking matrix."""
     expected_clients = BLOCKING_CLIENTS if blocking else CLIENTS
     if allow_legacy_snapshot:
-        legacy_clients = tuple(
-            client for client in expected_clients if client != "wreq_blocking_st"
-        )
-        if set(config.get("clients", ())) == set(legacy_clients):
-            expected_clients = legacy_clients
+        # Snapshots from before blocking CT, and from before blocking ST as well.
+        for missing in ({"wreq_blocking_ct"}, {"wreq_blocking_st", "wreq_blocking_ct"}):
+            legacy_clients = tuple(
+                client for client in expected_clients if client not in missing
+            )
+            if set(config.get("clients", ())) == set(legacy_clients):
+                expected_clients = legacy_clients
+                break
     axes = {
         "clients": expected_clients,
         "protocols": ("h1", "h2"),

bench/tests/test_benchmark.py

@@ -7,7 +7,7 @@
 import json
 import time
 import sys
-from types import ModuleType
+from types import ModuleType, SimpleNamespace
 
 import pytest
 
@@ -370,12 +370,13 @@ def test_blocking_st_shares_one_network_runtime(monkeypatch):
     runtime_module = ModuleType("wreq.runtime")
 
     def create_runtime(**options):
-        assert options == {"workers": 1, "work_steal": False}
+        assert options == {"scheduler": "per-worker", "workers": 1}
         runtime = object()
         runtimes.append(runtime)
         return runtime
 
     runtime_module.Runtime = create_runtime
+    runtime_module.Scheduler = SimpleNamespace(PER_WORKER="per-worker")
     monkeypatch.setitem(sys.modules, "wreq.runtime", runtime_module)
 
     @contextmanager

bench/tests/test_run.py

@@ -44,7 +44,7 @@ def candidate(tmp_path, full=False, blocking=False):
 
 
 def test_separate_publication_gates_reject_subsets_and_weak_batches():
-    assert len(BLOCKING_CLIENTS) == 8 and len(CLIENTS) == 17
+    assert len(BLOCKING_CLIENTS) == 9 and len(CLIENTS) == 18
     for blocking in (False, True):
         config = publication_config(blocking)
         require_publish(config, blocking=blocking)
@@ -76,15 +76,23 @@ def test_separate_publication_gates_reject_subsets_and_weak_batches():
 
 def test_default_suite_includes_wreq_mt_st_and_reads_legacy_snapshots():
     args = benchmark.parse_args(["--server", "server"])
-    assert {"wreq", "wreq_st", "wreq_blocking", "wreq_blocking_st"} <= set(args.clients)
-    assert len(args.clients) == 17
+    assert {
+        "wreq",
+        "wreq_st",
+        "wreq_blocking",
+        "wreq_blocking_st",
+        "wreq_blocking_ct",
+    } <= set(args.clients)
+    assert len(args.clients) == 18
     assert args.requests == args.warmup_requests == 300
     for blocking in (False, True):
         legacy = publication_config(blocking)
-        legacy["clients"].remove("wreq_blocking_st")
-        require_publish(legacy, blocking=blocking, allow_legacy_snapshot=True)
-        with pytest.raises(ValueError):
-            require_publish(legacy, blocking=blocking)
+        # Snapshots from before CT, then from before ST as well.
+        for client in ("wreq_blocking_ct", "wreq_blocking_st"):
+            legacy["clients"].remove(client)
+            require_publish(legacy, blocking=blocking, allow_legacy_snapshot=True)
+            with pytest.raises(ValueError):
+                require_publish(legacy, blocking=blocking)
         legacy["clients"].remove("wreq_blocking")
         with pytest.raises(ValueError):
             require_publish(legacy, blocking=blocking, allow_legacy_snapshot=True)
@@ -216,7 +224,8 @@ def test_full_publish_freezes_existing_blocking_selection(
     latest.write_bytes(b"previous full selection")
     latest_blocking = tmp_path / "latest-blocking.json"
     legacy_configuration = publication_config(True)
-    legacy_configuration["clients"].remove("wreq_blocking_st")
+    for client in ("wreq_blocking_st", "wreq_blocking_ct"):
+        legacy_configuration["clients"].remove(client)
     blocking_raw = json.dumps(make_document(legacy_configuration), indent=3).encode()
     latest_blocking.write_bytes(blocking_raw)
     monkeypatch.setattr(run, "LATEST", latest)

clippy.toml

@@ -0,0 +1,11 @@
+# Runtime operations go through `crate::runtime::Runtime`, which keeps blocking calls,
+# current-thread driving and the shared runtime consistent.
+disallowed-methods = [
+    { path = "tokio::task::spawn", reason = "use Runtime::spawn" },
+    { path = "tokio::task::spawn_blocking", reason = "use Runtime::spawn_blocking" },
+    { path = "tokio::task::yield_now", reason = "use Runtime::defer_wake" },
+    { path = "tokio::task::block_in_place", reason = "use Runtime::spawn_blocking, and Runtime::block_on_in_pool inside it" },
+    { path = "tokio::runtime::Handle::current", reason = "use Runtime" },
+    { path = "tokio::runtime::Handle::try_current", reason = "use Runtime" },
+]
+disallowed-types = [{ path = "tokio::runtime::Handle", reason = "use Runtime" }]

docs/source/api/runtime.md

@@ -1,7 +1,11 @@
 # wreq.runtime
 
-Runtime configuration for asynchronous and blocking clients. `Runtime` is also
-available as `wreq.Runtime`.
+Runtime configuration for asynchronous and blocking clients. `Runtime` and
+`Scheduler` are also available as `wreq.Runtime` and `wreq.Scheduler`.
+
+::: wreq.runtime.Scheduler
+    options:
+      show_root_heading: true
 
 ::: wreq.runtime.Runtime
     options: