Golangでゴルーチンにより再帰関数を並列処理

#ゴルーチンで再帰関数を並列処理する
フィボナッチ数列など再帰関数をゴルーチンを使って並列処理について調査したときの備忘録。
ただし、パフォーマンスの観点は考慮していない。

##フィボナッチ数列
まずはGoでフィボナッチ数列のサンプル。
負の数については今回は対象外とした。

sample1.go

package main

import (
	"fmt"
	"os"
)

func fibonacci(n int64) int64 {
	switch {
	case n == 0:
		return 0
	case n == 1:
		return 1
	case n < 0:
		fmt.Println("Parameter should be positive integers")
		os.Exit(1)
	}
	return fibonacci(n - 2) + fibonacci(n - 1)
}

func main() {
	fmt.Println(fibonacci(20))
}

上記のサンプルの結果はこちら。

出力結果

$ go run main.go
6765

##ゴルーチンを使って再帰関数を並行処理
フィボナッチ数列のサンプルをゴルーチンで並行処理をする。
値はチャンネル経由で受信するようにした。

sample2.go

package main

import (
	"fmt"
	"os"
)

func fibonacci(n int64) int64 {
	switch {
	case n == 0:
		return 0
	case n == 1:
		return 1
	case n < 0:
		fmt.Println("Parameter should be positive integers")
		os.Exit(1)
	}

	ch1 := make(chan int64)
	go func() {
		defer close(ch1)
		ch1 <- fibonacci(n-2)
	}()
	ch2 := make(chan int64)
	go func() {
		defer close(ch2)
		ch2 <- fibonacci(n-1)
	}()
	return <-ch1 + <-ch2
}

func main() {
	ch := make(chan int64)
	go func() {
		defer close(ch)
		ch <- fibonacci(20)
	}()

	fmt.Println(<-ch)
}

同じ結果が出力された。

出力結果

$ go run main.go
6765

##再帰関数の並行処理時のゴルーチンの同時並行最大数
###WaitGroupを使ったカウンター
まずは正確にカウントできるかどうかサンプルで確認する。
ひたすらインクリメントするゴルーチンに対してWaitGroupを使った集計方法を導入したサンプル。

counter.go

package main

import (
	"fmt"
	"sync"
)

type MyWaitGroup struct {
	Wg sync.WaitGroup
	Count int64
	Ch1 chan int64
	Ch2 chan int64
}

func (g *MyWaitGroup) Add() {
	g.Wg.Add(1)
	g.Count++
}

func (g *MyWaitGroup) Done() {
	g.Wg.Done()
}

func (g *MyWaitGroup) Wait() {
	g.Wg.Wait()
}

func main() {
	counter := 0
	g := &MyWaitGroup{ Count:0 }
	for i := 0; i < 1000; i++ {
		g.Add()
		go func() {
			defer g.Done()
			counter++
		}()
	}
	g.Wait()

	fmt.Printf("%d, %d\n", counter, g.Count)
}

結果は以下。
ひたすらインクリメントするゴルーチンは計算処理中にmainが終了するため、期待するカウント値は取得できないのに対してWaitGroupを使った集計では取得ができていることを確認。

結果

$ go run main.go
951, 1000

###ゴルーチンの同時並行最大数を集計
先程のサンプルを拡張し、フィボナッチ数列のサンプルを並行処理をしたときのゴルーチンの同時並行最大数を集計。

sample3.go

package main

import (
	"fmt"
	"os"
	"sync"
)

type MyWaitGroup struct {
	Wg sync.WaitGroup
	Count int64
	MaxCount int64
}

func (g *MyWaitGroup) Add() {
	g.Wg.Add(1)
	g.Count++
	if g.MaxCount < g.Count { g.MaxCount = g.Count }
}

func (g *MyWaitGroup) Done() {
	g.Wg.Done()
	g.Count--
}

func (g *MyWaitGroup) Wait() {
	g.Wg.Wait()
	fmt.Printf("Max:%d\n", g.MaxCount)
}

func (g *MyWaitGroup) fibonacci(n int64) int64 {
	switch {
	case n == 0:
		return 0
	case n == 1:
		return 1
	case n < 0:
		fmt.Println("Parameter should be positive integers")
		os.Exit(1)
	}

	ch1 := make(chan int64)
	g.Add()
	go func() {
		defer close(ch1)
		defer g.Done()
		ch1 <- g.fibonacci(n-2)
	}()

	ch2 := make(chan int64)
	g.Add()
	go func() {
		defer close(ch2)
		defer g.Done()
		ch2 <- g.fibonacci(n-1)
	}()
	return <-ch1 + <-ch2
}

