ヒヨコとはじめる、やさしいGo言語の並行処理

こんばんは、ねじねじおです。

Goの並行処理をヒヨコと一緒に学んでいきます。

今回のポイントは3つだけです。

1. goroutine で並行処理を実行する
2. WaitGroup で処理の終了を待つ
3. channel でキューをつくる

ここに10匹の泣いているヒヨコを慰めるプログラムを用意しました。このプログラムでは一人のヒヨコシッターが一匹ずつ順番にヒヨコを慰めます。一匹のヒヨコが泣き止むには1秒かかります。

// 泣いているヒヨコを1秒で慰める関数
func comfort(piyo string) string {
	time.Sleep(1 * time.Second)
	return strings.Replace(piyo, "（> e <）", "（・e・）ﾜｰｲ", 1)
}

func main() {
	start := time.Now()

	for i := 1; i <= 10; i++ {
		// 泣いてるヒヨコ
		piyo := fmt.Sprintf("[%v]（> e <）", i)
		fmt.Println(piyo)
		// 慰める
		piyo = comfort(piyo)
		fmt.Println(piyo)
	}

	fmt.Printf("%f秒\n", (time.Now().Sub(start)).Seconds())
}

出力結果

[1]（> e <）
[1]（・e・）ﾜｰｲ
[2]（> e <）
[2]（・e・）ﾜｰｲ
[3]（> e <）
[3]（・e・）ﾜｰｲ
... 省略 ...
[10]（> e <）
[10]（・e・）ﾜｰｲ
10.030122秒

すべてのヒヨコを慰め終わるまでに、約10秒かかります。
このプログラムを使って、並行処理の書き方を学んでいきます。

1. goroutine で並行処理を実行する

goroutine は、Goのランタイムに管理される軽量なスレッドです。並行処理の肝です。
その実行は簡単で、go の後に続けてスレッド化したい関数を呼び出すだけです。次のプログラムでは、泣いているヒヨコの数だけスレッドを作ってヒヨコシッターを増やすことができます。

func main() {
	start := time.Now()

	for i := 1; i <= 10; i++ {
		// 泣いてるヒヨコ
		piyo := fmt.Sprintf("[%v]（> e <）", i)
		// goroutine を作成
		go func(piyo string) {
			fmt.Println(piyo)
			// 慰める
			piyo = comfort(piyo)
			fmt.Println(piyo)
		}(piyo)
	}

	// すべての処理が終わるまで待つ
	time.Sleep(2 * time.Second)

	fmt.Printf("%f秒\n", (time.Now().Sub(start)).Seconds())
}

出力結果

[2]（> e <）
[7]（> e <）
[10]（> e <）
[1]（> e <）
[5]（> e <）
[3]（> e <）
[4]（> e <）
[8]（> e <）
[9]（> e <）
[6]（> e <）
[5]（・e・）ﾜｰｲ
[4]（・e・）ﾜｰｲ
[6]（・e・）ﾜｰｲ
[8]（・e・）ﾜｰｲ
[2]（・e・）ﾜｰｲ
[1]（・e・）ﾜｰｲ
[9]（・e・）ﾜｰｲ
[7]（・e・）ﾜｰｲ
[10]（・e・）ﾜｰｲ
[3]（・e・）ﾜｰｲ
2.005158秒

すべてのヒヨコが2秒で泣き止みました。同時に処理が実行されるのでヒヨコの順番はバラバラになります。
ちなみに、ねじおのMacのコア数は4です。

func main() {
	fmt.Println(runtime.NumCPU())
}
// 出力結果
// 4

2. WaitGroup で処理の終了を待つ

前のプログラムでは、goroutineの処理が終わるのを待つために最後に2秒のスリープを入れていました。しかし、実際には処理はもっと早く終わっているかもしれないし、もっと遅いかもしれません。もし、ヒヨコが泣き止む前に呼び出し元の処理が終了してしまうと、ヒヨコとヒヨコシッターは闇の世界に消えてしまいます。

そんな悲しい結末にならないように、sync.WaitGroupを使ってgoroutineの終了を確実に待機するようにします。
sync.WaitGroupの使い方は、4ステップです。

1. WaitGroupをつくる。 wg := sync.WaitGroup{}
2. goroutineを呼び出す前に、呼び出し元でインクリメントする。wg.Add(1)
3. goroutineの処理が終わったら、goroutine内でデクリメントする。wg.Done()
4. すべての処理が終わるのを待つ。wg.Wait()

それでは、main()を書き換えます。

