asyncioを使ってみよう

asyncioの概要

threadingモジュールは複数のスレッドを生成することで同時実行を実装し、multiprocessingモジュールはシステムプロセスを使用して同時実行を実装します。これに対して、asyncioは、アプリケーションの一部が協力して最適なタイミングでタスクを明示的に切り替えるシングルスレッドのシングルプロセスアプローチを使用します。 ほとんどの場合、このコンテキストスイッチングは、プログラムがデータの読み取りまたは書き込みの待機をブロックするときに発生しますが、asyncioには、システム信号を処理するために、あるコルーチンが別のコルーチンの完了を待機できるように、また、アプリケーションが作業内容を変更する理由となる可能性のある他のイベントを認識するために、特定の将来の時間に実行されるコードのスケジューリングのサポートも含まれています。

非同期処理で実行されるタスクをコルーチン関数、あるいは単にコルーチンといいます。

コルーチンはサブルーチンのより一般的な形式で、サブルーチンは決められた地点から入り、別の決められた地点から出るのに対して、コルーチンは多くの様々な地点から入る、出る、再開することができます。コルーチンは中断することができる関数と考えると理解が早いでしょう。

asyncioモジュールは、コルーチンを使用して並行アプリケーションを構築するためのツールを提供します。

Python 3.6 までは次のようにイベントループから実行します。

code: python

import asyncio

async def hello_world():

print("Hello")

await asyncio.sleep(1)

print("World!")

loop = asyncio.get_event_loop()

loop.run_until_complete(hello_world())

loop.close()

code: python

import asyncio

async def hello_world():

print('hello')

await asyncio.sleep(1)

print('world')

asyncio.run(hello_world())

code: python

import aiohttp

import asyncio

async def fetch(session, url):

async with session.get(url) as response:

return await response.text()

async def main():

async with aiohttp.ClientSession() as session:

print(html)

if __name__ == '__main__':

loop = asyncio.get_event_loop()

loop.run_until_complete(main())

この例では、与えたURLからデータ（つまりHTML)を取得して、出力しています。

非同期同時実行の概念

他の同時実行モデルを使用するほとんどのプログラムは直線的に記述され、言語ランタイムまたはオペレーティングシステムの基盤となるスレッドまたはプロセス管理に依存して、必要に応じてコンテキストを変更します。 asyncioに基づくアプリケーションでは、コンテキストの変更を明示的に処理するアプリケーションコードが必要で、それを正しく行うための手法を使用するには、相互に関連するいくつかの概念を理解する必要があります。

asyncioが提供するフレームワークは、I/Oイベント、システムイベント、およびアプリケーションコンテキストの変更を効率的に処理するためのファーストクラスオブジェクトであるイベントループを中心としています。オペレーティングシステムの機能を効率的に利用するために、いくつかのループ実装が提供されています。通常、適切なデフォルトが自動的に選択されますが、アプリケーション内から特定のイベントループ実装を選択することもできます。これは、たとえば、一部のループクラスがネットワークI/Oの効率を犠牲にするような方法で外部プロセスのサポートを追加するWindowsで役立ちます。

アプリケーションはイベントループと明示的に対話して実行するコードを登録し、リソースが利用可能になったときにイベントループがアプリケーションコードに必要な呼び出しを行うようにします。たとえば、ネットワークサーバーはソケットを開き、入力イベントが発生したときに通知されるようにソケットを登録します。イベントループは、新しい着信接続があるとき、または読み取るデータがあるときにサーバーコードに警告します。アプリケーションコードは、現在のコンテキストでこれ以上作業を行うことができない短い期間の後に、再び制御をもたらすことが期待されます。たとえば、ソケットから読み取るデータがこれ以上ない場合、サーバーは制御をイベントループに戻す必要があります。

制御をイベントループに戻すメカニズムは、Pythonのコルーチン、つまり状態を失うことなく呼び出し元に制御を渡す特別な関数に依存しています。コルーチンはジェネレーター関数に似ており、実際、ジェネレーターを使用して、コルーチンオブジェクトのネイティブサポートなしで3.5より前のバージョンのPythonでコルーチンを実装できます。 asyncioは、コルーチンを直接書き込む代わりに、コールバックを使用してコードを記述するためのプロトコルとトランスポートのクラスベースの抽象化レイヤーも提供します。クラスベースモデルとコルーチンモデルの両方で、イベントループに再び入ることによってコンテキストを明示的に変更することは、Pythonのスレッド実装における暗黙的なコンテキスト変更の代わりになります。

タスクは、コルーチンの実行をラップして管理する方法を知っているFutureのサブクラスです。タスクをイベントループでスケジュールして、必要なリソースが利用可能になったときに実行し、他のコルーチンで使用できる結果を生成できます。

コルーチンとの協調マルチタスク

コルーチンの開始

code: async_coroutine.py

import asyncio

async def coroutine():

print('in coroutine')

event_loop = asyncio.get_event_loop()

try:

print('starting coroutine')

coro = coroutine()

print('entering event loop')

event_loop.run_until_complete(coro)

finally:

print('closing event loop')

event_loop.close()

code: bash

$ python asyncio_coroutine.py

starting coroutine

entering event loop

in coroutine

closing event loop

コルーチンからの戻り値

コルーチンの戻り値は、コルーチンを開始して待機するコードに返されます。

code: asyncio_coroutine_return.py

import asyncio

async def coroutine():

print('in coroutine')

return 'result'

event_loop = asyncio.get_event_loop()

try:

return_value = event_loop.run_until_complete(

coroutine()

)

print('it returned: {!r}'.format(return_value))

finally:

event_loop.close()

code: bash

$ python asyncio_coroutine_return.py

in coroutine

it returned: 'result'

コルーチンの連鎖

1つのコルーチンが別のコルーチンを開始し、結果を待つことができます。 これにより、タスクを再利用可能な部分に簡単に分解できます。 次の例には、順番に実行する必要がある2つのフェーズがありますが、他の操作と同時に実行できます。

code: asyncio_coroutine_chain.py

import asyncio

async def outer():

print('in outer')

print('waiting for result1')

result1 = await phase1()

print('waiting for result2')

result2 = await phase2(result1)

return (result1, result2)

async def phase1():

print('in phase1')

return 'result1'

async def phase2(arg):

print('in phase2')

return f'result2 derived from {arg}'

event_loop = asyncio.get_event_loop()

try:

