[tor-commits] [stem/master] Fix deadlock when stopping from an async context

[atagar at torproject.org]

commit a2e0da98669a7a173e25a6abe6e3e2e996be8e67
Author: Damian Johnson <atagar at torproject.org>
Date:   Sun Jul 12 17:01:18 2020 -0700

    Fix deadlock when stopping from an async context
    
    Reentrant locks can only be acquired multiple from within the same thread. When
    our _run_async_method() invoked start() or stop() we deadlocked.
---
 stem/util/__init__.py         |  6 ++++--
 test/unit/util/synchronous.py | 14 ++++++++++++++
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/stem/util/__init__.py b/stem/util/__init__.py
index c147a5a4..fbd844f8 100644
--- a/stem/util/__init__.py
+++ b/stem/util/__init__.py
@@ -350,9 +350,11 @@ class Synchronous(object):
         async def convert_generator(generator: AsyncIterator) -> Iterator:
           return iter([d async for d in generator])
 
-        return asyncio.run_coroutine_threadsafe(convert_generator(func(self, *args, **kwargs)), self._loop).result()
+        future = asyncio.run_coroutine_threadsafe(convert_generator(func(self, *args, **kwargs)), self._loop)
       else:
-        return asyncio.run_coroutine_threadsafe(func(self, *args, **kwargs), self._loop).result()
+        future = asyncio.run_coroutine_threadsafe(func(self, *args, **kwargs), self._loop)
+
+    return future.result()
 
   def __iter__(self) -> Iterator:
     return self._run_async_method('__aiter__')
diff --git a/test/unit/util/synchronous.py b/test/unit/util/synchronous.py
index d4429f3e..bbf91d18 100644
--- a/test/unit/util/synchronous.py
+++ b/test/unit/util/synchronous.py
@@ -118,6 +118,20 @@ class TestSynchronous(unittest.TestCase):
     sync_test()
     asyncio.run(async_test())
 
+  def test_stop_from_async(self):
+    """
+    Ensure we can stop our instance from within an async method without
+    deadlock.
+    """
+
+    class AsyncStop(Synchronous):
+      async def call_stop(self):
+        self.stop()
+
+    instance = AsyncStop()
+    instance.call_stop()
+    self.assertRaises(RuntimeError, instance.call_stop)
+
   def test_resuming(self):
     """
     Resume a previously stopped instance.