gh-71019: document which objects connect_read_pipe/connect_write_pipe accept and add tests

Doc/library/asyncio-eventloop.rst

@@ -1263,7 +1263,9 @@ Working with pipes
    *protocol_factory* must be a callable returning an
    :ref:`asyncio protocol <asyncio-protocol>` implementation.
 
-   *pipe* is a :term:`file-like object <file object>`.
+   *pipe* is a :term:`file-like object <file object>`.  See
+   :ref:`Supported pipe objects <asyncio-pipe-objects>` for the objects
+   supported as *pipe*.
 
    Return pair ``(transport, protocol)``, where *transport* supports
    the :class:`ReadTransport` interface and *protocol* is an object
@@ -1280,7 +1282,9 @@ Working with pipes
    *protocol_factory* must be a callable returning an
    :ref:`asyncio protocol <asyncio-protocol>` implementation.
 
-   *pipe* is :term:`file-like object <file object>`.
+   *pipe* is a :term:`file-like object <file object>`.  See
+   :ref:`Supported pipe objects <asyncio-pipe-objects>` for the objects
+   supported as *pipe*.
 
    Return pair ``(transport, protocol)``, where *transport* supports
    :class:`WriteTransport` interface and *protocol* is an object
@@ -1289,6 +1293,33 @@ Working with pipes
    With :class:`SelectorEventLoop` event loop, the *pipe* is set to
    non-blocking mode.
 
+.. _asyncio-pipe-objects:
+
+.. rubric:: Supported pipe objects
+
+These methods only work with objects the operating system can poll for
+readiness or perform overlapped I/O on.  Regular files on disk are **not**
+supported on any platform.  There is no asynchronous file I/O in asyncio;
+use :meth:`loop.run_in_executor` to read and write regular files without
+blocking the event loop.
+
+On Unix, with :class:`SelectorEventLoop`, *pipe* must wrap one of the
+following:
+
+* a pipe, such as an end of an :func:`os.pipe` pair or a FIFO created with
+  :func:`os.mkfifo`;
+* a socket;
+* a character device, such as a terminal.
+
+On Windows, where only :class:`ProactorEventLoop` implements these methods,
+*pipe* must wrap a handle opened for overlapped I/O (that is, created with the
+``FILE_FLAG_OVERLAPPED`` flag), since the handle has to be associated with an
+I/O completion port.  Handles that were not opened for overlapped I/O are
+rejected.  In particular, the standard streams (:data:`sys.stdin`,
+:data:`sys.stdout` and :data:`sys.stderr`), console handles, and the pipes
+created by :func:`os.pipe` are **not** opened for overlapped I/O and therefore
+cannot be used with these methods.
+
 .. note::
 
    :class:`SelectorEventLoop` does not support the above methods on

Doc/library/asyncio-platforms.rst

@@ -19,6 +19,11 @@ All Platforms
 * :meth:`loop.add_reader` and :meth:`loop.add_writer`
   cannot be used to monitor file I/O.
 
+* :meth:`loop.connect_read_pipe` and :meth:`loop.connect_write_pipe`
+  cannot be used with regular files.  See :ref:`Supported pipe objects
+  <asyncio-pipe-objects>` for the objects that are accepted on each
+  platform.
+
 
 Windows
 =======
@@ -62,6 +67,10 @@ All event loops on Windows do not support the following methods:
 * The :meth:`loop.add_reader` and :meth:`loop.add_writer`
   methods are not supported.
 
+* :meth:`loop.connect_read_pipe` and :meth:`loop.connect_write_pipe` only
+  accept a handle opened for overlapped I/O.
+  See :ref:`Supported pipe objects <asyncio-pipe-objects>` for which objects are supported.
+
 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

Lib/asyncio/proactor_events.py