func main() {
	g := &MyWaitGroup{ Count:0, MaxCount:0 }
	ch := make(chan int64)
	g.Add()
	go func() {
		defer close(ch)
		defer g.Done()
		ch <- g.fibonacci(20)
	}()

	fmt.Println(<-ch)
	g.Wait()
}

結果はこちら。
実行するごとに最大値は異なるが、結構な数のゴルーチンが生成されていることがわかる。

結果

$ go run main.go
6765
Max:4729

##ゴルーチンのトータル発生数について
###計算方法
sample3.goにおけるゴルーチンのトータル発生数を計算する。
nが0と1の時はゴルーチンのトータル発生数は1、2のときは3となる。
nが3のときは5、4のときは9、5のときは15・・・となるので、ゴルーチンのトータル発生数は以下の漸化式となる。

f_c(0)=f_c(1)=1\\\
f_c(n)=f_c(n-2)+f_c(n-1)+1

再び再帰関数が生まれたので、サンプルを作成。

sample4.go

package main

import (
	"fmt"
	"os"
)

func fc(n int64) int64 {
	switch {
	case n == 0:
		return 1
	case n == 1:
		return 1
	case n < 0:
		fmt.Println("Parameter should be positive integers")
		os.Exit(1)
	}
	return fc(n - 2) + fc(n - 1) + 1
}

func main() {
	fmt.Println(fc(20))
}

出力結果は以下となり、21891のゴルーチンが累計で発生することになる。
そのため、sample3.goの場合、sync.Mutex等でゴルーチンの同時並行処理数を21891以下にするとデッドロックが発生する可能性がある。

出力結果

$ go run main.go
21891

###（補足）再帰関数からループに変更した場合
sample1.goを以下のように再帰関数を使わずにループ処理に変更すると直前のループ処理の結果が必ず必要になるため、ゴルーチンを使っても並行処理をする場合、ループを分けるなどの工夫が必要となる。

sample1a.go

package main

import (
	"fmt"
)

func main() {
	pv := [2]int64{1, 1}
	var fib int64
	for i := 2; i < 10; i++ {
		fib = pv[0] + pv[1]
		pv[0] = pv[1]
		pv[1] = fib
	}
	fmt.Println(fib)
}

##できるだけゴルーチンの最大並行実行数を制限する
チャンネルのcapacityを制限させることで（参考）、最大並行実行数を制限する。
fatal error: all goroutines are asleep - deadlock!が発生しないようにすることが前提。
以下のようにサンプルを変更する。チャンネルのcapacityをデフォルト（0）にした。

sample5.go

package main

import (
	"fmt"
	"os"
	"sync"
)

type MyWaitGroup struct {
	Wg sync.WaitGroup
	Count int64
	MaxCount int64
	Ch chan int64
}

func (g *MyWaitGroup) Add() {
	g.Wg.Add(1)
	g.Count++
	if g.MaxCount < g.Count { g.MaxCount = g.Count }
}

func (g *MyWaitGroup) Done() {
	g.Wg.Done()
	g.Count--
}

func (g *MyWaitGroup) Wait(cs chan int64) {
	g.Wg.Wait()
	close(g.Ch)
	cs <- g.MaxCount
}

func(g *MyWaitGroup) fibonacci(n int64) {
	switch {
	case n == 0:
		break
	case n == 1:
		g.Ch <- 1
	case n < 0:
		fmt.Println("Parameter should be positive integers")
		os.Exit(1)
	default:
		g.Add()
		go func() {
			defer g.Done()
			g.fibonacci(n - 2)
		}()
		g.Add()
		go func() {
			defer g.Done()
			g.fibonacci(n - 1)
		}()
	}
}

func printResult(cs <-chan int64, done chan<- bool) {
	var sum int64
	for s := range cs {
		sum += s
	}
	fmt.Println(sum)

	done <- true
}

func printMax(cs <-chan int64, done chan <- bool) {
	fmt.Printf("Max:%d\n", <-cs)
	done <- true
}

