From 8be876e44b2dffb96b551029454cbfb16d697992 Mon Sep 17 00:00:00 2001 From: Yury Selivanov Date: Tue, 11 Sep 2018 17:10:37 -0700 Subject: [PATCH] bpo-33649: Cleanup asyncio/streams and asyncio/synchronization docs (GH-9192) --- Doc/library/asyncio-stream.rst | 186 +++++++---------- Doc/library/asyncio-sync.rst | 360 +++++++++++++++++---------------- Doc/whatsnew/3.7.rst | 3 +- 3 files changed, 263 insertions(+), 286 deletions(-) diff --git a/Doc/library/asyncio-stream.rst b/Doc/library/asyncio-stream.rst index 27b5205f1c..0cfecda91e 100644 --- a/Doc/library/asyncio-stream.rst +++ b/Doc/library/asyncio-stream.rst @@ -7,10 +7,10 @@ Streams ======= Streams are high-level async/await-ready primitives to work with -network connections. Streams allow send and receive data without +network connections. Streams allow sending and receiving data without using callbacks or low-level protocols and transports. -Here's an example of a TCP echo client written using asyncio +Here is an example of a TCP echo client written using asyncio streams:: import asyncio @@ -31,6 +31,9 @@ streams:: asyncio.run(tcp_echo_client('Hello World!')) +See also the `Examples`_ section below. + + .. rubric:: Stream Functions The following top-level asyncio functions can be used to create @@ -43,7 +46,7 @@ and work with streams: server_hostname=None, ssl_handshake_timeout=None) Establish a network connection and return a pair of - ``(reader, writer)``. + ``(reader, writer)`` objects. The returned *reader* and *writer* objects are instances of :class:`StreamReader` and :class:`StreamWriter` classes. @@ -52,7 +55,8 @@ and work with streams: automatically when this method is awaited from a coroutine. *limit* determines the buffer size limit used by the - returned :class:`StreamReader` instance. + returned :class:`StreamReader` instance. By default the *limit* + is set to 64 KiB. The rest of the arguments are passed directly to :meth:`loop.create_connection`. @@ -84,7 +88,8 @@ and work with streams: automatically when this method is awaited from a coroutine. *limit* determines the buffer size limit used by the - returned :class:`StreamReader` instance. + returned :class:`StreamReader` instance. By default the *limit* + is set to 64 KiB. The rest of the arguments are passed directly to :meth:`loop.create_server`. @@ -93,6 +98,9 @@ and work with streams: The *ssl_handshake_timeout* and *start_serving* parameters. + +.. rubric:: Unix Sockets + .. coroutinefunction:: open_unix_connection(path=None, \*, loop=None, \ limit=None, ssl=None, sock=None, \ server_hostname=None, ssl_handshake_timeout=None) @@ -114,6 +122,7 @@ and work with streams: The *path* parameter can now be a :term:`path-like object` + .. coroutinefunction:: start_unix_server(client_connected_cb, path=None, \ \*, loop=None, limit=None, sock=None, \ backlog=100, ssl=None, ssl_handshake_timeout=None, \ @@ -121,7 +130,7 @@ and work with streams: Start a UNIX socket server. - Similar to :func:`start_server` but operates on UNIX sockets. + Similar to :func:`start_server` but works with UNIX sockets. See also the documentation of :meth:`loop.create_unix_server`. @@ -136,67 +145,47 @@ and work with streams: The *path* parameter can now be a :term:`path-like object`. -.. rubric:: Contents - -* `StreamReader`_ and `StreamWriter`_ -* `StreamReaderProtocol`_ -* `Examples`_ +--------- StreamReader ============ -.. class:: StreamReader(limit=_DEFAULT_LIMIT, loop=None) - - This class is :ref:`not thread safe `. - - The *limit* argument's default value is set to _DEFAULT_LIMIT which is 2**16 (64 KiB) - - .. method:: exception() - - Get the exception. - - .. method:: feed_eof() +.. class:: StreamReader - Acknowledge the EOF. + Represents a reader object that provides APIs to read data + from the IO stream. - .. method:: feed_data(data) - - Feed *data* bytes in the internal buffer. Any operations waiting - for the data will be resumed. - - .. method:: set_exception(exc) - - Set the exception. - - .. method:: set_transport(transport) - - Set the transport. + It is not recommended to instantiate *StreamReader* objects + directly; use :func:`open_connection` and :func:`start_server` + instead. .. coroutinemethod:: read(n=-1) Read up to *n* bytes. 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, + If an EOF was received and the internal buffer is empty, return an empty ``bytes`` object. .. coroutinemethod:: readline() - Read one line, where "line" is a sequence of bytes ending with ``\n``. + Read one line, where "line" is a sequence of bytes + ending with ``\n``. - If EOF is received, and ``\n`` was not found, the method will - return the partial read bytes. + If an EOF is received and ``\n`` was not found, the method + returns partially read data. - If the EOF was received and the internal buffer is empty, + If an EOF is received and the internal buffer is empty, return an empty ``bytes`` object. .. coroutinemethod:: readexactly(n) - Read exactly *n* bytes. Raise an :exc:`IncompleteReadError` if the end of - the stream is reached before *n* can be read, the - :attr:`IncompleteReadError.partial` attribute of the exception contains - the partial read bytes. + Read exactly *n* bytes. + + Raise an :exc:`IncompleteReadError` if an EOF reached before *n* + can be read. Use the :attr:`IncompleteReadError.partial` + attribute to get the partially read data. .. coroutinemethod:: readuntil(separator=b'\\n') @@ -231,105 +220,76 @@ StreamReader StreamWriter ============ -.. class:: StreamWriter(transport, protocol, reader, loop) +.. class:: StreamWriter - Wraps a Transport. + Represents a writer object that provides APIs to write data + to the IO stream. - This exposes :meth:`write`, :meth:`writelines`, :meth:`can_write_eof()`, - :meth:`write_eof`, :meth:`get_extra_info` and :meth:`close`. It adds - :meth:`drain` which returns an optional :class:`Future` on which you can - wait for flow control. It also adds a transport attribute which references - the :class:`Transport` directly. + It is not recommended to instantiate *StreamWriter* objects + directly; use :func:`open_connection` and :func:`start_server` + instead. - This class is :ref:`not thread safe `. + .. method:: write(data) - .. attribute:: transport + Write *data* to the stream. - Transport. + .. method:: writelines(data) - .. method:: can_write_eof() + Write a list (or any iterable) of bytes to the stream. + + .. coroutinemethod:: drain() + + Wait until it is appropriate to resume writing to the stream. + E.g.:: + + writer.write(data) + await writer.drain() - Return :const:`True` if the transport supports :meth:`write_eof`, - :const:`False` if not. See :meth:`WriteTransport.can_write_eof`. + This is a flow-control method that interacts with the underlying + IO write buffer. When the size of the buffer reaches + the high-water limit, *drain()* blocks until the size of the + buffer is drained down to the low-water limit and writing can + be resumed. When there is nothing to wait for, the :meth:`drain` + returns immediately. .. method:: close() - Close the transport: see :meth:`BaseTransport.close`. + Close the stream. .. method:: is_closing() - Return ``True`` if the writer is closing or is closed. + Return ``True`` if the stream is closed or in the process of + being closed. .. versionadded:: 3.7 .. coroutinemethod:: wait_closed() - Wait until the writer is closed. + Wait until the stream is closed. - Should be called after :meth:`close` to wait until the underlying - connection (and the associated transport/protocol pair) is closed. + Should be called after :meth:`close` to wait until the underlying + connection is closed. .. versionadded:: 3.7 - .. coroutinemethod:: drain() - - Let the write buffer of the underlying transport a chance to be flushed. - - The intended use is to write:: - - w.write(data) - await w.drain() - - When the size of the transport buffer reaches the high-water limit (the - protocol is paused), block until the size of the buffer is drained down - to the low-water limit and the protocol is resumed. When there is nothing - to wait for, the yield-from continues immediately. - - Yielding from :meth:`drain` gives the opportunity for the loop to - schedule the write operation and flush the buffer. It should especially - be used when a possibly large amount of data is written to the transport, - and the coroutine does not yield-from between calls to :meth:`write`. - - This method is a :ref:`coroutine `. - - .. method:: get_extra_info(name, default=None) - - Return optional transport information: see - :meth:`BaseTransport.get_extra_info`. - - .. method:: write(data) - - Write some *data* bytes to the transport: see - :meth:`WriteTransport.write`. - - .. method:: writelines(data) + .. method:: can_write_eof() - Write a list (or any iterable) of data bytes to the transport: - see :meth:`WriteTransport.writelines`. + Return *True* if the underlying transport supports + the :meth:`write_eof` method, *False* otherwise. .. method:: write_eof() - Close the write end of the transport after flushing buffered data: - see :meth:`WriteTransport.write_eof`. - - -StreamReaderProtocol -==================== + Close the write end of the stream after the buffered write + data is flushed. -.. class:: StreamReaderProtocol(stream_reader, client_connected_cb=None, \ - loop=None) + .. attribute:: transport - Trivial helper class to adapt between :class:`Protocol` and - :class:`StreamReader`. Subclass of :class:`Protocol`. + Return the underlying asyncio transport. - *stream_reader* is a :class:`StreamReader` instance, *client_connected_cb* - is an optional function called with (stream_reader, stream_writer) when a - connection is made, *loop* is the event loop instance to use. + .. method:: get_extra_info(name, default=None) - (This is a helper class instead of making :class:`StreamReader` itself a - :class:`Protocol` subclass, because the :class:`StreamReader` has other - potential uses, and to prevent the user of the :class:`StreamReader` from - accidentally calling inappropriate methods of the protocol.) + Access optional transport information; see + :meth:`BaseTransport.get_extra_info` for details. Examples diff --git a/Doc/library/asyncio-sync.rst b/Doc/library/asyncio-sync.rst index 574f70f069..8e01ca9440 100644 --- a/Doc/library/asyncio-sync.rst +++ b/Doc/library/asyncio-sync.rst @@ -1,172 +1,204 @@ .. currentmodule:: asyncio + .. _asyncio-sync: -Synchronization primitives ========================== +Synchronization Primitives +========================== + +asyncio synchronization primitives are designed to be similar to +those of the :mod:`threading` module with two important caveats: + +* asyncio primitives are not thread-safe, therefore they should not + be used for OS threads synchronization (use :mod:`threading` for + that); -**Source code:** :source:`Lib/asyncio/locks.py` +* methods of synchronization objects do not accept the *timeout* + argument; use the :func:`asyncio.wait_for` function to perform + operations with timeouts. -Locks: +asyncio has the following basic primitives: * :class:`Lock` * :class:`Event` * :class:`Condition` - -Semaphores: - * :class:`Semaphore` * :class:`BoundedSemaphore` -asyncio lock API was designed to be close to classes of the :mod:`threading` -module (:class:`~threading.Lock`, :class:`~threading.Event`, -:class:`~threading.Condition`, :class:`~threading.Semaphore`, -:class:`~threading.BoundedSemaphore`), but it has no *timeout* parameter. The -:func:`asyncio.wait_for` function can be used to cancel a task after a timeout. + +--------- Lock ----- +==== .. class:: Lock(\*, loop=None) - Primitive lock objects. + Implements a mutex lock for asyncio tasks. Not thread-safe. - A primitive lock is a synchronization primitive that is not owned by a - particular coroutine when locked. A primitive lock is in one of two states, - 'locked' or 'unlocked'. + An asyncio lock can be used to guarantee exclusive access to a + shared resource. - The lock is created in the unlocked state. - It has two basic methods, :meth:`acquire` and :meth:`release`. - When the state is unlocked, acquire() changes the state to - locked and returns immediately. When the state is locked, acquire() blocks - until a call to release() in another coroutine changes it to unlocked, then - the acquire() call resets it to locked and returns. The release() method - should only be called in the locked state; it changes the state to unlocked - and returns immediately. If an attempt is made to release an unlocked lock, - a :exc:`RuntimeError` will be raised. + The preferred way to use a Lock is an :keyword:`async with` + statement:: - When more than one coroutine is blocked in acquire() waiting for the state - to turn to unlocked, only one coroutine proceeds when a release() call - resets the state to unlocked; first coroutine which is blocked in acquire() - is being processed. + lock = asyncio.Lock() - :meth:`acquire` is a coroutine and should be called with ``await``. + # ... later + async with lock: + # access shared state - Locks support the :ref:`context management protocol `. + which is equivalent to:: - This class is :ref:`not thread safe `. + lock = asyncio.Lock() - .. method:: locked() - - Return ``True`` if the lock is acquired. + # ... later + await lock.acquire() + try: + # access shared state + finally: + lock.release() .. coroutinemethod:: acquire() - Acquire a lock. - - This method blocks until the lock is unlocked, then sets it to locked and - returns ``True``. + Acquire the lock. - This method is a :ref:`coroutine `. + This method waits until the lock is *unlocked*, sets it to + *locked* and returns ``True``. .. method:: release() - Release a lock. + Release the lock. - When the lock is locked, reset it to unlocked, and return. If any other - coroutines are blocked waiting for the lock to become unlocked, allow - exactly one of them to proceed. + When the lock is *locked*, reset it to *unlocked* and return. - When invoked on an unlocked lock, a :exc:`RuntimeError` is raised. + If the lock is *unlocked* a :exc:`RuntimeError` is raised. - There is no return value. + .. method:: locked() + + Return ``True`` if the lock is *locked*. Event ------ +===== .. class:: Event(\*, loop=None) - An Event implementation, asynchronous equivalent to :class:`threading.Event`. + An event object. Not thread-safe. - Class implementing event objects. An event manages a flag that can be set to - true with the :meth:`set` method and reset to false with the :meth:`clear` - method. The :meth:`wait` method blocks until the flag is true. The flag is - initially false. + An asyncio event can be used to notify multiple asyncio tasks + that some event has happened. - This class is :ref:`not thread safe `. + An Event object manages an internal flag that can be set to *true* + with the :meth:`set` method and reset to *false* with the + :meth:`clear` method. The :meth:`wait` method blocks until the + flag is set to *true*. The flag is set to *false* initially. - .. method:: clear() + Example:: - Reset the internal flag to false. Subsequently, coroutines calling - :meth:`wait` will block until :meth:`set` is called to set the internal - flag to true again. + async def waiter(event): + print('waiting ...') + await event.wait() + print('... got it!') - .. method:: is_set() + async def main(): + # Create an Event object. + event = asyncio.Event() - Return ``True`` if and only if the internal flag is true. + # Spawn a Task to wait until 'event' is set. + waiter_task = asyncio.create_task(waiter(event)) - .. method:: set() + # Sleep for 1 second and set the event. + await asyncio.sleep(1) + event.set() + + # Wait until the waiter task is finished. + await waiter_task - Set the internal flag to true. All coroutines waiting for it to become - true are awakened. Coroutine that call :meth:`wait` once the flag is true - will not block at all. + asyncio.run(main()) .. coroutinemethod:: wait() - Block until the internal flag is true. + Wait until the event is set. + + If the event is set, return ``True`` immediately. + Otherwise block until another task calls :meth:`set`. + + .. method:: set() + + Set the event. - If the internal flag is true on entry, return ``True`` immediately. - Otherwise, block until another coroutine calls :meth:`set` to set the - flag to true, then return ``True``. + All tasks waiting for event to be set will be immediately + awakened. + + .. method:: clear() - This method is a :ref:`coroutine `. + Clear (unset) the event. + + Tasks awaiting on :meth:`wait` will now block until the + :meth:`set` method is called again. + + .. method:: is_set() + + Return ``True`` if the event is set. Condition ---------- +========= .. class:: Condition(lock=None, \*, loop=None) - A Condition implementation, asynchronous equivalent to - :class:`threading.Condition`. + A Condition object. Not thread-safe. - This class implements condition variable objects. A condition variable - allows one or more coroutines to wait until they are notified by another - coroutine. + An asyncio condition primitive can be used by a task to wait for + some event to happen and then get an exclusive access to a shared + resource. - If the *lock* argument is given and not ``None``, it must be a :class:`Lock` - object, and it is used as the underlying lock. Otherwise, - a new :class:`Lock` object is created and used as the underlying lock. + In essence, a Condition object combines the functionality + of :class:`Event` and :class:`Lock`. It is possible to have many + Condition objects sharing one Lock, which allows to coordinate + exclusive access to a shared resource between different tasks + interested in particular states of that shared resource. - Conditions support the :ref:`context management protocol - `. + The optional *lock* argument must be a :class:`Lock` object or + ``None``. In the latter case a new Lock object is created + automatically. - This class is :ref:`not thread safe `. + The preferred way to use a Condition is an :keyword:`async with` + statement:: - .. coroutinemethod:: acquire() + cond = asyncio.Condition() - Acquire the underlying lock. + # ... later + async with cond: + await cond.wait() - This method blocks until the lock is unlocked, then sets it to locked and - returns ``True``. + which is equivalent to:: - This method is a :ref:`coroutine `. + cond = asyncio.Condition() - .. method:: notify(n=1) + # ... later + await lock.acquire() + try: + await cond.wait() + finally: + lock.release() - By default, wake up one coroutine waiting on this condition, if any. - If the calling coroutine has not acquired the lock when this method is - called, a :exc:`RuntimeError` is raised. + .. coroutinemethod:: acquire() - This method wakes up at most *n* of the coroutines waiting for the - condition variable; it is a no-op if no coroutines are waiting. + Acquire the underlying lock. + + This method waits until the underlying lock is *unlocked*, + sets it to *locked* and returns ``True``. + + .. method:: notify(n=1) - .. note:: + Wake up at most *n* tasks (1 by default) waiting on this + condition. The method is no-op if no tasks are waiting. - An awakened coroutine does not actually return from its :meth:`wait` - call until it can reacquire the lock. Since :meth:`notify` does not - release the lock, its caller should. + The lock must be acquired before this method is called and + released shortly after. If called with an *unlocked* lock + a :exc:`RuntimeError` error is raised. .. method:: locked() @@ -174,78 +206,87 @@ Condition .. method:: notify_all() - Wake up all coroutines waiting on this condition. This method acts like - :meth:`notify`, but wakes up all waiting coroutines instead of one. If the - calling coroutine has not acquired the lock when this method is called, a - :exc:`RuntimeError` is raised. + Wake up all tasks waiting on this condition. - .. method:: release() + This method acts like :meth:`notify`, but wakes up all waiting + tasks. - Release the underlying lock. + The lock must be acquired before this method is called and + released shortly after. If called with an *unlocked* lock + a :exc:`RuntimeError` error is raised. - When the lock is locked, reset it to unlocked, and return. If any other - coroutines are blocked waiting for the lock to become unlocked, allow - exactly one of them to proceed. + .. method:: release() - When invoked on an unlocked lock, a :exc:`RuntimeError` is raised. + Release the underlying lock. - There is no return value. + When invoked on an unlocked lock, a :exc:`RuntimeError` is + raised. .. coroutinemethod:: wait() Wait until notified. - If the calling coroutine has not acquired the lock when this method is + If the calling task has not acquired the lock when this method is called, a :exc:`RuntimeError` is raised. - This method releases the underlying lock, and then blocks until it is - awakened by a :meth:`notify` or :meth:`notify_all` call for the same - condition variable in another coroutine. Once awakened, it re-acquires - the lock and returns ``True``. - - This method is a :ref:`coroutine `. + This method releases the underlying lock, and then blocks until + it is awakened by a :meth:`notify` or :meth:`notify_all` call. + Once awakened, the Condition re-acquires its lock and this method + returns ``True``. .. coroutinemethod:: wait_for(predicate) - Wait until a predicate becomes true. - - The predicate should be a callable which result will be interpreted as a - boolean value. The final predicate value is the return value. + Wait until a predicate becomes *true*. - This method is a :ref:`coroutine `. + The predicate must be a callable which result will be + interpreted as a boolean value. The final value is the + return value. Semaphore ---------- +========= .. class:: Semaphore(value=1, \*, loop=None) - A Semaphore implementation. + A Semaphore object. Not thread-safe. A semaphore manages an internal counter which is decremented by each - :meth:`acquire` call and incremented by each :meth:`release` call. The - counter can never go below zero; when :meth:`acquire` finds that it is zero, - it blocks, waiting until some other coroutine calls :meth:`release`. + :meth:`acquire` call and incremented by each :meth:`release` call. + The counter can never go below zero; when :meth:`acquire` finds + that it is zero, it blocks, waiting until some task calls + :meth:`release`. + + The optional *value* argument gives the initial value for the + internal counter (``1`` by default). If the given value is + less than ``0`` a :exc:`ValueError` is raised. + + The preferred way to use a Semaphore is an :keyword:`async with` + statement:: + + sem = asyncio.Semaphore(10) - The optional argument gives the initial value for the internal counter; it - defaults to ``1``. If the value given is less than ``0``, :exc:`ValueError` - is raised. + # ... later + async with sem: + # work with shared resource - Semaphores support the :ref:`context management protocol - `. + which is equivalent to:: - This class is :ref:`not thread safe `. + sem = asyncio.Semaphore(10) + + # ... later + await sem.acquire() + try: + # work with shared resource + finally: + sem.release() .. coroutinemethod:: acquire() Acquire a semaphore. - If the internal counter is larger than zero on entry, decrement it by one - and return ``True`` immediately. If it is zero on entry, block, waiting - until some other coroutine has called :meth:`release` to make it larger - than ``0``, and then return ``True``. - - This method is a :ref:`coroutine `. + If the internal counter is greater than zero, decrement + it by one and return ``True`` immediately. If it is zero wait + until a :meth:`release` is called and return ``True``. .. method:: locked() @@ -253,53 +294,30 @@ Semaphore .. method:: release() - Release a semaphore, incrementing the internal counter by one. When it - was zero on entry and another coroutine is waiting for it to become - larger than zero again, wake up that coroutine. + Release a semaphore, incrementing the internal counter by one. + Can wake up a task waiting to acquire the semaphore. + + Unlike :class:`BoundedSemaphore`, :class:`Semaphore` allows + to make more ``release()`` calls than ``acquire()`` calls. BoundedSemaphore ----------------- +================ .. class:: BoundedSemaphore(value=1, \*, loop=None) - A bounded semaphore implementation. Inherit from :class:`Semaphore`. - - This raises :exc:`ValueError` in :meth:`~Semaphore.release` if it would - increase the value above the initial value. + A bounded semaphore object. Not thread-safe. - Bounded semaphores support the :ref:`context management - protocol `. + Bounded Semaphore is a version of :class:`Semaphore` that raises + a :exc:`ValueError` in :meth:`~Semaphore.release` if it + increases the internal counter above the initial *value*. - This class is :ref:`not thread safe `. +--------- -.. _async-with-locks: - -Using locks, conditions and semaphores in the :keyword:`async with` statement ------------------------------------------------------------------------------ - -:class:`Lock`, :class:`Condition`, :class:`Semaphore`, and -:class:`BoundedSemaphore` objects can be used in :keyword:`async with` -statements. - -The :meth:`acquire` method will be called when the block is entered, -and :meth:`release` will be called when the block is exited. Hence, -the following snippet:: - - async with lock: - # do something... - -is equivalent to:: - - await lock.acquire() - try: - # do something... - finally: - lock.release() .. deprecated:: 3.7 - Lock acquiring using ``await lock`` or ``yield from lock`` and + Acquiring a lock using ``await lock`` or ``yield from lock`` and/or :keyword:`with` statement (``with await lock``, ``with (yield from - lock)``) are deprecated. + lock)``) is deprecated. Use ``async with lock`` instead. diff --git a/Doc/whatsnew/3.7.rst b/Doc/whatsnew/3.7.rst index fbaa2cf15f..be4fef1819 100644 --- a/Doc/whatsnew/3.7.rst +++ b/Doc/whatsnew/3.7.rst @@ -1923,8 +1923,7 @@ asyncio Support for directly ``await``-ing instances of :class:`asyncio.Lock` and other asyncio synchronization primitives has been deprecated. An asynchronous context manager must be used in order to acquire and release -the synchronization resource. See :ref:`async-with-locks` for more -information. +the synchronization resource. (Contributed by Andrew Svetlov in :issue:`32253`.) The :meth:`asyncio.Task.current_task` and :meth:`asyncio.Task.all_tasks` -- 2.40.0