ゴルーチンによるorチャネルパターン

「Go言語による並行処理4.4 orチャネル」 で、複数のチャネルをまとめて、まとめたチャネルのうちどれか1つでも閉じられた場合、まとめたチャネルも閉じるようにしたい。という場合に、orチャネルというパターンがあることが紹介されている。

「この関数はかなり簡潔になっていて」という説明があるが、関数自体を自分で再実装しようと思った時に少し理解しづらいように感じたので、別パターンの実装も用意して考えてみた。

元のor関数の実装はここには記載しないが、元の関数の場合の具体的なケースを図にしてみる。これはs1~s6のチャネルを1つのチャネルにまとめあげた場合である。

元の関数で組み立てた時の依存グラフ

図にすると分かりやすい。ノードはチャネルを表しており、矢印の向きは監視しているチャネルを指す。あるチャネルが1つでも閉じられると、そのチャネルを監視対象とするチャネルが閉じられる。orDoneは循環しているので、どれか1つでも閉じられれば、全てのorDoneが閉じられる事になる。 つまりのところ、orDoneを循環させればいいという話である。末端のorDoneに閉じられたという情報が流れてくるので、末端のノードを全てのorDoneに繋げてしまえば全てのorDoneが閉じられることになる。

その点を踏まえて、一つ代替実装を書いてみた。末端のノードになる予定のorDoneチャネルを一つだけ先に用意しておく。再帰関数でチャネルを畳み込む時に、末端のorDoneチャネルから受け取れるように繋ぎ込んでおく。最後に、末端のorDoneチャネルだけは別途ゴルーチンで接続する。

var myOr func(channels ...<-chan interface{}) <-chan interface{}
myOr = func(channels ...<-chan interface{}) <-chan interface{} {
   switch len(channels) {
   case 0:
      return nil
   case 1:
      return channels[0]
   }

   lastOrDone := make(chan interface{})
   var joinRec func(channels ...<-chan interface{}) <-chan interface{}
   joinRec = func(channels ...<-chan interface{}) <-chan interface{} {
      if len(channels) == 1 {
         return channels[0]
      }
      orDone := make(chan interface{})
      go func() {
         defer close(orDone)
         select {
         case <-channels[0]:
         case <-lastOrDone: // 末端のノードのチャネルを繋ぎこむ
         case <-joinRec(channels[1:]...):
         }
      }()
      return orDone
   }

   joined := joinRec(channels[1:]...)
   go func() {
      defer close(lastOrDone)
      select {
      case <-channels[0]:
      case <-joined:
      }
   }()
   return lastOrDone
}

より直感的な書き方をすると、以下の図のような依存グラフが出来上がる。先ほどの元のor関数で作られた依存グラフとは少し異なる。

myOr関数で組み立てた時の依存グラフ

元の関数が、まとめる対象のチャネルの数 x に対して必要なゴルーチンの数 x/2 だったのに対し、今回書いたものは x-1 個のゴルーチンが必要になる。あきらかに効率が悪いことがわかる。

効率をあげるためには再帰関数の１ステップでまとめあげるチャネルの数を増やせば良い。まとめあげるチャネルの数がわかっていれば、当然まとめあげるチャネル全てから受信するselect文を用意すれば、ゴルーチンの数は1つで済む。そうでないので、部分的にまとめていかざるをえない。再帰関数１ステップでまとめるチャネルの数が増えれば増えるほど、関数は冗長になるのでトレードオフの結果、本では3つというようになっているのだろう。

特に学びのない投稿になった。