Skip to content

C
cpython
Commit 8be876e4 authored by Yury Selivanov's avatar Yury Selivanov Committed by GitHub
Browse files
Options

bpo-33649: Cleanup asyncio/streams and asyncio/synchronization docs (GH-9192)

parent fa7dfae3
Hide whitespace changes
Inline Side-by-side
Showing with 263 additions and 286 deletions
Doc/library/asyncio-stream.rst
View file @ 8be876e4
......@@ -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 <asyncio-multithreading>`.
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 <asyncio-multithreading>`.
.. 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 <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
......
Doc/library/asyncio-sync.rst
View file @ 8be876e4
.. 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 <async-with-locks>`.
which is equivalent to::
This class is :ref:`not thread safe <asyncio-multithreading>`.
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 <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 <asyncio-multithreading>`.
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 <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
<async-with-locks>`.
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 <asyncio-multithreading>`.
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 <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 <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 <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
<async-with-locks>`.
which is equivalent to::
This class is :ref:`not thread safe <asyncio-multithreading>`.
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 <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 <async-with-locks>`.
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 <asyncio-multithreading>`.
---------
.. _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.
Doc/whatsnew/3.7.rst
View file @ 8be876e4
......@@ -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`
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7