return_value = event_loop.run_until_complete(outer())

print('return value: {!r}'.format(return_value))

finally:

event_loop.close()

code: bash

$ python asyncio_coroutine_chain.py

in outer

waiting for result1

in phase1

waiting for result2

in phase2

return value: ('result1', 'result2 derived from result1')

コルーチンの代替としてジェネレーター

コルーチン関数は、asyncioの設計の重要なコンポーネントです。 これらは、プログラムの一部の実行を停止し、その呼び出しの状態を保持し、後で状態に再び入るための言語構造を提供します。これらはすべて、並行性フレームワークの重要な機能です。

code: asyncio_coroutine_generator.py

import asyncio

@asyncio.coroutine

def outer():

print('in outer')

print('waiting for result1')

result1 = yield from phase1()

print('waiting for result2')

result2 = yield from phase2(result1)

return (result1, result2)

@asyncio.coroutine

def phase1():

print('in phase1')

return 'result1'

@asyncio.coroutine

def phase2(arg):

print('in phase2')

return f'result2 derived from {arg}'

event_loop = asyncio.get_event_loop()

try:

return_value = event_loop.run_until_complete(outer())

print('return value: {!r}'.format(return_value))

finally:

event_loop.close()

code: bash

$ python asyncio_coroutine_generator.py

in outer

waiting for result1

in phase1

waiting for result2

in phase2

return value: ('result1', 'result2 derived from result1')

通常の関数への呼び出しのスケジューリング

コルーチンとI / Oコールバックの管理に加えて、asyncioイベントループは、ループに保持されているタイマー値に基づいて、通常の関数への呼び出しをスケジュールできます。

コールバックを「すぐに」スケジュールする

code: asyncio_call_soon.py

import asyncio

import functools

def callback(arg, *, kwarg='default'):

print(f'callback invoked with {arg} and {kwarg}')

async def main(loop):

print('registering callbacks')

loop.call_soon(callback, 1)

wrapped = functools.partial(callback, kwarg='not default')

loop.call_soon(wrapped, 2)

await asyncio.sleep(0.1)

event_loop = asyncio.get_event_loop()

try:

print('entering event loop')

event_loop.run_until_complete(main(event_loop))

finally:

print('closing event loop')

event_loop.close()

コールバックは、スケジュールされた順序で呼び出されます。

code: bash

$ python asyncio_call_soon.py

entering event loop

registering callbacks

callback invoked with 1 and default

callback invoked with 2 and not default

closing event loop

遅延のあるコールバックのスケジュール

code: asyncio_call_later.py

import asyncio

def callback(n):

print(f'callback {n} invoked')

async def main(loop):

print('registering callbacks')

loop.call_later(0.2, callback, 1)

loop.call_later(0.1, callback, 2)

loop.call_soon(callback, 3)

await asyncio.sleep(0.4)

event_loop = asyncio.get_event_loop()

try:

print('entering event loop')

event_loop.run_until_complete(main(event_loop))

finally:

print('closing event loop')

event_loop.close()

code: bash

$ python asyncio_call_later.py

entering event loop

registering callbacks

callback 3 invoked

callback 2 invoked

callback 1 invoked

closing event loop

特定の時間のコールバックのスケジュール

code: asyncio_call_at.py

import time

import asyncio

def callback(n, loop):

print('callback {} invoked at {}'.format(n, loop.time()))

async def main(loop):

now = loop.time()

print('clock time: {time.time()}')

print('loop time: {now}')

print('registering callbacks')

loop.call_at(now + 0.2, callback, 1, loop)

loop.call_at(now + 0.1, callback, 2, loop)

loop.call_soon(callback, 3, loop)

await asyncio.sleep(1)

event_loop = asyncio.get_event_loop()

try:

print('entering event loop')

event_loop.run_until_complete(main(event_loop))

finally:

print('closing event loop')

event_loop.close()

code: bash

$ python asyncio_call_at.py

entering event loop

clock time: {time.time()}

loop time: {now}

registering callbacks

callback 3 invoked at 0.082787363

callback 2 invoked at 0.185166158

callback 1 invoked at 0.285039865

closing event loop

非同期で結果を生成する

Futureオブジェクトを待つ

code: asyncio_future_event_loop.py

import asyncio

def mark_done(future, result):

print('setting future result to {!r}'.format(result))

future.set_result(result)

event_loop = asyncio.get_event_loop()

try:

all_done = asyncio.Future()

print('scheduling mark_done')

event_loop.call_soon(mark_done, all_done, 'the result')

print('entering event loop')

result = event_loop.run_until_complete(all_done)

print('returned result: {!r}'.format(result))

finally:

print('closing event loop')

event_loop.close()

print('future result: {!r}'.format(all_done.result()))

code: bash

$ python asyncio_future_event_loop.py

scheduling mark_done

entering event loop

setting future result to 'the result'

returned result: 'the result'

closing event loop

future result: 'the result'

code: asyncio_future_await.py

import asyncio

def mark_done(future, result):

print('setting future result to {!r}'.format(result))

future.set_result(result)

async def main(loop):

all_done = asyncio.Future()

print('scheduling mark_done')

loop.call_soon(mark_done, all_done, 'the result')

result = await all_done

print('returned result: {!r}'.format(result))

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(event_loop))

finally:

event_loop.close()

code: bash

$ python asyncio_future_await.py

scheduling mark_done

setting future result to 'the result'

returned result: 'the result'

Furetureオブジェクトのコールバック

code: asyncio_future_callback.py

import asyncio

import functools

def callback(future, n):

print('{}: future done: {}'.format(n, future.result()))

async def register_callbacks(all_done):

print('registering callbacks on future')

all_done.add_done_callback(functools.partial(callback, n=1))

all_done.add_done_callback(functools.partial(callback, n=2))

async def main(all_done):

await register_callbacks(all_done)

print('setting result of future')

all_done.set_result('the result')

event_loop = asyncio.get_event_loop()

try:

all_done = asyncio.Future()

event_loop.run_until_complete(main(all_done))

finally:

event_loop.close()

code: bash

$ python asyncio_future_callback.py

registering callbacks on future

setting result of future

1: future done: the result

2: future done: the result

タスクの同時実行

タスクの開始

code: asyncio_create_task.py

import asyncio

async def task_func():

print('in task_func')

return 'the result'

async def main(loop):

print('creating task')

task = loop.create_task(task_func())

