Add ProactorEventLoop.add_reader and friends

via background SelectorThread, imported from tornado

Author: minrk

Doc/library/asyncio-eventloop.rst

@@ -1051,6 +1051,9 @@ Watching file descriptors
 See also :ref:`Platform Support <asyncio-platform-support>` section
 for some limitations of these methods.
 
+.. versionchanged:: 3.15
+
+   Added support for these methods to :class:`ProactorEventLoop`.
 
 Working with socket objects directly
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Doc/library/asyncio-platforms.rst

@@ -16,7 +16,7 @@ due to the platforms' underlying architecture and capabilities.
 All Platforms
 =============
 
-* :meth:`loop.add_reader` and :meth:`loop.add_writer`
+* :meth:`~asyncio.loop.add_reader` and :meth:`~asyncio.loop.add_writer`
   cannot be used to monitor file I/O.
 
 
@@ -59,15 +59,19 @@ All event loops on Windows do not support the following methods:
 
 :class:`ProactorEventLoop` has the following limitations:
 
-* The :meth:`loop.add_reader` and :meth:`loop.add_writer`
-  methods are not supported.
+* :meth:`loop.add_reader` and :meth:`loop.add_writer` only accept
+  socket handles (e.g. pipe file descriptors are not supported).
+  When called, :func:`select.select` is run in an additional thread.
 
 The resolution of the monotonic clock on Windows is usually around 15.6
 milliseconds.  The best resolution is 0.5 milliseconds. The resolution depends on the
 hardware (availability of `HPET
 <https://en.wikipedia.org/wiki/High_Precision_Event_Timer>`_) and on the
 Windows configuration.
 
+.. versionadded:: 3.15
+
+   Support for :meth:`loop.add_reader`, :meth:`loop.add_writer` added to :class:`ProactorEventLoop`.
 
 .. _asyncio-windows-subprocess:
 

Lib/asyncio/proactor_events.py

@@ -23,6 +23,7 @@
 from . import transports
 from . import trsock
 from .log import logger
+from ._selector_thread import SelectorThread as _SelectorThread
 
 
 def _set_socket_extra(transport, sock):
@@ -633,6 +634,7 @@ def __init__(self, proactor):
         logger.debug('Using proactor: %s', proactor.__class__.__name__)
         self._proactor = proactor
         self._selector = proactor   # convenient alias
+        self._selector_thread = None
         self._self_reading_future = None
         self._accept_futures = {}   # socket file descriptor => Future
         proactor.set_loop(self)
@@ -641,6 +643,17 @@ def __init__(self, proactor):
             # wakeup fd can only be installed to a file descriptor from the main thread
             signal.set_wakeup_fd(self._csock.fileno())
 
+    def _get_selector_thread(self):
+        """Return the SelectorThread.
+
+        creating it on first request,
+        so no thread is created until/unless
+        the first call to `add_reader` and friends.
+        """
+        if self._selector_thread is None:
+            self._selector_thread = _SelectorThread(self)
+        return self._selector_thread
+
     def _make_socket_transport(self, sock, protocol, waiter=None,
                                extra=None, server=None):
         return _ProactorSocketTransport(self, sock, protocol, waiter,
@@ -697,10 +710,25 @@ def close(self):
         self._proactor.close()
         self._proactor = None
         self._selector = None
+        if self._selector_thread is not None:
+            self._selector_thread.close()
+            self._selector_thread = None
 
         # Close the event loop
         super().close()
 
+    def add_reader(self, fd, callback, *args):
+        return self._get_selector_thread().add_reader(fd, callback, *args)
+
+    def remove_reader(self, fd):
+        return self._get_selector_thread().remove_reader(fd)
+
+    def add_writer(self, fd, callback, *args):
+        return self._get_selector_thread().add_writer(fd, callback, *args)
+
+    def remove_writer(self, fd):
+        return self._get_selector_thread().remove_writer(fd)
+
     async def sock_recv(self, sock, n):
         return await self._proactor.recv(sock, n)
 

Lib/test/test_asyncio/test_windows_events.py

@@ -323,6 +323,58 @@ def threadMain():
         stop.set()
         thr.join()
 
+    def test_add_reader_invalid_argument(self):
+        def assert_raises():
+            return self.assertRaisesRegex(ValueError, r'Invalid file object')
+
+        def cb(sock):
+            return None
+
+        with assert_raises():
+            self.loop.add_reader(object(), cb)
+        with assert_raises():
+            self.loop.add_writer(object(), cb)
+
+        with assert_raises():
+            self.loop.remove_reader(object())
+        with assert_raises():
+            self.loop.remove_writer(object())
+
+    def test_selector_thread(self):
+        assert self.loop._selector_thread is None
+        a, b = socket.socketpair()
+        async def _test():
+            read_future = asyncio.Future()
+            sent = b"asdf"
+
+            def write():
+                a.sendall(sent)
+                self.loop.remove_writer(a)
+
+            def read():
+                msg = b.recv(100)
+                read_future.set_result(msg)
+
+            self.loop.add_reader(b, read)
+            _selector_thread = self.loop._selector_thread
+            assert b in _selector_thread._readers
+            assert _selector_thread is not None
+            self.loop.add_writer(a, write)
+            assert self.loop._selector_thread is _selector_thread
+            assert a in _selector_thread._writers
+            msg = await asyncio.wait_for(read_future, timeout=10)
+
+            self.loop.remove_writer(a)
+            assert a not in _selector_thread._writers
+            self.loop.remove_reader(b)
+            assert b not in _selector_thread._readers
+            a.close()
+            b.close()
+            assert self.loop._selector_thread is _selector_thread
+            assert msg == sent
+
+        self.loop.run_until_complete(_test())
+
 
 class WinPolicyTests(WindowsEventsTestCase):
 

Misc/NEWS.d/next/Windows/2025-11-21-14-44-48.gh-issue-81554.hNFGMW.rst

@@ -0,0 +1,2 @@
+Added support for :meth:`~asyncio.loop.add_reader` to :class:`~asyncio.ProactorEventLoop` on Windows by
+running :func:`select.select` in a background thread.