asyncio での開発¶

非同期プログラミングは、従来の「逐次的」なプログラミングとは異なります。

このページはよくある間違いや落とし穴を列挙し、それらを回避する方法を説明します。

デバッグモード¶

asyncio はデフォルトで本運用モードで実行されます。いっぽう、開発を容易にするために asyncio は "デバッグモード" を持っています。

asyncio のデバッグモードを有効化する方法はいくつかあります:

PYTHONASYNCIODEBUG 環境変数の値を 1 に設定する。
Python 開発モードを使う。
asyncio.run() 実行時に debug=True を設定する。
loop.set_debug() を呼び出す。

デバッグモードを有効化することに加え、以下も検討してください:

asyncio ロガーのログレベルを logging.DEBUG に設定します。例えばアプリケーションの起動時に以下を実行します:
```
logging.basicConfig(level=logging.DEBUG)
```
warnings モジュールが ResourceWarning 警告を表示するように設定します。やり方のひとつは -W default コマンドラインオプションを使うことです。

デバッグモードが有効化されたときの動作:

スレッドセーフでない asyncio APIs の多く (loop.call_soon() や loop.call_at() など) は、誤ったスレッドから呼び出されたときに例外を送出します。
I/O セレクタが I/O 処理を実行する時間が長すぎる場合、その実行時間が記録されます。
実行時間が 100 ミリ秒を超えるコールバックはログに記録されます。 loop.slow_callback_duration 属性で、「遅い」とみなす実行時間の最小値を秒単位で設定できます。

並行処理とマルチスレッド処理¶

イベントループはスレッド（典型的にはメインスレッド）内で動作し、すべてのコールバックとタスクをそのスレッド内で実行します。ひとつのタスクがイベントループ内で実行される間、他のタスクを同じスレッド内で実行することはできません。タスクが await 式を実行すると、実行中のタスクはサスペンドされ、イベントループは次のタスクを実行します。

別の OS スレッドからのコールバック (callback) をスケジュールする場合、 loop.call_soon_threadsafe() メソッドを使ってください。例:

loop.call_soon_threadsafe(callback, *args)

ほぼ全ての非同期オブジェクトはスレッドセーフではありませんが、タスクやコールバックの外側で非同期オブジェクトを使うコードが存在しない限り、それが問題にはなることはほとんどありません。もしそのような目的で低レベルの asyncio API を呼び出すようなコードを書く必要がある場合、 loop.call_soon_threadsafe() メソッドを使ってください。例:

loop.call_soon_threadsafe(fut.cancel)

別の OS スレッドからコルーチンオブジェクトをスケジュールする場合は、 run_coroutine_threadsafe() メソッドを使ってください。 run_coroutine_threadsafe() は結果にアクセスするための concurrent.futures.Future オブジェクトを返します:

async def coro_func():
     return await asyncio.sleep(1, 42)

# 後で、別の OS スレッド内で実行する:

future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
# 結果を待つ:
result = future.result()

シグナルの処理を行うには、イベントループはメインスレッド内で実行しなければなりません。

loop.run_in_executor() メソッドを concurrent.futures.ThreadPoolExecutor または InterpreterPoolExecutor と組み合わせると、イベントループの OS スレッドをブロックせずに、別の OS スレッドでブロッキングコードを実行できます。

現在のところ、 (たとえば multiprocessing で開始したような) 別のプロセスからコルーチンやコールバックを直接スケジュールすることはできません。イベントループのメソッド節では、イベントループをブロックすることなくパイプからの読み込みやファイルデスクリプタの監視ができる API のリストを掲載しています。さらに、 asyncio のサブプロセス API はイベントループからプロセスを開始したりプロセスと通信したりする方法を提供します。最後に、前述の loop.run_in_executor() メソッドは concurrent.futures.ProcessPoolExecutor とともに使用することで、別のプロセス内でコードを実行することもできます。

ブロッキングコードの実行¶