print('waiting for {!r}'.format(task))

return_value = await task

print('task completed {!r}'.format(task))

print('return value: {!r}'.format(return_value))

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(event_loop))

finally:

event_loop.close()

code: bash

$ python asyncio_create_task.py

creating task

waiting for <Task pending coro=<task_func() running at asyncio_create_task.py:3>>

in task_func

task completed <Task finished coro=<task_func() done, defined at asyncio_create_task.py:3> result='the result'>

return value: 'the result'

タスクのキャンセル

code: asyncio_cancel_task.py

import asyncio

async def task_func():

print('in task_func')

return 'the result'

async def main(loop):

print('creating task')

task = loop.create_task(task_func())

print('canceling task')

task.cancel()

print('canceled task {!r}'.format(task))

try:

await task

except asyncio.CancelledError:

print('caught error from canceled task')

else:

print('task result: {!r}'.format(task.result()))

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(event_loop))

finally:

event_loop.close()

code: bash

$ python asyncio_cancel_task.py

creating task

canceling task

canceled task <Task cancelling coro=<task_func() running at asyncio_cancel_task.py:3>>

caught error from canceled task

code: asyncio_cancel_task2.py

import asyncio

async def task_func():

print('in task_func, sleeping')

try:

await asyncio.sleep(1)

except asyncio.CancelledError:

print('task_func was canceled')

raise

return 'the result'

def task_canceller(t):

print('in task_canceller')

t.cancel()

print('canceled the task')

async def main(loop):

print('creating task')

task = loop.create_task(task_func())

loop.call_soon(task_canceller, task)

try:

await task

except asyncio.CancelledError:

print('main() also sees task as canceled')

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(event_loop))

finally:

event_loop.close()

例外をキャッチすると、必要に応じて、すでに実行された作業をクリーンアップする機会が提供されます。

code: bash

$ python asyncio_cancel_task2.py

creating task

in task_func, sleeping

in task_canceller

canceled the task

task_func was canceled

main() also sees task as canceled

コルーチンからのタスクの作成

code: asyncio_ensure_future.py

import asyncio

async def wrapped():

print('wrapped')

return 'result'

async def inner(task):

print('inner: starting')

print('inner: waiting for {!r}'.format(task))

result = await task

print('inner: task returned {!r}'.format(result))

async def starter():

print('starter: creating task')

task = asyncio.ensure_future(wrapped())

print('starter: waiting for inner')

await inner(task)

print('starter: inner returned')

event_loop = asyncio.get_event_loop()

try:

print('entering event loop')

result = event_loop.run_until_complete(starter())

finally:

event_loop.close()

code: bash

$ python asyncio_ensure_future.py

entering event loop

starter: creating task

starter: waiting for inner

inner: starting

inner: waiting for <Task pending coro=<wrapped() running at asyncio_ensure_future.py:3>>

wrapped

inner: task returned 'result'

starter: inner returned

制御構造を持つコルーチンの構成

複数のコルーチンを待つ

code: asyncio_wait.py

import asyncio

async def phase(i):

print(f'in phase {i}')

await asyncio.sleep(0.1 * i)

print(f'done with phase {i}')

return f'phase {i} result'

async def main(num_phases):

print('starting main')

phases = [

phase(i)

for i in range(num_phases)

]

print('waiting for phases to complete')

completed, pending = await asyncio.wait(phases)

print('results: {!r}'.format(results))

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(3))

finally:

event_loop.close()

code: bash

$ python asyncio_wait.py

starting main

waiting for phases to complete

in phase 0

in phase 2

in phase 1

done with phase 0

done with phase 1

done with phase 2

code: asyncio_wait_timeout.py

import asyncio

async def phase(i):

print(f'in phase {i}')

try:

await asyncio.sleep(0.1 * i)

except asyncio.CancelledError:

print(f'phase {i} canceled')

raise

else:

print(f'done with phase {i}')

return f'phase {i} result'

async def main(num_phases):

print('starting main')

phases = [

phase(i)

for i in range(num_phases)

]

print('waiting 0.1 for phases to complete')

completed, pending = await asyncio.wait(phases, timeout=0.1)

print('{} completed and {} pending'.format(

len(completed), len(pending),

))

# タスクを終了せずに終了するときにエラーが発生しないように

# 残りのタスクをキャンセルする

if pending:

print('canceling tasks')

for t in pending:

t.cancel()

print('exiting main')

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(3))

finally:

event_loop.close()

これらの残りのバックグラウンド操作は、キャンセルするか、それらを待つことによって終了する必要があります。 イベントループが続く間、それらを保留のままにしておくと、それらはさらに実行されます。これは、操作全体が中止されたと見なされる場合は望ましくない場合があります。 プロセスの最後にそれらを保留のままにしておくと、警告が報告されます。

code: bash

$ python asyncio_wait_timeout.py

starting main

waiting 0.1 for phases to complete

in phase 0

in phase 1

in phase 2

done with phase 0

1 completed and 2 pending

canceling tasks

exiting main

phase 1 canceled

phase 2 canceled

コルーチンからの結果を収集

code: asyncio_gather.py

import asyncio

async def phase1():

print('in phase1')

await asyncio.sleep(2)

print('done with phase1')

return 'phase1 result'

async def phase2():

print('in phase2')

await asyncio.sleep(1)

print('done with phase2')

return 'phase2 result'

async def main():

print('starting main')

print('waiting for phases to complete')

results = await asyncio.gather(

phase1(),

phase2(),

)

print('results: {!r}'.format(results))

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main())

finally:

event_loop.close()

code: bash

$ python asyncio_gather.py

starting main

waiting for phases to complete

in phase1

in phase2

done with phase2

done with phase1

バックグラウンド操作終了時の処理

code: asyncio_as_completed.py

import asyncio

async def phase(i):

print(f'in phase {i}')

await asyncio.sleep(0.5 - (0.1 * i))

print(f'done with phase {i}')

return f'phase {i} result'

async def main(num_phases):

print('starting main')

phases = [

phase(i)

for i in range(num_phases)

]

print('waiting for phases to complete')

results = []

for next_to_complete in asyncio.as_completed(phases):

answer = await next_to_complete

print('received answer {!r}'.format(answer))

results.append(answer)

print('results: {!r}'.format(results))

return results

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(3))

finally:

event_loop.close()

code: bash

$ python asyncio_as_completed.py

starting main

waiting for phases to complete

