`asyncio` の概念を理解する¶

この HOWTO では、asyncio の基本的な仕組みをしっかり理解できるように説明します。推奨される書き方がどのように動き、なぜ推奨されるのかを理解する助けになるでしょう。

asyncio の主要な概念について、疑問を持っているかもしれません。この記事を読み終えるころには、次の問いに無理なく答えられるようになるでしょう。

オブジェクトを await すると、その裏側では何が起きるのでしょうか？
asyncio は、ネットワーク要求やファイルの読み取りのように CPU 時間を必要としないタスクと、n の階乗を計算するように CPU 時間が必要なタスクを、どのように区別するのでしょうか？
非同期のスリープやデータベースへの要求など、ある操作の非同期版はどのように書けばよいのでしょうか？

参考

この HOWTO のもとになった、Alexander Nordin によるガイド。
Python コアチームのメンバー Łukasz Langa による、asyncio の詳しい YouTube チュートリアルシリーズ。
500 Lines or Less: A Web Crawler With asyncio Coroutines by A. Jesse Jiryu Davis and Guido van Rossum.

概念の説明、第 1 部: 全体像¶

第 1 部では、asyncio の主要な高水準の構成要素であるイベントループ、コルーチン関数、コルーチンオブジェクト、タスク、await を取り上げます。

イベントループ¶

asyncio では、すべてがイベントループを中心に動きます。イベントループは主役ですが、表に立つよりも裏方としてリソースを管理し、調整する役割を担います。オーケストラの指揮者のようなものです。明示的に与えられた権限もありますが、仕事を進める力の多くは、楽団員が指示に従い協力することから生まれます。

もう少し技術的にいうと、イベントループは実行すべきジョブの集合を持っています。ジョブには、自分で直接追加するものも、asyncio が間接的に追加するものもあります。イベントループは待機中のジョブを 1 つ選び、関数を呼び出すのと同じように起動します (つまり「制御を渡します」)。ジョブは実行を始め、中断または完了するとイベントループに制御を戻します。するとイベントループは別のジョブを選んで起動します。このジョブの集合は、おおまかには キューと考えられます。ジョブは追加された後、通常は順番に (ただし、常にそうとは限りません) 1 つずつ処理されます。イベントループはこの処理を繰り返し続けます。実行待ちのジョブがなくなると、CPU 時間を無駄にしないよう休止します。 I/O 操作の完了やタイマーの期限到来などによって新たな仕事ができると、また動き始めます。

効率よく実行するには、ジョブ同士が処理の機会を分け合い、協力する必要があります。欲張りなジョブが制御を独占すると、ほかのジョブが実行できなくなり、イベントループという仕組み全体がほとんど役に立たなくなってしまいます。

import asyncio

# イベントループを作り、そのジョブの集合を
# 繰り返し処理し続ける。
event_loop = asyncio.new_event_loop()
event_loop.run_forever()

非同期関数とコルーチン¶

これはごく普通の Python 関数です。

def hello_printer():
    print(
        "Hi, I am a lowly, simple printer, though I have all I "
        "need in life -- \nfresh paper and my dearly beloved octopus "
        "partner in crime."
    )

通常の関数を呼び出すと、その本体の処理が実行されます。

>>> hello_printer()
Hi, I am a lowly, simple printer, though I have all I need in life --
fresh paper and my dearly beloved octopus partner in crime.

単なる def の代わりに async def を使うと、非同期関数 (「コルーチン関数」) になります。これを呼び出すと、コルーチンオブジェクトが作られて返されます。

async def loudmouth_penguin(magic_number: int):
    print(
     "I am a super special talking penguin. Far cooler than that printer. "
     f"By the way, my lucky number is: {magic_number}."
    )

非同期関数 loudmouth_penguin を呼び出しても、print 文は実行されません。代わりに、コルーチンオブジェクトが作られます。

>>> loudmouth_penguin(magic_number=3)
<coroutine object loudmouth_penguin at 0x104ed2740>

