https://github.com/python/cpython/commit/e2818a74c65f0937eec071ae3fc0032e59b80c09
commit: e2818a74c65f0937eec071ae3fc0032e59b80c09
branch: 3.14
author: Kumar Aditya <[email protected]>
committer: kumaraditya303 <[email protected]>
date: 2026-07-14T08:22:43Z
summary:
[3.14] gh-71019: document which objects connect_read_pipe/connect_write_pipe
accept and add tests (GH-153660) (#153686)
gh-71019: document which objects connect_read_pipe/connect_write_pipe accept
and add tests (GH-153660)
(cherry picked from commit 7fb315b0cd53277c6191b4bfd8019f999d49f1f0)
files:
M Doc/library/asyncio-eventloop.rst
M Doc/library/asyncio-platforms.rst
M Lib/asyncio/proactor_events.py
M Lib/test/test_asyncio/test_events.py
M Lib/test/test_asyncio/test_windows_events.py
diff --git a/Doc/library/asyncio-eventloop.rst
b/Doc/library/asyncio-eventloop.rst
index 865f5e3be39a868..1caecb1c143a238 100644
--- a/Doc/library/asyncio-eventloop.rst
+++ b/Doc/library/asyncio-eventloop.rst
@@ -1274,7 +1274,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
@@ -1291,7 +1293,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
@@ -1300,6 +1304,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
diff --git a/Doc/library/asyncio-platforms.rst
b/Doc/library/asyncio-platforms.rst
index a2a3114ad6e4c5d..0b7c17a2c584795 100644
--- a/Doc/library/asyncio-platforms.rst
+++ b/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
diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
index 6b94975d0046ea3..3cd8ceadaf272f5 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/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.
diff --git a/Lib/test/test_asyncio/test_events.py
b/Lib/test/test_asyncio/test_events.py
index 919d543b0329e9f..43d1fb1047e7e4e 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/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
@@ -3172,5 +3173,100 @@ def test_get_loop(self):
events.AbstractServer().get_loop()
[email protected](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()
diff --git a/Lib/test/test_asyncio/test_windows_events.py
b/Lib/test/test_asyncio/test_windows_events.py
index 0af3368627afca7..4a85b1a09ee0e24 100644
--- a/Lib/test/test_asyncio/test_windows_events.py
+++ b/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
@@ -359,5 +363,110 @@ async def main():
asyncio.events._set_event_loop_policy(old_policy)
+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()
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