in phase 0

in phase 2

in phase 1

done with phase 2

received answer 'phase 2 result'

done with phase 1

received answer 'phase 1 result'

done with phase 0

received answer 'phase 0 result'

同期プリミティブ

asyncioアプリケーションは通常、シングルスレッドプロセスとして実行されますが、それでも並行アプリケーションとして構築されます。 各コルーチンまたはタスクは、I / Oおよびその他の外部イベントからの遅延および割り込みに基づいて、予測できない順序で実行される場合があります。 安全な並行性をサポートするために、asyncioには、スレッドモジュールとマルチプロセッシングモジュールにあるのと同じ低レベルプリミティブのいくつかの実装が含まれています。

ロック

ロックを使用して、共有リソースへのアクセスを保護できます。 ロックの所有者のみがリソースを使用できます。 ロックを取得しようと複数回試行するとブロックされるため、一度に1つのホルダーしかありません。

code: async_lock.py

import asyncio

import functools

def unlock(lock):

print('callback releasing lock')

lock.release()

async def coro1(lock):

print('coro1 waiting for the lock')

async with lock:

print('coro1 acquired lock')

print('coro1 released lock')

async def coro2(lock):

print('coro2 waiting for the lock')

await lock.acquire()

try:

print('coro2 acquired lock')

finally:

print('coro2 released lock')

lock.release()

async def main(loop):

# 共有ロックを作成して取得します。

lock = asyncio.Lock()

print('acquiring the lock before starting coroutines')

await lock.acquire()

print('lock acquired: {}'.format(lock.locked()))

# ロックを解除するためのコールバックをスケジュールする

loop.call_later(0.1, functools.partial(unlock, lock))

# ロックを使用するコルーチンを実行

print('waiting for coroutines')

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(event_loop))

finally:

event_loop.close()

code: bash

$ python asyncio_lock.py

acquiring the lock before starting coroutines

lock acquired: True

waiting for coroutines

coro2 waiting for the lock

coro1 waiting for the lock

callback releasing lock

coro2 acquired lock

coro2 released lock

coro1 acquired lock

coro1 released lock

Eventクラス

code: asyncio_event.py

import asyncio

import functools

def set_event(event):

print('setting event in callback')

event.set()

async def coro1(event):

print('coro1 waiting for event')

await event.wait()

print('coro1 triggered')

async def coro2(event):

print('coro2 waiting for event')

await event.wait()

print('coro2 triggered')

async def main(loop):

# Create a shared event

event = asyncio.Event()

print(f'event start state: {event.is_set()}')

loop.call_later(

0.1, functools.partial(set_event, event)

)

print(f'event end state: {event.is_set()}')

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(event_loop))

finally:

event_loop.close()

code: bash

$ python asyncio_event.py

event start state: False

coro2 waiting for event

coro1 waiting for event

setting event in callback

coro2 triggered

coro1 triggered

event end state: True

Conditionオブジェクト

code: asyncio_condition.py

import asyncio

async def consumer(condition, n):

async with condition:

print(f'consumer {n} is waiting')

await condition.wait()

print(f'consumer {n} triggered')

print(f'ending consumer {n}')

async def manipulate_condition(condition):

print('starting manipulate_condition')

# 一時停止してコンシューマーが開始できるようにする

await asyncio.sleep(0.1)

for i in range(1, 3):

async with condition:

print(f'notifying {i} consumers')

condition.notify(n=i)

await asyncio.sleep(0.1)

async with condition:

print('notifying remaining consumers')

condition.notify_all()

print('ending manipulate_condition')

async def main(loop):

# Create a condition

condition = asyncio.Condition()

# Set up tasks watching the condition

consumers = [

consumer(condition, i)

for i in range(5)

]

# condition変数を操作するタスクをスケジュールする

loop.create_task(manipulate_condition(condition))

# コンシューマーが終了するのを待つ

await asyncio.wait(consumers)

event_loop = asyncio.get_event_loop()

try:

result = event_loop.run_until_complete(main(event_loop))

finally:

event_loop.close()

code: bash

$ python asyncio_condition.py

starting manipulate_condition

consumer 2 is waiting

consumer 3 is waiting

consumer 0 is waiting

consumer 4 is waiting

consumer 1 is waiting

notifying 1 consumers

consumer 2 triggered

ending consumer 2

notifying 2 consumers

consumer 3 triggered

ending consumer 3

consumer 0 triggered

ending consumer 0

notifying remaining consumers

ending manipulate_condition

consumer 4 triggered

ending consumer 4

consumer 1 triggered

ending consumer 1

Queuesクラス

code: async_queues.py

import asyncio

async def consumer(n, q):

print(f'consumer {n}: starting')

while True:

print(f'consumer {n}: waiting for item')

item = await q.get()

print(f'consumer {n}: has item {item}')

if item is None:

# None は終了させるためのシグナル

q.task_done()

break

else:

await asyncio.sleep(0.01 * item)

q.task_done()

print(f'consumer {n}: ending')

async def producer(q, num_workers):

print('producer: starting')

# ジョブをシミュレートするためにキューにいくつかの番号を追加

for i in range(num_workers * 3):

await q.put(i)

print(f'producer: added task {i} to the queue')

# キューにNoneエントリを追加して、

# コンシューマーに終了するように通知

print('producer: adding stop signals to the queue')

for i in range(num_workers):

await q.put(None)

print('producer: waiting for queue to empty')

await q.join()

print('producer: ending')

async def main(loop, num_consumers):

# 固定サイズのキューを作成して、

# コンシューマーがいくつかのアイテムを引き出すまで

# プロデューサーがブロックする

q = asyncio.Queue(maxsize=num_consumers)

# Scheduled the consumer tasks.

consumers = [

loop.create_task(consumer(i, q))

for i in range(num_consumers)

]

# プロデューサータスクをスケジュール

prod = loop.create_task(producer(q, num_consumers))

# すべてのコルーチンが終了するのを待つ

event_loop = asyncio.get_event_loop()

try:

event_loop.run_until_complete(main(event_loop, 2))

finally:

event_loop.close()

code: bash

$ python asyncio_queueus.py

consumer 0: starting

consumer 0: waiting for item

consumer 1: starting

consumer 1: waiting for item

producer: starting

producer: added task 0 to the queue

producer: added task 1 to the queue

consumer 0: has item 0

consumer 1: has item 1

producer: added task 2 to the queue