「コルーチン関数」と「コルーチンオブジェクト」は、どちらも単にコルーチンと呼ばれることがあり、混乱を招きます。この記事では、コルーチンはコルーチンオブジェクト、より正確には types.CoroutineType のインスタンス (ネイティブコルーチン) を指します。コルーチンは collections.abc.Coroutine のインスタンスとして存在することもあります。この違いは型チェックで重要になります。

コルーチンは関数の本体、つまり処理内容を表します。コルーチンは明示的に開始する必要があり、繰り返しになりますが、作成しただけでは実行されません。特に重要なのは、関数本体のさまざまな位置で実行を中断し、再開できる点です。この中断と再開の能力が、非同期の振る舞いを可能にします。

コルーチンとコルーチン関数は、ジェネレータとジェネレータ関数の機能を活用して作られました。ジェネレータ関数とは、次のように yield を使う関数です。

def get_random_number():
    # これでは乱数生成器としては役に立たない！
    print("Hi")
    yield 1
    print("Hello")
    yield 7
    print("Howdy")
    yield 4
    ...

コルーチン関数と同様に、ジェネレータ関数を呼び出しても本体は実行されません。代わりに、ジェネレータオブジェクトが作られます。

>>> get_random_number()
<generator object get_random_number at 0x1048671c0>

組み込み関数 next() を使うと、ジェネレータを次の yield まで進められます。つまり、ジェネレータは実行されてから中断します。例えば、次のようになります。

>>> generator = get_random_number()
>>> next(generator)
Hi
1
>>> next(generator)
Hello
7

Task¶

おおまかにいうと、タスクはイベントループに結び付けられたコルーチンです (コルーチン関数ではありません)。タスクはコールバック関数のリストも保持しています。この重要性は、後で await を説明するときに明らかになります。

タスクを作ると、実行のために自動的にスケジュールされます。具体的には、タスクを実行するコールバックが、イベントループの処理待ちリスト、つまりジョブの集合に追加されます。タスクの作成には asyncio.create_task() を使うことが推奨されます。

asyncio はタスクとイベントループを自動的に関連付けます。この自動的な関連付けは、簡単に使えるようにするための意図的な設計です。これがなければ、イベントループオブジェクトを自分で管理し、タスクを作るすべてのコルーチン関数に渡す必要があり、asyncio を使うコードが冗長で煩雑になります。

coroutine = loudmouth_penguin(magic_number=5)
# Task オブジェクトを作り、イベントループで実行するようにスケジュールする。
task = asyncio.create_task(coroutine)

先ほどはイベントループを手動で作り、実行し続けるようにしました。実際には asyncio.run() を使うことが推奨され、よく使われています。この関数はイベントループを管理し、渡したコルーチンが完了してから次の処理に進むようにします。例えば、多くの非同期プログラムは次のような構成になっています。

import asyncio

async def main():
    # さまざまな非同期処理を行う...
    ...

if __name__ == "__main__":
    asyncio.run(main())
    # main() コルーチンが完了するまで、
    # 次の print 文には進まない。
    print("coroutine main() is done!")

It's important to be aware that the task itself is not added to the event loop, only a callback to the task is. This matters if the task object you created is garbage collected before it's called by the event loop. For example, consider this program:

async def hello():
    print("hello!")

async def main():
    asyncio.create_task(hello())
    # しばらく実行した後にイベントループへ制御を渡す、
    # ほかの非同期処理...
    ...

asyncio.run(main())

Because there's no reference to the task object created on line 5, it might be garbage collected before the event loop invokes it. Later instructions in the coroutine main() hand control back to the event loop so it can invoke other jobs. When the event loop eventually tries to run the task, it might fail and discover the task object does not exist! This can also happen even if a coroutine keeps a reference to a task but completes before that task finishes. When the coroutine exits, local variables go out of scope and may be subject to garbage collection. In practice, asyncio and Python's garbage collector work pretty hard to ensure this sort of thing doesn't happen. But that's no reason to be reckless!

await¶

await は Python のキーワードで、一般に次の 2 通りの使い方があります。

await task
await coroutine

await の動作は、待機する対象のオブジェクトの型によって、重要な点で異なります。

タスクの待機¶

Awaiting a task will cede control from the current task or coroutine to the event loop. In the process of relinquishing control, a few important things happen. We'll use the following code example to illustrate:

