From 5fd308704ced3e4277c527d3ceff77786f9465d3 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Mon, 10 Sep 2018 11:11:13 -0700 Subject: [PATCH 01/11] Add a stub for new streams --- Lib/asyncio/streams2.py | 93 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) create mode 100644 Lib/asyncio/streams2.py diff --git a/Lib/asyncio/streams2.py b/Lib/asyncio/streams2.py new file mode 100644 index 00000000000000..48525da7da07a7 --- /dev/null +++ b/Lib/asyncio/streams2.py @@ -0,0 +1,93 @@ +from . import events +from . import protocols +from . import streams + +_DEFAULT_LIMIT = 2 ** 16 # 64 KiB + + +async def connect(host=None, port=None, *, + loop=None, limit=_DEFAULT_LIMIT, **kwds): + if loop is None: + loop = events.get_running_loop() + + stream = Stream(limit=limit, loop=loop) + + +async def serve(client_connected_cb, host=None, port=None, *, + loop=None, limit=_DEFAULT_LIMIT, **kwds): + if loop is None: + loop = events.get_running_loop() + + def factory(): + reader = Stream(limit=limit, loop=loop) + protocol = StreamReaderProtocol(reader, client_connected_cb, + loop=loop) + return protocol + + return await loop.create_server(factory, host, port, **kwds) + + +class Stream: + + def __init__(self, limit, loop): + self._limit = limit + self._loop = loop + self._transport = None + self._protocol = None + + def _on_connection_made(self, transport): + self._transport = transport + + def _on_connection_lost(self, exc): + pass + + def _on_pause_writing(self): + pass + + def _on_resume_writing(self): + pass + + def _on_get_buffer(self, sizehint): + pass + + def _on_buffer_updated(self, nbytes): + pass + + def _on_eof(self): + pass + + +class _BaseStreamProtocol(protocols.BufferedProtocol): + def __init__(self, stream): + self._stream = stream + + def connection_made(self, transport): + self._stream._on_connection_made(transport) + + def connection_lost(self, exc): + self._stream._on_connection_lost(exc) + + def pause_writing(self): + self._stream._on_pause_writing() + + def resume_writing(self): + self._stream._on_resume_writing() + + def get_buffer(self, sizehint): + return self._stream._on_get_buffer(sizehint) + + def buffer_updated(self, nbytes): + self._stream._on_buffer_updated(nbytes) + + def eof_received(self): + self._stream._on_eof() + + +class _ClientStreamProtocol(_BaseStreamProtocol): + def __init__(self, stream): + self._stream = stream + + +class _ServerStreamProtocol(_BaseStreamProtocol): + def __init__(self, stream): + self._stream = stream From c47ca7f649d5cf4c1ed16f77e69ca1dab6386cab Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Tue, 11 Sep 2018 09:34:36 -0700 Subject: [PATCH 02/11] Work on --- Lib/asyncio/streams2.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/Lib/asyncio/streams2.py b/Lib/asyncio/streams2.py index 48525da7da07a7..13f9f3c5177b8a 100644 --- a/Lib/asyncio/streams2.py +++ b/Lib/asyncio/streams2.py @@ -60,6 +60,9 @@ def _on_eof(self): class _BaseStreamProtocol(protocols.BufferedProtocol): def __init__(self, stream): self._stream = stream + self._paused = False + self._drain_waiter = None + self._connection_lost = False def connection_made(self, transport): self._stream._on_connection_made(transport) From c4cebd221f04466b161e4e62add608c910c07a49 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Tue, 11 Sep 2018 09:52:35 -0700 Subject: [PATCH 03/11] Work on --- Lib/asyncio/streams2.py | 61 ++++++++++++++++++++++++++++++++--------- 1 file changed, 48 insertions(+), 13 deletions(-) diff --git a/Lib/asyncio/streams2.py b/Lib/asyncio/streams2.py index 13f9f3c5177b8a..b72af0b7f2b727 100644 --- a/Lib/asyncio/streams2.py +++ b/Lib/asyncio/streams2.py @@ -1,6 +1,6 @@ from . import events +from . import exceptions from . import protocols -from . import streams _DEFAULT_LIMIT = 2 ** 16 # 64 KiB @@ -20,8 +20,8 @@ async def serve(client_connected_cb, host=None, port=None, *, def factory(): reader = Stream(limit=limit, loop=loop) - protocol = StreamReaderProtocol(reader, client_connected_cb, - loop=loop) + protocol = _StreamProtocol(reader, client_connected_cb, + loop=loop) return protocol return await loop.create_server(factory, host, port, **kwds) @@ -41,12 +41,6 @@ def _on_connection_made(self, transport): def _on_connection_lost(self, exc): pass - def _on_pause_writing(self): - pass - - def _on_resume_writing(self): - pass - def _on_get_buffer(self, sizehint): pass @@ -63,18 +57,53 @@ def __init__(self, stream): self._paused = False self._drain_waiter = None self._connection_lost = False + self._over_ssl = False def connection_made(self, transport): - self._stream._on_connection_made(transport) + self._stream.set_transport(transport) + self._over_ssl = transport.get_extra_info('sslcontext') is not None + if self._client_connected_cb is not None: + self._stream_writer = StreamWriter(transport, self, + self._stream_reader, + self._loop) + res = self._client_connected_cb(self._stream_reader, + self._stream_writer) + if coroutines.iscoroutine(res): + self._loop.create_task(res) def connection_lost(self, exc): - self._stream._on_connection_lost(exc) + self._connection_lost = True + # Wake up the writer if currently paused. + if not self._paused: + return + waiter = self._drain_waiter + if waiter is None: + return + self._drain_waiter = None + if waiter.done(): + return + if exc is None: + waiter.set_result(None) + else: + waiter.set_exception(exc) def pause_writing(self): - self._stream._on_pause_writing() + assert not self._paused + self._paused = True + if self._loop.get_debug(): + logger.debug("%r pauses writing", self) def resume_writing(self): - self._stream._on_resume_writing() + assert self._paused + self._paused = False + if self._loop.get_debug(): + logger.debug("%r resumes writing", self) + + waiter = self._drain_waiter + if waiter is not None: + self._drain_waiter = None + if not waiter.done(): + waiter.set_result(None) def get_buffer(self, sizehint): return self._stream._on_get_buffer(sizehint) @@ -84,6 +113,12 @@ def buffer_updated(self, nbytes): def eof_received(self): self._stream._on_eof() + if self._over_ssl: + # Prevent a warning in SSLProtocol.eof_received: + # "returning true from eof_received() + # has no effect when using ssl" + return False + return True class _ClientStreamProtocol(_BaseStreamProtocol): From 5af9af19a7a385dd6fcd86985875ac54fb60a9d6 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Tue, 11 Sep 2018 10:53:26 -0700 Subject: [PATCH 04/11] Work on --- Lib/asyncio/streams2.py | 341 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 330 insertions(+), 11 deletions(-) diff --git a/Lib/asyncio/streams2.py b/Lib/asyncio/streams2.py index b72af0b7f2b727..c2a229f16c0638 100644 --- a/Lib/asyncio/streams2.py +++ b/Lib/asyncio/streams2.py @@ -13,7 +13,7 @@ async def connect(host=None, port=None, *, stream = Stream(limit=limit, loop=loop) -async def serve(client_connected_cb, host=None, port=None, *, +async def serve(callback, host=None, port=None, *, loop=None, limit=_DEFAULT_LIMIT, **kwds): if loop is None: loop = events.get_running_loop() @@ -30,11 +30,328 @@ def factory(): class Stream: def __init__(self, limit, loop): + if limit <= 0: + raise ValueError('Limit cannot be <= 0') + self._limit = limit self._loop = loop + self._buffer = bytearray() + + self._eof = False # Whether we're done. + self._waiter = None # A future used by _wait_for_data() + self._exception = None self._transport = None + self._paused = False + self._protocol = None + def __repr__(self): + info = [self.__class__.__name__] + if self._buffer: + info.append(f'{len(self._buffer)} bytes') + if self._eof: + info.append('eof') + if self._limit != _DEFAULT_LIMIT: + info.append(f'limit={self._limit}') + if self._waiter: + info.append(f'waiter={self._waiter!r}') + if self._exception: + info.append(f'exception={self._exception!r}') + if self._transport: + info.append(f'transport={self._transport!r}') + if self._paused: + info.append('paused') + return '<{}>'.format(' '.join(info)) + + def exception(self): + # the main reason for the method is + # a compatibility with old stream reader API + return self._exception + + def at_eof(self): + """Return True if the buffer is empty and 'feed_eof' was called.""" + return self._eof and not self._buffer + + async def _wait_for_data(self, func_name): + """Wait until feed_data() or feed_eof() is called. + + If stream was paused, automatically resume it. + """ + # StreamReader uses a future to link the protocol feed_data() method + # to a read coroutine. Running two read coroutines at the same time + # would have an unexpected behaviour. It would not possible to know + # which coroutine would get the next data. + if self._waiter is not None: + raise RuntimeError( + f'{func_name}() called while another coroutine is ' + f'already waiting for incoming data') + + assert not self._eof, '_wait_for_data after EOF' + + # Waiting for data while paused will make deadlock, so prevent it. + # This is essential for readexactly(n) for case when n > self._limit. + if self._paused: + self._paused = False + self._transport.resume_reading() + + self._waiter = self._loop.create_future() + try: + await self._waiter + finally: + self._waiter = None + + async def readline(self): + """Read chunk of data from the stream until newline (b'\n') is found. + + On success, return chunk that ends with newline. If only partial + line can be read due to EOF, return incomplete line without + terminating newline. When EOF was reached while no bytes read, empty + bytes object is returned. + + If limit is reached, ValueError will be raised. In that case, if + newline was found, complete line including newline will be removed + from internal buffer. Else, internal buffer will be cleared. Limit is + compared against part of the line without newline. + + If stream was paused, this function will automatically resume it if + needed. + """ + sep = b'\n' + seplen = len(sep) + try: + line = await self.readuntil(sep) + except IncompleteReadError as e: + return e.partial + except LimitOverrunError as e: + if self._buffer.startswith(sep, e.consumed): + del self._buffer[:e.consumed + seplen] + else: + self._buffer.clear() + self._maybe_resume_transport() + raise ValueError(e.args[0]) + return line + + async def readuntil(self, separator=b'\n'): + """Read data from the stream until ``separator`` is found. + + On success, the data and separator will be removed from the + internal buffer (consumed). Returned data will include the + separator at the end. + + Configured stream limit is used to check result. Limit sets the + maximal length of data that can be returned, not counting the + separator. + + If an EOF occurs and the complete separator is still not found, + an IncompleteReadError exception will be raised, and the internal + buffer will be reset. The IncompleteReadError.partial attribute + may contain the separator partially. + + If the data cannot be read because of over limit, a + LimitOverrunError exception will be raised, and the data + will be left in the internal buffer, so it can be read again. + """ + seplen = len(separator) + if seplen == 0: + raise ValueError('Separator should be at least one-byte string') + + if self._exception is not None: + raise self._exception + + # Consume whole buffer except last bytes, which length is + # one less than seplen. Let's check corner cases with + # separator='SEPARATOR': + # * we have received almost complete separator (without last + # byte). i.e buffer='some textSEPARATO'. In this case we + # can safely consume len(separator) - 1 bytes. + # * last byte of buffer is first byte of separator, i.e. + # buffer='abcdefghijklmnopqrS'. We may safely consume + # everything except that last byte, but this require to + # analyze bytes of buffer that match partial separator. + # This is slow and/or require FSM. For this case our + # implementation is not optimal, since require rescanning + # of data that is known to not belong to separator. In + # real world, separator will not be so long to notice + # performance problems. Even when reading MIME-encoded + # messages :) + + # `offset` is the number of bytes from the beginning of the buffer + # where there is no occurrence of `separator`. + offset = 0 + + # Loop until we find `separator` in the buffer, exceed the buffer size, + # or an EOF has happened. + while True: + buflen = len(self._buffer) + + # Check if we now have enough data in the buffer for `separator` to + # fit. + if buflen - offset >= seplen: + isep = self._buffer.find(separator, offset) + + if isep != -1: + # `separator` is in the buffer. `isep` will be used later + # to retrieve the data. + break + + # see upper comment for explanation. + offset = buflen + 1 - seplen + if offset > self._limit: + raise LimitOverrunError( + 'Separator is not found, and chunk exceed the limit', + offset) + + # Complete message (with full separator) may be present in buffer + # even when EOF flag is set. This may happen when the last chunk + # adds data which makes separator be found. That's why we check for + # EOF *ater* inspecting the buffer. + if self._eof: + chunk = bytes(self._buffer) + self._buffer.clear() + raise IncompleteReadError(chunk, None) + + # _wait_for_data() will resume reading if stream was paused. + await self._wait_for_data('readuntil') + + if isep > self._limit: + raise LimitOverrunError( + 'Separator is found, but chunk is longer than limit', isep) + + chunk = self._buffer[:isep + seplen] + del self._buffer[:isep + seplen] + self._maybe_resume_transport() + return bytes(chunk) + + async def read(self, n=-1): + """Read up to `n` bytes from the stream. + + If n is not provided, or set to -1, read until EOF and return all read + bytes. If the EOF was received and the internal buffer is empty, return + an empty bytes object. + + If n is zero, return empty bytes object immediately. + + If n is positive, this function try to read `n` bytes, and may return + less or equal bytes than requested, but at least one byte. If EOF was + received before any byte is read, this function returns empty byte + object. + + Returned value is not limited with limit, configured at stream + creation. + + If stream was paused, this function will automatically resume it if + needed. + """ + + if self._exception is not None: + raise self._exception + + if n == 0: + return b'' + + if n < 0: + # This used to just loop creating a new waiter hoping to + # collect everything in self._buffer, but that would + # deadlock if the subprocess sends more than self.limit + # bytes. So just call self.read(self._limit) until EOF. + blocks = [] + while True: + block = await self.read(self._limit) + if not block: + break + blocks.append(block) + return b''.join(blocks) + + if not self._buffer and not self._eof: + await self._wait_for_data('read') + + # This will work right even if buffer is less than n bytes + data = bytes(self._buffer[:n]) + del self._buffer[:n] + + self._maybe_resume_transport() + return data + + async def readexactly(self, n): + """Read exactly `n` bytes. + + Raise an IncompleteReadError if EOF is reached before `n` bytes can be + read. The IncompleteReadError.partial attribute of the exception will + contain the partial read bytes. + + if n is zero, return empty bytes object. + + Returned value is not limited with limit, configured at stream + creation. + + If stream was paused, this function will automatically resume it if + needed. + """ + if n < 0: + raise ValueError('readexactly size can not be less than zero') + + if self._exception is not None: + raise self._exception + + if n == 0: + return b'' + + while len(self._buffer) < n: + if self._eof: + incomplete = bytes(self._buffer) + self._buffer.clear() + raise IncompleteReadError(incomplete, n) + + await self._wait_for_data('readexactly') + + if len(self._buffer) == n: + data = bytes(self._buffer) + self._buffer.clear() + else: + data = bytes(self._buffer[:n]) + del self._buffer[:n] + self._maybe_resume_transport() + return data + + def __aiter__(self): + return self + + async def __anext__(self): + val = await self.readline() + if val == b'': + raise StopAsyncIteration + return val + + def _wakeup_waiter(self): + """Wakeup read*() functions waiting for data or EOF.""" + waiter = self._waiter + if waiter is not None: + self._waiter = None + if not waiter.cancelled(): + waiter.set_result(None) + + def _set_transport(self, transport): + assert self._transport is None, 'Transport already set' + self._transport = transport + + def _maybe_resume_transport(self): + if self._paused and len(self._buffer) <= self._limit: + self._paused = False + self._transport.resume_reading() + + def _on_eof(self): + self._eof = True + self._wakeup_waiter() + + def _set_exception(self, exc): + self._exception = exc + + waiter = self._waiter + if waiter is not None: + self._waiter = None + if not waiter.cancelled(): + waiter.set_exception(exc) + def _on_connection_made(self, transport): self._transport = transport @@ -62,14 +379,6 @@ def __init__(self, stream): def connection_made(self, transport): self._stream.set_transport(transport) self._over_ssl = transport.get_extra_info('sslcontext') is not None - if self._client_connected_cb is not None: - self._stream_writer = StreamWriter(transport, self, - self._stream_reader, - self._loop) - res = self._client_connected_cb(self._stream_reader, - self._stream_writer) - if coroutines.iscoroutine(res): - self._loop.create_task(res) def connection_lost(self, exc): self._connection_lost = True @@ -127,5 +436,15 @@ def __init__(self, stream): class _ServerStreamProtocol(_BaseStreamProtocol): - def __init__(self, stream): - self._stream = stream + def __init__(self, stream, callback): + super().__init__(stream) + self._callback = callback + self._tasks = None + + def connection_made(self, transport): + super().connection_made(transport) + self._stream_writer = StreamWriter(transport, self, + self._stream_reader, + self._loop) + # TODO: store created task somewhere + self._loop.create_task(self._callback(self._stream)) From 05358d495a269eda86b878bc7a01875399ec1230 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Tue, 11 Sep 2018 14:05:37 -0700 Subject: [PATCH 05/11] Implement asyncio.connect() --- Lib/asyncio/streams.py | 51 ++- Lib/asyncio/streams2.py | 450 -------------------------- Lib/test/test_asyncio/test_streams.py | 23 ++ 3 files changed, 71 insertions(+), 453 deletions(-) delete mode 100644 Lib/asyncio/streams2.py diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py index 9dab49b35e46e8..8b4582d00ccf20 100644 --- a/Lib/asyncio/streams.py +++ b/Lib/asyncio/streams.py @@ -1,11 +1,13 @@ __all__ = ( 'StreamReader', 'StreamWriter', 'StreamReaderProtocol', - 'open_connection', 'start_server') + 'open_connection', 'start_server', + 'connect') import socket if hasattr(socket, 'AF_UNIX'): - __all__ += ('open_unix_connection', 'start_unix_server') + __all__ += ('open_unix_connection', 'start_unix_server', + 'unix_connect') from . import coroutines from . import events @@ -18,6 +20,23 @@ _DEFAULT_LIMIT = 2 ** 16 # 64 KiB +async def connect(host=None, port=None, *, + loop=None, limit=_DEFAULT_LIMIT, **kwds): + if loop is None: + loop = events.get_running_loop() + stream = Stream(limit=limit, loop=loop) + protocol = StreamReaderProtocol(stream, loop=loop) + stream._set_protocol(protocol) + transport, _ = await loop.create_connection( + lambda: protocol, host, port, **kwds) + return stream + + +async def serve(client_connected_cb, host=None, port=None, *, + loop=None, limit=_DEFAULT_LIMIT, **kwds): + pass + + async def open_connection(host=None, port=None, *, loop=None, limit=_DEFAULT_LIMIT, **kwds): """A wrapper for create_connection() returning a (reader, writer) pair. @@ -85,6 +104,17 @@ def factory(): if hasattr(socket, 'AF_UNIX'): # UNIX Domain Sockets are supported on this platform + async def unix_connect(path=None, *, + loop=None, limit=_DEFAULT_LIMIT, **kwds): + if loop is None: + loop = events.get_running_loop() + stream = Stream(limit=limit, loop=loop) + protocol = StreamReaderProtocol(stream, loop=loop) + stream._set_protocol(protocol) + transport, _ = await loop.create_unix_connection( + lambda: protocol, path, **kwds) + return stream + async def open_unix_connection(path=None, *, loop=None, limit=_DEFAULT_LIMIT, **kwds): """Similar to `open_connection` but works with UNIX Domain Sockets.""" @@ -338,7 +368,7 @@ def __init__(self, limit=_DEFAULT_LIMIT, loop=None): self._paused = False def __repr__(self): - info = ['StreamReader'] + info = [self.__class__.__name__] if self._buffer: info.append(f'{len(self._buffer)} bytes') if self._eof: @@ -663,3 +693,18 @@ async def __anext__(self): if val == b'': raise StopAsyncIteration return val + + +class Stream(StreamReader, StreamWriter): + def __init__(self, limit, loop): + StreamReader.__init__(self, limit, loop) + # A trick for emulating StreamWriter ctor without an actual call + self._reader = self + self._protocol = None # setup the attribute in _set_protocol() + + def _set_protocol(self, protocol): + # a post-init method to set protocol instance + self._protocol = protocol + + def __repr__(self): + return StreamReader.__repr__(self) diff --git a/Lib/asyncio/streams2.py b/Lib/asyncio/streams2.py deleted file mode 100644 index c2a229f16c0638..00000000000000 --- a/Lib/asyncio/streams2.py +++ /dev/null @@ -1,450 +0,0 @@ -from . import events -from . import exceptions -from . import protocols - -_DEFAULT_LIMIT = 2 ** 16 # 64 KiB - - -async def connect(host=None, port=None, *, - loop=None, limit=_DEFAULT_LIMIT, **kwds): - if loop is None: - loop = events.get_running_loop() - - stream = Stream(limit=limit, loop=loop) - - -async def serve(callback, host=None, port=None, *, - loop=None, limit=_DEFAULT_LIMIT, **kwds): - if loop is None: - loop = events.get_running_loop() - - def factory(): - reader = Stream(limit=limit, loop=loop) - protocol = _StreamProtocol(reader, client_connected_cb, - loop=loop) - return protocol - - return await loop.create_server(factory, host, port, **kwds) - - -class Stream: - - def __init__(self, limit, loop): - if limit <= 0: - raise ValueError('Limit cannot be <= 0') - - self._limit = limit - self._loop = loop - self._buffer = bytearray() - - self._eof = False # Whether we're done. - self._waiter = None # A future used by _wait_for_data() - self._exception = None - self._transport = None - self._paused = False - - self._protocol = None - - def __repr__(self): - info = [self.__class__.__name__] - if self._buffer: - info.append(f'{len(self._buffer)} bytes') - if self._eof: - info.append('eof') - if self._limit != _DEFAULT_LIMIT: - info.append(f'limit={self._limit}') - if self._waiter: - info.append(f'waiter={self._waiter!r}') - if self._exception: - info.append(f'exception={self._exception!r}') - if self._transport: - info.append(f'transport={self._transport!r}') - if self._paused: - info.append('paused') - return '<{}>'.format(' '.join(info)) - - def exception(self): - # the main reason for the method is - # a compatibility with old stream reader API - return self._exception - - def at_eof(self): - """Return True if the buffer is empty and 'feed_eof' was called.""" - return self._eof and not self._buffer - - async def _wait_for_data(self, func_name): - """Wait until feed_data() or feed_eof() is called. - - If stream was paused, automatically resume it. - """ - # StreamReader uses a future to link the protocol feed_data() method - # to a read coroutine. Running two read coroutines at the same time - # would have an unexpected behaviour. It would not possible to know - # which coroutine would get the next data. - if self._waiter is not None: - raise RuntimeError( - f'{func_name}() called while another coroutine is ' - f'already waiting for incoming data') - - assert not self._eof, '_wait_for_data after EOF' - - # Waiting for data while paused will make deadlock, so prevent it. - # This is essential for readexactly(n) for case when n > self._limit. - if self._paused: - self._paused = False - self._transport.resume_reading() - - self._waiter = self._loop.create_future() - try: - await self._waiter - finally: - self._waiter = None - - async def readline(self): - """Read chunk of data from the stream until newline (b'\n') is found. - - On success, return chunk that ends with newline. If only partial - line can be read due to EOF, return incomplete line without - terminating newline. When EOF was reached while no bytes read, empty - bytes object is returned. - - If limit is reached, ValueError will be raised. In that case, if - newline was found, complete line including newline will be removed - from internal buffer. Else, internal buffer will be cleared. Limit is - compared against part of the line without newline. - - If stream was paused, this function will automatically resume it if - needed. - """ - sep = b'\n' - seplen = len(sep) - try: - line = await self.readuntil(sep) - except IncompleteReadError as e: - return e.partial - except LimitOverrunError as e: - if self._buffer.startswith(sep, e.consumed): - del self._buffer[:e.consumed + seplen] - else: - self._buffer.clear() - self._maybe_resume_transport() - raise ValueError(e.args[0]) - return line - - async def readuntil(self, separator=b'\n'): - """Read data from the stream until ``separator`` is found. - - On success, the data and separator will be removed from the - internal buffer (consumed). Returned data will include the - separator at the end. - - Configured stream limit is used to check result. Limit sets the - maximal length of data that can be returned, not counting the - separator. - - If an EOF occurs and the complete separator is still not found, - an IncompleteReadError exception will be raised, and the internal - buffer will be reset. The IncompleteReadError.partial attribute - may contain the separator partially. - - If the data cannot be read because of over limit, a - LimitOverrunError exception will be raised, and the data - will be left in the internal buffer, so it can be read again. - """ - seplen = len(separator) - if seplen == 0: - raise ValueError('Separator should be at least one-byte string') - - if self._exception is not None: - raise self._exception - - # Consume whole buffer except last bytes, which length is - # one less than seplen. Let's check corner cases with - # separator='SEPARATOR': - # * we have received almost complete separator (without last - # byte). i.e buffer='some textSEPARATO'. In this case we - # can safely consume len(separator) - 1 bytes. - # * last byte of buffer is first byte of separator, i.e. - # buffer='abcdefghijklmnopqrS'. We may safely consume - # everything except that last byte, but this require to - # analyze bytes of buffer that match partial separator. - # This is slow and/or require FSM. For this case our - # implementation is not optimal, since require rescanning - # of data that is known to not belong to separator. In - # real world, separator will not be so long to notice - # performance problems. Even when reading MIME-encoded - # messages :) - - # `offset` is the number of bytes from the beginning of the buffer - # where there is no occurrence of `separator`. - offset = 0 - - # Loop until we find `separator` in the buffer, exceed the buffer size, - # or an EOF has happened. - while True: - buflen = len(self._buffer) - - # Check if we now have enough data in the buffer for `separator` to - # fit. - if buflen - offset >= seplen: - isep = self._buffer.find(separator, offset) - - if isep != -1: - # `separator` is in the buffer. `isep` will be used later - # to retrieve the data. - break - - # see upper comment for explanation. - offset = buflen + 1 - seplen - if offset > self._limit: - raise LimitOverrunError( - 'Separator is not found, and chunk exceed the limit', - offset) - - # Complete message (with full separator) may be present in buffer - # even when EOF flag is set. This may happen when the last chunk - # adds data which makes separator be found. That's why we check for - # EOF *ater* inspecting the buffer. - if self._eof: - chunk = bytes(self._buffer) - self._buffer.clear() - raise IncompleteReadError(chunk, None) - - # _wait_for_data() will resume reading if stream was paused. - await self._wait_for_data('readuntil') - - if isep > self._limit: - raise LimitOverrunError( - 'Separator is found, but chunk is longer than limit', isep) - - chunk = self._buffer[:isep + seplen] - del self._buffer[:isep + seplen] - self._maybe_resume_transport() - return bytes(chunk) - - async def read(self, n=-1): - """Read up to `n` bytes from the stream. - - If n is not provided, or set to -1, read until EOF and return all read - bytes. If the EOF was received and the internal buffer is empty, return - an empty bytes object. - - If n is zero, return empty bytes object immediately. - - If n is positive, this function try to read `n` bytes, and may return - less or equal bytes than requested, but at least one byte. If EOF was - received before any byte is read, this function returns empty byte - object. - - Returned value is not limited with limit, configured at stream - creation. - - If stream was paused, this function will automatically resume it if - needed. - """ - - if self._exception is not None: - raise self._exception - - if n == 0: - return b'' - - if n < 0: - # This used to just loop creating a new waiter hoping to - # collect everything in self._buffer, but that would - # deadlock if the subprocess sends more than self.limit - # bytes. So just call self.read(self._limit) until EOF. - blocks = [] - while True: - block = await self.read(self._limit) - if not block: - break - blocks.append(block) - return b''.join(blocks) - - if not self._buffer and not self._eof: - await self._wait_for_data('read') - - # This will work right even if buffer is less than n bytes - data = bytes(self._buffer[:n]) - del self._buffer[:n] - - self._maybe_resume_transport() - return data - - async def readexactly(self, n): - """Read exactly `n` bytes. - - Raise an IncompleteReadError if EOF is reached before `n` bytes can be - read. The IncompleteReadError.partial attribute of the exception will - contain the partial read bytes. - - if n is zero, return empty bytes object. - - Returned value is not limited with limit, configured at stream - creation. - - If stream was paused, this function will automatically resume it if - needed. - """ - if n < 0: - raise ValueError('readexactly size can not be less than zero') - - if self._exception is not None: - raise self._exception - - if n == 0: - return b'' - - while len(self._buffer) < n: - if self._eof: - incomplete = bytes(self._buffer) - self._buffer.clear() - raise IncompleteReadError(incomplete, n) - - await self._wait_for_data('readexactly') - - if len(self._buffer) == n: - data = bytes(self._buffer) - self._buffer.clear() - else: - data = bytes(self._buffer[:n]) - del self._buffer[:n] - self._maybe_resume_transport() - return data - - def __aiter__(self): - return self - - async def __anext__(self): - val = await self.readline() - if val == b'': - raise StopAsyncIteration - return val - - def _wakeup_waiter(self): - """Wakeup read*() functions waiting for data or EOF.""" - waiter = self._waiter - if waiter is not None: - self._waiter = None - if not waiter.cancelled(): - waiter.set_result(None) - - def _set_transport(self, transport): - assert self._transport is None, 'Transport already set' - self._transport = transport - - def _maybe_resume_transport(self): - if self._paused and len(self._buffer) <= self._limit: - self._paused = False - self._transport.resume_reading() - - def _on_eof(self): - self._eof = True - self._wakeup_waiter() - - def _set_exception(self, exc): - self._exception = exc - - waiter = self._waiter - if waiter is not None: - self._waiter = None - if not waiter.cancelled(): - waiter.set_exception(exc) - - def _on_connection_made(self, transport): - self._transport = transport - - def _on_connection_lost(self, exc): - pass - - def _on_get_buffer(self, sizehint): - pass - - def _on_buffer_updated(self, nbytes): - pass - - def _on_eof(self): - pass - - -class _BaseStreamProtocol(protocols.BufferedProtocol): - def __init__(self, stream): - self._stream = stream - self._paused = False - self._drain_waiter = None - self._connection_lost = False - self._over_ssl = False - - def connection_made(self, transport): - self._stream.set_transport(transport) - self._over_ssl = transport.get_extra_info('sslcontext') is not None - - def connection_lost(self, exc): - self._connection_lost = True - # Wake up the writer if currently paused. - if not self._paused: - return - waiter = self._drain_waiter - if waiter is None: - return - self._drain_waiter = None - if waiter.done(): - return - if exc is None: - waiter.set_result(None) - else: - waiter.set_exception(exc) - - def pause_writing(self): - assert not self._paused - self._paused = True - if self._loop.get_debug(): - logger.debug("%r pauses writing", self) - - def resume_writing(self): - assert self._paused - self._paused = False - if self._loop.get_debug(): - logger.debug("%r resumes writing", self) - - waiter = self._drain_waiter - if waiter is not None: - self._drain_waiter = None - if not waiter.done(): - waiter.set_result(None) - - def get_buffer(self, sizehint): - return self._stream._on_get_buffer(sizehint) - - def buffer_updated(self, nbytes): - self._stream._on_buffer_updated(nbytes) - - def eof_received(self): - self._stream._on_eof() - if self._over_ssl: - # Prevent a warning in SSLProtocol.eof_received: - # "returning true from eof_received() - # has no effect when using ssl" - return False - return True - - -class _ClientStreamProtocol(_BaseStreamProtocol): - def __init__(self, stream): - self._stream = stream - - -class _ServerStreamProtocol(_BaseStreamProtocol): - def __init__(self, stream, callback): - super().__init__(stream) - self._callback = callback - self._tasks = None - - def connection_made(self, transport): - super().connection_made(transport) - self._stream_writer = StreamWriter(transport, self, - self._stream_reader, - self._loop) - # TODO: store created task somewhere - self._loop.create_task(self._callback(self._stream)) diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 66d18738b31626..467896acb59e37 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/Lib/test/test_asyncio/test_streams.py @@ -893,6 +893,29 @@ def test_wait_closed_on_close_with_unread_data(self): wr.close() self.loop.run_until_complete(wr.wait_closed()) + def _basetest_connect(self, connect_fut): + stream = self.loop.run_until_complete(connect_fut) + stream.write(b'GET / HTTP/1.0\r\n\r\n') + f = stream.readline() + data = self.loop.run_until_complete(f) + self.assertEqual(data, b'HTTP/1.0 200 OK\r\n') + f = stream.read() + data = self.loop.run_until_complete(f) + self.assertTrue(data.endswith(b'\r\n\r\nTest message')) + stream.close() + self.loop.run_until_complete(stream.wait_closed()) + + def test_connect(self): + with test_utils.run_test_server() as httpd: + connect_fut = asyncio.connect(*httpd.address, loop=self.loop) + self._basetest_connect(connect_fut) + + @support.skip_unless_bind_unix_socket + def test_unix_connect(self): + with test_utils.run_test_unix_server() as httpd: + connect_fut = asyncio.unix_connect(httpd.address, loop=self.loop) + self._basetest_connect(connect_fut) + if __name__ == '__main__': unittest.main() From 6aaf7d1e2b45f0fde37ce940860b73023e7166fc Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Tue, 11 Sep 2018 15:02:34 -0700 Subject: [PATCH 06/11] Work on --- Lib/asyncio/streams.py | 25 +++++++++++++++---------- Lib/test/test_asyncio/test_streams.py | 15 ++++++++------- 2 files changed, 23 insertions(+), 17 deletions(-) diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py index 8b4582d00ccf20..3f6c19eb727d11 100644 --- a/Lib/asyncio/streams.py +++ b/Lib/asyncio/streams.py @@ -7,7 +7,7 @@ if hasattr(socket, 'AF_UNIX'): __all__ += ('open_unix_connection', 'start_unix_server', - 'unix_connect') + 'connect_unix') from . import coroutines from . import events @@ -21,9 +21,9 @@ async def connect(host=None, port=None, *, - loop=None, limit=_DEFAULT_LIMIT, **kwds): - if loop is None: - loop = events.get_running_loop() + limit=_DEFAULT_LIMIT, **kwds): + assert 'loop' not in kwds + loop = events.get_running_loop() stream = Stream(limit=limit, loop=loop) protocol = StreamReaderProtocol(stream, loop=loop) stream._set_protocol(protocol) @@ -104,10 +104,10 @@ def factory(): if hasattr(socket, 'AF_UNIX'): # UNIX Domain Sockets are supported on this platform - async def unix_connect(path=None, *, - loop=None, limit=_DEFAULT_LIMIT, **kwds): - if loop is None: - loop = events.get_running_loop() + async def connect_unix(path=None, *, + limit=_DEFAULT_LIMIT, **kwds): + assert 'loop' not in kwds + loop = events.get_running_loop() stream = Stream(limit=limit, loop=loop) protocol = StreamReaderProtocol(stream, loop=loop) stream._set_protocol(protocol) @@ -698,10 +698,15 @@ async def __anext__(self): class Stream(StreamReader, StreamWriter): def __init__(self, limit, loop): StreamReader.__init__(self, limit, loop) - # A trick for emulating StreamWriter ctor without an actual call - self._reader = self + # Emulate StreamWriter ctor without an actual call self._protocol = None # setup the attribute in _set_protocol() + @property + def _reader(self): + # A trick for making StreamWriter work: the class requires + # self._reader attribute + return self + def _set_protocol(self, protocol): # a post-init method to set protocol instance self._protocol = protocol diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 467896acb59e37..b9a155b0957c0b 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/Lib/test/test_asyncio/test_streams.py @@ -893,8 +893,7 @@ def test_wait_closed_on_close_with_unread_data(self): wr.close() self.loop.run_until_complete(wr.wait_closed()) - def _basetest_connect(self, connect_fut): - stream = self.loop.run_until_complete(connect_fut) + def _basetest_connect(self, stream): stream.write(b'GET / HTTP/1.0\r\n\r\n') f = stream.readline() data = self.loop.run_until_complete(f) @@ -907,14 +906,16 @@ def _basetest_connect(self, connect_fut): def test_connect(self): with test_utils.run_test_server() as httpd: - connect_fut = asyncio.connect(*httpd.address, loop=self.loop) - self._basetest_connect(connect_fut) + stream = self.loop.run_until_complete( + asyncio.connect(*httpd.address)) + self._basetest_connect(stream) @support.skip_unless_bind_unix_socket - def test_unix_connect(self): + def test_connect_unix(self): with test_utils.run_test_unix_server() as httpd: - connect_fut = asyncio.unix_connect(httpd.address, loop=self.loop) - self._basetest_connect(connect_fut) + stream = self.loop.run_until_complete( + asyncio.connect_unix(httpd.address)) + self._basetest_connect(stream) if __name__ == '__main__': From 3a566c298af43c1f6b541e65738ca76f1d0510a0 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Wed, 12 Sep 2018 14:04:43 -0700 Subject: [PATCH 07/11] Fix merge conflict --- Lib/test/test_asyncio/test_streams.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 1033ab73a7f450..7428cec248d050 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/Lib/test/test_asyncio/test_streams.py @@ -965,6 +965,9 @@ def test_del_stream_before_connection_made(self): messages[0]['message']) def _basetest_connect(self, stream): + messages = [] + self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx)) + stream.write(b'GET / HTTP/1.0\r\n\r\n') f = stream.readline() data = self.loop.run_until_complete(f) @@ -975,6 +978,8 @@ def _basetest_connect(self, stream): stream.close() self.loop.run_until_complete(stream.wait_closed()) + self.assertEqual([], messages) + def test_connect(self): with test_utils.run_test_server() as httpd: stream = self.loop.run_until_complete( From 59f5866f098241c20c27a8aaec80daa147466b79 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Fri, 14 Sep 2018 13:10:34 -0700 Subject: [PATCH 08/11] Implement sendfile --- Lib/asyncio/streams.py | 17 ++++++++++++ Lib/test/test_asyncio/test_streams.py | 39 +++++++++++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py index 423d583d818e04..ba911f79228678 100644 --- a/Lib/asyncio/streams.py +++ b/Lib/asyncio/streams.py @@ -792,3 +792,20 @@ def _set_protocol(self, protocol): def __repr__(self): return StreamReader.__repr__(self) + + async def sendfile(self, file, offset=0, count=None, *, fallback=True): + await self.drain() + return await self._loop.sendfile(self._transport, file, + offset, count, fallback=fallback) + + async def start_tls(self, sslcontext, *, + server_hostname=None, + ssl_handshake_timeout=None): + server_side = self._protocol.self._client_connected_cb is not None + await self.drain() + transport = await self._loop.start_tls( + self._transport, self._protocol, sslcontext, + server_side=server_side, server_hostname=server_hostname, + ssl_handshake_timeout=ssl_handshake_timeout) + self._transport = transport + self._protocol._transport = transport diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 05393ec7a8b4b9..0ad730e909c61a 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/Lib/test/test_asyncio/test_streams.py @@ -1015,6 +1015,45 @@ def test_connect_unix(self): asyncio.connect_unix(httpd.address)) self._basetest_connect(stream) + def test_sendfile(self): + messages = [] + self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx)) + + with open(support.TESTFN, 'wb') as fp: + fp.write(b'data\n') + self.addCleanup(support.unlink, support.TESTFN) + + async def do_serve(reader, writer): + data = await reader.readline() + self.assertEqual(data, b'begin\n') + data = await reader.readline() + self.assertEqual(data, b'data\n') + data = await reader.readline() + self.assertEqual(data, b'end\n') + await writer.awrite(b'done\n') + await writer.aclose() + + server = self.loop.run_until_complete( + asyncio.start_server(do_serve, 'localhost', 0, loop=self.loop)) + + host, port = server.sockets[0].getsockname() + + async def do_connect(): + stream = await asyncio.connect(host, port) + stream.write(b'begin\n') + with open(support.TESTFN, 'rb') as fp: + await stream.sendfile(fp) + stream.write(b'end\n') + data = await stream.readline() + self.assertEqual(data, b'done\n') + await stream.aclose() + + self.loop.run_until_complete(do_connect()) + server.close() + self.loop.run_until_complete(server.wait_closed()) + + self.assertEqual([], messages) + if __name__ == '__main__': unittest.main() From b26a78f39284dce6398169ccc25ae3db105e3bb0 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Fri, 14 Sep 2018 14:00:40 -0700 Subject: [PATCH 09/11] Add stream.start_tls --- Lib/asyncio/streams.py | 2 +- Lib/test/test_asyncio/test_streams.py | 13 +++++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py index ba911f79228678..174d29db56df80 100644 --- a/Lib/asyncio/streams.py +++ b/Lib/asyncio/streams.py @@ -801,7 +801,7 @@ async def sendfile(self, file, offset=0, count=None, *, fallback=True): async def start_tls(self, sslcontext, *, server_hostname=None, ssl_handshake_timeout=None): - server_side = self._protocol.self._client_connected_cb is not None + server_side = self._protocol._client_connected_cb is not None await self.drain() transport = await self._loop.start_tls( self._transport, self._protocol, sslcontext, diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 0ad730e909c61a..a2b0d1c582781f 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/Lib/test/test_asyncio/test_streams.py @@ -1054,6 +1054,19 @@ async def do_connect(): self.assertEqual([], messages) + @unittest.skipIf(ssl is None, 'No ssl module') + def test_connect_start_tls(self): + with test_utils.run_test_server(use_ssl=True) as httpd: + # connect without SSL but upgrade to TLS just after + # connection is established + stream = self.loop.run_until_complete( + asyncio.connect(*httpd.address)) + + self.loop.run_until_complete( + stream.start_tls( + sslcontext=test_utils.dummy_ssl_context())) + self._basetest_connect(stream) + if __name__ == '__main__': unittest.main() From 2290409f69c55dbca7a6f40e091ad20589ef09eb Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Fri, 14 Sep 2018 15:17:16 -0700 Subject: [PATCH 10/11] Add docs for sendfile and start_tls --- Doc/library/asyncio-stream.rst | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/Doc/library/asyncio-stream.rst b/Doc/library/asyncio-stream.rst index 80b76253d06569..cc1d0343bd3a62 100644 --- a/Doc/library/asyncio-stream.rst +++ b/Doc/library/asyncio-stream.rst @@ -315,6 +315,28 @@ StreamWriter .. versionadded:: 3.7 + .. coroutinemethod:: sendfile(file, offset=0, count=None, \*, \ + fallback=True) + + Send a *file* to a peer. Return the total number of bytes + sent. + + For more details about arguments and implementation see + :meth:`loop.sendfile`. + + .. versionadded:: 3.8 + + .. coroutinemethod:: start_tls(sslcontext, \*, \ + server_hostname=None, \ + ssl_handshake_timeout=None) + + Upgrade an existing transport-based connection to TLS. + + For more details about arguments and implementation see + :meth:`loop.start_tls`. + + .. versionadded:: 3.8 + Examples ======== From 35e29dff51f39f53690ee4113a25a811cb2590c3 Mon Sep 17 00:00:00 2001 From: Andrew Svetlov Date: Sat, 15 Sep 2018 03:03:30 -0700 Subject: [PATCH 11/11] Fix SSL warning --- Lib/asyncio/streams.py | 1 + 1 file changed, 1 insertion(+) diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py index 174d29db56df80..8817828ee57a8e 100644 --- a/Lib/asyncio/streams.py +++ b/Lib/asyncio/streams.py @@ -809,3 +809,4 @@ async def start_tls(self, sslcontext, *, ssl_handshake_timeout=ssl_handshake_timeout) self._transport = transport self._protocol._transport = transport + self._over_ssl = True