producer: added task 3 to the queue

consumer 0: waiting for item

consumer 0: has item 2

producer: added task 4 to the queue

consumer 1: waiting for item

consumer 1: has item 3

producer: added task 5 to the queue

producer: adding stop signals to the queue

consumer 0: waiting for item

consumer 0: has item 4

consumer 1: waiting for item

consumer 1: has item 5

producer: waiting for queue to empty

consumer 0: waiting for item

consumer 0: waiting for item

consumer 0: has item None

consumer 0: ending

consumer 1: waiting for item

consumer 1: has item None

consumer 1: ending

producer: ending

Semaphoreクラス

この時点で、別のワーカーがSemaphoreに到達すると、実行が拒否されます。 セマフォは、リソースが過剰に使用されないように保護する有益な方法です。

次の例では、Semaphoreの単純なインスタンスを作成してから、そのセマフォを取得しようとする3つのワーカー関数を作成します。 このセマフォの初期値は2であるため、2つのワーカー関数がセマフォを正常に取得してから解放し、3番目のワーカーがセマフォを取得できるようにします。

code: async_semaphore.py

import asyncio

import time

async def myWorker(semaphore, i):

await semaphore.acquire()

print(f"Worker {i}: セマフォの取得に成功")

await asyncio.sleep(3)

print(f"Worker {i}: セマフォを開放")

semaphore.release()

async def main(loop):

mySemaphore = asyncio.Semaphore(value=2)

await asyncio.wait([myWorker(mySemaphore, 1),

myWorker(mySemaphore, 2),

myWorker(mySemaphore, 3)])

print("メイン・コルーチン")

loop = asyncio.get_event_loop()

loop.run_until_complete(main(loop))

print("すべてのワーカーが終了")

loop.close()

これを実行すると、最初に２つのワーカー（Worker１とWorker３）がセマフォを取得してから解放し、その後、３番目のワーカー(Worker2)が続き、セマフォを取得して開放しています。

セマフォのカウンタの初期値で与えた数が同時にセマフォを取得できるようになります。

code: bash

% python asyncio_semaphore.py

Worker 3: セマフォの取得に成功

Worker 1: セマフォの取得に成功

Worker 3: セマフォを開放

Worker 1: セマフォを開放

Worker 2: セマフォの取得に成功

Worker 2: セマフォを開放

メイン・コルーチン

すべてのワーカーが終了

制限付きセマフォ

code: async_boundedsemaphore.py

import asyncio

import time

async def myWorker(semaphore, i):

await semaphore.acquire()

print(f"Worker {i}: セマフォの取得に成功")

await asyncio.sleep(3)

print(f"Worker {i}: セマフォを開放")

semaphore.release()

print(f"Worker {i}: セマフォを開放 ２回目")

semaphore.release()

async def main(loop):

mySemaphore = asyncio.BoundedSemaphore(value=2)

await asyncio.wait([myWorker(mySemaphore, 1),

myWorker(mySemaphore, 2),

myWorker(mySemaphore, 3)])

print("メイン・コルーチン")

loop = asyncio.get_event_loop()

loop.run_until_complete(main(loop))

print("すべてのワーカーが終了")

loop.close()

code: bash

% python asyncio_bounded_semaphore.py

Worker 3: セマフォの取得に成功

Worker 1: セマフォの取得に成功

Worker 3: セマフォを開放

Worker 3: セマフォを開放 ２回目

Worker 1: セマフォを開放

Worker 2: セマフォの取得に成功

Worker 2: セマフォを開放

Worker 2: セマフォを開放 ２回目

Task exception was never retrieved

future: <Task finished coro=<myWorker() done, defined at asyncio_bounded_semaphore.py:4> exception=ValueError('BoundedSemaphore released too many times')>

Traceback (most recent call last):

File "asyncio_bounded_semaphore.py", line 11, in myWorker

semaphore.release()

File "/Users/goichiiisaka/anaconda3/envs/parallel/lib/python3.7/asyncio/locks.py", line 506, in release

raise ValueError('BoundedSemaphore released too many times')

ValueError: BoundedSemaphore released too many times

Task exception was never retrieved

future: <Task finished coro=<myWorker() done, defined at asyncio_bounded_semaphore.py:4> exception=ValueError('BoundedSemaphore released too many times')>

Traceback (most recent call last):

File "asyncio_bounded_semaphore.py", line 9, in myWorker

semaphore.release()

File "/Users/goichiiisaka/anaconda3/envs/parallel/lib/python3.7/asyncio/locks.py", line 506, in release

raise ValueError('BoundedSemaphore released too many times')

ValueError: BoundedSemaphore released too many times

メイン・コルーチン

すべてのワーカーが終了

プロトコルクラスの抽象化を使用した非同期I/O

この時点まで、例はすべて、一度に1つの概念に焦点を合わせるために並行性とI / O操作を混在させることを避けてきました。 ただし、I / Oブロック時にコンテキストを切り替えることは、非同期の主な使用例の1つです。 このセクションでは、すでに紹介した並行性の概念に基づいて、ソケットセクションとソケットサーバーセクションで使用されている例と同様に、単純なエコーサーバーとクライアントを実装する2つのサンプルプログラムについて説明します。 クライアントはサーバーに接続し、データを送信してから、応答と同じデータを受信できます。 I / O操作が開始されるたびに、実行中のコードはイベントループへの制御を放棄し、I / Oの準備ができるまで他のタスクを実行できるようにします。

エコーサーバー

サーバーは、非同期とロギングを設定するために必要なモジュールをインポートすることから始め、次にイベントループオブジェクトを作成します。

code: asyncio_echo_server_protocol.py

import sys

import logging

import asyncio

SERVER_ADDRESS = ('localhost', 10000)

logging.basicConfig(

level=logging.DEBUG,

format='%(name)s: %(message)s',

stream=sys.stderr,

)

log = logging.getLogger('main')

event_loop = asyncio.get_event_loop()

code: python

class EchoServer(asyncio.Protocol):

code: python

def connection_made(self, transport):

self.transport = transport

self.address = transport.get_extra_info('peername')

self.log = logging.getLogger(

'EchoServer_{}_{}'.format(*self.address)

)

self.log.debug('connection accepted')

code: python

def data_received(self, data):

self.log.debug('received {!r}'.format(data))

self.transport.write(data)

self.log.debug('sent {!r}'.format(data))

code: python

