fix: a test that only held locally, and a cap gevent should never have had

CI caught the first one on every Linux runner: the skewed-version test
spelled an "impossible" coordinator version, 0.1.2, and a wheel built from
a checkout without tags is 0.1.dev1 -- which is exactly what CI installs.
The versions matched, the worker started and served, and the test blocked
on its output until it timed out.  The skew is derived from the real
version now, the way the unit tests around it already did.

The second was mine too, in the message I had just written: it points at
profile=gevent as the way past the exec cap, and profile=gevent was capped
right along with thread.  Waiting for an exec that runs elsewhere -- the
main thread, a greenlet -- parked a pool thread on a threading.Event, so
execs that cost no thread each held one anyway, and a gevent worker was
rationed to 20 concurrent greenlets.  That is a cap on the one thing the
profile is for.  The wait is a trio.Event woken from the exec's own thread
now, and capacity is the strategy's to declare: None where execs are tasks
or greenlets, half the thread budget where they are threads.

Also drops the last of the review's stale wording: Channel.send has not
blocked on a full queue since 2.1's write lock, and says what it does do.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: RonnyPfannschmidt

CHANGELOG.rst

@@ -67,8 +67,15 @@ series, once the consumers that need them have released without them.
   Admitting it instead made request 41 wait for a slot only a finishing exec
   could free, which from the coordinator is indistinguishable from a hung
   ``remote_exec``. ``remote_status()`` gained ``execcapacity`` (``None``
-  under ``profile=trio``, whose execs are tasks and are not bounded this
-  way), and ``numexecuting`` now counts what is really running.
+  under ``profile=trio`` and ``profile=gevent``, whose execs are tasks and
+  greenlets and are not bounded this way), and ``numexecuting`` now counts
+  what is really running.
+* **A ``profile=gevent`` worker is no longer limited to as many concurrent
+  execs as it has threads.** Waiting for a greenlet to finish parked a pool
+  thread on a ``threading.Event``, so execs that cost no thread each held
+  one anyway -- capping exactly the concurrency the profile exists to
+  provide. The wait is a ``trio.Event`` woken from the exec's own thread
+  now; same for the main-thread exec under ``profile=thread``.
 * **An exec that finishes after its connection died no longer takes the
   worker down.** Closing the channel is how an exec reports it finished, and
   a connection that went away first makes that raise; the exception reached

HANDOFF.md

