🏃 ホストの入出力¶

ホストはネットワークを走らせるプログラムで、mujou.run が返す Run を通して実行とやり取りする。入力を書き、出力を読み、待つ。このページではそれぞれと、Port の初期値を説明する。

書く¶

ホストが書くのは、実行の Port のうち、受け取る型を持ちつながっていないもの。ほかの Port にはノードが書く。それ以外の Port に書くと ValueError になる。

t = run.write({net.prev: x, net.teacher: y})  # 書き込みの時刻を返す

書いた値は次に書くまで残る。入力は時間とともに変わる信号で、ノードは同じ入力を何度も読む
1 回の write の値は一緒に切り替わる。ノードが片方の新しい値ともう片方の古い値を読むことはない。入力とその目標のように組になる値は 1 回の write で書く（か、1 つの DictSpec のメッセージにする）
値はコピーされる。書いた後でテンソルを書き換えてもネットワークには届かない
shape や dtype の違う値は ValueError、種別の違う値（IntSpec にテンソルなど）やキーの足りない dict、余分なキーのある dict は TypeError、IntSpec の範囲外の int は OverflowError になる。どれも Port を名指しする

初期値¶

方法	読む側が受け取るもの	その時刻
何もしない	ゼロ（スカラーは `0`、`False` など）	`None`
`Port(send=..., initial=value)`	Port が最初に送るまで `value`	`None`
`mujou.run(net, inputs={port: value})`	最初の更新から `value`	実行が始まった時刻
`run.write({port: value})`	書き込みの後は `value`	`write` が返した時刻

initial はノードが送る Port のためのもので、inputs と write はホストが書く Port のためのもの。mujou.run が戻るとノードはすぐに更新を始めるので、その直後に write しても、それより前の何回かの更新ではゼロを読むことがある。最初の更新から入力を見せるなら inputs で渡す。

入力をしばらく見せる¶

実行のタイマーで run.timer.sleep(seconds) で待つ。こちらは実行の時間の速さと一時停止に従う。

for x in stream:
    run.write({net.prev: x})
    run.timer.sleep(0.1)  # x を実行の時間で 0.1 秒見せる
    y = run.read(net.next)  # 見せ終わる時点の出力

出力は入力が変わってから遅れて追いつく。遅れはネットワークのダイナミクス（ノードの時定数や段数）で決まるので、十分な見せる時間を選ぶ。write の直後に読むと前の入力に対する出力が返る。期待する出力が分かっているなら、代わりに wait_for で待つ。

読む¶

方法	返すもの	向いている用途
`run.read(port)`	最新の値	好きなときに読む
`run.read_with_time(port)`	最新の値と送られた時刻の組 `(value, Timestamp \\| None)`	値がどれだけ新しいかを知る
`run.subscribe(port)`	前より新しい `(value, Timestamp \\| None)` を順に返すイテレータ	新しい出力ごとに処理する
`run.wait_for(port, until)`	`until` を満たす最初の `(value, Timestamp \\| None)`	期待した出力が届くまで待つ

どれも実行の送る Port を、つながっていてもいなくても読める。つながっている Port を読めば、ネットワークの途中の値を変えずに観測できる。返すのは、その Port が最後に send したものの CPU 上の新しいコピー（それまでは初期値で、時刻は None）。

subscribe は最新の値をすぐに返し、その後は新しい値を 1 つずつ待つ。ホストが読むのが遅ければ間の値は飛ばされる。実行が正常に終了すると反復が終わる。break でループを抜けるか、timeout= を渡す。

clock = Clock()  # a node runs by itself, without a network
with mujou.run(clock) as run:
    received: list[tuple[float, float]] = []
    # timeout: give up after 5 s of real time instead of waiting forever.
    for value, t in run.subscribe(clock.next, timeout=5.0):
        if t is None:  # the zeros the port holds until the clock first sends
            continue
        received.append((value.item(), float(t)))
        if len(received) == 20:
            break

wait_for は最新の値から順に新しい値ごとに条件を確かめ、満たした最初の値を返す。先に実行が正常に終了すると RuntimeError を送出する。

y, t = run.wait_for(net.next, lambda y: torch.equal(y, expected), timeout=5.0)

timeout は実時間の秒で、既定は無期限。時間切れになると TimeoutError を送出し、実行は走り続ける（例外と終了）。

終了させる¶

方法	すること
`with mujou.run(...) as run:`	ブロックを抜けるときに `shutdown()` を呼ぶ。通常はこれを使う
`run.shutdown(timeout=...)`	全ノードに終了を求め、全ノードの `on_stop` が戻るまで待つ
`run.join(timeout=...)`	終了を求めずに終了まで待つ。別のスレッドが `shutdown` を呼んだ後や、ノードの例外を待つときに使う

ホストが Run を捨てても、shutdown かインタプリタの終了まで実行は走り続ける。実行は自分で終了させる。正常に終了した後の shutdown は何もせず、write は RuntimeError になり、run.state_dict() は最後の状態を取得できる。ノードが例外を投げると、以後の呼び出しは NetworkError を送出する（例外と終了）。