call_soon() now detects automatically switch to call_soon_threadsafe() if

selector.select() is running. Similar change for call_at(): reschedule the call
with call_soon_threadsafe() if selector.select() is running.

3 files changed, 25 insertions, 24 deletions

diff --git a/README b/README
index 000d705..27ea8c9 100644
--- a/README
+++ b/README
@@ -72,15 +72,6 @@ Hello World::
     loop.close()
 
 
-Eventlet and aiogreen
----------------------
-
-Using the event loop from greenthreads is not safe: calls to the event loop
-must be passed to ``call_soon_threadsafe()``. Example to stop the event loop:
-
-    eventlet.spawn(loop.call_soon_threadsafe, loop.stop)
-
-
 Installation
 ============
 
diff --git a/aiogreen.py b/aiogreen.py
index 6c44d00..5c45635 100644
--- a/aiogreen.py
+++ b/aiogreen.py
@@ -189,6 +189,24 @@ class EventLoop(asyncio.SelectorEventLoop):
         if eventlet.patcher.is_monkey_patched('thread'):
             self._default_executor = _TpoolExecutor(self)
 
+    def call_soon(self, callback, *args):
+        if self._selector._event:
+            # selector.select() is running: use the thread-safe version to wake
+            # up select().
+            return self.call_soon_threadsafe(callback, *args)
+        else:
+            # Fast-path: no need to wake up th eevent loop.
+            return super(EventLoop, self).call_soon(callback, *args)
+
+    def call_at(self, when, callback, *args):
+        if self._selector._event:
+            # selector.select() is running: use the thread-safe version to wake
+            # up select().
+            return self.call_soon_threadsafe(super(EventLoop, self).call_at, when, callback, *args)
+        else:
+            # Fast-path: no need to wake up th eevent loop.
+            return super(EventLoop, self).call_at(when, callback, *args)
+
     def set_debug(self, debug):
         super(EventLoop, self).set_debug(debug)
 
@@ -201,14 +219,6 @@ class EventLoop(asyncio.SelectorEventLoop):
     def time(self):
         return self._hub.clock()
 
-    def _assert_is_current_event_loop(self):
-        super(EventLoop, self)._assert_is_current_event_loop()
-        if self._selector._event:
-            # call_soon() must not be called while selector.select() is
-            # running, it does not wake up the event loop
-            raise RuntimeError(
-                "Non-thread-safe operation invoked from a greenthread")
-
 
 class EventLoopPolicy(asyncio.DefaultEventLoopPolicy):
     _loop_factory = EventLoop
diff --git a/tests/test_eventlet.py b/tests/test_eventlet.py
index a77f772..24c23f5 100644
--- a/tests/test_eventlet.py
+++ b/tests/test_eventlet.py
@@ -24,9 +24,9 @@ try:
                 try:
                     value = gt.wait()
                 except Exception as exc:
-                    loop.call_soon_threadsafe(fut.set_exception, exc)
+                    loop.call_soon(fut.set_exception, exc)
                 else:
-                    loop.call_soon_threadsafe(fut.set_result, value)
+                    loop.call_soon(fut.set_result, value)
             gt.link(copy_result)
 
             value = yield from fut
@@ -69,9 +69,9 @@ except ImportError:
             try:
                 value = gt.wait()
             except Exception as exc:
-                loop.call_soon_threadsafe(fut.set_exception, exc)
+                loop.call_soon(fut.set_exception, exc)
             else:
-                loop.call_soon_threadsafe(fut.set_result, value)
+                loop.call_soon(fut.set_result, value)
         gt.link(copy_result)
 
         value = yield From(fut)
@@ -104,7 +104,7 @@ except ImportError:
 class EventletTests(tests.TestCase):
     def test_stop(self):
         def func():
-            eventlet.spawn(self.loop.call_soon_threadsafe, self.loop.stop)
+            eventlet.spawn(self.loop.call_soon, self.loop.stop)
 
         def schedule_greenthread():
             eventlet.spawn(func)
@@ -117,7 +117,7 @@ class EventletTests(tests.TestCase):
 
         def func():
             result.append("spawn")
-            self.loop.call_soon_threadsafe(self.loop.stop)
+            self.loop.call_soon(self.loop.stop)
 
         def schedule_greenthread():
             eventlet.spawn(func)
@@ -134,7 +134,7 @@ class EventletTests(tests.TestCase):
 
         def func2():
             result.append("spawn_after")
-            self.loop.call_soon_threadsafe(self.loop.stop)
+            self.loop.call_soon(self.loop.stop)
 
         def schedule_greenthread():
             eventlet.spawn(func1)