try piggy-backing on tornado for proactor loop support

zmq/asyncio.py

@@ -10,18 +10,58 @@
 from asyncio import SelectorEventLoop, Future
 import selectors
 import warnings
+from weakref import WeakKeyDictionary
 
 import zmq as _zmq
 from zmq import _future
 
+# registry of asyncio loop : selector loop
+_selector_loops: WeakKeyDictionary = WeakKeyDictionary()
+
+
+def _get_selector_loop(io_loop):
+    """Get selector-compatible loop
+
+    Workaround Windows proactor removal of
+    *reader methods, which we need.
+    """
+    if io_loop in _selector_loops:
+        return _selector_loops[io_loop]
+
+    # detect add_reader instead of checking for proactor?
+    if hasattr(asyncio, "ProactorEventLoop") and isinstance(
+        io_loop, asyncio.ProactorEventLoop
+    ):
+        try:
+            from tornado.platform.asyncio import AddThreadSelectorEventLoop
+        except ImportError:
+            raise RuntimeError(
+                "pyzmq requires tornado 6.1 for compatibility with ProactorEventLoop on Windows."
+                " Install tornado 6.1 or use `asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())`"
+            )
+        warnings.warn(
+            "zmq sockets are not compatible with Windows IOCP (Proactor)."
+            " Registering an additional selector thread."
+            " Use `asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())`"
+            " to avoid this warning.",
+            RuntimeWarning,
+            # stacklevel 5 matches most likely zmq.asyncio.Context().socket()
+            stacklevel=5,
+        )
+        selector_loop = AddThreadSelectorEventLoop(io_loop)
+        _selector_loops[io_loop] = selector_loop
+        return selector_loop
+    else:
+        return io_loop
+
 
 class _AsyncIO(object):
     _Future = Future
     _WRITE = selectors.EVENT_WRITE
     _READ = selectors.EVENT_READ
 
     def _default_loop(self):
-        return asyncio.get_event_loop()
+        return _get_selector_loop(asyncio.get_event_loop())
 
 
 class Poller(_AsyncIO, _future._AsyncPoller):

zmq/tests/conftest.py

@@ -1,15 +1 @@
 """pytest configuration and fixtures"""
-
-import sys
-
-import pytest
-
-
-@pytest.fixture(scope='session', autouse=True)
-def win_py38_asyncio():
-    """fix tornado compatibility on py38"""
-    if sys.version_info < (3, 8) or not sys.platform.startswith('win'):
-        return
-    import asyncio
-
-    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())