trioによる並行処理①（非同期開始、open_nursery、await、atomic）

目次

trioとは

• 並行処理（複数処理の実行）ができるpythonライブラリ
  • 並列処理ではないので処理速度が上がるわけではない
  • CPU並列したいならtrio-parallelという似たやつもある（あるいは他の言語を使った方が良いかも）
• 似たライブラリにthreadingがあるが、そちらはOSのスレッドを使うため協調的なタスクが難しいと思われる（メモリ共有やrace conditionが問題になりやすい）
• 似たライブラリにasyncioがあり、恐らくそちらの方がスタンダードだが、trioは構造化にこだわっていて人間工学的に有利との噂
  • いわゆるgoto文が許された言語とそうでない言語の違いくらい有利らしい。下記が参考になりそう
    • Nathaniel J. Smith: Notes on structured concurrency, or: Go statement considered harmful
      • gotta_dive_into_python氏による訳
    • John Belmonte: “Concurrency made easy”: coming soon to a programming language near you
  • 私はasyncioの経験がないから実感ないけど、並行処理の割に可読性の高いコードを書きやすくて完成度の高いライブラリだと感じています
  • (追記)asyncioにTaskGroupというコンテキストマネージャーが追加されたらしいです。trioと同じような書き方ができるようなので、ここら辺の有利面は薄れたのかな（今までの書き方もできるから上記事で言われてるgotoが許されてるという状況は変わって無さそうだけど）

導入方法

$ pip install trio

並行処理の基本

サンプル

import trio

async def func1():
    print("func1 start")
    await trio.sleep(2)
    print("func1 done")

async def func2(a, b):
    print(f"func2 start (a={a}, b={b})")
    await trio.sleep(1)
    print("func2 done")

async def main():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(func1)
        nursery.start_soon(func2, 2, 3)  # このように値を渡せる
        print("end of nursery")
    print("all done")

trio.run(main)

実行結果例（func2 startとfunc1 startは入れ替わる可能性があります）

end of nursery
func2 start (a=2, b=3)
func1 start
func2 done
func1 done
all done

• trio.run()で非同期処理を開始
• 非同期処理の関数定義時にasyncを付ける
• async with trio.open_nursery() as nursery:のようにwithブロックを作ることができる（trioの特徴）
  • nursery.start_soon()で非同期な子タスクを実行できる
  • このwithブロックは子タスクが全て終わるまで基本的に抜けない
• trio.sleep()は非同期関数であり、指定した秒数だけ待機する
• 非同期関数はnursery.start_soon(非同期関数)で呼び出す場合と、await 非同期関数()で呼び出す場合がある
  • 子タスクとしてスレッドを分けて呼びたい場合はnursery.start_soon(非同期関数)
    • 非同期関数の開始を待たずに、その下の処理が実行される
  • スレッドを分けない場合はawait 非同期関数()
    • 非同期関数が終わるまで、その下の処理が実行されない
  • 例えば↑のサンプルでfunc2をnursery.start_soon(func2, 2, 3)ではなくawait func2(2, 3)で呼んだ場合、下記のようにfunc2が終わるまでend of nurseryとならない

    func2 start (a=2, b=3)
    func1 start
    func2 done
    end of nursery
    func1 done
    all done
    

重要だと思われる基本的なこと（awaitとatomic）

• awaitで呼び出された非同期関数は、他の処理を優先して待機する（awaitって言ってるくらいだし）
  • 例えばawait trio.sleep(1)はただちに他の処理を優先して1秒待機
• awaitしてない処理は他の処理を優先しない（いわゆるatomicな操作）

よって

import trio

async def func1():
    print("func1 start")
    await trio.sleep(0)
    print("func1 done")

async def func2():
    print("func2 start")
    await trio.sleep(0)
    print("func2 calc total")
    total = 0
    for i in range(1000):
        total += i
    print(f"func2 done (total={total})")

async def main():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(func1)
        nursery.start_soon(func2)
        print("end of nursery")
    print("all done")

trio.run(main)

実行結果例（func2 startとfunc1 startは入れ替わる可能性があります）

end of nursery
func2 start
func1 start
func1 done
func2 calc total
func2 done (total=499500)
all done

重要なのは

• func1とfunc2のどちらが先に開始されるかは運次第だが、どちらであってもawait trio.sleep(0)で一旦待機してもう片方の開始を待つ（awaitって付いてるから）
• func2 calc totalからfunc2 done (total=499500)までの間はawaitの付いた呼び出しが無いから、この間は他の処理を優先することなく連続で実行される（よってこの間にfunc1 doneがprintされる事はない）

with trio.open_nursery()によるブロックの中断方法

with trio.open_nursery()のブロックは子タスクが終わるまで基本的に抜けないと上述しましたが、nursery.cancel_scope.cancel()を使えば中断できます。

サンプル

import trio

async def func():
    print("func start")
    while True:
        await trio.sleep(1)
    print("func done")

async def main():
    async with trio.open_nursery() as nursery:
        nursery.start_soon(func)
        await trio.sleep(0.5)
        nursery.cancel_scope.cancel()
        print("end of nursery")
    print("all done")

trio.run(main)

実行結果

func start
end of nursery
all done

このように子タスクがwhile Trueの無限ループで終了しないとしても、nursery.cancel_scope.cancel()で終了できていることが分かります。

参考記事

• Python + Trio の非同期コーディングパターン
• 非同期入出力library Trioの調べ事