Merge pull request #47 from MHFU-ModKit/fix/client-close-race

fix(ppsspp-debug): Client.close() is safe from several threads

Author: sp00ktober

packages/ppsspp-debug/src/ppsspp_debug/_sync.py

@@ -5,7 +5,7 @@
 import functools
 import threading
 from collections.abc import Callable, Coroutine, Iterator
-from contextlib import AbstractAsyncContextManager, AbstractContextManager, contextmanager
+from contextlib import AbstractAsyncContextManager, AbstractContextManager, contextmanager, suppress
 from types import TracebackType
 from typing import Any, Concatenate, Generic, ParamSpec, TypeVar
 
@@ -27,15 +27,18 @@ def __init__(self) -> None:
         )
         self._thread.start()
         self._stopping = False
+        # orders scheduling against stop: nothing reaches the loop once it is stopping
+        self._lock = threading.Lock()
 
     def run(self, coro: Coroutine[Any, Any, R]) -> R:
         if threading.current_thread() is self._thread:
             coro.close()
             raise RuntimeError("a blocking call on the client's own loop would deadlock")
-        if self._stopping:
-            coro.close()
-            raise Disconnected("the client is closed")
-        future = asyncio.run_coroutine_threadsafe(coro, self._loop)
+        with self._lock:
+            if self._stopping:
+                coro.close()
+                raise Disconnected("the client is closed")
+            future = asyncio.run_coroutine_threadsafe(coro, self._loop)
         try:
             return future.result()
         except concurrent.futures.CancelledError:
@@ -51,16 +54,25 @@ def stopped(self) -> bool:
         return self._stopping
 
     def stop(self) -> None:
-        """Cancel what still runs, so a call waiting in another thread fails instead of hanging."""
-        self._stopping = True
+        """Cancel what still runs, so a call waiting in another thread fails instead of hanging.
+
+        Safe from any thread and more than once; every caller returns once the loop is down."""
 
         async def cancel_all() -> None:
             tasks = asyncio.all_tasks() - {asyncio.current_task()}
             for task in tasks:
                 task.cancel()
             await asyncio.gather(*tasks, return_exceptions=True)
 
-        asyncio.run_coroutine_threadsafe(cancel_all(), self._loop).result()
+        with self._lock:
+            cancelled = None
+            if not self._stopping:
+                self._stopping = True
+                cancelled = asyncio.run_coroutine_threadsafe(cancel_all(), self._loop)
+        if cancelled is None:  # another thread is stopping it
+            self._thread.join()
+            return
+        cancelled.result()
         self._loop.call_soon_threadsafe(self._loop.stop)
         self._thread.join()
         self._loop.close()
@@ -162,8 +174,9 @@ def connect(
     def close(self) -> None:
         """Close the connection; breakpoints stay armed in PPSSPP."""
         if not self._loop.stopped:
-            self._loop.run(self._async.close())
-            self._loop.stop()
+            with suppress(Disconnected):  # another thread closed it meanwhile
+                self._loop.run(self._async.close())
+        self._loop.stop()
 
     def __enter__(self) -> Client:
         return self

packages/ppsspp-debug/tests/test_sync.py

@@ -90,6 +90,27 @@ def wait():
         c.status()
 
 
+def test_close_from_many_threads(served):
+    errors: list[BaseException] = []
+
+    def close(c: Client) -> None:
+        try:
+            c.close()
+        except BaseException as e:
+            errors.append(e)
+
+    for _ in range(40):
+        c = Client.connect(port=served.fake.port)
+        closers = [threading.Thread(target=close, args=(c,), daemon=True) for _ in range(4)]
+        for t in closers:
+            t.start()
+        for t in closers:
+            t.join(2)
+        assert not any(t.is_alive() for t in closers)
+        assert c.closed
+    assert errors == []
+
+
 def test_every_async_method_is_mirrored():
     public = {n for n, v in vars(AsyncClient).items() if not n.startswith("_") and callable(v)}
     assert public <= set(vars(Client))