func main() {
	start := time.Now()

	wg := sync.WaitGroup{} //// 1. WaitGroupを作成
	for i := 1; i <= 10; i++ {
		// 泣いてるヒヨコ
		piyo := fmt.Sprintf("[%v]（> e <）", i)
		// goroutine を作成
		wg.Add(1) //// 2. WaitGroupをインクリメント
		go func(piyo string) {
			defer wg.Done() //// 3. WaitGroupをデクリメント

			fmt.Println(piyo)
			// 慰める
			piyo = comfort(piyo)
			fmt.Println(piyo)
		}(piyo)
	}

	wg.Wait() //// 4. すべての処理が終わるまで待つ

	fmt.Printf("%f秒\n", (time.Now().Sub(start)).Seconds())
}

出力結果

[10]（> e <）
[8]（> e <）
[9]（> e <）
[2]（> e <）
[4]（> e <）
[5]（> e <）
[6]（> e <）
[1]（> e <）
[7]（> e <）
[3]（> e <）
[7]（・e・）ﾜｰｲ
[3]（・e・）ﾜｰｲ
[10]（・e・）ﾜｰｲ
[8]（・e・）ﾜｰｲ
[9]（・e・）ﾜｰｲ
[2]（・e・）ﾜｰｲ
[6]（・e・）ﾜｰｲ
[4]（・e・）ﾜｰｲ
[5]（・e・）ﾜｰｲ
[1]（・e・）ﾜｰｲ
1.000854秒

これで無駄なく確実にgoroutineの終了を待てるようになりました。

3. channel でキューをつくる

さて、これまでのプログラムでは10匹のヒヨコに対して10人のヒヨコシッターがいました。実際にはそんな贅沢なことは稀です。ヒヨコシッターが2人しかいないときはどうすればよいでしょうか？

ここでは、泣いているヒヨコに一列に並んでもらい、何人かのヒヨコシッターが先頭のヒヨコから順番に慰めていくことにします。所謂、キューってやつですね。

Goにはchannelというスレッド間で通信する仕組みがあり、これを使うと簡単にキューを作ることができます。

キューを作成する

queue := make(chan データ型, キューのサイズ)

キューに値を追加する

queue <- val

キューに空き枠がない場合、この処理は空き枠ができるまでブロック（待機）します。

キューから値を取り出す

val, ok <-queue
if !ok {
	// queue is closed.
}

キューが空で、かつ、キューの入り口が閉じていない場合、この処理はキューに値が入るまでブロック（待機）します。
キューが空で、かつ、キューの入り口が閉じている場合、ok に false が入ります。

キューの入り口を閉じる

close(queue)

キューの入り口が閉じれるとキューに値を追加できなくなります。
また、キューの受信者は、キューが空のとき、キューが閉じられていること（もう値が入ってくることはないこと）を知ることができます。

では、main()を書き換えます。ヒヨコシッターは2人です。

const (
	thread = 2 // スレッド数
)

func main() {
	start := time.Now()

	// キューを作成
	queue := make(chan string, thread*2)

	// キューを処理するスレッドを作成
	wg := sync.WaitGroup{}
	for i := 0; i < thread; i++ {
		// goroutine を作成
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				// キューからヒヨコを取り出す
				piyo, ok := <-queue
				if !ok {
					break
				}
				fmt.Println(piyo)
				// 慰める
				piyo = comfort(piyo)
				fmt.Println(piyo)
			}
		}()
	}

	// キューに泣いているヒヨコを追加
	for i := 1; i <= 10; i++ {
		piyo := fmt.Sprintf("[%v]（> e <）", i)
		queue <- piyo
	}
	close(queue) // キューを閉じる

	wg.Wait() // すべての処理が終わるまで待つ

	fmt.Printf("%f秒\n", (time.Now().Sub(start)).Seconds())
}

出力結果

[2]（> e <）
[1]（> e <）
[1]（・e・）ﾜｰｲ
[3]（> e <）
[2]（・e・）ﾜｰｲ
[4]（> e <）
[3]（・e・）ﾜｰｲ
[5]（> e <）
[4]（・e・）ﾜｰｲ
[6]（> e <）
[5]（・e・）ﾜｰｲ
[7]（> e <）
[6]（・e・）ﾜｰｲ
[8]（> e <）
[7]（・e・）ﾜｰｲ
[9]（> e <）
[8]（・e・）ﾜｰｲ
[10]（> e <）
[10]（・e・）ﾜｰｲ
[9]（・e・）ﾜｰｲ
5.008001秒

うまくいったようです。2人で分担することで、約5秒で10匹のヒヨコを泣き止ますことができました。

さて、
退勤時間になったのでヒヨコシッターに仕事を中断させたいときにはどうすればよいのだろう？
ヒヨコが暴れだしてヒヨコシッターがパニックを起こしたらどうすれよいのだろう？
など疑問は残しつつも、今回はこれで終わります。

OK。
ねじねじおでした。