ブロッキングコード (CPU バウンドなコード) を直接呼び出すべきではありません。たとえば、 CPU 負荷の高い関数を1秒実行したとすると、並行に処理されている全ての非同期タスクと I/O 処理は1秒遅れる可能性があります。

Executor を使うと、別のスレッド (別のインタープリタを含む) や別のプロセスで処理を実行し、イベントループの OS スレッドがブロックされるのを避けられます。詳細は loop.run_in_executor() メソッドを参照してください。

ログ記録¶

asyncio は logging モジュールを利用し、全てのログ記録は "asyncio" ロガーを通じて行われます。

デフォルトのログレベルは logging.INFO ですが、これは簡単に調節できます:

logging.getLogger("asyncio").setLevel(logging.WARNING)

ネットワークログ記録は、イベントループをブロックし得ます。ログ処理のスレッドを分離するか、ノンブロッキング IO を使用することを推奨します。例えば、ブロックする handler を扱うを見てください。

待ち受け処理を伴わないコルーチンの検出¶

コルーチン関数を呼び出しただけで、await も asyncio.create_task() によるスケジュールも行わなかった場合 (例えば await coro() の代わりに coro() と書いた場合)、asyncio は RuntimeWarning を出します。

import asyncio

async def test():
    print("never scheduled")

async def main():
    test()

asyncio.run(main())

出力:

test.py:7: RuntimeWarning: coroutine 'test' was never awaited
  test()

デバッグモードの出力:

test.py:7: RuntimeWarning: coroutine 'test' was never awaited
Coroutine created at (most recent call last)
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

  < .. >

  File "../t.py", line 7, in main
    test()
  test()

通常の修正方法はコルーチンを待ち受ける (await) か、 asyncio.create_task() 関数を呼び出すことです:

async def main():
    await test()

回収されない例外の検出¶

もし Future.set_exception() メソッドが呼び出されても、その Future オブジェクトを待ち受けていなければ、例外は決してユーザーコードまで伝播しません。この場合 asyncio は、 Future オブジェクトがガベージコレクションの対象となったときにログメッセージを送出することがあります。

処理されない例外の例:

import asyncio

async def bug():
    raise Exception("not consumed")

async def main():
    asyncio.create_task(bug())

asyncio.run(main())

出力:

Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
  exception=Exception('not consumed')>

Traceback (most recent call last):
  File "test.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed

タスクが生成された箇所を特定するには、デバッグモードを有効化してトレースバックを取得してください:

asyncio.run(main(), debug=True)

デバッグモードの出力:

Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
    exception=Exception('not consumed') created at asyncio/tasks.py:321>

source_traceback: Object created at (most recent call last):
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

< .. >

Traceback (most recent call last):
  File "../t.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed

非同期ジェネレーターを使う際の推奨事項¶

正しく効率的な asyncio コードを書くには、いくつかの落とし穴を知っておく必要があります。この節では、長時間のデバッグを避けるために役立つ重要な実践方法を説明します。

非同期ジェネレーターを明示的に閉じる¶

非同期ジェネレーターは明示的に閉じることを推奨します。例えば async for ループの本体で例外が起きるなどして、途中でジェネレーターの使用を終えると、非同期の後始末処理が予期しないコンテキストで実行される可能性があります。依存するタスクが完了した後や、イベントループの終了中に非同期ジェネレーターのガベージコレクション用フックが呼び出されたときに、後始末が行われることがあります。

これを避けるには、aclose() メソッドを呼び出して明示的にジェネレーターを閉じるか、contextlib.aclosing() コンテキストマネージャーを使ってください。

import asyncio
import contextlib

async def gen():
  yield 1
  yield 2

async def func():
  async with contextlib.aclosing(gen()) as g:
    async for x in g:
      break  # 最後までイテレーションしない

asyncio.run(func())

前述のとおり、これらの非同期ジェネレーターの後始末処理は後回しにされます。次の例は、非同期ジェネレーターの終了処理が予期しない順序で行われる場合を示しています。

import asyncio
work_done = False