async def plant_a_tree():
    dig_the_hole_task = asyncio.create_task(dig_the_hole())
    await dig_the_hole_task

    # 木を植えるためのその他の処理。
    ...

この例では、イベントループからコルーチン plant_a_tree() の先頭へ制御が渡ったと考えてください。コルーチンは上記のとおりタスクを作り、それを await します。 await dig_the_hole_task は、plant_a_tree() を再開するためのコールバックを、dig_the_hole_task オブジェクトのコールバックのリストに追加します。その後、イベントループに制御を渡します。しばらくすると、イベントループは dig_the_hole_task に制御を渡し、タスクは必要な処理を終えます。タスクが完了すると、登録されているコールバックがイベントループに追加されます。この場合は、plant_a_tree() を再開する呼び出しです。

一般に、待機していたタスク (dig_the_hole_task) が完了すると、もとのタスクやコルーチン (plant_a_tree()) がイベントループの処理待ちリストに戻され、再開されます。

これは基本的ですが、頼りになる考え方です。実際の制御の受け渡しはもう少し複雑ですが、大きくは違いません。第 2 部では、これを可能にする仕組みを詳しく説明します。

コルーチンの待機¶

Unlike tasks, awaiting a coroutine does not hand control back to the event loop! Wrapping a coroutine in a task first, then awaiting that would cede control. The behavior of await coroutine is effectively the same as invoking a regular, synchronous Python function. Consider this program:

import asyncio

async def coro_a():
   print("I am coro_a(). Hi!")

async def coro_b():
   print("I am coro_b(). I sure hope no one hogs the event loop...")

async def main():
   task_b = asyncio.create_task(coro_b())
   num_repeats = 3
   for _ in range(num_repeats):
      await coro_a()
   await task_b

asyncio.run(main())

コルーチン main() の最初の文は task_b を作り、イベントループで実行されるようにスケジュールします。その後、coro_a() が繰り返し await されます。この間はイベントループに制御が戻らないため、coro_a() の 3 回分の出力がすべて表示された後に、coro_b() の出力が表示されます。

I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_b(). I sure hope no one hogs the event loop...

await coro_a() を await asyncio.create_task(coro_a()) に変えると、動作が変わります。コルーチン main() はこの文でイベントループに制御を渡します。イベントループは待機中の処理を進め、task_b、次に coro_a() を包むタスクを呼び出してから、コルーチン main() を再開します。

I am coro_b(). I sure hope no one hogs the event loop...
I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_a(). Hi!

この await coroutine の動作でつまずく人は少なくありません。この例が示すように、await coroutine だけを使っていると、意図せずほかのタスクから制御を奪い、実質的にイベントループを止めてしまうことがあります。 asyncio.run() の debug=True を指定してデバッグモードを有効にすると、こうした問題の検出に役立ちます。このモードでは、ほかの情報に加え、100 ミリ秒以上実行を独占するコルーチンがログに記録されます。

The design intentionally trades off some conceptual clarity around usage of await for improved performance. Each time a task is awaited, control needs to be passed all the way up the call stack to the event loop. Then, the event loop needs to manage its internal state and work through its processing logic to resume the next job. That might sound minor, but in a large program with many awaits, that overhead can add up to a non-negligible performance drag.

概念の説明、第 2 部: 内部の仕組み¶

第 2 部では、asyncio が制御の流れを管理する仕組みを詳しく説明します。ここに仕掛けの核心があります。この節を読むと、await の裏側で何が起きるか、独自の非同期操作をどう作るかが分かるでしょう。

コルーチンの内部動作¶

asyncio は Python の 4 つの構成要素を使って制御を受け渡します。

coroutine.send(arg) は、コルーチンを開始または再開するメソッドです。中断していたコルーチンを再開する場合、引数 arg は、中断した位置の yield 文の戻り値として渡されます。再開ではなく初めてコルーチンを動かす場合は、arg は None でなければなりません。

class Rock:
    def __await__(self):
        value_sent_in = yield 7
        print(f"Rock.__await__ resuming with value: {value_sent_in}.")
        return value_sent_in

