mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-22 09:45:06 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

bpo-32193: Convert asyncio to async/await usage (#4753)

Browse source 
* Convert asyncio/tasks.py to async/await

* Convert asyncio/queues.py to async/await

* Convert asyncio/test_utils.py to async/await

* Convert asyncio/base_subprocess.py to async/await

* Convert asyncio/subprocess.py to async/await

* Convert asyncio/streams.py to async/await

* Fix comments

* Convert asyncio/locks.py to async/await

* Convert asyncio.sleep to async def

* Add a comment

* Add missing news

* Convert stubs from AbstrctEventLoop to async functions

* Convert subprocess_shell/subprocess_exec

* Convert connect_read_pipe/connect_write_pip to async/await syntax

* Convert create_datagram_endpoint

* Convert create_unix_server/create_unix_connection

* Get rid of old style coroutines in unix_events.py

* Convert selector_events.py to async/await

* Convert wait_closed and create_connection

* Drop redundant line

* Convert base_events.py

* Code cleanup

* Drop redundant comments

* Fix indentation

* Add explicit tests for compatibility between old and new coroutines

* Convert windows event loop to use async/await

* Fix double awaiting of async function

* Convert asyncio/locks.py

* Improve docstring

* Convert tests to async/await

* Convert more tests

* Convert more tests

* Convert more tests

* Convert tests

* Improve test
This commit is contained in:
Andrew Svetlov 2017-12-09 00:23:48 +02:00 • committed by GitHub
parent ede157331b
commit 5f841b5538
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
22 changed files with 647 additions and 771 deletions
Download patch file Download diff file Expand all files Collapse all files

78
Lib/asyncio/streams.py
View file

@ -14,8 +14,8 @@ if hasattr(socket, 'AF_UNIX'):
from . import coroutines
from . import events
from . import protocols
from .coroutines import coroutine
from .log import logger
from .tasks import sleep
_DEFAULT_LIMIT = 2 ** 16
@ -52,9 +52,8 @@ class LimitOverrunError(Exception):
return type(self), (self.args[0], self.consumed)
@coroutine
def open_connection(host=None, port=None, *,
loop=None, limit=_DEFAULT_LIMIT, **kwds):
async def open_connection(host=None, port=None, *,
loop=None, limit=_DEFAULT_LIMIT, **kwds):
"""A wrapper for create_connection() returning a (reader, writer) pair.
The reader returned is a StreamReader instance; the writer is a
@ -76,15 +75,14 @@ def open_connection(host=None, port=None, *,
loop = events.get_event_loop()
reader = StreamReader(limit=limit, loop=loop)
protocol = StreamReaderProtocol(reader, loop=loop)
transport, _ = yield from loop.create_connection(
transport, _ = await loop.create_connection(
lambda: protocol, host, port, **kwds)
writer = StreamWriter(transport, protocol, reader, loop)
return reader, writer
@coroutine
def start_server(client_connected_cb, host=None, port=None, *,
loop=None, limit=_DEFAULT_LIMIT, **kwds):
async def start_server(client_connected_cb, host=None, port=None, *,
loop=None, limit=_DEFAULT_LIMIT, **kwds):
"""Start a socket server, call back for each client connected.
The first parameter, `client_connected_cb`, takes two parameters:
@ -115,28 +113,26 @@ def start_server(client_connected_cb, host=None, port=None, *,
loop=loop)
return protocol
return (yield from loop.create_server(factory, host, port, **kwds))
return await loop.create_server(factory, host, port, **kwds)
if hasattr(socket, 'AF_UNIX'):
# UNIX Domain Sockets are supported on this platform
@coroutine
def open_unix_connection(path=None, *,
loop=None, limit=_DEFAULT_LIMIT, **kwds):
async def open_unix_connection(path=None, *,
loop=None, limit=_DEFAULT_LIMIT, **kwds):
"""Similar to `open_connection` but works with UNIX Domain Sockets."""
if loop is None:
loop = events.get_event_loop()
reader = StreamReader(limit=limit, loop=loop)
protocol = StreamReaderProtocol(reader, loop=loop)
transport, _ = yield from loop.create_unix_connection(
transport, _ = await loop.create_unix_connection(
lambda: protocol, path, **kwds)
writer = StreamWriter(transport, protocol, reader, loop)
return reader, writer
@coroutine
def start_unix_server(client_connected_cb, path=None, *,
loop=None, limit=_DEFAULT_LIMIT, **kwds):
async def start_unix_server(client_connected_cb, path=None, *,
loop=None, limit=_DEFAULT_LIMIT, **kwds):
"""Similar to `start_server` but works with UNIX Domain Sockets."""
if loop is None:
loop = events.get_event_loop()
@ -147,7 +143,7 @@ if hasattr(socket, 'AF_UNIX'):
loop=loop)
return protocol
return (yield from loop.create_unix_server(factory, path, **kwds))
return await loop.create_unix_server(factory, path, **kwds)
class FlowControlMixin(protocols.Protocol):
@ -203,8 +199,7 @@ class FlowControlMixin(protocols.Protocol):
else:
waiter.set_exception(exc)
@coroutine
def _drain_helper(self):
async def _drain_helper(self):
if self._connection_lost:
raise ConnectionResetError('Connection lost')
if not self._paused:
@ -213,7 +208,7 @@ class FlowControlMixin(protocols.Protocol):
assert waiter is None or waiter.cancelled()
waiter = self._loop.create_future()
self._drain_waiter = waiter
yield from waiter
await waiter
class StreamReaderProtocol(FlowControlMixin, protocols.Protocol):
@ -313,14 +308,13 @@ class StreamWriter:
def get_extra_info(self, name, default=None):
return self._transport.get_extra_info(name, default)
@coroutine
def drain(self):
async def drain(self):
"""Flush the write buffer.
The intended use is to write
w.write(data)
yield from w.drain()
await w.drain()
"""
if self._reader is not None:
exc = self._reader.exception()
@ -331,11 +325,11 @@ class StreamWriter:
# Yield to the event loop so connection_lost() may be
# called. Without this, _drain_helper() would return
# immediately, and code that calls
# write(...); yield from drain()
# write(...); await drain()
# in a loop would never call connection_lost(), so it
# would not see an error when the socket is closed.
yield
yield from self._protocol._drain_helper()
await sleep(0, loop=self._loop)
await self._protocol._drain_helper()
class StreamReader:
@ -436,8 +430,7 @@ class StreamReader:
else:
self._paused = True
@coroutine
def _wait_for_data(self, func_name):
async def _wait_for_data(self, func_name):
"""Wait until feed_data() or feed_eof() is called.
If stream was paused, automatically resume it.
@ -460,12 +453,11 @@ class StreamReader:
self._waiter = self._loop.create_future()
try:
yield from self._waiter
await self._waiter
finally:
self._waiter = None
@coroutine
def readline(self):
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
@ -484,7 +476,7 @@ class StreamReader:
sep = b'\n'
seplen = len(sep)
try:
line = yield from self.readuntil(sep)
line = await self.readuntil(sep)
except IncompleteReadError as e:
return e.partial
except LimitOverrunError as e:
@ -496,8 +488,7 @@ class StreamReader:
raise ValueError(e.args[0])
return line
@coroutine
def readuntil(self, separator=b'\n'):
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
@ -577,7 +568,7 @@ class StreamReader:
raise IncompleteReadError(chunk, None)
# _wait_for_data() will resume reading if stream was paused.
yield from self._wait_for_data('readuntil')
await self._wait_for_data('readuntil')
if isep > self._limit:
raise LimitOverrunError(
@ -588,8 +579,7 @@ class StreamReader:
self._maybe_resume_transport()
return bytes(chunk)
@coroutine
def read(self, n=-1):
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
@ -623,14 +613,14 @@ class StreamReader:
# bytes. So just call self.read(self._limit) until EOF.
blocks = []
while True:
block = yield from self.read(self._limit)
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:
yield from self._wait_for_data('read')
await self._wait_for_data('read')
# This will work right even if buffer is less than n bytes
data = bytes(self._buffer[:n])
@ -639,8 +629,7 @@ class StreamReader:
self._maybe_resume_transport()
return data
@coroutine
def readexactly(self, n):
async def readexactly(self, n):
"""Read exactly `n` bytes.
Raise an IncompleteReadError if EOF is reached before `n` bytes can be
@ -670,7 +659,7 @@ class StreamReader:
self._buffer.clear()
raise IncompleteReadError(incomplete, n)
yield from self._wait_for_data('readexactly')
await self._wait_for_data('readexactly')
if len(self._buffer) == n:
data = bytes(self._buffer)
@ -684,9 +673,8 @@ class StreamReader:
def __aiter__(self):
return self
@coroutine
def __anext__(self):
val = yield from self.readline()
async def __anext__(self):
val = await self.readline()
if val == b'':
raise StopAsyncIteration
return val