Pony 試乗記(4) アクターでたらい回し

# はじめに

たらい回し関数[Wikipedia](https://ja.wikipedia.org/wiki/%E7%AB%B9%E5%86%85%E9%96%A2%E6%95%B0)は、こんな感じの内容です。

~~~pony
fun tarai(x: I64, y: I64, z: I64): I64 =>
  if x <= y then
    y
  else
    tarai(tarai(x - 1, y, z),
          tarai(y - 1, z, x),
          tarai(z - 1, x, y))
  end
~~~

これをアクターを使って高速化できないか、と考えました。

# 非同期たらい回し

単純に上記の tarai を、引数 x, y, z を一度に渡してもらう actor にしてしまうとただ単にメッセージのキューイングやスレッド起床などのコストがかさむだけです。

しかし、tarai は x <= y の場合には、z がなくても戻り値はy と確定できます。
なので、z の値の取得を後回しにできれば計算量を削減できる可能性があります。
アクターを使ってメッセージパッシングで z の値の通知を遅らせることを考えました。

まず、tarai の結果を受信する、あるいは tarai を実装するインターフェース DFrameを用意しました。

~~~pony
interface DFrame
  be unify(index: U8, value: I64)
~~~

DFrame の名前の由来は、並列 Prolog の一種である EPILOG における dframe （ダイナミックフレーム/離散フレーム）です。

tarai の引数は DFrame 上の変数として格納し、その変数の外部からの設定操作を unify で個別に行います。このため、x, y の値が設定されているが z の値は未設定、という状態もありえます。x, y の値のみで結果が確定できる場合には z の値の設定を待たずに結果を確定させる、といったことも可能です。

確定した結果は予め設定されたリスナーに通知します。リスナーは通常、親 DFrame なので、コンストラクタで与えることができます。

DFrame の木構造は計算木そのもので、木の根となる DFrame には tarai 全体の結果を表示する結果表示用アクターとします。

図にすると

1. まず、根となる DFrame である結果表示アクター (PrintResult) を作成します。
2. 結果表示アクターをリスナーとしてゴールとなる tarai を計算する DFrame を作ります。最初は引数の値は未設定なので Tarai(_, _, _) の状態です。ここで、矢印は結果値が通知される向きを表します。


![tarai_dframe0.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/412308/714ff53a-6c82-6d69-b1fa-bca70a19e5db.png)

3. Tarai(_, _, _) のそれぞれの引数に値を設定していきます。この図の例だと 7, 6, 0 に設定しています。

![tarai_dframe1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/412308/90bb3b80-7091-1b92-ff11-dbcc159ad0bc.png)

4. 値が設定されたことを契機として処理が開始します。x <= y でないので、サブゴールが起動されます。
5. サブゴール tarai(tarai(7-1, 6, 0), tarai(6-1, 0, 7), tarai(0-1,7,6)) を含む DFrame の木構造ができます。tarai(7-1, 6, 0) および tarai(0-1,7,6) は引数の値が設定されたときに値が確定できるので、根の方に向かって確定した値を通知します。


![tarai_dframe2.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/412308/50fa6aee-4850-e698-3adf-dc1dd31c50e1.png)

このような手順で木を成長させていくと、最終的に木全体の未確定な値が全て確定します。

実際には木が成長しきるのを待つ必要もなければ、全ての値の確定を待つ必要もなく、結果表示アクターに値が通知されれば全ての DFrame は破棄して構いません。なので、結果が確定すると自分のサブゴールに対して停止を通知するようにしました。

## 実装

### 結果表示アクター

~~~pony
actor PrintResult is DFrame
  let logger: Logger[String]
  new create(logger': Logger[String]) =>
    logger = logger'
  be unify(index: U8, value: I64) =>
    logger(Info) and logger.log("Result: " + value.string())
~~~

### たらい回しアクター

※このソースは改善後のものです。

~~~pony
actor Tarai is DFrame
  """
  Guarded Horn Clauses で書くと
  tarai(X, Y, Z, R) :- integer(X), integer(Y), X =< Y | R = Y.
  tarai(X, Y, Z, R) :- integer(X), integer(Y), X > Y
                       | X1 is X -1, Y1 is Y - 1, Z1 is Z - 1,
                         tarai(X1, Y, Z, X2),
                         tarai(Y1, Z, X, Y2),
                         tarai(Z1, X, Y, Z2),
                         tarai(X2, Y2, Z2, R).
  """
  let logger: Logger[String]
  // --- リスナーの情報
  let watchIndex: U8  // listener へ通知する変数のインデックス
  let listener: DFrame tag  //
  let listenerIndex: U8
  // --- フレーム変数
  var x: (I64 | None) = None  // インデックス[1] - 引数 x
  var y: (I64 | None) = None  // インデックス[2] - 引数 y
  var z: (I64 | None) = None  // インデックス[3] - 引数 z
  var r: (I64 | None) = None  // インデックス[4] - 結果の値
  // --- 不要処理の終了
  var stopped: Bool = false   // 処理終了
  var subgoals: Array[Tarai tag] = Array[Tarai tag]
  new create(watchIndex': U8, listener': DFrame tag, listenerIndex': U8,
      logger': Logger[String]) =>
    logger = logger'
    watchIndex = watchIndex'
    listener = listener'
    listenerIndex = listenerIndex'
  be unify(index:U8, value: I64) =>
    """
    引数の値、あるいは結果の値を受け取り設定する。
    """
    logger(Fine) and logger.log(string()
    + ": unify(" + index.string() + ", " + value.string() + ")")
    if stopped == true then
      logger(Fine) and logger.log(string() + " already stopped")
    else
      match index
      | 1 => x = value
      | 2 => y = value
      | 3 => z = value
      | 4 => _unify_r(value)
      end
      match r
      | None =>
        _action()
      end
    end
  fun ref _unify_r(r': I64): None =>
    """
    結果の値を設定する。サブゴール全てに停止を指示し、リスナーに結果を通知する。
    """
    r = r'
    if stopped == false then
      stop()
      logger(Fine) and logger.log(string() + ": notify result")
      listener.unify(listenerIndex, r')
    end
  fun ref _action(): None =>
    """
    x の値が確定しているか調べる。確定していれば y の値のチェックに進む。
    """
    match x
    | None =>
      logger(Fine) and logger.log(string() + ": waiting argument value x")
    | let x': I64 =>
      _action_x(x')
    end
  fun ref _action_x(x': I64): None =>
    """
    x の値が確定している前提で、y の値が確定しているか調べる。
    確定していれば x, y の値を元に結果を確定できるかの確認に進む。
    """
    match y
    | None =>
      logger(Fine) and logger.log(string() + ": waiting argument value y")
    | let y': I64 =>
      _action_x_y(x', y')
    end
  fun ref _action_x_y(x': I64, y': I64): None =>
    """
    x, y の値が確定している前提で、
    x <= y か確認する。x <= y なら、y を結果の値とする。
    そうでないならサブゴール起動に進む。
    """
    if x' <= y' then
      match r
      | None =>
        logger(Fine) and logger.log(string() + ": x =< y | r = y")
        _unify_r(y')
      end
    else
      match z
      | None =>
        logger(Fine) and logger.log(
          string() + ": x > y | waiting argument value z")
      | let z': I64 =>
        _action_x_y_z(x', y', z')
      end
    end
  fun ref _action_x_y_z(x': I64, y': I64, z': I64): None =>
    """
    サブゴール起動
    """
    logger(Fine) and logger.log(string() + ": x > y | calling subgoals")
    //let t0: Tarai = Tarai(4, listener, listenerIndex, logger)
    let t0: Tarai = Tarai(4, this, 4, logger)
    let t1: Tarai = Tarai(4, t0, 1, logger)
    t1.unify(1, x' -1)
    t1.unify(2, y')
    t1.unify(3, z')
    let t2: Tarai = Tarai(4, t0, 2, logger)
    t2.unify(1, y' -1)
    t2.unify(2, z')
    t2.unify(3, x')
    let t3: Tarai = Tarai(4, t0, 3, logger)
    t3.unify(1, z' -1)
    t3.unify(2, x')
    t3.unify(3, y')
    subgoals.push(t0)
    subgoals.push(t1)
    subgoals.push(t2)
    subgoals.push(t3)
  be stop() =>
    """
    今後の処理は不要であることをフラグに記録する。
    また、サブゴール起動を行っていたならばサブゴールに stop を通知する。
    """
    logger(Fine) and logger.log(string() + " stopping")
    stopped = true
    let subgoals': Array[Tarai tag] = subgoals = Array[Tarai tag]
    for subgoal in subgoals'.values() do
      subgoal.stop()
    end
  fun string(): String =>
    "Tarai(" + x.string() + "," + y.string() + "," + z.string() + ", "
    + r.string() + ")"
~~~

### メインアクター

~~~pony
actor Main
  new create(env: Env) =>
    let logger = StringLogger(Info, env.out, DateTimeLogFormatter)
    logger(Info) and logger.log("Start")
    try
      let tarai: Tarai = Tarai(4, PrintResult(logger), 1, logger)
      tarai.unify(1, env.args(1)?.i64()?)
      tarai.unify(2, env.args(2)?.i64()?)
      tarai.unify(3, env.args(3)?.i64()?)
    else
      try
        env.out.print("usage: " + env.args(0)? + " x y z")
      end
    end
~~~

# 実行結果（改善前）

[前回書いたログ記録方法](https://qiita.com/tadashi9e/items/745f2f5caa71eb3ddc4f) でタイムスタンプを付けて試しに実行してみると、

~~~
$ time ./tarai 4 3 0
15:22:25.473404415: tarai.pony:128: Start
15:22:25.473844481: tarai.pony:123: Result: 4
15:22:25.473847356: tarai.pony:123: Result: 4
~~~

と一見うまく動いているように見えます（結果が複数回出力されているのはとりあえず気にしない）。

しかし、結果を表示したあとも延々動き続けて最終的な実行時間はこんな感じでした:

~~~
real	8m49.684s
user	9m22.661s
sys	0m2.931s
~~~

もう少し引数の値を増やしてみます。

~~~
$ time ./tarai 7 6 0
15:35:51.311397303: tarai.pony:128: Start
15:35:51.556375968: tarai.pony:123: Result: 7
15:35:51.556380737: tarai.pony:123: Result: 7
15:35:51.556381609: tarai.pony:123: Result: 7
15:35:51.556382140: tarai.pony:123: Result: 7
15:35:51.556382751: tarai.pony:123: Result: 7
15:35:51.556384454: tarai.pony:123: Result: 7
15:35:51.556384935: tarai.pony:123: Result: 7
15:35:51.556385687: tarai.pony:123: Result: 7
~~~

これも結果は即座に得られていますが、

~~~
Killed

real	2m7.662s
user	0m43.001s
sys	0m51.924s
~~~

と使用リソース量が多すぎたためか Kill されてしまいました。

どうやら stop 通知があまり有効に機能していないようです。stop 通知も他のメッセージを追い越せず順番に処理されるためだと思われます。

この be は他の be よりも緊急で処理すべきもの、とアノテーションする方法があれば解決できそうな問題ではあります。

stop を非同期 (be) でなく同期 (fun) にできればよいのですが、Tarai の Reference Capability が tag なので fun で呼び出すことができません。

ならばと

~~~pony
actor Signal
  var stopped: Bool
  new create() =>
    """
    初期状態は stopped = false
    """
    stopped = false
  be stop() =>
    """
    stopped にする
    """
    stopped = true
  fun isStopped(): Bool =>
    """
    stopped になったかチェックする
    """
    stopped
~~~

というものを作って全部の DFrame から同じ Signal インスタンスの isStopped をチェックさせればいいだろう、と思ったのですが、複数の DFrame にインスタンスの参照を渡すと Reference Capability は tag なので isStopped は tag にするしかなく、isStopped が tag だと stopped は参照できない、ということでこれもダメそうです。

そろそろ Reference Capabilities について無知な状態では苦しくなってきたのので、本腰を入れてドキュメントを読んでいこうと思います。

# 実行結果（改善後）

上記の実行結果と考察を書いたときは Reference Capabilities についての理解不足により（とりあえずこうするとコンパイルが通るから、というわりと安直な理由で）、be から呼び出しているサブルーチン的処理を全て be で書いていました。
期待通りに動作していないのはこれのせいなのでは？と考えて fun ref に変更したところ、動作は劇的に改善してしまいました。

~~~
$ time ./tarai 12 6 0
21:12:43.834302879: tarai.pony:156: Start
21:12:43.870687187: tarai.pony:11: Result: 12

real	0m0.122s
user	0m0.266s
sys	0m0.040s
~~~

stop が正常に働くようになり、無駄な結果の表示の繰り返しはなくなった上に、結果表示後に無駄に動作し続けることもなくなりました。
闇雲になんでも非同期並列にすればいいというものではない、という事実を再確認することになりました。