async def main():
    print("Beginning coroutine main().")
    rock = Rock()
    print("Awaiting rock...")
    value_from_rock = await rock
    print(f"Coroutine received value: {value_from_rock} from rock.")
    return 23

coroutine = main()
intermediate_result = coroutine.send(None)
print(f"Coroutine paused and returned intermediate value: {intermediate_result}.")

print(f"Resuming coroutine and sending in value: 42.")
try:
    coroutine.send(42)
except StopIteration as e:
    returned_value = e.value
print(f"Coroutine main() finished and provided value: {returned_value}.")

yield は通常どおり実行を中断し、呼び出し元に制御を戻します。上の例では、11 行目の ... = await rock によって 3 行目の yield が呼ばれます。より一般的にいうと、await は与えられたオブジェクトの __await__() メソッドを呼び出します。 await には、もう 1 つ特別な役割があります。受け取った yield を呼び出しの連鎖をさかのぼって伝播させる (「受け渡す」) のです。この場合は、16 行目の ... = coroutine.send(None) に戻ります。

21 行目の coroutine.send(42) を呼び出すと、コルーチンが再開されます。コルーチンは 3 行目の yield で中断した位置から動き始め、本体の残りの文を実行します。コルーチンが完了すると、戻り値を value 属性に持つ StopIteration 例外が送出されます。

このコードは次の出力を生成します。

Beginning coroutine main().
Awaiting rock...
Coroutine paused and returned intermediate value: 7.
Resuming coroutine and sending in value: 42.
Rock.__await__ resuming with value: 42.
Coroutine received value: 42 from rock.
Coroutine main() finished and provided value: 23.

ここで少し立ち止まり、制御や値がどのように受け渡されたかを追えているか確認してください。重要な考え方をいくつも取り上げたので、確実に理解しておくとよいでしょう。

コルーチンから yield する (実質的に制御を渡す) 唯一の方法は、__await__ メソッド内で yield するオブジェクトを await することです。奇妙に思えるかもしれません。次のような疑問を持ったのではないでしょうか。

1. What about a yield directly within the coroutine function? The coroutine function becomes an async generator function, a different beast entirely.

2. What about a yield from within the coroutine function to a (plain) generator? That causes the error: SyntaxError: yield from not allowed in a coroutine. This was intentionally designed for the sake of simplicity -- mandating only one way of using coroutines. Despite that, yield from and await effectively do the same thing. Initially yield was barred as well, but was re-accepted to allow for async generators.

Future¶

Future は、計算の状態と結果を表すためのオブジェクトです。この名前は、これから起こること、まだ起きていないことを指しています。 Future オブジェクトは、その状況を把握する手段です。

Future には重要な属性がいくつかあります。 1 つは状態で、「待機中」「キャンセル済み」「完了」のいずれかになります。もう 1 つは結果で、状態が完了に変わるときに設定されます。コルーチンと異なり、Future は実行する計算そのものを表しません。赤・黄・緑の状態ランプや表示器のように、その計算の状態と結果を表します。

asyncio.Task は、これらの機能を得るために asyncio.Future を継承しています。前の節ではタスクがコールバックのリストを保持すると説明しましたが、厳密には正確ではありません。実際にこの処理を実装するのは Future クラスで、Task はそれを継承しています。

Future はタスクを介さずに直接使うこともできます。タスクは自身のコルーチンが完了すると、自分の状態を完了にします。 Future はもっと柔軟で、指定したタイミングで完了にできます。そのため、待機や再開の条件を独自に定めるための柔軟なインターフェースになります。

asyncio.sleep を自作する¶

Future を使い、asyncio.sleep() をまねた独自の非同期スリープ async_sleep を作る例を見ていきましょう。

次のコードはイベントループにいくつかのタスクを登録してから、async_sleep(3) コルーチンを包む、asyncio.create_task で作ったタスクを await します。ほかのタスクの実行を妨げずに、3 秒経過してからこのタスクを完了させたいと考えています。

async def other_work():
    print("I like work. Work work.")

