AI-translated. Read the original in KoreanEmbedded slides, videos and images remain in their original language.
Breaking Down asyncio, Part 2: Working with Futures
Software Engineer
Used well, Futures enable more varied control flow than simply applying await. The previous post introduced coroutines and the event loop, asyncio's core components. This time, we will look at creating Futures, registering callbacks, and putting them to work.
- Breaking Down asyncio, Part 1: Coroutines and the Event Loop
- (This post) Breaking Down asyncio, Part 2: Working with Futures
- Breaking Down asyncio, Part 3: Handling SIGTERM (CTRL+C) Properly
- (WIP) Breaking Down asyncio, Part 4: Graceful Shutdown in APScheduler
What is a Future?
A Future is not specific to Python. It is a widely used concept in asynchronous programming, usually representing an object that stores the result of an asynchronous operation. If you have encountered it in Java or JavaScript, it may already be familiar.
A synchronous function blocks and then returns a result when it finishes, so you can use that value directly. With an asynchronous function, you cannot immediately obtain the return value unless the caller deliberately waits for the operation to finish. A Future is the object handed to the caller to say, “I don't have the value now, but I will fill it in here later.”
Do I really need to know about Futures?
As the official documentation explains, exposing Future objects is discouraged. If you only use high-level asyncio APIs, you may rarely encounter them. Still, they are an important asynchronous programming concept, and understanding them is essential for efficient low-level asyncio code.
Futures in Python
Python actually has two different Future classes: concurrent.futures.Future and asyncio.Future. Every Future mentioned below refers to the latter. These classes were introduced at different times for different purposes and are not interchangeable. You can, however, convert the former into the latter using asyncio.wrap_future().
Futures in asyncio
As mentioned above, asyncio rarely exposes Futures directly. loop.create_future() and asyncio.create_task() are the most common encounters, but the latter is usually used to register a coroutine with the event loop rather than to manipulate the returned object.
What a Future can do
For reference, here is a quick summary of its capabilities.
- It is awaitable, so
awaitcan wait for its result. done()checks whether the operation has finished.set_result()sets the result.cancel()cancels the operation.add_done_callback()registers a callback.
Using Futures
Example 1: callbacks
Let's add a few conditions to the requirements from the previous post and implement them using Futures.
Requirements
Existing requirements:
- Create a new job every ten seconds; each job takes a random amount of time.
- A new job must be created every ten seconds regardless of whether the previous job takes longer than that.
- Run jobs concurrently.
New requirements:
- Print each job's total elapsed time when it finishes.
- Print this whether the job succeeds or fails with an exception.
Implementation 1: change the job
The simplest approach is to modify the job so it prints its own elapsed time.
import asyncio
import time
from random import randint
async def run_job() -> None:
before = time.time()
try:
delay = randint(5, 15)
await asyncio.sleep(delay) # 5~15초 동안 잠자기
finally:
after = time.time()
print(f'Job duration: {after - before} sec')
async def main() -> None:
while True:
asyncio.create_task(run_job())
await asyncio.sleep(10)
asyncio.run(main())
For a small example like this, that is fine. In an application, however, it can create problems. This approach usually leads to code that violates SOLID's Single Responsibility Principle. Imagine this code is a library that periodically schedules jobs: users would have to modify their jobs just to enable monitoring.
Implementation 2: use a callback
import asyncio
import functools
import time
from random import randint
async def run_job() -> None:
delay = randint(5, 15)
await asyncio.sleep(delay) # 5~15초 동안 잠자기
def print_wall_time(before: float, _: asyncio.Future[None]) -> None:
after = time.time()
print(f'Job duration: {after - before} sec')
async def main() -> None:
while True:
before = time.time()
future = asyncio.create_task(run_job())
future.add_done_callback(functools.partial(print_wall_time, before))
await asyncio.sleep(10)
asyncio.run(main())
Here is the same behavior implemented with Future callbacks. run_job() is unchanged from the previous post; main() now registers a monitoring callback. The monitoring logic also lives in a separate function, print_wall_time().
Future.add_done_callback() runs whenever the Future finishes, whether successfully, with an exception, or through cancel(). That achieves our goal without code such as try-finally.
The Future passes itself to the callback, so we can also monitor the job's result if we want.
import asyncio
import functools
import time
from random import randint
async def run_job() -> int:
delay = randint(5, 15)
await asyncio.sleep(delay) # 5~15초 동안 잠자기
return randint(0, 1000) # 임의의 연산 결과
def print_wall_time(before: float, future: asyncio.Future[int]) -> None:
after = time.time()
if future.exception() is not None:
print(f'Error occurred: {future.exception()}, duration: {after - before} sec')
else:
print(f'Result: {future.result()}, duration: {after - before} sec')
async def main() -> None:
while True:
before = time.time()
future = asyncio.create_task(run_job())
future.add_done_callback(functools.partial(print_wall_time, before))
await asyncio.sleep(10)
asyncio.run(main())
Things to keep in mind
When do registered callbacks run, and in what order?
The official documentation does not appear to specify this. Since an event loop can provide its own optimized Future implementation, it seems difficult to rely on a universal rule. The default implementation, however, uses loop.call_soon() to schedule callbacks. After a Future finishes, it schedules them for the next event loop iteration, apparently beginning execution in registration order. In short, if timing or ordering must be guaranteed, use a separate mechanism rather than relying on an unspecified behavior.
How can I register an async function—a coroutine function—as a callback?
Future.add_done_callback() expects a regular function, so it cannot execute a coroutine function directly. It can instead schedule a coroutine on the event loop:
Future.add_done_callback(lambda fut: asyncio.create_task(some_async_func(fut))).
Because this uses asyncio.create_task(), you cannot wait for that coroutine to complete with await within this callback arrangement. See the previous post for the reasoning.
Example 2: creating Futures
Many asyncio utilities—including Queue, gather, and as_completed—are implemented internally using Futures. Let's explore another use by implementing a simple Semaphore. The interface is provided first; I recommend trying your own implementation before reading the solution.
Interface and test code
import asyncio
from abc import ABCMeta
from datetime import datetime
# 인터페이스
class Semaphore(metaclass=ABCMeta):
_value: int
def __init__(self, initial_value: int = 1) -> None:
self._value = initial_value
async def acquire(self) -> None:
raise NotImplementedError
def release(self) -> None:
raise NotImplementedError
# 테스트 코드
async def run_job(sem: Semaphore, job_id: int) -> None:
await sem.acquire()
print(f'{datetime.now()} - start job {job_id}')
await asyncio.sleep(1)
print(f'{datetime.now()} - job {job_id} finished')
sem.release()
async def main() -> None:
sem = SomeSemaphore(2) # 구현체로 변경 필요
await asyncio.gather( # 한번에 2개의 Job 묶음이 1초 간격으로 실행되어야함
run_job(sem, 1),
run_job(sem, 2),
run_job(sem, 3),
run_job(sem, 4),
run_job(sem, 5),
)
asyncio.run(main())
Implementation 1: busy waiting
Busy waiting is probably the first approach that comes to mind.
class BusyWaitingSemaphore(Semaphore):
async def acquire(self) -> None:
while self._value <= 0:
await asyncio.sleep(0.1)
self._value -= 1
def release(self) -> None:
self._value += 1
Given the title of this post, the absence of a Future is a clue that something is wrong. This code is inefficient because it polls, sleeping and periodically checking whether a release has occurred. With unlucky timing, acquire may return only 0.1 seconds after release is called. Reducing the sleep interval to 0.01 seconds decreases that delay but increases polling frequency and event loop overhead.
Implementation 2: creating Futures
With Futures, we can remove polling and build a far more efficient asynchronous semaphore. Until now, we have awaited or attached callbacks to Futures returned by asyncio or related libraries. This time, we create and manage the Future ourselves.
import asyncio
from collections import deque
class FutureSemaphore(Semaphore):
_waiters: deque[asyncio.Future[None]]
def __init__(self, initial_value: int = 1) -> None:
super().__init__(initial_value)
self._waiters = deque()
async def acquire(self) -> None:
if self._value <= 0:
loop = asyncio.get_running_loop()
future = loop.create_future()
self._waiters.append(future)
await future
self._value -= 1
def release(self) -> None:
self._value += 1
if len(self._waiters) > 0:
fut = self._waiters.popleft()
fut.set_result(None)
The while loop in BusyWaitingSemaphore.acquire() has been replaced with a Future. We create an empty Future and use await to wait for it to finish. When should it wake up? When someone calls release() and a semaphore slot becomes available.
To make that happen, acquire() adds the newly created Future to _waiters, recording that it is waiting. At the end of release(), we look for a waiting Future and complete it with set_result().
The real asyncio.Semaphore is implemented in a similar way, although it is more complex to handle cases that this example does not cover.
Things to keep in mind
How to create a Future
As the official documentation recommends, use loop.create_future() rather than constructing asyncio.Future() directly. The event loop may provide an implementation better optimized for itself.
Thread safety of the example code
None of the examples above are thread-safe, but they are safe within a single thread. The same is true of asyncio.Semaphore, because an event loop runs in one thread. For why changing _value and _waiters in release() is safe in that setting even without atomic operations, see the cooperative multitasking section in the previous post.
Closing thoughts
This post explored Future use cases that go beyond simply applying await. Next, we will look at event loop signal handling and how to handle termination signals such as SIGINT and SIGTERM correctly.