trio-utilによるreadableな並行処理①（AsyncValue）

目次

trio-utilの紹介

これまでpythonの並行処理ライブラリtrioについて紹介してきました。
trio-utilはtrioについて便利な機能を提供するライブラリです（ソースコードはこれ）。
READMEには次のような事が書かれていますが、まさにその特性を持ったライブラリだと思います。

What attributes make a good utility function or class?
(良いutil関数やクラスとはどのような特性なのか？)

- of general use, intuitive, hard to use incorrectly
  (一般的で、直感的で、使い方を間違えにくい)
- makes code more readable and reduces cognitive load
  (コードの可読性を高め、認識負荷を軽減する)
- already vetted for a length of time within a project, ideally used by multiple developers
  (既に何かのプロジェクト内で長期間検証されていて、複数の開発者によって使用された)

導入方法

$ pip install trio-util

AsyncValue

AsyncValueは任意の型をラップして特定の値や遷移を待つ機能を提供するクラスです（以下では簡単のためにオブジェクトの事を値と呼びます）。
コンストラクタには初期値を渡します。

メンバ	内容
value	値のset/getに使う。
await wait_value(値 or 関数)	引数が値の場合、valueが引数の値になるまで待つ。引数が関数の場合、その関数がTrueを返すまで待つ（呼び出し時やvalueが変化する度にvalueが引数の関数に渡される）。
await wait_transition(空 or 値 or 関数)	引数が空の場合、valueが変化するまで待つ。引数が値の場合、その値に遷移するまで待つ。引数が関数の場合、その関数がTrueを返すまで待つ（valueが変化するたびに引数の関数に現value&旧valueが渡される）。
async for val in eventual_value(空 or 値 or 関数)	条件を満たす度にループを回す（条件確認は呼び出し時とvalueが変化する度に行われる）。引数が空の場合、常に条件を満たす。引数が値の場合、valueがその値であれば条件を満たす。引数が関数の場合、その関数がTrueを返せば条件を満たす（条件確認時に引数の関数にvalueが渡される）。
async for val, old in transitions(空 or 値 or 関数)	eventual_value()と同様に条件を満たす度にループを回す。ただし、条件確認はvalueが変化した時のみであり、呼び出し時には行われない。

wait_value(値)の例

下記サンプルは①〜⑤の順に処理されます。

import trio
import trio_util

async def update_event(event: trio_util.AsyncValue):
    await trio.sleep(1)  # ④ 1秒待機
    event.value = 7  # ⑤ eventに7がsetされて③を抜ける


async def main():
    async with trio.open_nursery() as nursery:
        event = trio_util.AsyncValue(3)  # ① インスタンス化（初期値として3をset）
        nursery.start_soon(update_event, event)  # ② 子タスクupdate_event開始

        await event.wait_value(7)  # ③ eventに7がsetされるまで待機

trio.run(main)

wait_value(関数)の例

下記サンプルは①〜⑥の順に処理されます。

import trio
import trio_util

async def update_event(event: trio_util.AsyncValue):
    await trio.sleep(1)  # ④ 1秒待機
    event.value = "Hello World"  # ⑤ eventに11文字がsetされて③を抜ける

async def main():
    async with trio.open_nursery() as nursery:
        event = trio_util.AsyncValue("Hello")  # ① インスタンス化（初期値として"Hello"をset）
        nursery.start_soon(update_event, event)  # ② 子タスクupdate_event開始

        s = await event.wait_value(lambda v: len(v) == 11)  # ③ eventに11文字がsetされるまで待機
        print(s)  # ⑥ eventの中身を出力（event.valueも同じ値）

trio.run(main)

実行結果

Hello World

wait_transition(関数)の例

下記サンプルは①〜⑥の順に処理されます。

import trio
import trio_util

async def update_event(event: trio_util.AsyncValue):
    await trio.sleep(1)  # ④ 1秒待機
    event.value = "Hello World"  # ⑤ eventに11文字がsetされて③を抜ける

async def main():
    async with trio.open_nursery() as nursery:
        event = trio_util.AsyncValue("Hello")  # ① インスタンス化（初期値として"Hello"をset）
        nursery.start_soon(update_event, event)  # ② 子タスクupdate_event開始

        val, old = await event.wait_transition(lambda v, _: len(v) == 11)  # ③ eventの変化後に11文字になるまで待機
        print(old)  # ⑥ eventの中身を出力
        print(val)

trio.run(main)

実行結果

Hello
Hello World

eventual_values(空)の例

下記サンプルは①〜⑦の順に処理されます。

import trio
import trio_util

async def update_event(event: trio_util.AsyncValue):
    await trio.sleep(1)  # ④ 1秒待機
    event.value = "Hello World"  # ⑤ 値がsetされたため③のループが回る
    await trio.sleep(1)  # ⑥ 1秒待機
    event.value = "Hello Universe"  # ⑦ 値がsetされたため③のループが回る

async def main():
    async with trio.open_nursery() as nursery:
        event = trio_util.AsyncValue("Hello")  # ① インスタンス化（初期値として"Hello"をset）
        nursery.start_soon(update_event, event)  # ② 子タスクupdate_event開始

        async for v in event.eventual_values():  # ③ 呼び出し時やeventが変化する度にループを回す
            print(v)

trio.run(main)

実行結果

Hello
Hello World
Hello Universe

eventual_values(値)の例

eventual_values(空)の例でeventual_values()の引数を"Hello World"とすると下記結果となります。

実行結果

Hello World

eventual_values(関数)の例

eventual_values(空)の例でeventual_values()の引数を

lambda v: len(v) > 5

とすると下記の結果となります。

実行結果

Hello World
Hello Universe

transitions(関数)の例

import trio
import trio_util

async def update_event(event: trio_util.AsyncValue):
    await trio.sleep(1)  # ④ 1秒待機
    event.value = "Hello World"  # ⑤ 値がsetされて③の条件を満たすためループが回る
    await trio.sleep(1)  # ⑥ 1秒待機
    event.value = "Hello Universe"  # ⑦ 値がsetされたが③の条件を満たさないためループが回らない

async def main():
    async with trio.open_nursery() as nursery:
        event = trio_util.AsyncValue("Hello")  # ① インスタンス化（初期値として"Hello"をset）
        nursery.start_soon(update_event, event)  # ② 子タスクupdate_event開始

        async for v in event.transitions(lambda v, _: v[-1] == "d"):  # ③ eventが変化して引数がTrueを返す度にループを回す
            print(v)

trio.run(main)

('Hello World', 'Hello')

wait_value()とwait_transition()の違い

wait_value()は最初から条件を満たしていれば待機しません。
wait_transition()は最初から条件を満たしていても待機し、一度別の値になってから再度条件を満たすと抜けます。

event = trio_util.AsyncValue(7)

await event.wait_value(7)  # 最初から条件を満たすため待機しない

await event.wait_transition(7)  # 最初から条件を満たすが、一度別の値に変わってから7になるまで待機（あくまでも7への遷移を待つ）