18.5.9. Develop with asyncio

    The implementation of asyncio has been written for performance. In order to ease the development of asynchronous code, you may wish to enable debug mode.

    To enable all debug checks for an application:

    • Enable the asyncio debug mode globally by setting the environment variable to , or by calling AbstractEventLoop.set_debug().

    • Set the log level of the to logging.DEBUG. For example, call logging.basicConfig(level=logging.DEBUG) at startup.

    • Configure the warnings module to display warnings. For example, use the -Wdefault command line option of Python to display them.

    Examples debug checks:

    • Log coroutines defined but never “yielded from”

    • and call_at() methods raise an exception if they are called from the wrong thread.

    • Log the execution time of the selector

    • Log callbacks taking more than 100 ms to be executed. The AbstractEventLoop.slow_callback_duration attribute is the minimum duration in seconds of “slow” callbacks.

    • warnings are emitted when transports and event loops are not closed explicitly.

    参见

    The method and the asyncio logger.

    18.5.9.2. Cancellation

    Cancellation of tasks is not common in classic programming. In asynchronous programming, not only is it something common, but you have to prepare your code to handle it.

    Futures and tasks can be cancelled explicitly with their Future.cancel() method. The function cancels the waited task when the timeout occurs. There are many other cases where a task can be cancelled indirectly.

    Don’t call set_result() or method of Future if the future is cancelled: it would fail with an exception. For example, write:

    Don’t schedule directly a call to the or the set_exception() method of a future with : the future can be cancelled before its method is called.

    If you wait for a future, you should check early if the future was cancelled to avoid useless operations. Example:

    1. @coroutine
    2. def slow_operation(fut):
    3.     if fut.cancelled():
    4.         return
    5.     # ... slow computation ...
    6.     yield from fut
    7.     # ...

    The shield() function can also be used to ignore cancellation.

    18.5.9.3. Concurrency and multithreading

    An event loop runs in a thread and executes all callbacks and tasks in the same thread. While a task is running in the event loop, no other task is running in the same thread. But when the task uses yield from, the task is suspended and the event loop executes the next task.

    1. loop.call_soon_threadsafe(callback, *args)

    Most asyncio objects are not thread safe. You should only worry if you access objects outside the event loop. For example, to cancel a future, don’t call directly its Future.cancel() method, but:

    1. loop.call_soon_threadsafe(fut.cancel)

    为了能够处理信号和执行子进程，事件循环必须运行于主线程中。

    To schedule a coroutine object from a different thread, the function should be used. It returns a concurrent.futures.Future to access the result:

    1. future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
    2. result = future.result(timeout)  # Wait for the result with a timeout

    The method can be used with a thread pool executor to execute a callback in different thread to not block the thread of the event loop.

    参见

    The Synchronization primitives section describes ways to synchronize tasks.

    The section lists asyncio limitations to run subprocesses from different threads.

    Blocking functions should not be called directly. For example, if a function blocks for 1 second, other tasks are delayed by 1 second which can have an important impact on reactivity.

    For networking and subprocesses, the asyncio module provides high-level APIs like .

    An executor can be used to run a task in a different thread or even in a different process, to not block the thread of the event loop. See the AbstractEventLoop.run_in_executor() method.

    参见

    The section details how the event loop handles time.

    18.5.9.5. 日志记录

    The module logs information with the logging module in the logger 'asyncio'.

    The default log level for the module is logging.INFO. For those not wanting such verbosity from asyncio the log level can be changed. For example, to change the level to logging.WARNING:

    1. logging.getLogger('asyncio').setLevel(logging.WARNING)

    18.5.9.6. Detect coroutine objects never scheduled

    When a coroutine function is called and its result is not passed to ensure_future() or to the method, the execution of the coroutine object will never be scheduled which is probably a bug. Enable the debug mode of asyncio to to detect it.

    Example with the bug:

    调试模式的输出:

    1. Coroutine test() at test.py:3 was never yielded from
    2. Coroutine object created at (most recent call last):
    3.   File "test.py", line 7, in <module>
    4.     test()

    The fix is to call the ensure_future() function or the method with the coroutine object.

    参见

    Pending task destroyed.

    Example of unhandled exception:

    1. import asyncio
    3. @asyncio.coroutine
    4. def bug():
    5.     raise Exception("not consumed")
    7. loop = asyncio.get_event_loop()
    8. asyncio.ensure_future(bug())
    9. loop.run_forever()
    10. loop.close()

    输出:

    1. Task exception was never retrieved
    2. Traceback (most recent call last):
    3.   File "asyncio/tasks.py", line 237, in _step
    4.     result = next(coro)
    5.   File "asyncio/coroutines.py", line 141, in coro
    6.     res = func(*args, **kw)
    7.   File "test.py", line 5, in bug
    8.     raise Exception("not consumed")
    9. Exception: not consumed

    to get the traceback where the task was created. Output in debug mode:

    1. Task exception was never retrieved
    3. source_traceback: Object created at (most recent call last):
    4.   File "test.py", line 8, in <module>
    5.     asyncio.ensure_future(bug())
    6. Traceback (most recent call last):
    7.   File "asyncio/tasks.py", line 237, in _step
    8.     result = next(coro)
    9.   File "asyncio/coroutines.py", line 79, in __next__
    10.     return next(self.gen)
    11.   File "asyncio/coroutines.py", line 141, in coro
    12.     res = func(*args, **kw)
    13.   File "test.py", line 5, in bug
    14.     raise Exception("not consumed")
    15. Exception: not consumed

    There are different options to fix this issue. The first option is to chain the coroutine in another coroutine and use classic try/except:

    1. @asyncio.coroutine
    2. def handle_exception():
    3.     try:
    4.         yield from bug()
    5.     except Exception:
    6.         print("exception consumed")
    8. loop = asyncio.get_event_loop()
    9. asyncio.ensure_future(handle_exception())
    10. loop.run_forever()
    11. loop.close()

    Another option is to use the AbstractEventLoop.run_until_complete() function:

    参见

    The method.

    18.5.9.8. Chain coroutines correctly

    When a coroutine function calls other coroutine functions and tasks, they should be chained explicitly with yield from. Otherwise, the execution is not guaranteed to be sequential.

    Example with different bugs using to simulate slow operations:

    1. import asyncio
    3. @asyncio.coroutine
    4. def create():
    5.     yield from asyncio.sleep(3.0)
    6.     print("(1) create file")
    8. @asyncio.coroutine
    9. def write():
    10.     yield from asyncio.sleep(1.0)
    11.     print("(2) write into file")
    12. @asyncio.coroutine
    13. def close():
    14.     print("(3) close file")
    16. @asyncio.coroutine
    17. def test():
    18.     asyncio.ensure_future(create())
    19.     asyncio.ensure_future(write())
    21.     yield from asyncio.sleep(2.0)
    22.     loop.stop()
    24. loop = asyncio.get_event_loop()
    25. asyncio.ensure_future(test())
    26. loop.run_forever()
    27. print("Pending tasks at exit: %s" % asyncio.Task.all_tasks(loop))
    28. loop.close()

    Expected output:

    1. (1) create file
    2. (2) write into file
    3. (3) close file
    4. Pending tasks at exit: set()

    Actual output:

    1. (3) close file
    2. (2) write into file
    3. Pending tasks at exit: {<Task pending create() at test.py:7 wait_for=<Future pending cb=[Task._wakeup()]>>}
    4. Task was destroyed but it is pending!
    5. task: <Task pending create() done at test.py:5 wait_for=<Future pending cb=[Task._wakeup()]>>

    The loop stopped before the create() finished, close() has been called before write(), whereas coroutine functions were called in this order: create(), write(), close().

    To fix the example, tasks must be marked with yield from:

    1. @asyncio.coroutine
    2. def test():
    3.     yield from asyncio.ensure_future(create())
    4.     yield from asyncio.ensure_future(write())
    5.     yield from asyncio.ensure_future(close())
    6.     yield from asyncio.sleep(2.0)
    7.     loop.stop()

    Or without asyncio.ensure_future():

    1. @asyncio.coroutine
    2. def test():
    3.     yield from create()
    4.     yield from write()
    5.     yield from close()
    6.     yield from asyncio.sleep(2.0)
    7.     loop.stop()

    18.5.9.9. Pending task destroyed

    If a pending task is destroyed, the execution of its wrapped did not complete. It is probably a bug and so a warning is logged.

    Example of log:

    Enable the debug mode of asyncio to get the traceback where the task was created. Example of log in debug mode:

    1. Task was destroyed but it is pending!
    2. source_traceback: Object created at (most recent call last):
    3.   File "test.py", line 15, in <module>
    4.     task = asyncio.ensure_future(coro, loop=loop)
    5. task: <Task pending coro=<kill_me() done, defined at test.py:5> wait_for=<Future pending cb=[Task._wakeup()] created at test.py:7> created at test.py:15>

    参见

    .

    When a transport is no more needed, call its close() method to release resources. Event loops must also be closed explicitly.