🚨 例外と終了¶

ノードの hook が例外を投げたときに何が起き、ホストがそれをどこで受け取るか、ホストの呼び出しのタイムアウトを説明する。

1 つのノードの例外で全体が止まる¶

どのノードのどの hook（on_update・on_start・state_dict などどれでも）が例外を投げても、実行全体が止まる。

mujou がノードのパス・hook・traceback を logger "mujou" に ERROR レベルで出す
各ノードは実行中の hook を終えて止まり、ほかのノードを待たずに自分の on_exception() を呼ぶ。例外を投げていないノードでも呼ばれる
全ノードの on_exception() が戻ったら、全ノードの on_stop() を呼ぶ。例外を投げたノードでも呼ばれる
ホストは NetworkError（RuntimeError の派生）を受け取る。__cause__ は最初の例外。ほぼ同時に投げられたほかのノードの例外はログに出るだけ

例外を投げたノードだけを止めることはしない。そうすると、隣のノードがその古い値を読み続け、気づかないうちに平衡から外れてしまう。

"""What happens when a node raises.

The whole run stops: every node's ``on_exception``, then its ``on_stop``, is
called, and the host gets a ``NetworkError`` whose ``__cause__`` is the node's
exception. An exception in ``on_start`` comes from ``mujou.run`` itself.
"""

import logging
import sys
from typing import override

import torch

import mujou
from mujou import NetworkError, Node, Port, TensorSpec

SCALAR = TensorSpec(())

# mujou also logs the node's exception, with its traceback, to the "mujou" logger.
# Show the log on stdout here (without a handler, Python prints it to stderr).
logging.basicConfig(stream=sys.stdout, format="log: %(message)s")


class Checked(Node):
    """Passes its input on, and raises once it is negative."""

    def __init__(self, fail_on_start: bool = False) -> None:
        super().__init__()
        self.fail_on_start = fail_on_start
        self.prev = Port(recv=SCALAR)
        self.next = Port(send=SCALAR)

    @override
    def on_start(self) -> None:
        super().on_start()
        if self.fail_on_start:
            raise RuntimeError("cannot start")

    @override
    def on_update(self, dt: float) -> None:
        super().on_update(dt)
        u = self.prev.recv()
        if u < 0:
            raise ValueError(f"negative input: {u.item()}")
        self.next.send(u)

    @override
    def on_exception(self) -> None:
        super().on_exception()
        print("on_exception: called on every node")

    @override
    def on_stop(self) -> None:
        super().on_stop()
        print("on_stop: called after on_exception")


class Single(mujou.Network):
    def __init__(self, fail_on_start: bool = False) -> None:
        super().__init__()
        self.checked = Checked(fail_on_start)
        self.prev = self.checked.prev


# An exception while the run goes on reaches the host at its next call.
net = Single()
run = mujou.run(net)
run.write({net.prev: torch.tensor(-1.0)})
try:
    run.join(timeout=10.0)  # waits until the run stops (here: by the exception)
except NetworkError as error:
    cause = error.__cause__
    print(f"NetworkError, caused by {type(cause).__name__}: {cause}")
    assert isinstance(cause, ValueError)

# An exception in on_start comes from mujou.run.
try:
    mujou.run(Single(fail_on_start=True))
except NetworkError as error:
    print(f"mujou.run raised NetworkError, caused by {error.__cause__!r}")
    assert isinstance(error.__cause__, RuntimeError)

ホストが受け取る場所¶

例外が起きたとき	`NetworkError` を送出するもの
起動中（`on_start` など）	`mujou.run`。終了処理（`on_exception`・`on_stop`）を済ませてから送出する
起動の後	次の `write`・`read`・`read_with_time`・`subscribe`（とそのイテレータ）・`wait_for`・`state_dict`・`shutdown`・`join`、または `ManualRun` の `step`

起動後の例外はホストの次の呼び出しで届くので、待つだけのホストは気づかない。run.timer.sleep ではなく run.join(timeout) で待つ。with ブロックの終わりでは shutdown が送出する。ブロックの本体も例外を投げたときは本体の例外が伝わり、NetworkError は捨てられる（ノードの例外はログに残る）。

`on_exception` と `on_stop`¶

hook	呼ばれるとき	書くこと
`on_exception()`	例外による終了時。全ノードで先に	アクチュエータの安全停止など、異常終了に固有の処理
`on_stop()`	終了のたびに 1 回。正常終了でも例外でも	I/O などの資源の解放。状態は変えない

on_exception は引数を取らないので、ノードはどの例外で止まったかを知らない。原因に応じた対処はホストが __cause__ を見て行う
ノードの中で例外に対処するなら、その hook の中で try で捕まえる。捕まえた例外は実行を止めない
on_exception や on_stop が例外を投げてもログに出るだけで、終了処理は続く
これらの hook の中では self.timer が進まない（ノードが守る規則）

ログ¶

ノードの例外は traceback 付きで logger "mujou" に出る。ログを設定していないプログラムでは標準エラー出力に出る。出力先は Python の logging で変える。

# mujou also logs the node's exception, with its traceback, to the "mujou" logger.
# Show the log on stdout here (without a handler, Python prints it to stderr).
logging.basicConfig(stream=sys.stdout, format="log: %(message)s")

例外の後の状態¶

例外で終了した実行の状態は残らず、run.state_dict() も NetworkError を送出する。残したい状態は実行中に、長い実行では定期的に取得する。もう一度走らせるときは新しいインスタンスを作る。

タイムアウト¶

shutdown・join・subscribe・wait_for・state_dict は実時間の秒で timeout を取る。既定の math.inf は必要なだけ待つ。

呼び出し	時間切れのとき
`run.shutdown(timeout=...)`	`TimeoutError`。終了の要求は取り消されず、ノードはもう更新しない。終了処理は続くので `join` で待つ
`run.join(timeout=...)`	`TimeoutError`。何も変わらない
`run.subscribe(port, timeout=...)`	イテレータが `TimeoutError` を送出する。時間はループの本体を含む反復全体で数える。実行は走り続ける
`run.wait_for(port, condition, timeout=...)`	`TimeoutError`。実行は走り続ける
`run.state_dict(timeout=...)`	止まらないノードがあると `TimeoutError`（各ノードは実行中の `on_update` を終えてから止まる）。状態は取得せず、実行は走り続ける

1 回の on_update が長いノード（重い BP など）は、状態の取得のための一時停止をいつも遅らせる。

インタプリタの終了¶

実行中にインタプリタが終了すると、ほかの daemon でないスレッドが終わった後に、mujou が実行を終了させて全ノードの on_stop を待つ。subscribe を回しているスレッドは実行が終わるまで終了を待たせるので、実行は shutdown か with ブロックで自分で終了させる。