@@ -62,6 +62,7 @@ def __init__(self, loop, sock, protocol, waiter=None,
         self._closing = False  # Set when close() called.
         self._called_connection_lost = False
         self._eof_written = False
+        self._empty_waiter = None
         if self._server is not None:
             self._server._attach(self)
         self._loop.call_soon(self._protocol.connection_made, self)
@@ -331,10 +332,6 @@ class _ProactorBaseWritePipeTransport(_ProactorBasePipeTransport,
 
     _start_tls_compatible = True
 
-    def __init__(self, *args, **kw):
-        super().__init__(*args, **kw)
-        self._empty_waiter = None
-
     def write(self, data):
         if not isinstance(data, (bytes, bytearray, memoryview)):
             raise TypeError(
@@ -465,7 +462,6 @@ class _ProactorDatagramTransport(_ProactorBasePipeTransport,
     def __init__(self, loop, sock, protocol, address=None,
                  waiter=None, extra=None):
         self._address = address
-        self._empty_waiter = None
         self._buffer_size = 0
         # We don't need to call _protocol.connection_made() since our base
         # constructor does it for us.

Lib/test/test_asyncio/test_events.py

@@ -33,6 +33,7 @@
 from multiprocessing.util import _cleanup_tests as multiprocessing_cleanup_tests
 from test.test_asyncio import utils as test_utils
 from test import support
+from test.support import os_helper
 from test.support import socket_helper
 from test.support import threading_helper
 from test.support import ALWAYS_EQ, LARGEST, SMALLEST
@@ -3134,5 +3135,100 @@ def test_get_loop(self):
             events.AbstractServer().get_loop()
 
 
+@unittest.skipIf(sys.platform == 'win32', 'Unix pipe transport semantics')
+class UnixPipeObjectSupportTests(unittest.TestCase):
+    def setUp(self):
+        super().setUp()
+        self.loop = asyncio.SelectorEventLoop()
+        self.addCleanup(self.loop.close)
+
+    def check_accepted(self, connect, pipeobj):
+        lost = self.loop.create_future()
+
+        class Proto(asyncio.Protocol):
+            def connection_lost(self, exc):
+                if not lost.done():
+                    lost.set_result(exc)
+
+        async def run():
+            transport, protocol = await connect(Proto, pipeobj)
+            self.assertIsInstance(protocol, Proto)
+            self.assertFalse(pipeobj.closed)
+            transport.close()
+            self.assertIsNone(await lost)
+
+        self.loop.run_until_complete(run())
+        self.assertTrue(pipeobj.closed)
+
+    def check_rejected(self, connect, pipeobj):
+        self.addCleanup(pipeobj.close)
+        with self.assertRaisesRegex(ValueError, 'Pipe transport is'):
+            self.loop.run_until_complete(connect(asyncio.Protocol, pipeobj))
+
+    def test_read_pipe(self):
+        rfd, wfd = os.pipe()
+        self.addCleanup(os.close, wfd)
+        self.check_accepted(self.loop.connect_read_pipe, open(rfd, 'rb', 0))
+
+    def test_write_pipe(self):
+        rfd, wfd = os.pipe()
+        self.addCleanup(os.close, rfd)
+        self.check_accepted(self.loop.connect_write_pipe, open(wfd, 'wb', 0))
+
+    def test_read_fifo(self):
+        path = os_helper.TESTFN
+        os.mkfifo(path)
+        self.addCleanup(os_helper.unlink, path)
+        rfd = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
+        wfd = os.open(path, os.O_WRONLY)
+        self.addCleanup(os.close, wfd)
+        self.check_accepted(self.loop.connect_read_pipe, open(rfd, 'rb', 0))
+
+    def test_write_fifo(self):
+        path = os_helper.TESTFN
+        os.mkfifo(path)
+        self.addCleanup(os_helper.unlink, path)
+        rfd = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
+        self.addCleanup(os.close, rfd)
+        wfd = os.open(path, os.O_WRONLY)
+        self.check_accepted(self.loop.connect_write_pipe, open(wfd, 'wb', 0))
+
+    def test_read_socket(self):
+        rsock, wsock = socket.socketpair()
+        self.addCleanup(wsock.close)
+        self.check_accepted(self.loop.connect_read_pipe,
+                            open(rsock.detach(), 'rb', 0))
+
+    def test_write_socket(self):
+        rsock, wsock = socket.socketpair()
+        self.addCleanup(rsock.close)
+        self.check_accepted(self.loop.connect_write_pipe,
+                            open(wsock.detach(), 'wb', 0))
+
+    @unittest.skipUnless(hasattr(os, 'openpty'), 'need os.openpty()')
+    def test_read_character_device(self):
+        master, slave = os.openpty()
+        self.addCleanup(os.close, slave)
+        self.check_accepted(self.loop.connect_read_pipe, open(master, 'rb', 0))
+
+    @unittest.skipUnless(hasattr(os, 'openpty'), 'need os.openpty()')
+    def test_write_character_device(self):
+        master, slave = os.openpty()
+        self.addCleanup(os.close, master)
+        self.check_accepted(self.loop.connect_write_pipe, open(slave, 'wb', 0))
+
+    def test_read_regular_file(self):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        with open(os_helper.TESTFN, 'wb') as f:
+            f.write(b'spam')
+        self.check_rejected(self.loop.connect_read_pipe,
+                            open(os_helper.TESTFN, 'rb', 0))
+
+    def test_write_regular_file(self):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        self.check_rejected(self.loop.connect_write_pipe,
+                            open(os_helper.TESTFN, 'wb', 0))
+
+
 if __name__ == '__main__':
     unittest.main()

Lib/test/test_asyncio/test_windows_events.py

@@ -5,6 +5,7 @@
 import time
 import threading
 import unittest
+import warnings
 from unittest import mock
 
 if sys.platform != 'win32':
@@ -15,6 +16,9 @@
 
 import asyncio
 from asyncio import windows_events
+from asyncio import windows_utils
+from test import support
+from test.support import os_helper
 from test.test_asyncio import utils as test_utils
 
 
@@ -324,5 +328,110 @@ def threadMain():
         thr.join()
 
 
+class ProactorPipeObjectSupportTests(unittest.TestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.loop = asyncio.ProactorEventLoop()
+        self.addCleanup(self.loop.close)
+        self.errors = []
+        self.loop.set_exception_handler(
+            lambda loop, context: self.errors.append(context))
+
+    def check_read_rejected(self, pipe):
+        self.addCleanup(pipe.close)
+        lost = self.loop.create_future()
+
+        class Proto(asyncio.Protocol):
+            def connection_lost(self, exc):
+                if not lost.done():
+                    lost.set_result(exc)
+
+        async def run():
+            transport, _ = await self.loop.connect_read_pipe(Proto, pipe)
+            exc = await lost
+            transport.close()
+            return exc
+
+        exc = self.loop.run_until_complete(run())
+        self.assertIsInstance(exc, OSError)
+
+    def check_write_rejected(self, pipe):
+        self.addCleanup(pipe.close)
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore', ResourceWarning)
+            with self.assertRaises(OSError):
+                self.loop.run_until_complete(
+                    self.loop.connect_write_pipe(asyncio.BaseProtocol, pipe))
+            support.gc_collect()
+
+    def check_accepted(self, rpipe, wpipe):
+        self.addCleanup(rpipe.close)
+        self.addCleanup(wpipe.close)
+
+        chunks = []
+        lost = self.loop.create_future()
+
+        class ReadProto(asyncio.Protocol):
+            def data_received(self, data):
+                chunks.append(data)
+
+            def connection_lost(self, exc):
+                if not lost.done():
+                    lost.set_result(exc)
+
+        async def run():
+            rtransport, _ = await self.loop.connect_read_pipe(ReadProto, rpipe)
+            wtransport, _ = await self.loop.connect_write_pipe(
+                asyncio.BaseProtocol, wpipe)
+            wtransport.write(b'spam')
+            wtransport.close()
+            await lost
+            rtransport.close()
+
+        self.loop.run_until_complete(run())
+        self.assertEqual(b''.join(chunks), b'spam')
+        self.assertFalse(self.errors)
+
+    def test_overlapped_pipe(self):
+        rhandle, whandle = windows_utils.pipe(duplex=True, overlapped=(True, True))
+        self.check_accepted(windows_utils.PipeHandle(rhandle),
+                            windows_utils.PipeHandle(whandle))
+
+    def test_socketpair(self):
+        rsock, wsock = socket.socketpair()
+        self.check_accepted(rsock, wsock)
+
+    def test_read_non_overlapped_pipe(self):
+        rhandle, whandle = windows_utils.pipe(overlapped=(False, False))
+        self.addCleanup(windows_utils.PipeHandle(whandle).close)
+        self.check_read_rejected(windows_utils.PipeHandle(rhandle))
+
+    def test_write_non_overlapped_pipe(self):
+        rhandle, whandle = windows_utils.pipe(overlapped=(False, False))
+        self.addCleanup(windows_utils.PipeHandle(rhandle).close)
+        self.check_write_rejected(windows_utils.PipeHandle(whandle))
+
+    def test_read_regular_file(self):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        with open(os_helper.TESTFN, 'wb') as f:
+            f.write(b'spam')
+        self.check_read_rejected(open(os_helper.TESTFN, 'rb', 0))
+
+    def test_write_regular_file(self):
+        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
+        self.check_write_rejected(open(os_helper.TESTFN, 'wb', 0))
+
+    def test_read_os_pipe(self):
+        rfd, wfd = os.pipe()
+        self.addCleanup(os.close, wfd)
+        self.check_read_rejected(open(rfd, 'rb', 0))
+
+    def test_write_os_pipe(self):
+        rfd, wfd = os.pipe()
+        self.addCleanup(os.close, rfd)
+        self.check_write_rejected(open(wfd, 'wb', 0))
+
+
 if __name__ == '__main__':
     unittest.main()