func main() {
	g := &MyWaitGroup{ Count:0, MaxCount:0, Ch:make(chan int64) }
	g.Add()
	go func() {
		defer func() {
			g.Done()
		}()
		g.fibonacci(20)
	}()

	done1 := make(chan int64)
	go g.Wait(done1)

	done2 := make(chan bool)
	go printResult(g.Ch, done2)
	<-done2

	done3 := make(chan bool)
	go printMax(done1, done3)
	<-done3
}

出力結果は以下のようにゴルーチンの同時並行最大数を制限できていることを確認。

出力結果

$ go run main.go
6765
Max:1792

sample3.goよりもsample5.goの方が同時並行実行されるゴルーチンの最大値よりも少なくなる可能性が高くなる。
フィボナッチ数列の計算量が増えれば増えるほど（サンプルでは20だが100等に増やした場合）、その可能性は高くなっていく。

##（補足）ゴルーチンの処理が完了しないケース
###sync.WaitGroupを使った同期
以下のようなサンプルの場合、ゴルーチンの処理が完了する前にmainが終了するため、正確なカウントができない。

calc1.go

package main

import (
	"fmt"
)

func main() {
	counter := 0
	for i := 0; i <= 100; i++ {
		go func(t int) {
			counter += i
		} (i)
	}
	fmt.Printf("%d\n", counter)
}

出力結果

$ go run main.go
4794

sync.WaitGroupでワーカーを用意し、同期させることでゴルーチンを正確にカウントする（計算を完了させる）ことができる。

calc2.go

package main

import (
	"sync"
	"fmt"
)

func main() {
	var wg sync.WaitGroup
	ch := make(chan int, 10)
	done := make(chan bool, 1)
	for i := 0; i <= 100; i++ {
		wg.Add(1)
		go func(t int) {
			defer wg.Done()
			ch <- t
		}(i)
	}
	s := 0
	go func () {
		for {
			select {
			case i := <- ch:
				s += i
			case <- done:
				return
			}
		}
	}()

	wg.Wait()
	fmt.Println(s)
}

出力結果

$ go run main.go
5050

###その他
以下の方法でもゴルーチンを正確を正確にカウントする（計算を完了させる）ことができる。
一旦数値をチャンネルに渡す、すべてのゴルーチンを起動した直後に、Waitしているので、WaitGroupカウンターが0になるのを待ち、チャンネルをクローズするゴルーチンを起動している。そうすることで、rangeでチャンネルの受信待ちをしている箇所でgoroutine 1 [chan receive]といったデッドロックのエラーを回避している。
なお、チャンネルのキャパシティをデフォルトの0ではなく、make(chan int, 100)と100等にすると処理速度の向上が期待できる。

calc3.go

package main

import (
	"sync"
	"fmt"
)

func main() {
	var wg sync.WaitGroup
	ch := make(chan int)
	for i := 0; i <= 100; i++ {
		wg.Add(1)
		go func(t int) {
			defer wg.Done()
			ch <- t
		}(i)
	}

	go func() {
		wg.Wait()
		close(ch)
	}()

	sum := 0
	for v := range ch {
		sum += v
	}
	fmt.Println(sum)
}

出力結果

$ go run main.go
5050

以下のように、チャンネルのキャパシティを10などにして、先にWaitGroupカウンターが0になるのを待ち、チャンネルをクローズするゴルーチンを起動すると、数値をチャンネルに渡すゴルーチンの処理状況においては、すべてのゴルーチンが起動する前にWaitGroupカウンターが0になる場合があるので、panic: send on closed channelといったエラーが発生する。

calc4.go

package main

import (
	"sync"
	"fmt"
	"time"
)

func main() {
	var wg sync.WaitGroup
	ch := make(chan int, 10)

	go func() {
		wg.Wait()
		close(ch)
	}()

	for i := 0; i <= 10; i++ {
		wg.Add(1)
		go func(t int) {
			defer wg.Done()
			ch <- t
		}(i)
		time.Sleep(time.Second)
	}

	sum := 0
	for v := range ch {
		sum += v
	}
	fmt.Println(sum)
}

出力結果

$ go run main.go
panic: send on closed channel

goroutine 17 [running]:
main.main.func2(0xc420014090, 0xc420086000, 0x1)
	/Users/hiroyuki/Documents/work/qiita/golang/fibonacci/main.go:22 +0x65
created by main.main
	/Users/hiroyuki/Documents/work/qiita/golang/fibonacci/main.go:20 +0xdb
exit status 2