diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst index 865f5e3be39a86..1caecb1c143a23 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 ` implementation. - *pipe* is a :term:`file-like object `. + *pipe* is a :term:`file-like object `. See + :ref:`Supported 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 ` implementation. - *pipe* is :term:`file-like object `. + *pipe* is a :term:`file-like object `. See + :ref:`Supported 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 a2a3114ad6e4c5..0b7c17a2c58479 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 + ` 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 ` 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 cf2902b4c76559..f18a7fe5855815 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 919d543b0329e9..43d1fb1047e7e4 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() +@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() diff --git a/Lib/test/test_asyncio/test_windows_events.py b/Lib/test/test_asyncio/test_windows_events.py index 0af3368627afca..4a85b1a09ee0e2 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()