Repository: python-trio/trio-asyncio (language: python)

Issue:
TrioExecutor double-acquires the limiter, which can deadlock
It makes its own acquire() call and then calls `to_thread.run_sync` which will acquire() again. If many tasks are submitted to the executor near-simultaneously, then they might all complete the first acquire() and then all block waiting for the second one to succeed, producing deadlock.
TrioExecutor double-acquires the limiter, which can deadlock
It makes its own acquire() call and then calls `to_thread.run_sync` which will acquire() again. If many tasks are submitted to the executor near-simultaneously, then they might all complete the first acquire() and then all block waiting for the second one to succeed, producing deadlock.


Discussion / hints:

Fix patch:
diff --git a/newsfragments/143.bugfix.rst b/newsfragments/143.bugfix.rst
new file mode 100644
index 00000000..f7ddabec
--- /dev/null
+++ b/newsfragments/143.bugfix.rst
@@ -0,0 +1,3 @@
+``TrioExecutor.submit``, called from :meth:`asyncio.loop.run_in_executor`, no longer acquires a token from its `~trio.CapacityLimiter` before calling `~trio.to_thread.run_sync` (which already does its own ``acquire()``).
+
+The previous behaviour led to a double-acquire, leading each worker thread to require two tokens to run instead of one. Tasks could get stuck having acquired the first token but unable to acquire the second as part of `~trio.to_thread.run_sync`, leading to a deadlock.
\ No newline at end of file
diff --git a/trio_asyncio/_base.py b/trio_asyncio/_base.py
index a20603d2..a30112aa 100644
--- a/trio_asyncio/_base.py
+++ b/trio_asyncio/_base.py
@@ -84,14 +84,8 @@ def __init__(self, limiter=None, thread_name_prefix=None, max_workers=None):
     async def submit(self, func, *args):
         if not self._running:  # pragma: no cover
             raise RuntimeError("Executor is down")
-        lim = self._limiter
-        if lim is not None:
-            await lim.acquire()
-        try:
-            return await trio.to_thread.run_sync(func, *args, limiter=self._limiter)
-        finally:
-            if lim is not None:
-                lim.release()
+        # there is no need to call self._limiter.acquire() here, run_sync does it
+        return await trio.to_thread.run_sync(func, *args, limiter=self._limiter)
 
     def shutdown(self, wait=None):
         self._running = False


Test patch:
diff --git a/tests/test_trio_asyncio.py b/tests/test_trio_asyncio.py
index 45ed2f45..5afb17a5 100644
--- a/tests/test_trio_asyncio.py
+++ b/tests/test_trio_asyncio.py
@@ -176,3 +176,18 @@ async def unshield_later():
     ]
     assert trio.current_time() == 1.5 + (shield * 0.5)
     assert scope.cancelled_caught == (not shield)
+
+
+@pytest.mark.trio
+async def test_executor_limiter_deadlock():
+    def noop():
+        pass
+
+    # capacity of 1 to catch a double-acquire
+    limiter = trio.CapacityLimiter(1)
+    executor = trio_asyncio.TrioExecutor(limiter=limiter)
+    async with trio_asyncio.open_loop() as loop:
+        with trio.move_on_after(1) as scope:
+            await trio_asyncio.aio_as_trio(loop.run_in_executor)(executor, noop)
+
+    assert not scope.cancelled_caught