async def cursor():
    try:
        yield 1
    finally:
        assert work_done

async def rows():
    global work_done
    try:
        yield 2
    finally:
        await asyncio.sleep(0.1) # 非同期の処理を模倣する
        work_done = True


async def main():
    async for c in cursor():
        async for r in rows():
            break
        break

asyncio.run(main())

この例の出力は次のようになります。

unhandled exception during asyncio.run() shutdown
task: <Task finished name='Task-3' coro=<<async_generator_athrow without __name__>()> exception=AssertionError()>
Traceback (most recent call last):
  File "example.py", line 6, in cursor
    yield 1
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "example.py", line 8, in cursor
    assert work_done
           ^^^^^^^^^
AssertionError

cursor() 非同期ジェネレーターの終了処理が rows ジェネレーターより先に行われています。これは意図しない動作です。

この例は、cursor と rows の非同期ジェネレーターを明示的に閉じることで修正できます。

async def main():
    async with contextlib.aclosing(cursor()) as cursor_gen:
        async for c in cursor_gen:
            async with contextlib.aclosing(rows()) as rows_gen:
                async for r in rows_gen:
                    break
            break

Create asynchronous generators only when the event loop is running¶

It is recommended to create asynchronous generators only after the event loop has been created.

非同期ジェネレーターを確実に閉じるため、イベントループは sys.set_asyncgen_hooks() 関数でコールバック関数を登録します。これらのコールバックは、実行中の非同期ジェネレーターの一覧を更新し、整合性を保ちます。

loop.shutdown_asyncgens() 関数が呼び出されると、実行中のジェネレーターの終了処理を行い、一覧を空にします。

非同期ジェネレーターは、最初のイテレーションで対応するシステムフックを呼び出します。同時に、フックを呼び出したことを記録し、以後は再び呼び出しません。

Therefore, if iteration begins before the event loop is created, the event loop will not be able to add the generator to its list of active generators because the hooks are set after the generator attempts to call them. Consequently, the event loop will not be able to terminate the generator if necessary.

次に示す例について考えてみてください。:

import asyncio

async def agenfn():
    try:
        yield 10
    finally:
        await asyncio.sleep(0)


with asyncio.Runner() as runner:
    agen = agenfn()
    print(runner.run(anext(agen)))
    del agen

出力:

10
Exception ignored while closing generator <async_generator object agenfn at 0x000002F71CD10D70>:
Traceback (most recent call last):
  File "example.py", line 13, in <module>
    del agen
        ^^^^
RuntimeError: async generator ignored GeneratorExit

この例は次のように修正できます。

import asyncio

async def agenfn():
    try:
        yield 10
    finally:
        await asyncio.sleep(0)

async def main():
    agen = agenfn()
    print(await anext(agen))
    del agen

asyncio.run(main())

同じジェネレーターのイテレーションとクローズを並行して行わない¶

Async generators may be reentered while another __anext__() / athrow() / aclose() call is in progress. This may lead to an inconsistent state of the async generator and can cause errors.

次の例を考えてみてください。

import asyncio

async def consumer():
    for idx in range(100):
        await asyncio.sleep(0)
        message = yield idx
        print('received', message)

async def amain():
    agenerator = consumer()
    await agenerator.asend(None)

    fa = asyncio.create_task(agenerator.asend('A'))
    fb = asyncio.create_task(agenerator.asend('B'))
    await fa
    await fb

asyncio.run(amain())

出力:

received A
Traceback (most recent call last):
  File "test.py", line 38, in <module>
    asyncio.run(amain())
    ~~~~~~~~~~~^^^^^^^^^
  File "Lib/asyncio/runners.py", line 204, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "Lib/asyncio/runners.py", line 127, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "Lib/asyncio/base_events.py", line 719, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File "test.py", line 36, in amain
    await fb
RuntimeError: anext(): asynchronous generator is already running

そのため、並行して動くタスクや複数のイベントループで非同期ジェネレーターを共有して使うことは避けるよう推奨します。