async def main():
    # 非同期でスリープしている間に実行する処理があるように、
    # イベントループにほかのタスクをいくつか追加する。
    work_tasks = [
        asyncio.create_task(other_work()),
        asyncio.create_task(other_work()),
        asyncio.create_task(other_work())
    ]
    print(
        "Beginning asynchronous sleep at time: "
        f"{datetime.datetime.now().strftime("%H:%M:%S")}."
    )
    await asyncio.create_task(async_sleep(3))
    print(
        "Done asynchronous sleep at time: "
        f"{datetime.datetime.now().strftime("%H:%M:%S")}."
    )
    # asyncio.gather は実質的に集合内の各タスクを await する。
    await asyncio.gather(*work_tasks)

以下では、Future を使ってタスクが完了するタイミングを独自に制御します。 Future を完了させるメソッド future.set_result() が呼ばれなければ、このタスクはいつまでも完了しません。また、経過時間を監視して、適切なタイミングで future.set_result() を呼び出す別のタスクも使います。そのタスクは後で紹介します。

async def async_sleep(seconds: float):
    future = asyncio.Future()
    time_to_wake = time.time() + seconds
    # 監視タスクをイベントループに追加する。
    watcher_task = asyncio.create_task(_sleep_watcher(future, time_to_wake))
    # Future が完了するまで待機する。
    await future

以下では、ごく簡単な YieldToEventLoop() オブジェクトを使います。このオブジェクトは __await__ メソッド内で yield し、イベントループに制御を渡します。これは実質的には asyncio.sleep(0) を呼ぶのと同じですが、この方法のほうが仕組みがはっきりします。それに、asyncio.sleep の実装方法を示すのに asyncio.sleep 自体を使うのは、少しずるいでしょう。

As usual, the event loop cycles through its tasks, giving them control and receiving control back when they pause or finish. The watcher_task, which runs the coroutine _sleep_watcher(...), will be invoked once per full cycle of the event loop. On each resumption, it'll check the time and if not enough has elapsed, then it'll pause once again and hand control back to the event loop. Once enough time has elapsed, _sleep_watcher(...) marks the future as done and completes by exiting its infinite while loop. Given this helper task is only invoked once per cycle of the event loop, you'd be correct to note that this asynchronous sleep will sleep at least three seconds, rather than exactly three seconds. Note this is also true of asyncio.sleep.

class YieldToEventLoop:
    def __await__(self):
        yield

async def _sleep_watcher(future, time_to_wake):
    while True:
        if time.time() >= time_to_wake:
            # Future を完了させる。
            future.set_result(None)
            break
        else:
            await YieldToEventLoop()

プログラム全体の出力は次のとおりです。

$ python custom-async-sleep.py
Beginning asynchronous sleep at time: 14:52:22.
I like work. Work work.
I like work. Work work.
I like work. Work work.
Done asynchronous sleep at time: 14:52:25.

この非同期スリープの実装は、必要以上に複雑だと感じたかもしれません。確かにそのとおりです。この例の目的は、より複雑な要件にも応用できる単純な例で、Future の柔軟さを示すことでした。参考までに、Future を使わずに次のように実装することもできます。

async def simpler_async_sleep(seconds):
    time_to_wake = time.time() + seconds
    while True:
        if time.time() >= time_to_wake:
            return
        else:
            await YieldToEventLoop()

今回はここまでです。自信を持って非同期プログラミングに取り組んだり、ほかのドキュメント で高度な話題を調べたりする準備ができていれば幸いです。

`asyncio` の概念を理解する¶

概念の説明、第 1 部: 全体像¶

イベントループ¶

非同期関数とコルーチン¶

Task¶

await¶

タスクの待機¶

コルーチンの待機¶

概念の説明、第 2 部: 内部の仕組み¶

コルーチンの内部動作¶

Future¶

asyncio.sleep を自作する¶

目次

前のトピックへ

次のトピックへ

このページ

asyncio の概念を理解する¶

概念の説明、第 1 部: 全体像¶

イベントループ¶

非同期関数とコルーチン¶

Task¶

await¶

タスクの待機¶

コルーチンの待機¶

概念の説明、第 2 部: 内部の仕組み¶

コルーチンの内部動作¶

Future¶

asyncio.sleep を自作する¶

`asyncio` の概念を理解する¶