Learn Python Coroutines in Less Than 5 Minutes
B efore I explain Coroutines in Python I would like to explain the subject. We implement applications using classes and functions. These functions execute business logic as a sequence of machine instructions. The functions in turn call other functions to deliver the complete solution. It is important to note that all code gets executed without any suspension. These general purpose functions are also knows as subroutines. Let’s build clarity with a simple program which generates a countdown and performs a divisibility test on the numbers.
First Coroutine
On the other hand coroutines are functions which can be suspended and resumed multiple times. The suspended execution enables Coroutines to build concurrency in an application. Coroutines have use cases in state-machine, Actors, Event loops, Server listeners etc.
Concurrency is defined the concept of execution of multiple functions in an interleaved manner. It is different from parallelism, which is defined as the simultaneous execution by utilising multi-core CPU architecture. Concurrency is often achieved with asynchronous data transfer or by using event based callbacks.
Python supports coroutines since 2.5 version. It improved the support in 3.5 by introducing async and await syntax. Coroutines are executed by using the asyncio package.
- The async keyword is added to the method definition. It signifies that the method can suspend execution to wait for new events.
- The await keyword suspends method execution and yields back the CPU to execute next task. The method resumes execution from the await statement after the invoking event has occurred.
Let’s now re-implement the above program using Coroutines.
We created a coroutine for the countdown function. Coroutines can’t be invoked directly as function calls. They need to be invoked using an external implementation like asyncio. Try to invoke tasks function and it will fail with the error coroutine ‘tasks’ was never awaited
The tasks coroutine provides important details of coroutine execution.. In order to execute our countdown coroutine we had to build a asyncio task. We then waited for the completion of the task. The tasks were executed by the asyncio engine after we invoked the run method. It is important to note that both the tasks get executed sequentially, one after other. Since the tasks do not give back control so there is no element of concurrency here.
Enable Concurrency
Concurrency can only be added if the methods suspend execution and give up CPU. In order to give-up cpu we add a sleep to the divisibility method. We can timer.sleep as it would consume the cpu. Instead we use asyncio.sleep method for the same. As we do this we convert the divisibleBy subroutine into a coroutine. It would need to have async keyword in its definitions and the invocation will need to be awaited.
The above code produces concurrent output from both the tasks.
Producer-consumer Coroutines
The asyncio python package is quite versatile. It not only offers invocation of coroutines, but also supports various execution scenarios. We can perform data transfer between various coroutines using Queues. Lets re-look our example using queue.
The above code changes the solution of countdown and divisibleBy to a producer-consumer scenario. The countdown function is a producer which adds number to the queue. Next, we added two divisibleBy functions which are consumers of the numbers on the queue. In the end we are no longer awaiting for tasks to finish. Instead we have joined on the queue. The join functions makes sure that we await until the queue becomes empty. But, we await on the countdown task to finish. This would make sure that all numbers have been added to queue. If we do not await on it the program would exit, without any result. This is because no numbers will be added when the queue is joined on.
I have shown on queues but asyncio also have events and synchronisation constructs to support various other use-cases like state-transitions. It also enables us to build worker based event loops to execute tasks as well as processes. I had just scratched the surface of asyncio package. The above example was not very useful. It was only intended to explain the concepts. The package provides a number of APIs which can be used to build many use-cases like network listeners, batch processing engine, state-machines etc.
Coroutines and Tasks¶
This section outlines high-level asyncio APIs to work with coroutines and Tasks.
Coroutines¶
Coroutines declared with the async/await syntax is the preferred way of writing asyncio applications. For example, the following snippet of code prints “hello”, waits 1 second, and then prints “world”:
Note that simply calling a coroutine will not schedule it to be executed:
To actually run a coroutine, asyncio provides the following mechanisms:
The asyncio.run() function to run the top-level entry point “main()” function (see the above example.)
Awaiting on a coroutine. The following snippet of code will print “hello” after waiting for 1 second, and then print “world” after waiting for another 2 seconds:
The asyncio.create_task() function to run coroutines concurrently as asyncio Tasks .
Let’s modify the above example and run two say_after coroutines concurrently:
Note that expected output now shows that the snippet runs 1 second faster than before:
The asyncio.TaskGroup class provides a more modern alternative to create_task() . Using this API, the last example becomes:
The timing and output should be the same as for the previous version.
Awaitables¶
We say that an object is an awaitable object if it can be used in an await expression. Many asyncio APIs are designed to accept awaitables.
There are three main types of awaitable objects: coroutines, Tasks, and Futures.
Python coroutines are awaitables and therefore can be awaited from other coroutines:
In this documentation the term “coroutine” can be used for two closely related concepts:
a coroutine function: an async def function;
a coroutine object: an object returned by calling a coroutine function.
Tasks are used to schedule coroutines concurrently.
When a coroutine is wrapped into a Task with functions like asyncio.create_task() the coroutine is automatically scheduled to run soon:
A Future is a special low-level awaitable object that represents an eventual result of an asynchronous operation.
When a Future object is awaited it means that the coroutine will wait until the Future is resolved in some other place.
Future objects in asyncio are needed to allow callback-based code to be used with async/await.
Normally there is no need to create Future objects at the application level code.
Future objects, sometimes exposed by libraries and some asyncio APIs, can be awaited:
A good example of a low-level function that returns a Future object is loop.run_in_executor() .
Creating Tasks¶
asyncio. create_task ( coro , * , name = None , context = None ) ¶
Wrap the coro coroutine into a Task and schedule its execution. Return the Task object.
If name is not None , it is set as the name of the task using Task.set_name() .
An optional keyword-only context argument allows specifying a custom contextvars.Context for the coro to run in. The current context copy is created when no context is provided.
The task is executed in the loop returned by get_running_loop() , RuntimeError is raised if there is no running loop in current thread.
asyncio.TaskGroup.create_task() is a newer alternative that allows for convenient waiting for a group of related tasks.
Save a reference to the result of this function, to avoid a task disappearing mid-execution. The event loop only keeps weak references to tasks. A task that isn’t referenced elsewhere may get garbage collected at any time, even before it’s done. For reliable “fire-and-forget” background tasks, gather them in a collection:
New in version 3.7.
Changed in version 3.8: Added the name parameter.
Changed in version 3.11: Added the context parameter.
Task Cancellation¶
Tasks can easily and safely be cancelled. When a task is cancelled, asyncio.CancelledError will be raised in the task at the next opportunity.
It is recommended that coroutines use try/finally blocks to robustly perform clean-up logic. In case asyncio.CancelledError is explicitly caught, it should generally be propagated when clean-up is complete. Most code can safely ignore asyncio.CancelledError .
The asyncio components that enable structured concurrency, like asyncio.TaskGroup and asyncio.timeout() , are implemented using cancellation internally and might misbehave if a coroutine swallows asyncio.CancelledError . Similarly, user code should not call uncancel .
Task Groups¶
Task groups combine a task creation API with a convenient and reliable way to wait for all tasks in the group to finish.
class asyncio. TaskGroup ¶
An asynchronous context manager holding a group of tasks. Tasks can be added to the group using create_task() . All tasks are awaited when the context manager exits.
New in version 3.11.
Create a task in this task group. The signature matches that of asyncio.create_task() .
The async with statement will wait for all tasks in the group to finish. While waiting, new tasks may still be added to the group (for example, by passing tg into one of the coroutines and calling tg.create_task() in that coroutine). Once the last task has finished and the async with block is exited, no new tasks may be added to the group.
The first time any of the tasks belonging to the group fails with an exception other than asyncio.CancelledError , the remaining tasks in the group are cancelled. No further tasks can then be added to the group. At this point, if the body of the async with statement is still active (i.e., __aexit__() hasn’t been called yet), the task directly containing the async with statement is also cancelled. The resulting asyncio.CancelledError will interrupt an await , but it will not bubble out of the containing async with statement.
Once all tasks have finished, if any tasks have failed with an exception other than asyncio.CancelledError , those exceptions are combined in an ExceptionGroup or BaseExceptionGroup (as appropriate; see their documentation) which is then raised.
Two base exceptions are treated specially: If any task fails with KeyboardInterrupt or SystemExit , the task group still cancels the remaining tasks and waits for them, but then the initial KeyboardInterrupt or SystemExit is re-raised instead of ExceptionGroup or BaseExceptionGroup .
If the body of the async with statement exits with an exception (so __aexit__() is called with an exception set), this is treated the same as if one of the tasks failed: the remaining tasks are cancelled and then waited for, and non-cancellation exceptions are grouped into an exception group and raised. The exception passed into __aexit__() , unless it is asyncio.CancelledError , is also included in the exception group. The same special case is made for KeyboardInterrupt and SystemExit as in the previous paragraph.
Sleeping¶
Block for delay seconds.
If result is provided, it is returned to the caller when the coroutine completes.
sleep() always suspends the current task, allowing other tasks to run.
Setting the delay to 0 provides an optimized path to allow other tasks to run. This can be used by long-running functions to avoid blocking the event loop for the full duration of the function call.
Example of coroutine displaying the current date every second for 5 seconds:
Changed in version 3.10: Removed the loop parameter.
Running Tasks Concurrently¶
Run awaitable objects in the aws sequence concurrently.
If any awaitable in aws is a coroutine, it is automatically scheduled as a Task.
If all awaitables are completed successfully, the result is an aggregate list of returned values. The order of result values corresponds to the order of awaitables in aws.
If return_exceptions is False (default), the first raised exception is immediately propagated to the task that awaits on gather() . Other awaitables in the aws sequence won’t be cancelled and will continue to run.
If return_exceptions is True , exceptions are treated the same as successful results, and aggregated in the result list.
If gather() is cancelled, all submitted awaitables (that have not completed yet) are also cancelled.
If any Task or Future from the aws sequence is cancelled, it is treated as if it raised CancelledError – the gather() call is not cancelled in this case. This is to prevent the cancellation of one submitted Task/Future to cause other Tasks/Futures to be cancelled.
A more modern way to create and run tasks concurrently and wait for their completion is asyncio.TaskGroup .
If return_exceptions is False, cancelling gather() after it has been marked done won’t cancel any submitted awaitables. For instance, gather can be marked done after propagating an exception to the caller, therefore, calling gather.cancel() after catching an exception (raised by one of the awaitables) from gather won’t cancel any other awaitables.
Changed in version 3.7: If the gather itself is cancelled, the cancellation is propagated regardless of return_exceptions.
Changed in version 3.10: Removed the loop parameter.
Deprecated since version 3.10: Deprecation warning is emitted if no positional arguments are provided or not all positional arguments are Future-like objects and there is no running event loop.
Shielding From Cancellation¶
If aw is a coroutine it is automatically scheduled as a Task.
is equivalent to:
except that if the coroutine containing it is cancelled, the Task running in something() is not cancelled. From the point of view of something() , the cancellation did not happen. Although its caller is still cancelled, so the “await” expression still raises a CancelledError .
If something() is cancelled by other means (i.e. from within itself) that would also cancel shield() .
If it is desired to completely ignore cancellation (not recommended) the shield() function should be combined with a try/except clause, as follows:
Save a reference to tasks passed to this function, to avoid a task disappearing mid-execution. The event loop only keeps weak references to tasks. A task that isn’t referenced elsewhere may get garbage collected at any time, even before it’s done.
Changed in version 3.10: Removed the loop parameter.
Deprecated since version 3.10: Deprecation warning is emitted if aw is not Future-like object and there is no running event loop.
Timeouts¶
An asynchronous context manager that can be used to limit the amount of time spent waiting on something.
delay can either be None , or a float/int number of seconds to wait. If delay is None , no time limit will be applied; this can be useful if the delay is unknown when the context manager is created.
In either case, the context manager can be rescheduled after creation using Timeout.reschedule() .
If long_running_task takes more than 10 seconds to complete, the context manager will cancel the current task and handle the resulting asyncio.CancelledError internally, transforming it into an asyncio.TimeoutError which can be caught and handled.
The asyncio.timeout() context manager is what transforms the asyncio.CancelledError into an asyncio.TimeoutError , which means the asyncio.TimeoutError can only be caught outside of the context manager.
The context manager produced by asyncio.timeout() can be rescheduled to a different deadline and inspected.
class asyncio. Timeout ¶
An asynchronous context manager that limits time spent inside of it.
New in version 3.11.
when ( ) → float | None ¶
Return the current deadline, or None if the current deadline is not set.
The deadline is a float, consistent with the time returned by loop.time() .
reschedule ( when : float | None ) ¶
Change the time the timeout will trigger.
If when is None , any current deadline will be removed, and the context manager will wait indefinitely.
If when is a float, it is set as the new deadline.
if when is in the past, the timeout will trigger on the next iteration of the event loop.
expired ( ) → bool ¶
Return whether the context manager has exceeded its deadline (expired).
Timeout context managers can be safely nested.
New in version 3.11.
Similar to asyncio.timeout() , except when is the absolute time to stop waiting, or None .
New in version 3.11.
Wait for the aw awaitable to complete with a timeout.
If aw is a coroutine it is automatically scheduled as a Task.
timeout can either be None or a float or int number of seconds to wait for. If timeout is None , block until the future completes.
If a timeout occurs, it cancels the task and raises TimeoutError .
To avoid the task cancellation , wrap it in shield() .
The function will wait until the future is actually cancelled, so the total wait time may exceed the timeout. If an exception happens during cancellation, it is propagated.
If the wait is cancelled, the future aw is also cancelled.
Changed in version 3.10: Removed the loop parameter.
Changed in version 3.7: When aw is cancelled due to a timeout, wait_for waits for aw to be cancelled. Previously, it raised TimeoutError immediately.
Changed in version 3.10: Removed the loop parameter.
Waiting Primitives¶
Run Future and Task instances in the aws iterable concurrently and block until the condition specified by return_when.
The aws iterable must not be empty.
Returns two sets of Tasks/Futures: (done, pending) .
timeout (a float or int), if specified, can be used to control the maximum number of seconds to wait before returning.
Note that this function does not raise TimeoutError . Futures or Tasks that aren’t done when the timeout occurs are simply returned in the second set.
return_when indicates when this function should return. It must be one of the following constants:
The function will return when any future finishes or is cancelled.
The function will return when any future finishes by raising an exception. If no future raises an exception then it is equivalent to ALL_COMPLETED .
The function will return when all futures finish or are cancelled.
Unlike wait_for() , wait() does not cancel the futures when a timeout occurs.
Changed in version 3.10: Removed the loop parameter.
Changed in version 3.11: Passing coroutine objects to wait() directly is forbidden.
Run awaitable objects in the aws iterable concurrently. Return an iterator of coroutines. Each coroutine returned can be awaited to get the earliest next result from the iterable of the remaining awaitables.
Raises TimeoutError if the timeout occurs before all Futures are done.
Changed in version 3.10: Removed the loop parameter.
Changed in version 3.10: Removed the loop parameter.
Deprecated since version 3.10: Deprecation warning is emitted if not all awaitable objects in the aws iterable are Future-like objects and there is no running event loop.
Running in Threads¶
Asynchronously run function func in a separate thread.
Any *args and **kwargs supplied for this function are directly passed to func. Also, the current contextvars.Context is propagated, allowing context variables from the event loop thread to be accessed in the separate thread.
Return a coroutine that can be awaited to get the eventual result of func.
This coroutine function is primarily intended to be used for executing IO-bound functions/methods that would otherwise block the event loop if they were run in the main thread. For example:
Directly calling blocking_io() in any coroutine would block the event loop for its duration, resulting in an additional 1 second of run time. Instead, by using asyncio.to_thread() , we can run it in a separate thread without blocking the event loop.
Due to the GIL , asyncio.to_thread() can typically only be used to make IO-bound functions non-blocking. However, for extension modules that release the GIL or alternative Python implementations that don’t have one, asyncio.to_thread() can also be used for CPU-bound functions.
New in version 3.9.
Scheduling From Other Threads¶
Submit a coroutine to the given event loop. Thread-safe.
Return a concurrent.futures.Future to wait for the result from another OS thread.
This function is meant to be called from a different OS thread than the one where the event loop is running. Example:
If an exception is raised in the coroutine, the returned Future will be notified. It can also be used to cancel the task in the event loop:
See the concurrency and multithreading section of the documentation.
Unlike other asyncio functions this function requires the loop argument to be passed explicitly.
New in version 3.5.1.
Introspection¶
Return the currently running Task instance, or None if no task is running.
If loop is None get_running_loop() is used to get the current loop.
New in version 3.7.
Return a set of not yet finished Task objects run by the loop.
If loop is None , get_running_loop() is used for getting current loop.
New in version 3.7.
Task Object¶
A Future-like object that runs a Python coroutine . Not thread-safe.
Tasks are used to run coroutines in event loops. If a coroutine awaits on a Future, the Task suspends the execution of the coroutine and waits for the completion of the Future. When the Future is done, the execution of the wrapped coroutine resumes.
Event loops use cooperative scheduling: an event loop runs one Task at a time. While a Task awaits for the completion of a Future, the event loop runs other Tasks, callbacks, or performs IO operations.
Use the high-level asyncio.create_task() function to create Tasks, or the low-level loop.create_task() or ensure_future() functions. Manual instantiation of Tasks is discouraged.
To cancel a running Task use the cancel() method. Calling it will cause the Task to throw a CancelledError exception into the wrapped coroutine. If a coroutine is awaiting on a Future object during cancellation, the Future object will be cancelled.
cancelled() can be used to check if the Task was cancelled. The method returns True if the wrapped coroutine did not suppress the CancelledError exception and was actually cancelled.
Tasks support the contextvars module. When a Task is created it copies the current context and later runs its coroutine in the copied context.
Changed in version 3.7: Added support for the contextvars module.
Changed in version 3.8: Added the name parameter.
Deprecated since version 3.10: Deprecation warning is emitted if loop is not specified and there is no running event loop.
Return True if the Task is done.
A Task is done when the wrapped coroutine either returned a value, raised an exception, or the Task was cancelled.
Return the result of the Task.
If the Task is done, the result of the wrapped coroutine is returned (or if the coroutine raised an exception, that exception is re-raised.)
If the Task has been cancelled, this method raises a CancelledError exception.
If the Task’s result isn’t yet available, this method raises a InvalidStateError exception.
Return the exception of the Task.
If the wrapped coroutine raised an exception that exception is returned. If the wrapped coroutine returned normally this method returns None .
If the Task has been cancelled, this method raises a CancelledError exception.
If the Task isn’t done yet, this method raises an InvalidStateError exception.
add_done_callback ( callback , * , context = None ) ¶
Add a callback to be run when the Task is done.
This method should only be used in low-level callback-based code.
See the documentation of Future.add_done_callback() for more details.
Remove callback from the callbacks list.
This method should only be used in low-level callback-based code.
See the documentation of Future.remove_done_callback() for more details.
Return the list of stack frames for this Task.
If the wrapped coroutine is not done, this returns the stack where it is suspended. If the coroutine has completed successfully or was cancelled, this returns an empty list. If the coroutine was terminated by an exception, this returns the list of traceback frames.
The frames are always ordered from oldest to newest.
Only one stack frame is returned for a suspended coroutine.
The optional limit argument sets the maximum number of frames to return; by default all available frames are returned. The ordering of the returned list differs depending on whether a stack or a traceback is returned: the newest frames of a stack are returned, but the oldest frames of a traceback are returned. (This matches the behavior of the traceback module.)
print_stack ( * , limit = None , file = None ) ¶
Print the stack or traceback for this Task.
This produces output similar to that of the traceback module for the frames retrieved by get_stack() .
The limit argument is passed to get_stack() directly.
The file argument is an I/O stream to which the output is written; by default output is written to sys.stdout .
Return the coroutine object wrapped by the Task .
New in version 3.8.
Return the name of the Task.
If no name has been explicitly assigned to the Task, the default asyncio Task implementation generates a default name during instantiation.
New in version 3.8.
Set the name of the Task.
The value argument can be any object, which is then converted to a string.
In the default Task implementation, the name will be visible in the repr() output of a task object.
New in version 3.8.
Request the Task to be cancelled.
This arranges for a CancelledError exception to be thrown into the wrapped coroutine on the next cycle of the event loop.
The coroutine then has a chance to clean up or even deny the request by suppressing the exception with a try … … except CancelledError … finally block. Therefore, unlike Future.cancel() , Task.cancel() does not guarantee that the Task will be cancelled, although suppressing cancellation completely is not common and is actively discouraged.
Changed in version 3.9: Added the msg parameter.
Changed in version 3.11: The msg parameter is propagated from cancelled task to its awaiter.
The following example illustrates how coroutines can intercept the cancellation request:
Return True if the Task is cancelled.
The Task is cancelled when the cancellation was requested with cancel() and the wrapped coroutine propagated the CancelledError exception thrown into it.
Decrement the count of cancellation requests to this Task.
Returns the remaining number of cancellation requests.
Note that once execution of a cancelled task completed, further calls to uncancel() are ineffective.
New in version 3.11.
This method is used by asyncio’s internals and isn’t expected to be used by end-user code. In particular, if a Task gets successfully uncancelled, this allows for elements of structured concurrency like Task Groups and asyncio.timeout() to continue running, isolating cancellation to the respective structured block. For example:
While the block with make_request() and make_another_request() might get cancelled due to the timeout, unrelated_code() should continue running even in case of the timeout. This is implemented with uncancel() . TaskGroup context managers use uncancel() in a similar fashion.
Return the number of pending cancellation requests to this Task, i.e., the number of calls to cancel() less the number of uncancel() calls.
Note that if this number is greater than zero but the Task is still executing, cancelled() will still return False . This is because this number can be lowered by calling uncancel() , which can lead to the task not being cancelled after all if the cancellation requests go down to zero.
This method is used by asyncio’s internals and isn’t expected to be used by end-user code. See uncancel() for more details.
Python AsyncIO Awaitables: Coroutine, Future, and Task
Python asyncio is a library for efficient single-thread concurrent applications. In my last blog post “Python AsyncIO Event Loop”, we have understood what an event loop is in Python asyncio by looking at the Python source code. This seems to be effective to understand how Python asyncio works.
In this blog post, I would like to take one step further and discuss the mechanisms of the three key asyncio awaitables, including Coroutine , Future , and Task , by looking at the Python source code again.
Coroutine
Starting from Python 3.5, coroutine functions are defined using async def and Coroutine objects are created by calling coroutine functions. The abstracted class of Coroutine is just as follows. It does not have method overloading because the derived class and method overload is generated by Python interpreter for the coroutine functions defined using async def . The key method for Coroutine class is send . It is trying to mimic the behavior of trampoline.
“Fortunately”, Python asyncio coroutine was once implemented using a @asyncio.coroutine decorator on a Python generator in Python 3.4. Hopefully the logic of the coroutine in Python 3.5+ is similar to the coroutine in Python 3.4 that it yields sub coroutine upon calling.
A typical coroutine could be implemented using a Python generator just like the follows.
The @asyncio.coroutine decorator implementation is as follows.
Without looking into the details, this @asyncio.coroutine decorator almost does not change the generator at all, since most likely wrapper $\approx$ coro .
When we tried to run coroutine with loop.run_until_complete, we see from the comment that if the argument is a coroutine then it would be converted to a Task in the first place, and loop.run_until_complete is actually scheduling Task s. So we would look into Task shortly.
Future
Future has closed relationship with Task , so let’s look at Future first.
Future use has an event loop. By default, it is the event loop in the main thread.
The key method of Future is future.set_result . Let’s check what will happen if we call future.set_result .
Once future.set_result is called, it would trigger self.__schedule_callbacks asking the even loop to call all the callback s related to the Future as soon as possible. These Future related callback s are added or removed by future.add_done_callback or future.remove_done_callback . If no Future related callback s, no more callback s are scheduled in the event loop.
So we have known what will happen after the Future got result. What happens when the Future is scheduled in the event loop?
From the last blog post “Python AsyncIO Event Loop”, we have seen the Future was scheduled into the event loop via loop.ensure_future . “If the argument is a Future, it is returned directly.” So when the Future is scheduled in the event loop, there is almost no callback scheduled, until the future.set_result is called. (I said almost no callback because there is a default callback _run_until_complete_cb added as we have seen in the last blog post.)
Because _PyFuture = Future , Task is just a derived class of Future . The task of a Task is to wrap a coroutine in a Future .
In the constructor, we see that the Task schedules a callback self.__step in the event loop. The task.__step is a long method, but we should just pay attention to the try block and the else block since these two are the ones mostly likely to be executed.
Here we see the coroutine.send method again. Each time we call coroutine.send in the try block, we get a result . In the else blcok, we always have another self._loop.call_soon call. We do this in a trampoline fashion until Coroutine runs out of results to send .
Trampoline Function
The flavor of the wrapping of Task to Coroutine is somewhat similar to trampoline. Every time we call coroutine.send , we got some returned values and scheduled another callback .
Conclusion
The implementation of asyncio is complicated and I don’t expect I could know all the details. But trying to understand more about the low-level design might be useful for implementing low-level asyncio libraries and prevent stupid mistakes in high-level asyncio applications.
The key to scheduling the key asyncio awaitables, Coroutine , Future , and Task , are that the awaitables are all wrapped into Future in some way under the hood of asyncio interface.
Полное руководство по модулю asyncio в Python. Часть 3
Сегодня публикуем третью часть (первая, вторая) перевода учебного руководства по модулю asyncio в Python. Здесь представлены разделы оригинала №5, 6 и 7.
5. Определение, создание и запуск корутин
В Python-программах можно определять корутины — так же, как определяют новые подпрограммы (функции).
Функция корутины, после того, как она определена, может быть использована для создания объекта корутины.
Модуль asyncio даёт нам средства для запуска объектов корутин в цикле событий, который представляет собой среду выполнения для корутин.
5.1. Как определить корутину
Корутину можно определить посредством выражения async def .
Это — расширение выражения def , предназначенного для определения подпрограмм.
Оно определяет корутину, которая может быть создана, и возвращает объект корутины.
Корутину, определённую с помощью async def , называют функцией корутины.
Функция корутины: функция, которая возвращает объект корутины. Функцию корутины можно определить, пользуясь командой async def, она может содержать ключевые слова await, async for и async with.
Python glossary
Затем в пределах корутины могут использоваться выражения, специфичные для корутин, такие, как await , async for и async with .
Выполнение Python-корутины может быть приостановлено и возобновлено во многих местах. Выражения await, async for и async with могут быть использованы только в теле функции корутины.
Coroutine function definition
5.2. Как создать корутину
После того, как корутина определена, её можно создать.
Выглядит это как вызов функции.
В ходе работы этой команды корутина не запускается.
Эта команда возвращает объект корутины.
Функцию корутины можно рассматривать как фабрику для создания объектов корутины; точнее — не забывайте о том, что вызов функции корутины не приводит к выполнению кода, написанного пользователем. Вместо этого в ходе такого вызова лишь создаётся и возвращается объект корутины.
Python in a Nutshell, 2017, с. 516
У Python-объекта корутины есть методы — такие, как send() и close() . Он имеет тип coroutine .
Продемонстрировать это можно, создав экземпляр корутины и выведя сведения о его типе, воспользовавшись встроенной Python-функцией type() :
Выполнение кода этого примера приводит к выводу сообщения о том, что созданная корутина относится к классу coroutine .
Нам, кроме того, сообщают об ошибке RuntimeError , так как корутина была создана, но не запускалась. Мы исследуем этот вопрос в следующем разделе.
Объект корутины — это объект, допускающий ожидание.
Это значит, что он представляет Python-тип, реализующий метод await() .
Объекты, допускающие ожидание, обычно реализуют метод await(). Объект корутины, возвращаемый из функции, объявленной с использованием выражения async def — это объект, допускающий ожидание.
Awaitable objects
Подробности об объектах, допускающих ожидание, можно найти здесь.
5.3. Как запустить корутину из Python-кода
Корутины можно определять и создавать в обычном Python-коде, но запускать их можно только в цикле событий.
Цикл событий — это база любого asyncio-приложения. Цикл событий выполняет асинхронные задачи и коллбэки, сетевые операции ввода/вывода, подпроцессы.
Event Loop
Цикл событий, выполняющий корутины, организует работу кооперативной многозадачности, применяемой корутинами.
Код объекта корутины может выполняться лишь тогда, когда работает цикл событий.
Python in a Nutshell, 2017, с. 517
Типичный способ запуска цикла событий для корутин заключается в использовании функции asyncio.run().
Эта функция принимает одну корутину и возвращает значение корутины. Предоставленная ей корутина может быть использована как точка входа в программу, основанную на корутинах.
Теперь, когда мы знаем о том, как определять, создавать и запускать корутины — поближе познакомимся с циклом событий.
6. Цикл событий asyncio
Цикл событий — это сердце программ, основанных на asyncio .
В этом разделе мы поговорим о цикле событий.
6.1. Что такое цикл событий asyncio
Цикл событий — это среда для выполнения корутин в одном потоке.
Asyncio — это библиотека для выполнения этих корутин в асинхронной манере с использованием модели конкурентности, известной под названием «однопоточный цикл событий».
Python Concurrency with asyncio, 2022, с. 3
Цикл событий — это важнейший элемент asyncio-программы.
Он отвечает за решение множества задач. Вот некоторые из них:
Выполнение сетевых операций ввода/вывода.
«Цикл событий» — это распространённый паттерн проектирования, который стал весьма популярным в наши дни благодаря его использованию в JavaScript.
В JavaScript имеется модель среды выполнения, основанная на цикле событий, который отвечает за выполнение кода, за сбор и обработку событий, за выполнение подзадач, поставленных в очередь. Эта модель сильно отличается от моделей из других языков, таких как C и Java.
The event loop, Mozilla
Цикл событий, как видно из его названия, это — цикл. Он управляет списком задач (корутин) и стремится продвинуть выполнение каждой из них в определённой последовательности на каждой своей итерации. Он, кроме того, выполняет и другие задачи — наподобие выполнения коллбэков и обработки операций ввода/вывода.
Модуль asyncio даёт нам функции для доступа к циклу событий и для организации взаимодействия с ним.
При разработке типичных Python-приложений это не нужно.
Вместо этого доступ к циклу событий нацелен на разработчиков фреймворков, на тех, кто хочет разрабатывать свои проекты на базе модуля asyncio или хочет дать возможность работы с asyncio пользователям своих библиотек.
Разработчикам приложений обычно следует использовать высокоуровневые функции asyncio, такие, как asyncio.run(). У них редко будет возникать необходимость пользоваться ссылкой на объект цикла или вызов его методов.
Event Loop
Модуль asyncio позволяет работать с низкоуровневым API для получения доступа к текущему объекту цикла событий. Этот модуль так же содержит набор методов, которые можно применять для взаимодействия с циклом событий.
Низкоуровневый API предназначен для разработчиков фреймворков, которые могут расширять и дополнять возможности asyncio и интегрировать этот модуль в свои библиотеки.
Обычным разработчикам редко нужно взаимодействовать с циклом событий в программах, основанных на asyncio . Вместо этого они, как правило, применяют высокоуровневый API модуля.
Но, как бы то ни было, мы вполне можем кратко обсудить вопрос использования объекта цикла событий.
6.2. Запуск цикла событий и получение ссылки на его объект
Обычно в asyncio-приложениях ссылки на объекты циклов событий получают, вызывая функцию asyncio.run() .
Эта функция всегда создаёт новый цикл событий и в конце завершает его работу. Её следует использовать как основную точку входа для asyncio-программ, в идеале её нужно вызывать в программах лишь один раз.
Asyncio Coroutines and Tasks
Эта функция принимает корутину и выполняет её до завершения её работы.
Обычно этой функции передают главную корутину, с которой начинается выполнение программы.
Существуют и низкоуровневые функции для создания цикла событий и для работы с ним.
Функция asyncio.new_event_loop() создаёт новый цикл событий и возвращает ссылку на него.
Можно продемонстрировать это всё на рабочем примере.
Здесь мы создаём новый цикл событий и сообщаем сведения о нём.
Выполнение этого кода приведёт к созданию цикла событий и к выводу сведений об его объекте.
В данном случае можно видеть, что цикл событий имеет тип _UnixSelectorEventLoop , и что он не выполняется, но и не является закрытым.
Если цикл событий asyncio уже выполняется — доступ к нему можно получить посредством функции asyncio.get_running_loop().
Возвращает выполняющийся цикл событий в текущем потоке ОС. Если в потоке нет цикла событий — выдаётся ошибка RuntimeError. Эта функция может быть вызвана только из корутины или из коллбэка.
Event Loop
Есть ещё функция, предназначенная для получения или запуска цикла событий. Это — asyncio.get_event_loop(). Но она, в Python 3.10, была признана устаревшей. Пользоваться ей не стоит.
6.3. Подробности об объекте цикла событий
Цикл событий реализован в виде Python-объекта.
Этот объект определяет реализацию цикла событий, он предоставляет стандартный API, предназначенный для взаимодействия с циклом, описанный в классе AbstractEventLoop.
Существуют различные реализации цикла событий для разных платформ.
Например, в ОС семейства Windows и Unix цикл событий будет реализован по-разному из-за различных внутренних механизмов реализации неблокирующего ввода/вывода на этих платформах.
SelectorEventLoop — это цикл событий, используемый по умолчанию в ОС, основанных на Unix — наподобие Linux и macOS.
ProactorEventLoop — это цикл событий, по умолчанию используемый в Windows.
Сторонние библиотеки могут содержать собственные реализации цикла событий ради его оптимизации под специфические задачи.
6.4. Зачем может понадобиться доступ к циклу событий
Зачем нам обращаться к циклу событий за пределами asyncio-программы?
Это может быть нужно по многим причинам.
Для мониторинга хода выполнения задач.
Для выдачи и получения результатов работы задач.
Для запуска одноразовых задач.
Цикл событий asyncio может использоваться в программах как альтернатива пулу потоков, рассчитанная на работу с задачами, основанными на корутинах.
Цикл событий, кроме того, может быть встроен в обычную asyncio-программу, к нему можно обращаться тогда, когда это нужно.
Теперь, когда мы немного познакомились с циклом событий — перейдём к asyncio-задачам.
7. Создание и запуск asyncio-задач
Объекты Task (задачи) в asyncio-программах можно создавать из корутин.
Задачи предоставляют инструменты, предназначенные для независимого планирования и выполнения корутин. Они позволяют ставить задачи в очередь и отменять их, получать в нужное время результаты их работы и выданные ими исключения.
Цикл событий asyncio управляет задачами. Получается, что все корутины в цикле событий становятся задачами, работа с ними тоже ведётся как с задачами.
Поговорим об asyncio-задачах.
7.1. Что такое asyncio-задача
Task — это объект, который отвечает за планирование выполнения asyncio-корутин и за их независимый запуск.
Задача предоставляет средства, предназначенные для планирования выполнения корутин. К этим средствам asyncio-программа может обращаться для получения сведений о корутинах, с их помощью она может взаимодействовать с корутинами.
Задачи создают из корутин. Для создания задачи нужен объект корутины. Задача оборачивает корутину, планирует её выполнение и даёт средства для взаимодействия с ней.
Задачи выполняются независимо друг от друга. Это значит, что их выполнение планируется в цикле событий asyncio , и что они выполняются независимо от того, что ещё происходит в создавшей их корутине. Это отличается от прямого выполнения корутины, когда вызывающая сторона должна дождаться её завершения.
Задачи используются для планирования конкурентного выполнения корутин. Когда корутину оборачивают в объект Task, пользуясь функцией наподобие asyncio.create_task(), выполнение корутины автоматически планируется на ближайшее время.
Coroutines and Tasks
Класс asyncio.Task расширяет класс asyncio.Future, его экземпляры являются объектами, допускающими ожидание.
Future — это низкоуровневый класс. Он представляет собой сущность, которая рано или поздно вернёт результат.
Future — это особый низкоуровневый объект, допускающий ожидание, который представляет конечный результат асинхронной операции.
Coroutines and Tasks
Классы, которые расширяют класс Future , часто называют Future-подобными классами.
Так как Task — это объект, допускающий ожидание, получается, что корутина может подождать завершения задачи с использованием выражения await .
Теперь, когда мы разобрались с тем, что собой представляют asyncio-задачи, поговорим о том, как их создавать.
7.2. Как создать задачу
Задачи создают с использованием экземпляра корутины, предоставленного соответствующему механизму.
Вспомните — корутину определяют, используя выражение async def . Она выглядит как функция.
Задачу можно создать и запланировать на выполнение только внутри корутины.
Есть два основных способа создания и планирования задач:
Создать объект Task с использованием высокоуровневого API (предпочтительный способ).
Создать объект Task с помощью низкоуровневого API.
Рассмотрим подробнее каждый из этих способов.
Создание объекта Task с использованием высокоуровневого API
Задачу можно создать, прибегнув к функции asyncio.create_task().
Эта функция принимает экземпляр класса корутины и необязательное имя задачи, а возвращает экземпляр класса asyncio.Task .
Сделать это можно в одной строке, с помощью сложного выражения.
Вот что здесь происходит:
Корутина оборачивается в экземпляр Task .
Планируется выполнение задачи в текущем цикле событий.
Возвращается экземпляр Task .
Ссылку на экземпляр Task можно и не сохранять. С задачей можно взаимодействовать посредством методов, её выполнения можно ожидать в корутине.
Это — предпочтительный способ создания объектов Task из корутин в asyncio-программах.
Создание объекта Task с использованием низкоуровневого API
Задачи можно создавать из корутин и с использованием низкоуровневого API asyncio .
Первый способ такого создания задач заключается в использовании функции asyncio.ensure_future().
Эта функция принимает объект Task , Future , или Future-подобный объект, такой, как корутина, и, необязательно — цикл событий, в котором нужно запланировать выполнение соответствующего объекта.
Если цикл событий не предоставлен — выполнение объекта будет запланировано в текущем цикле событий.
Если этой функции предоставлена корутина — она автоматически оборачивается в экземпляр Task , который и возвращает эта функция.
Ещё одна низкоуровневая функция, которую можно использовать для создания объектов Task и для планирования их выполнения, представлена методом loop.create_task().
Этот метод требует доступа к конкретному циклу событий, в котором планируется выполнять корутину как задачу.
Можно получить ссылку на экземпляр текущего цикла событий, используемого в asyncio-программе, прибегнув к функции asyncio.get_event_loop().
Затем можно вызвать метод create_task() этого экземпляра цикла для создания экземпляра Task и для планирования его выполнения.
7.3. Когда запускаются задачи?
Распространённый вопрос о работе с задачами звучит так: «Когда, после того, как задача создана, она запускается?».
И это — хороший вопрос.
Хотя мы можем планировать независимый запуск корутин в виде задач, пользуясь функцией create_task() , задача может не запуститься немедленно.
На самом деле, задача не будет запущена до тех пор, пока у цикла событий не появится возможность её запустить.
Это произойдёт тогда, когда все другие корутины перестанут выполняться и настанет очередь интересующей нас задачи.
Например, имеется asyncio-программа с одной корутиной, которую создали и выполнение которой, виде задачи, запланировали. Запланированная задача не будет выполнена до тех пор, пока вызывающая корутина, создавшая эту задачу, не будет приостановлена.
Это может произойти в том случае, если вызывающая корутина решит приостановить работу, подождать выполнения другой корутины или задачи, или решит подождать выполнения новой задачи, выполнение которой было только что запланировано.
О, а приходите к нам работать? �� ��
Мы в wunderfund.io занимаемся высокочастотной алготорговлей с 2014 года. Высокочастотная торговля — это непрерывное соревнование лучших программистов и математиков всего мира. Присоединившись к нам, вы станете частью этой увлекательной схватки.
Мы предлагаем интересные и сложные задачи по анализу данных и low latency разработке для увлеченных исследователей и программистов. Гибкий график и никакой бюрократии, решения быстро принимаются и воплощаются в жизнь.