@@ -11,7 +11,7 @@ This is the doc to read first.  Two companions:
 ## How to work here
 
 ```
-uv run pytest testing/          # 582 passed, 66 skipped
+uv run pytest testing/          # 583 passed, 66 skipped
 uv run pytest testing/ -n 12    # must stay green (~8s)
 uv run pre-commit run -a        # never grep-filter its output
 uv run tox -e docs              # sphinx -W, then doctests all of doc/
@@ -119,9 +119,12 @@ request rather than letting a gateway hang.
 
 `TrioWorkerExec` is a FIFO admission pump delegating to strategy objects
 (`WORKER_EXEC_STRATEGIES`); subinterpreters are a future strategy slot, not
-built.  Admission is **bounded** (`exec_capacity()`, half the trio thread
-limiter) and a request over the line is refused on its channel, not
-queued — reported as `remote_status().execcapacity`.
+built.  Admission is **bounded** for the thread-shaped strategies
+(`exec_capacity()`, half the trio thread limiter) and a request over the
+line is refused on its channel, not queued — reported as
+`remote_status().execcapacity`, `None` where execs are tasks or greenlets
+and cost no thread.  Nothing may wait for an exec by parking a pool
+thread: that spends the budget it is rationing.
 `AsyncGroup.makegateway` defaults workers to `thread` — the coordinator's
 shape does not dictate the worker's.
 

doc/basics.rst

@@ -301,18 +301,19 @@ How many at once
 
 .. versionadded:: 3.0
 
-``thread`` and ``gevent`` execs each need a thread of the worker's thread
-budget, which also has to serve channel callbacks and the worker's own
-protocol work -- so a worker admits **half that budget** in concurrent
+Under ``thread`` each exec needs a thread of the worker's thread budget,
+which also has to serve channel callbacks and the worker's own protocol
+work -- so a worker admits **half that budget** in concurrent
 ``remote_exec`` calls (20, unless the worker changed trio's default
 limiter) and *refuses* the one after that with a ``RemoteError`` naming the
 limit.  ``remote_status().execcapacity`` reports the number.
 
 Refusing rather than queueing is deliberate: a request waiting for a thread
 that only a finishing exec can free is indistinguishable, from the
 coordinator, from an exec that hung.  For genuine fan-out use more gateways
--- that is what a ``Group`` is for -- or ``profile=trio``, whose execs are
-tasks on the worker's loop and are not bounded this way.
+-- that is what a ``Group`` is for -- or a profile whose execs are not
+threads.  ``trio`` and ``gevent`` are unbounded here (``execcapacity`` is
+``None``): their execs are tasks and greenlets, and spend no thread.
 
 
 Transports

src/execnet/_channel.py

@@ -327,12 +327,17 @@ def waitclose(self, timeout: float | None = None) -> None:
             raise error
 
     def send(self, item: object) -> None:
-        """Sends the given item to the other side of the channel,
-        possibly blocking if the sender queue is full.
+        """Sends the given item to the other side of the channel.
 
         The item must be a simple Python type and will be
         copied to the other side by value.
 
+        Returns once the data has reached the OS write, which is not the
+        same as the peer having read it: there is no flow control, so a
+        peer that never receives buffers everything sent to it rather than
+        pushing back.  Sending unboundedly to one is a memory leak in *its*
+        process.
+
         OSError is raised if the write pipe was prematurely closed.
         """
         # before the state check: an unusable gateway (inherited by a fork,

src/execnet/_trio_worker.py

@@ -47,6 +47,8 @@ class PoolExec:
 
     #: whether _run_worker must hand this strategy the process main thread
     needs_primary_thread = False
+    #: whether each concurrent exec occupies a thread of the worker's budget
+    exec_costs_a_thread = True
 
     def __init__(self, gateway: WorkerGateway) -> None:
         self.gateway = gateway
@@ -69,6 +71,28 @@ def trigger_shutdown(self) -> None:
         pass
 
 
+def _thread_signal() -> tuple[trio.Event, Callable[[], None]]:
+    """A loop-side event plus the callable that sets it from a foreign thread.
+
+    Waiting for an exec that runs *elsewhere* -- the main thread, a greenlet
+    -- must not park a pool thread on a ``threading.Event``: that thread is
+    part of the budget exec placement is rationed against, so an exec that
+    costs no thread would still spend one waiting for itself.
+
+    Must be built on the loop (it captures the trio token).
+    """
+    done = trio.Event()
+    token = trio.lowlevel.current_trio_token()
+
+    def signal() -> None:
+        # posted callbacks must not raise, and a loop that ended while the
+        # exec ran leaves nobody to wake
+        with suppress(trio.RunFinishedError):
+            token.run_sync_soon(done.set)
+
+    return done, signal
+
+
 class PrimaryThreadPump:
     """Runs exec requests handed to it on the process main thread.
 
@@ -79,10 +103,14 @@ class PrimaryThreadPump:
 
     #: whether _run_worker must hand this strategy the process main thread
     needs_primary_thread = True
+    #: whether each concurrent exec occupies a thread of the worker's budget
+    #: (see TrioWorkerExec.capacity): the primary one does not, but the
+    #: overflow HybridExec sends to the pool does
+    exec_costs_a_thread = True
 
     def __init__(self, gateway: WorkerGateway) -> None:
         self.gateway = gateway
-        self._primary: Mailbox[tuple[Channel, ExecItem, threading.Event] | None] = (
+        self._primary: Mailbox[tuple[Channel, ExecItem, Callable[[], None]] | None] = (
             Mailbox()
         )
 
@@ -91,9 +119,9 @@ async def admit(self, channel: Channel, item: ExecItem) -> bool:
         return True
 
     async def run(self, channel: Channel, item: ExecItem) -> None:
-        done = threading.Event()
-        self._primary.put((channel, item, done))
-        await trio.to_thread.run_sync(done.wait, abandon_on_cancel=True)
+        done, signal = _thread_signal()
+        self._primary.put((channel, item, signal))
+        await done.wait()
 
     def released(self) -> None:
         """Hook: the main thread is free again (called on it, before done)."""
@@ -104,14 +132,14 @@ def integrate_as_primary_thread(self) -> None:
             task = self._primary.get()
             if task is None:
                 break
-            channel, item, done = task
+            channel, item, signal = task
             try:
                 self.gateway.executetask((channel, item))
             finally:
                 # Release before signalling: the next request should see the
                 # main thread free as early as we can make it.
                 self.released()
-                done.set()
+                signal()
 
     def trigger_shutdown(self) -> None:
         self._primary.put(None)
@@ -186,34 +214,39 @@ class GreenletExec:
     """
 
     needs_primary_thread = True
+    #: greenlets, not threads: concurrent execs here cost the thread budget
+    #: nothing, so they are not rationed against it
+    exec_costs_a_thread = False
 
     def __init__(self, gateway: WorkerGateway) -> None:
         from ._boundary import make_wakener
 
         self.gateway = gateway
         # The integrate loop blocks in get() on the hub thread: a gevent
         # wakener parks only its root greenlet, letting exec greenlets run.
-        self._primary: Mailbox[tuple[Channel, ExecItem, threading.Event] | None] = (
+        self._primary: Mailbox[tuple[Channel, ExecItem, Callable[[], None]] | None] = (
             Mailbox(make_wakener("gevent"))
         )
 
     async def admit(self, channel: Channel, item: ExecItem) -> bool:
         return True
 
     async def run(self, channel: Channel, item: ExecItem) -> None:
-        done = threading.Event()
-        self._primary.put((channel, item, done))
-        await trio.to_thread.run_sync(done.wait, abandon_on_cancel=True)
+        done, signal = _thread_signal()
+        self._primary.put((channel, item, signal))
+        await done.wait()
 
     def integrate_as_primary_thread(self) -> None:
         """Run the hub on the main thread, spawning a greenlet per exec."""
         import gevent
 
-        def run_exec(channel: Channel, item: ExecItem, done: threading.Event) -> None:
+        def run_exec(
+            channel: Channel, item: ExecItem, signal: Callable[[], None]
+        ) -> None:
             try:
                 self.gateway.executetask((channel, item))
             finally:
-                done.set()
+                signal()
 
         while True:
             task = self._primary.get()
@@ -373,14 +406,19 @@ def active_count(self) -> int:
         with self._lock:
             return self._running
 
-    def capacity(self) -> int:
-        """Concurrent execs this worker admits (loop thread only).
+    def capacity(self) -> int | None:
+        """Concurrent execs this worker admits, or None for unbounded.
 
-        Resolved on first use rather than in ``__init__``, which runs before
-        there is a loop to read the thread limiter from.  Reported by
-        ``remote_status()`` as ``execcapacity``, because a coordinator that
-        just had a request refused wants the number it hit.
+        Only the thread-shaped strategies are rationed: a greenlet exec
+        spends no thread, so bounding it would cap the very thing
+        ``profile=gevent`` exists to provide.  Resolved on first use rather
+        than in ``__init__``, which runs before there is a loop to read the
+        thread limiter from.  Reported by ``remote_status()`` as
+        ``execcapacity``, because a coordinator that just had a request
+        refused wants the number it hit.
         """
+        if not self.strategy.exec_costs_a_thread:
+            return None
         if self._capacity is None:
             self._capacity = exec_capacity()
         return self._capacity
@@ -403,7 +441,7 @@ def schedule(self, channel: Channel, sourcetask: bytes) -> None:
             if self._shutting_down:
                 channel.close("execution disallowed")
                 return
-            if self._running >= capacity:
+            if capacity is not None and self._running >= capacity:
                 full = True
             else:
                 full = False
@@ -416,7 +454,7 @@ def schedule(self, channel: Channel, sourcetask: bytes) -> None:
                 " concurrency limit (half its thread budget; the rest serves"
                 " channel callbacks and protocol work). Use more gateways, or"
                 " a profile whose execs are not threads (profile=trio,"
-                " profile=gevent)."
+                " profile=gevent), which are not bounded this way."
             )
             return
         # Already on the Trio host thread (Message handler).

testing/test_cli.py

@@ -132,6 +132,15 @@ def test_the_override_also_comes_from_the_config_env(self, capfd) -> None:
 
     @posix_only
     def test_a_skewed_worker_exits_with_a_reason(self) -> None:
+        from execnet import _trio_worker
+
+        # derived, never spelled out: a literal "impossible" version is only
+        # impossible until an environment has it.  A wheel built from a
+        # checkout without tags is 0.1.dev1, which is what CI installs -- so
+        # a hardcoded 0.1.2 matched there, the worker started, and this
+        # blocked on its output until the test timed out.
+        major, _ = _trio_worker._rough_version(execnet.__version__)
+        skewed = f"{major + 1}.0.0"
         ours, theirs = socket.socketpair()
         out = subprocess.run(
             [
@@ -142,7 +151,7 @@ def test_a_skewed_worker_exits_with_a_reason(self) -> None:
                 "--protocol-fd",
                 str(theirs.fileno()),
                 "--config",
-                worker_config(coordinator_version="0.1.2"),
+                worker_config(coordinator_version=skewed),
             ],
             pass_fds=(theirs.fileno(),),
             capture_output=True,

testing/test_gevent.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import threading
+from contextlib import suppress
 
 import pytest
 
@@ -179,6 +180,35 @@ def test_execs_cooperate_via_gevent(self, worker_gw) -> None:
     def test_status_reports_gevent(self, worker_gw) -> None:
         assert worker_gw.remote_status().execmodel == "gevent"
 
+    def test_execs_are_not_rationed_against_the_thread_budget(self, worker_gw) -> None:
+        """Greenlets cost no thread, so nothing caps them at the thread limit.
+
+        The exec-admission bound exists because a thread-shaped exec spends
+        a thread the callbacks and protocol work also need.  A greenlet
+        spends none -- but waiting for one used to park a pool thread, so a
+        gevent worker was rationed to 20 concurrent execs, which is a cap on
+        exactly what the profile is for.
+        """
+        import trio
+
+        from execnet import _trio_worker
+
+        async def thread_bound_capacity() -> int:
+            return _trio_worker.exec_capacity()
+
+        assert worker_gw.remote_status().execcapacity is None
+        wanted = trio.run(thread_bound_capacity) + 5
+        channels = [
+            worker_gw.remote_exec("channel.send('go'); channel.receive()")
+            for _ in range(wanted)
+        ]
+        try:
+            assert [ch.receive(TESTTIMEOUT) for ch in channels] == ["go"] * wanted
+        finally:
+            for ch in channels:
+                with suppress(OSError):
+                    ch.send(None)
+
 
 def test_provisioning_adds_gevent_requirement() -> None:
     from execnet import XSpec