非同期入出力library Trioの調べ事

Trioを使おうとした所、日本語情報があまり無かったので調べました。

非同期codeのへの入り口

一秒後に'3'が出力される

import trio

async def add(a, b):
    await trio.sleep(1)
    return a + b

result = trio.run(add, 1, 2)
print(result)

trio.run()がtrioにおいて同期codeから最初の**async関数(非同期code)**を呼ぶ手段のようです。

複数の仕事を同時にこなす

非同期codeでは複数の事を同時にやらないと意味がありません。複数の仕事があって初めて例えば、ある非同期codeがネットワークからの受信を待っている間に別の非同期codeを進めるという風に効率よくCPUを使えるはずなので。trioでそれを担うのはnursery(直訳:保育所、託児所)です。

import trio

async def delayed_print(i):
    await trio.sleep(1)
    print(i)

async def main():
    async with trio.open_nursery() as nursery:
        for i in range(10):
            nursery.start_soon(delayed_print, i)
    print('全ての仕事が終わりました')

trio.run(main)

出力結果(数字の順番は毎回変わる)

2
0
1
4
8
3
7
6
5
9
全ての仕事が終わりました

delayed_print()は一秒休んでから数字を出力する関数で、これを10回呼んでいるにも関わらずかかった時間は10秒ではなく約一秒なので、非同期処理は上手くいったようです。

中断と時間制限

ネットワークからの受信を永遠に待ち続けるわけにはいきません。何らかの制限時間を設けるのが普通だと思います。trioにおいてそのような事をしたい時は以下のようにするようです。

import trio

async def main():
    with trio.move_on_after(1) as cancel_scope:
        await trio.sleep(.7)
        await trio.sleep(.7)
    if cancel_scope.cancelled_caught:
        print('眠りが妨げられました')  # <= こっちが出力される
    else:
        print('ぐっすり眠れました')

trio.run(main)

with trio.move_on_after(1)によってwith block内のcode全体に対して1秒の制限時間が設けられます。上のcodeでは個々のsleep()が一秒以内に完了可能でも合計が1秒を超えるため二度目のsleep()の最中に中断され、眠りが妨げられましたと出力されます。ここでcancelled_caughtという属性名に違和感を持ちましたが、どうやらtrioは内部でCancelledという例外を使って中断を実現しているようなので、その例外を捕まえたんだという風に考えると覚えやすそうです。因みにこの例外をこちらが捕らえることは推奨されていません。

move_on_after()の兄弟たち

もし時間切れになった時に何らかの例外で教えて欲しいのならtrio.fail_after()が使うようです。

import trio

async def main():
    try:
        with trio.fail_after(1):
            await trio.sleep(10)
        print('ぐっすり眠れました')
    except trio.TooSlowError:
        print('眠りが妨げられました')  # <= こっちが出力される

trio.run(main)

中断要求

cancel_scope.cancel()を呼ぶことで制限時間とは関係無く中断を要求できます。

import trio


async def main():
    with trio.move_on_after(1) as cancel_scope:
        await trio.sleep(.5) # A
        cancel_scope.cancel()
        print('これは出力される')
        await trio.sleep(.5)  # A
        print('これは出力されない')
    if cancel_scope.cancelled_caught:
        print('眠りが妨げられました')
    else:
        print('ぐっすり眠れました')

trio.run(main)

これは出力される
眠りが妨げられました

あくまで中断を要求するのであって、即座に中断するわけではない点がややこしいです。(時間制限による場合も含めてですが)中断は中断可能地点に入らないと起きません。そして中断可能地点は基本的にtrioのasync関数になります。(上の例ではA行です)。なので次のcodeだと

import trio

async def main():
    with trio.move_on_after(1) as cancel_scope:
        await trio.sleep(.5)
        cancel_scope.cancel()

    if cancel_scope.cancelled_caught:
        print('眠りが妨げられました')
    else:
        print('ぐっすり眠れました')  # <= こっちが出力される

trio.run(main)

中断は要求しているものの、その後中断可能地点には入ることなくwith blockを抜けているので中断は起きず、結果ぐっすり眠れましたと出力されます。勿論以下のcodeでも同じです。

import trio
import time

async def main():
    with trio.move_on_after(1) as cancel_scope:
        await trio.sleep(.7)
        time.sleep(.7)  # 同期版のsleep!!
    if cancel_scope.cancelled_caught:
        print('眠りが妨げられました')
    else:
        print('ぐっすり眠れました')  # <= こっちが出力される

trio.run(main)

一応cancel_scopeにはcancel_calledという属性もあり、cancelled_caughtと合わせて参照する事でwith blockをどのように抜けたのか分かるのかもしれません。

import trio

def print_state(cancel_scope):
    print(cancel_scope.cancel_called)
    print(cancel_scope.cancelled_caught)

async def main():
    print('時間切れの場合')
    with trio.move_on_after(1) as cancel_scope:
        await trio.sleep(.7)
        await trio.sleep(.7)
    print_state(cancel_scope)

    print('中断要求により中断された場合')
    with trio.move_on_after(1) as cancel_scope:
        await trio.sleep(.7)
        cancel_scope.cancel()
        await trio.sleep(0)
    print_state(cancel_scope)

    print('中断を要求したものの、実際にされなかった場合')
    with trio.move_on_after(1) as cancel_scope:
        await trio.sleep(.7)
        cancel_scope.cancel()
    print_state(cancel_scope)

    print('時間内に処理を終えた場合')
    with trio.move_on_after(1) as cancel_scope:
        await trio.sleep(.7)
    print_state(cancel_scope)

trio.run(main)

出力結果

時間切れの場合
True
True
中断要求により中断された場合
True
True
中断を要求したものの、実際にされなかった場合
True
False
時間内に処理を終えた場合
False
False

nurseryへ加える子に時間制限をかける際の注意点

以下のcodeは

import trio

async def main():
    start_time = trio.current_time()
    async with trio.open_nursery() as nursery:
        with trio.move_on_after(1):  # A
            nursery.start_soon(trio.sleep, 2)
    print(trio.current_time() - start_time, '秒経ちました')

trio.run(main)

出力結果

2.0023307930096053 秒経ちました

となり、A行でかけた時間制限が機能しません。そうさせる為には

import trio

async def main():
    start_time = trio.current_time()
    with trio.move_on_after(1):
        async with trio.open_nursery() as nursery:
            nursery.start_soon(trio.sleep, 2)
    print(trio.current_time() - start_time, '秒経ちました')

trio.run(main)

出力結果

1.0017503799754195 秒経ちました

という風にcancel_scopeをnurseryの外側に置く必要が有ります。

asyncioとの比較

trioにしかできないと思う事

上に出てきたようにtrioにおいて複数の仕事を同時にこなしたい時はnurseryを使います。asyncioで同等の物はasyncio.gather()だと思うのですが、この二つには決定的な違いがあるように思えます。それはasyncio.gather()にはあらかじめ用意しておいた仕事を一括で渡さないといけないのに対し、nurseryには後からいくらでも仕事を加えられます。

asyncioにしかできないと思う事

asyncio.gather()は渡された各仕事の戻り値をlistに纏めて返してくれますが、trioのnurseryにはそのような機能は見当たりません。