def eof_received(self):

self.log.debug('received EOF')

if self.transport.can_write_eof():

self.transport.write_eof()

code: python

def connection_lost(self, error):

if error:

self.log.error('ERROR: {}'.format(error))

else:

self.log.debug('closing')

super().connection_lost(error)

code: python

# サーバーを作成し、実際のイベントループを開始する前に、ループでコルーチンを終了させる

factory = event_loop.create_server(EchoServer, *SERVER_ADDRESS)

server = event_loop.run_until_complete(factory)

log.debug('starting up on {} port {}'.format(*SERVER_ADDRESS))

code: python

# すべての接続を処理するために、イベントループに永続的に入る

try:

event_loop.run_forever()

finally:

log.debug('closing server')

server.close()

event_loop.run_until_complete(server.wait_closed())

log.debug('closing event loop')

event_loop.close()

エコークライアント

code: asyncio_echo_client_protocl.py

import sys

import logging

import functools

import asyncio

MESSAGES = [

b'This is the message. ',

b'It will be sent ',

b'in parts.',

]

SERVER_ADDRESS = ('localhost', 10000)

logging.basicConfig(

level=logging.DEBUG,

format='%(name)s: %(message)s',

stream=sys.stderr,

)

log = logging.getLogger('main')

event_loop = asyncio.get_event_loop()

code: pyton

class EchoClient(asyncio.Protocol):

def __init__(self, messages, future):

super().__init__()

self.messages = messages

self.log = logging.getLogger('EchoClient')

self.f = future

クライアントがサーバーに正常に接続すると、すぐに通信を開始します。 基になるネットワークコードが複数のメッセージを1つの送信に結合する場合がありますが、メッセージのシーケンスは一度に1つずつ送信されます。 すべてのメッセージが使い果たされると、EOFが送信されます。

code: python

def connection_made(self, transport):

self.transport = transport

self.address = transport.get_extra_info('peername')

self.log.debug(

'connecting to {} port {}'.format(*self.address)

)

# これは、送信されているメッセージの各部分を表示するのが難しくなることを除いて、

# transport.writelines()である可能性があります。

for msg in self.messages:

transport.write(msg)

self.log.debug('sending {!r}'.format(msg))

if transport.can_write_eof():

transport.write_eof()

サーバーからの応答を受信すると、ログに記録されます。

code: python

def data_received(self, data):

self.log.debug('received {!r}'.format(data))

また、ファイルの終わりマーカーを受信するか、サーバー側から接続を閉じると、ローカルトランスポートオブジェクトが閉じられ、結果を設定することで、将来のオブジェクトに完了のマークが付けられます。

code: python

def eof_received(self):

self.log.debug('received EOF')

self.transport.close()

if not self.f.done():

self.f.set_result(True)

def connection_lost(self, exc):

self.log.debug('server closed connection')

self.transport.close()

if not self.f.done():

self.f.set_result(True)

super().connection_lost(exc)

code: python

client_completed = asyncio.Future()

client_factory = functools.partial(

EchoClient,

messages=MESSAGES,

future=client_completed,

)

factory_coroutine = event_loop.create_connection(

client_factory,

*SERVER_ADDRESS,

)

code: python

log.debug('waiting for client to complete')

try:

event_loop.run_until_complete(factory_coroutine)

event_loop.run_until_complete(client_completed)

finally:

log.debug('closing event loop')

event_loop.close()

出力

サーバーを1つのウィンドウで実行し、クライアントを別のウィンドウで実行すると、次の出力が生成されます。

code: bash

% python asyncio_echo_client_protocl.py

asyncio: Using selector: KqueueSelector

main: waiting for client to complete

EchoClient: connecting to ::1 port 10000

EchoClient: sending b'This is the message. '

EchoClient: sending b'It will be sent '

EchoServer_::1_55405: connection accepted

EchoClient: sending b'in parts.'

EchoServer_::1_55405: received b'This is the message. It will be sent in parts.'

EchoServer_::1_55405: sent b'This is the message. It will be sent in parts.'

EchoServer_::1_55405: received EOF

EchoServer_::1_55405: closing

EchoClient: received b'This is the message. It will be sent in parts.'

EchoClient: received EOF

EchoClient: server closed connection

main: closing event loop

$ python asyncio_echo_client_protocl.py

asyncio: Using selector: KqueueSelector

main: waiting for client to complete

EchoServer_::1_55410: connection accepted

EchoClient: connecting to ::1 port 10000

EchoClient: sending b'This is the message. '

EchoServer_::1_55410: received b'This is the message. '

EchoClient: sending b'It will be sent '

EchoServer_::1_55410: sent b'This is the message. '

EchoClient: sending b'in parts.'

EchoServer_::1_55410: received b'It will be sent in parts.'

EchoServer_::1_55410: sent b'It will be sent in parts.'

EchoClient: received b'This is the message. '

EchoServer_::1_55410: received EOF

EchoClient: received b'It will be sent in parts.'

EchoServer_::1_55410: closing

EchoClient: received EOF

EchoClient: server closed connection

main: closing event loop

$ python asyncio_echo_client_protocl.py

asyncio: Using selector: KqueueSelector

main: waiting for client to complete

EchoClient: connecting to ::1 port 10000

EchoServer_::1_55415: connection accepted

EchoClient: sending b'This is the message. '

EchoClient: sending b'It will be sent '

EchoServer_::1_55415: received b'This is the message. '

EchoClient: sending b'in parts.'

EchoServer_::1_55415: sent b'This is the message. '

EchoClient: received b'This is the message. '

EchoServer_::1_55415: received b'It will be sent in parts.'

EchoServer_::1_55415: sent b'It will be sent in parts.'

EchoClient: received b'It will be sent in parts.'

EchoServer_::1_55415: received EOF

EchoServer_::1_55415: closing

EchoClient: received EOF

EchoClient: server closed connection

main: closing event loop

クライアントは常にメッセージを個別に送信しますが、クライアントが最初に実行されたときに、サーバーは1つの大きなメッセージを受信し、それをクライアントにエコーバックします。 これらの結果は、ネットワークのビジー状態と、すべてのデータが準備される前にネットワークバッファーがフラッシュされるかどうかに基づいて、後続の実行で異なります。

code: bash

$ python asyncio_echo_server_protocol.py

asyncio: Using selector: KqueueSelector

main: starting up on localhost port 10000

EchoServer_::1_55431: connection accepted

