*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
*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
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
* :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
=======
* 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
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)
_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(
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.
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
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()
import time
import threading
import unittest
+import warnings
from unittest import mock
if sys.platform != 'win32':
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
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()
projects / thirdparty / Python / cpython.git / commitdiff