EchoServer_::1_55431: received b'This is the message. It will be sent in parts.'

EchoServer_::1_55431: sent b'This is the message. It will be sent in parts.'

EchoServer_::1_55431: received EOF

EchoServer_::1_55431: closing

EchoServer_::1_55434: connection accepted

EchoServer_::1_55434: received b'This is the message. It will be sent in parts.'

EchoServer_::1_55434: sent b'This is the message. It will be sent in parts.'

EchoServer_::1_55434: received EOF

EchoServer_::1_55434: closing

EchoServer_::1_55435: connection accepted

EchoServer_::1_55435: received b'This is the message. It will be sent in parts.'

EchoServer_::1_55435: sent b'This is the message. It will be sent in parts.'

EchoServer_::1_55435: received EOF

EchoServer_::1_55435: closing

サブプロセスの操作

他のプログラムやプロセスを操作したり、既存のコードを書き直さずに利用したり、Python内から利用できないライブラリや機能にアクセスしたりする必要があることがよくあります。 ネットワークI/Oと同様に、asyncioには、別のプログラムを開始してからそれと通信するための2つの抽象化が含まれています。

サブプロセスでのプロトコル抽象化の使用

code: async_subprocess_protocol.py

import asyncio

import functools

from dfprotocol import DFProtocol

async def run_df(loop):

print('in run_df')

cmd_done = asyncio.Future(loop=loop)

factory = functools.partial(DFProtocol, cmd_done)

proc = loop.subprocess_exec(

factory,

'df', '-hl',

stdin=None,

stderr=None,

)

try:

print('launching process')

transport, protocol = await proc

print('waiting for process to complete')

await cmd_done

finally:

transport.close()

return cmd_done.result()

event_loop = asyncio.get_event_loop()

try:

return_code, results = event_loop.run_until_complete(

run_df(event_loop)

)

finally:

event_loop.close()

if return_code:

print(f'error exit {return_code}')

else:

print('\nFree space:')

for r in results:

print('{Mounted:25}: {Avail}'.format(**r))

code: dfprotocol.py

import asyncio

class DFProtocol(asyncio.SubprocessProtocol):

def __init__(self, done_future):

self.done = done_future

self.buffer = bytearray()

super().__init__()

def connection_made(self, transport):

print(f'process started {transport.get_pid()}')

self.transport = transport

def pipe_data_received(self, fd, data):

if fd == 1:

self.buffer.extend(data)

def process_exited(self):

print('process exited')

return_code = self.transport.get_returncode()

print(f'return code {return_code}')

if not return_code:

cmd_output = bytes(self.buffer).decode()

results = self._parse_results(cmd_output)

else:

results = []

self.done.set_result((return_code, results))

def parse_results(output):

print('parsing results')

# 出力には1行のヘッダーがあり、すべて1ワード

# 残りの行はファイルシステムごとに1つで 列はヘッダーと一致する

# マウントポイントの名前に空白がないことが前提

if not output:

return []

lines = output.splitlines()

results = [

dict(zip(headers, line.split()))

for line in devices

]

return results

コマンド出力は、ヘッダー名を出力の各行の値にマッピングする一連の辞書に解析され、結果のリストが返されます。

以下の出力は、実行された一連の手順と、それが実行されたシステム上のディスクの空き領域を示しています。

code: bash

% python asyncio_subprocess_protocol.py

in run_df

launching process

process started 31517

waiting for process to complete

read 520 bytes from stdout

process exited

return code 0

parsing results

Free space:

/ : 198Gi

/System/Volumes/VM : 198Gi

/System/Volumes/Preboot : 198Gi

/System/Volumes/Update : 198Gi

/System/Volumes/Data : 198Gi

コルーチンとストリームを使用したサブプロセスの呼び出し

code: asyncio_subprocess_coroutine.py

import asyncio

import asyncio.subprocess

from dfprotocol import parse_results

async def run_df():

print('in run_df')

buffer = bytearray()

create = asyncio.create_subprocess_exec(

'df', '-hl',

stdout=asyncio.subprocess.PIPE,

)

print('launching process')

proc = await create

print(f'process started {proc.pid}')

while True:

line = await proc.stdout.readline()

print('read {!r}'.format(line))

if not line:

print('no more output from command')

break

buffer.extend(line)

print('waiting for process to complete')

await proc.wait()

return_code = proc.returncode

print(f'return code {return_code}')

if not return_code:

cmd_output = bytes(buffer).decode()

results = parse_results(cmd_output)

else:

results = []

return (return_code, results)

event_loop = asyncio.get_event_loop()

try:

return_code, results = event_loop.run_until_complete(

run_df()

)

finally:

event_loop.close()

if return_code:

print(f'error exit {return_code}')

else:

print('\nFree space:')

for r in results:

print('{Mounted:25}: {Avail}'.format(**r))

その時点で、終了ステータスを調べて、出力を解析するか、出力が生成されなかったためにエラーを処理するかを決定できます。 解析ロジックは前の例と同じですが、非表示にするプロトコルクラスがないため、スタンドアロン関数（ここには示されていません）にあります。

データが解析された後、結果と終了コードは発信者に返されます。

dfからの出力は一度に1行ずつ読み取ることができるため、プログラムの進行状況を示すためにエコーされます。 それ以外の場合、出力は前の例と同様になります。

code: bash

$ python asyncio_subprocess_coroutine.py

in run_df

launching process

process started 40792

read b'Filesystem Size Used Avail Capacity iused ifree %iused Mounted on\n'

read b'/dev/disk1s5s1 466Gi 14Gi 199Gi 7% 567557 4881885323 0% /\n'

read b'/dev/disk1s4 466Gi 5.0Gi 199Gi 3% 6 4882452874 0% /System/Volumes/VM\n'

read b'

no more output from command

waiting for process to complete

return code 0

parsing results

Free space:

/ : 199Gi

/System/Volumes/VM : 199Gi

/System/Volumes/Preboot : 199Gi

/System/Volumes/Update : 199Gi

/System/Volumes/Data : 199Gi

サブプロセスへのデータ送信

前の例は両方とも、2番目のプロセスからデータを読み取るために単一の通信チャネルのみを使用していました。 多くの場合、処理のためにデータをコマンドに送信する必要があります。 この例では、入力ストリーム内の文字を変換するためのUnixコマンドtrを実行するコルーチンを定義します。 この場合、trは小文字を大文字に変換するために使用されます。

code: asyncio_subprocess_coroutine_write.py

import asyncio

import asyncio.subprocess

async def to_upper(input):

print('in to_upper')

create = asyncio.create_subprocess_exec(

stdout=asyncio.subprocess.PIPE,

stdin=asyncio.subprocess.PIPE,

)

print('launching process')

proc = await create

print('pid {}'.format(proc.pid))

print('communicating with process')

stdout, stderr = await proc.communicate(input.encode())

print('waiting for process to complete')

await proc.wait()

return_code = proc.returncode

print('return code {}'.format(return_code))

if not return_code:

results = bytes(stdout).decode()

else:

results = ''

return (return_code, results)

MESSAGE = """

This message will be converted

to all caps.

"""

event_loop = asyncio.get_event_loop()

try:

return_code, results = event_loop.run_until_complete(

to_upper(MESSAGE)

)

finally:

event_loop.close()

if return_code:

print('error exit {}'.format(return_code))

else:

print('Original: {!r}'.format(MESSAGE))

print('Changed : {!r}'.format(results))

次に、to_upper（）は、プロセスのcommunicate（）メソッドを使用して、入力文字列をコマンドに送信し、結果のすべての出力を非同期で読み取ります。 同じメソッドのsubprocess.Popenバージョンと同様に、communicate（）は完全な出力バイト文字列を返します。 コマンドがメモリに快適に収まるよりも多くのデータを生成する可能性がある場合、入力を一度に生成できないか、出力を段階的に処理する必要があります。

communication（）を呼び出す代わりに、プロセスのstdin、stdout、およびstderrハンドルを直接使用することができます。

I / Oが完了したら、プロセスが完全に終了するのを待つことで、プロセスが適切にクリーンアップされます。

プログラムの主要部分は、変換されるメッセージ文字列を確立し、to_upper（）を実行するようにイベントループを設定して、結果を出力します。

出力には、操作のシーケンスと、単純なテキストメッセージがどのように変換されるかが表示されます。

code: bash

$ python asyncio_subprocess_coroutine_write.py

in to_upper

launching process

pid 41849

communicating with process

waiting for process to complete

return code 0

Original: '\nThis message will be converted\nto all caps.\n'

Changed : '\nTHIS MESSAGE WILL BE CONVERTED\nTO ALL CAPS.\n'

Unixシグナルの受信

Unixシステムのイベント通知は通常、アプリケーションを中断し、ハンドラーをトリガーします。 asyncioとともに使用すると、シグナルハンドラーのコールバックは、イベントループによって管理される他のコルーチンおよびコールバックとインターリーブされます。 これにより、中断される機能が少なくなり、その結果、不完全な操作をクリーンアップするためのセーフガードを提供する必要があります。

シグナルハンドラーは、コルーチンではなく、通常の呼び出し可能である必要があります。

code: asyncio_signals.py

import asyncio

import functools

import os

import signal

def signal_handler(name):

print('signal_handler({!r})'.format(name))

event_loop = asyncio.get_event_loop()

event_loop.add_signal_handler(

signal.SIGHUP,

functools.partial(signal_handler, name='SIGHUP'),

)

event_loop.add_signal_handler(

signal.SIGUSR1,

functools.partial(signal_handler, name='SIGUSR1'),

)

event_loop.add_signal_handler(

signal.SIGINT,

functools.partial(signal_handler, name='SIGINT'),

)

async def send_signals():

pid = os.getpid()

print('starting send_signals for {}'.format(pid))

print(f'sending {name}')

os.kill(pid, getattr(signal, name))

# シグナルがプログラムフローを中断しないため、

# シグナルハンドラを実行できるようにするための歩留まり調整

print('yielding control')

await asyncio.sleep(0.01)

return

try:

event_loop.run_until_complete(send_signals())

finally:

event_loop.close()

code: bash

% python asyncio_signals.py

starting send_signals for 42591

sending SIGHUP

yielding control

signal_handler('SIGHUP')

sending SIGHUP

yielding control

signal_handler('SIGHUP')

sending SIGUSR1

yielding control

signal_handler('SIGUSR1')

sending SIGINT

yielding control

signal_handler('SIGINT')

asyncioを使用したデバッグ

asyncioにはいくつかの便利なデバッグ機能が組み込まれています。

code: asyncio_debug.py

import sys

import time

import argparse

import asyncio

import logging

import warnings

parser = argparse.ArgumentParser('debugging asyncio')

parser.add_argument(

'-v',

dest='verbose',

default=False,

action='store_true',

)

args = parser.parse_args()

logging.basicConfig(

level=logging.DEBUG,

format='%(levelname)7s: %(message)s',

stream=sys.stderr,

)

LOG = logging.getLogger('')

async def inner():

LOG.info('inner starting')

# Use a blocking sleep to simulate

# doing work inside the function.

time.sleep(0.1)

LOG.info('inner completed')

async def outer(loop):

LOG.info('outer starting')

await asyncio.ensure_future(loop.create_task(inner()))

LOG.info('outer completed')

event_loop = asyncio.get_event_loop()

if args.verbose:

LOG.info('enabling debugging')

# デバッグを有効にする

event_loop.set_debug(True)

# Make the threshold for "slow" tasks very very small for

# illustration. The default is 0.1, or 100 milliseconds.

# 説明のために、「遅い」タスクのしきい値を非常に小さくする

# デフォルトは0.1、つまり100ミリ秒

event_loop.slow_callback_duration = 0.001

# 非同期リソースの管理に関するすべての間違いを報告する

warnings.simplefilter('always', ResourceWarning)

LOG.info('entering event loop')

event_loop.run_until_complete(outer(event_loop))

デバッグを有効にせずに実行すると、このアプリケーションではすべてが正常に表示されます。

code: bash

$ python asyncio_debug.py

DEBUG: Using selector: KqueueSelector

INFO: entering event loop

INFO: outer starting

INFO: inner starting

INFO: inner completed

INFO: outer completed

code: bash

% python asyncio_debug.py -v

DEBUG: Using selector: KqueueSelector

INFO: enabling debugging

INFO: entering event loop

INFO: outer starting

INFO: inner starting

INFO: inner completed

WARNING: Executing <Task finished coro=<inner() done, defined at asyncio_debug.py:24> result=None created at asyncio_debug.py:33> took 0.102 seconds

INFO: outer completed

参考