Zenn
🐭

goの基本文法 No.7

2020/10/01に公開
Go
並行処理
基本
ゴルーチン
tech

2020-10-01執筆
A Tour of Goに沿ってGo言語の文法をまとめます。
これまでにJavaScriptやPythonなどのリッチ言語しか使ったことがなかった自分のためにまとめたものです。
なのでこれまでに上のようなリッチ言語を触ったことがあり、これからGo言語を触ってみようと考えている人の参考になればと思います。
文章の構成は、基本的に

  1. 概要
  2. コード
  3. コードの説明

の構成にしているつもりです。概要で簡単な説明をし、コードで例を示して、さらにコードの説明で細かい説明をするといった構成です。

並行処理

ここでは、goroutineを扱います。Go独特のものなので難しく感じるかもしれません。今回の内容を参考にし疑問に思ったことは自分でもplaygroundなどでいろいろ試して理解を深めることをオススメします。

並行処理

並行処理は異なるタスクを同時に実行する処理のことをいいます。例えばウェブサーバはいろいろクライアントからリクエストが届きます。前の処理が終わるまで次の処理ができないとなると簡単な処理なのに待ち時間が長くなってしまいます。そのようなとき変更処理があるとリクエストが届いたらすぐに処理をすることができます。
また、昨今のCPUはコア数がどんどん増えています。複数のコアを十二分に使い効率の良い処理をするためにも並行処理は重要なファクターになっているようです。

goroutine

goroutineはGoランタイムによって管理されています。多言語では並行処理の記述は難解になりやすいらしいのですが、このgoroutineによって簡単に並行処理ができます。

go f1()
go f2()
go f3()

上のようにgoroutineは関数の前にgoを付けるだけでf1()~f3()を並行処理できます。例をみてみましょう。

// code:7-1
func say(s string) {
	for i := 0; i < 5; i++ {
		time.Sleep(1000 * time.Millisecond)
		fmt.Println(s)
	}
}

func main() {
	go say("w")
	go say("o")
	go say("r")
	go say("l")
	go say("d")
	say("end")
}
/*実行結果
d
end
o
r
w
略
w
o
r
end
*/

code:7-1は最後をのぞいてsay関数を並行処理しています。1秒刻みで5回引数に与えられた単語を表示する関数を「w, o, r, l, d」のそれぞれの文字で並行処理を行い、その後並行処理とは別にendに対して行ってます。
実行結果をみたらわかるように、同時に処理を行い処理が終わったものから表示しているのでバラバラに表示されています。

Channel

ゴルーティンは同じアドレス空間で動作するので、共有メモリへのアクセスは同期されていなければなりません。

// code:7-2
func inc(x *int) {
	*x++
}

func main() {
	var x int
	for h := 0; h < 5; h++ {
		x = 0
		for i := 0; i < 1000; i++ {
			go inc(&x)
		}
		fmt.Println("x[", h, "]:", x)
	}
}
/*実行結果
x[ 0 ]: 981
x[ 1 ]: 966
x[ 2 ]: 910
x[ 3 ]: 943
x[ 4 ]: 1000
*/

code:7-2は与えられた引数をインクリメントするinc関数を実装し、goroutineで1000回実行しています。比較のためにそれをさらにループで5回繰り返し、5回の処理結果を比べます。
実行結果をみてわかるように、5回全て結果が異なります。本来は全て1000になるべきです。この違いの原因は、インクリメントのタイミングにあります。このプログラムでは1000回のインクリメントをgoroutineで同時に実行しています。簡単にインクリメントと言っていますが、実際は

  • 値が格納されているメモリを参照する
  • 現在の値を取得する
  • 現在の値に１加える
  • １加えた値を元のメモリに格納する

という作業を行っています。環境によってそれぞれのインクリメントのタイミングが多少前後しますが1000回が一斉に実行されるので中には値を取得するタイミング被ったりし、インクリメントが反映されない状況が出てきます。これは変数xという共有メモリへのアクセスが同期されていないから起こっています。
これを同期させ目的の結果を出させるために、goroutineどうしでコミュニケーションを取り合うシステムがチャンネルです。

ch := make(chan 型, サイズ)
ch <- 変数 // :Sender チャンネルに送信
変数 <- ch // :Reciever チャンネルから受信

チャンネルは、goroutine間で値をやりとりするためのものです。チャンネルはmake関数で作成します。このときやりとりする値の型とチャンネルが一度に持てる値の数(サイズ)も指定します。また値のやりとりは上のように<-で行います。
まずは以下の例をみてみましょう。

// code:7-3
func inc(x *int, ch chan string) {
	*x++
	ch <- "done" // ②
}

func main() {
	var x int
	ch := make(chan string) // ①
	for h := 0; h < 5; h++ {
		x = 0
		for i := 0; i < 1000; i++ {
			go inc(&x, ch)
			<-ch // ③
		}
		fmt.Println("x[", h, "]:", x)
	}
}
/*実行結果
x[ 0 ]: 1000
x[ 1 ]: 1000
x[ 2 ]: 1000
x[ 3 ]: 1000
x[ 4 ]: 1000
*/

code:7-3は先ほどのcode:7-2にチャンネルを導入したものです。

  • ①でまずチャンネル変数chを宣言します。サイズは指定していませんので一つの値しか受信できません。なので一つ受信したら別の変数に送信するまでは他のgoroutineから値を受信できないことになります。
  • for文の中でinc関数が並行処理されます。inc関数内でchは"done"というメッセージを受け取ります。
  • この時点で、chは１つの値を持っているので他のgoroutineがchに値を送信することができません。なので他のgoroutineは待ち状態になります。
  • メッセージを受け取ると③で変数に"done"というメッセージをチャンネルから受信できます。今回はメッセージを扱うことはないので受信する変数は必要ないので<-chとしています。
  • chから受信したのでchは空になり、次のgoroutineから値を受け取ることができます。

これによってそれぞれのgoroutineが同期され正しい値が出力されます。
まだ、わからない部分が多いと思うのでいくつか例を出します。

// code:7-4
func task1(v string, ch chan int) {
	fmt.Println(v)
	ch <- 1
}

func task2(v string, ch chan int) {
	fmt.Println(v)
	ch <- 2
}

func main() {
	ch1 := make(chan int)
	ch2 := make(chan int)
	go task2("task2", ch2)
	go task1("task1", ch1)
	fmt.Println(<-ch1)
	fmt.Println(<-ch2)
}

/*実行結果
1回目
task1
task2
1
2

2回目
task1
1
task2
2
*/

code:7-4の例では２つのgoroutineに対して２つのチャンネルを定義しています。goroutineは並行処理なので実行の度に結果の順番が異なります。

// code:7-5
func main() {
	ch := make(chan int, 2)
	ch <- 1
	ch <- 2
	ch <- 3
	fmt.Println(<-ch)
	fmt.Println(<-ch)
	fmt.Println(<-ch)
}
/*
実行結果
:~/go/src/tour_of_go/channel/buffer (master)
$ ./buffer 
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()
        /Users/go/src/tour_of_go/channel/buffer/buffer.go:11 +0x9b
*/

func main() {
	ch := make(chan int, 2)
	ch <- 1
	ch <- 2
	fmt.Println(<-ch)
	ch <- 3
	fmt.Println(<-ch)
	fmt.Println(<-ch)
}
/*実行結果
1
2
3
*/

code:7-5ではチャンネルの持てる量を２つにしています。上ではチャンネルが３つの値を送信を挟まずに受信してしまっているので同時に３つの値を保持することになりエラーになっています。下では３つ目を受信する前に送信を行っているので一つ空きができ受信することが出来ます。

チャンネルを使ったやりとりは一般的にメッセージパッシングというらしいです。詳しくはメッセージパッシングとかCSP(Communicasing Sequential Processes)とかで調べてください。

Range and Close

Close

チャンネルはclose関数で閉じることができます。チャンネルを閉じると値からチャンネルに送信することができません。また、これまではチャンネルからは値を受信することしかしませんでしたが、値とチャンネルが閉じているかどうかの真偽値も受け取ることができます。

close(ch) // チャンネルを閉じる
v, ok := <-ch // チャンネルから値と閉じてるかどうかの真偽値を受け取る。

真偽値はチャンネルが閉じられていて、チャンネルから受信できる値がなければfalseです。

// code:7-6
func a(c chan int) {
	c <- 1
}

func main() {
	ch := make(chan int, 10)
	a(ch)
	close(ch)
	v, ok := <-ch

	if ok {
		fmt.Println(v, ok)
	} else {
		fmt.Println("channel closed.")
	}
}
/*実行結果
1 true
*/

// 閉じた後にチャンネルに値を送信した場合
func a(c chan int) {
	c <- 1
}

func main() {
	ch := make(chan int, 10)
	a(ch)
	close(ch)
	a(ch)
	v, ok := <-ch

	if ok {
		fmt.Println(v, ok)
	} else {
		fmt.Println("channel closed.")
	}
}
/*実行結果
panic: send on closed channel

goroutine 1 [running]:
main.a(...)
	/tmp/sandbox643135727/prog.go:8
main.main()
	/tmp/sandbox643135727/prog.go:20 +0x91
*/

code:7-6はこれまで説明したチャンネルの閉じ方と真偽値を受け取った例です。下のコードはチャンネルを閉じた後にa関数を実行しチャンネルに値を送信しています。このような場合はパニックを起こします。

// code:7-7
func b(c chan int) {
	c <- 2
	close(c)
}

func main() {
	ch := make(chan int, 10)
	b(ch)
	b(ch)
	v, ok := <-ch

	if ok {
		fmt.Println(v, ok)
	} else {
		fmt.Println("channel closed.")
	}
}
/*実行結果
panic: send on closed channel

goroutine 1 [running]:
main.b(...)
	/tmp/sandbox201782908/prog.go:12
main.main()
	/tmp/sandbox201782908/prog.go:19 +0x91
*/

code:7-7のようにmain関数ではなく呼び出した関数からでも閉じることができます。この場合は１度目のb関数の実行ですでに閉じているので2回目のb関数では値をチャンネルに送信できずパニックになります。

Range

複数の値を送信から受け取っている場合はrangeでループを回すことが出来きます。この時ループはチャンネルが閉じられているところまで繰り返されます。なのでrangeで回す時は、その前にclose()で必ず閉じておく必要があります。閉じていなかった場合は、パニックを起こします。

// code:7-8
func c(c chan int) {
	for i := 0; i < 10; i++ {
		c <- i
	}
}

func main() {
	ch := make(chan int, 10)
	c(ch)
	close(ch)
	for c := range ch {
		fmt.Println(c)
	}
}
/*実行結果
0
1
2
3
4
5
6
7
8
9
*/

code:7-8は10個の値が送信されたチャンネルをrangeでループして全ての値をfmt.Printlnで出力しています。この場合<-は必要ないことに注意してください。

注）チャネルを閉じる必要があるのは送信側だけで、受信側は閉じないでください。閉じたチャネルに送信すると、パニックが発生します。
別のメモ：チャネルはファイルとは異なります。通常は閉じる必要はありません。閉じる必要があるのは、rangeループを終了するなど、受信側に値が来ないことを通知する必要がある場合のみです。

Select

select文はswitch文と似ていますがswitch文と異なり、チャンネルが送信できるか受信できるかによって操作を分岐する文です。goroutineが複数の通信操作を待機できるようになります。
selectは、そのケースの1つが実行可能になるまでブロックし、その後そのケースを実行します。複数の準備ができている場合は、ランダムに1つを選択します。

下の例では以前やったフィボナッチ数列をgoroutine, channel, selectを使ってやっています。

// code:7-9
func fibonacci(c, quit chan int) {
    x, y := 0, 1
	for {
        select {
            case <-quit: // *1
			time.Sleep(200 * time.Millisecond)
			fmt.Println("quit")
			return
		case c <- x: // *2
			time.Sleep(200 * time.Millisecond)
			x, y = y, x+y
		}
	}
}

func main() {
    fmt.Println("go start.")
	c := make(chan int)
	quit := make(chan int)
	go func() {
        for i := 0; i < 10; i++ {
            fmt.Println(<-c)
		}
		quit <- 0
	}()
	fibonacci(c, quit)
}
/*実行結果
go start.
0
1
略
34
quit
*/

コードからは処理の流れが読み取りにくいですが次のようなことをやっています。

  1. チャンネルc, quitの定義
  2. 即時関数の実行
  3. fibonacci()の実行:quitは何も受信してないため、quitから送信できない。cは受信可能なので*2が実行される
  4. cからの送信が可能になるため即時関数内のループのi=0回目を実行し、Println()でcから受信した値を表示
  5. 3.を実行
  6. 4.を実行
    以後これをi=9まで合計10回繰り返す。
  7. 即時関数のforループが終了したので、quitが0を受信する。
  8. fibonacci()の実行:quitが0を受信しているので*1が実行され、returnでfibonacci()は終了される。
// code:7-10

// func fibobacci() {} 略
func main() {
	c := make(chan int)
	quit := make(chan int)

	fmt.Println("go start.")

	fibonacci(c, quit)
	go func() {
		for i := 0; i < 10; i++ {
			fmt.Println(<-c)
		}
		quit <- 0
	}()

}
/*
実行結果
$ ./select 
go start.
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [select]:
main.fibonacci(0xc0000200c0, 0xc000020120)
        /go/src/tour_of_go/channel/select/select.go:11 +0xe8
main.main()
        /go/src/tour_of_go/channel/select/select.go:29 +0xd7
*/

code:7-10の場合だと、fibonacci()を実行後に即時関数を実行する手順になっています。fibonacchi()でチャンネルcから受信する必要がありますが、cの中身はからなのでdeadlockが発生しエラーになります。並行処理の後に持ってくる必要があるので注意しましょう。

// code:7-11
func fibonacci(c, quit chan int) {
	x, y := 0, 1
	for i := 0; i < 10; i++ {
		time.Sleep(200 * time.Millisecond)
		c <- x
		x, y = y, x+y
	}
	quit <- 0
}

func main() {
	c := make(chan int)
	quit := make(chan int)
	fmt.Println("go start.")

	go fibonacci(c, quit)
	func() {
		for {
			select {
			case v := <-c:
				fmt.Println(v)
			case <-quit:
				fmt.Println("quit")
				return
			}
		}
	}()
}
/*
実行結果(略)
*/

code:7-11のようにfibonacci関数側ではなく即時関数側にselect文を書いて同じ実装も可能です。

defaultを使うと、他のケースの準備ができていないときに行う処理を実装できます。

// code:7-12
package main

import (
	"fmt"
	"time"
)

func main() {
	tick := time.Tick(1000 * time.Millisecond)
	boom := time.After(3000 * time.Millisecond)
	for {
		select {
		case <-tick:
			fmt.Println("tick.")
		case <-boom:
			fmt.Println("BOOM!")
			return
		default:
			fmt.Println("    .")
			time.Sleep(500 * time.Millisecond)
		}
	}
}
/*実行結果
    .
    .
tick.
    .
    .
tick.
    .
    .
BOOM!
(3秒をカウント！)
*/

code:7-12では1秒ごとにtick.を表示し3秒後にBOOM!を表示します。それ以外の場合は0.5秒感覚で" ."を表示します(defaultの処理)。このようにチャンネルから受信・送信するまでに別の処理をさせておくこともできます。

Exercise: Equivalent Binary Trees

最後に該当するエクササイズの概要と私の解答を載せておきます。参考までに。
私の解答

sync.Mutex

並行処理でのゴルーチン間のコミュニケーションにチャンネルが有効だというのはこれまでの内容でわかったと思います。
しかし、コミュニケーションを必要としないが処理の衝突を避けたい場合があります。その時使うのがmutexです。
mutexはロックとアンロックができます。
mutexによってロックされている間は他の処理が実行されません。なので以下の例はcode:7-3をmutexで書き換えたものですが、複数のゴルーチンが同時にインクリメントを実行し値がその時々で変化することはありません。ただし、これはロックを行っているinc関数内での話であり、fmt.Printlnは関係ない。なのでPrintlnで表示する前にtime.Sleepで遅延をおこす必要があります。

// code:7-12
func inc(x *int, m *sync.Mutex) {
	m.Lock()
	*x++
	m.Unlock()
}

func main() {
	var x int
	var m sync.Mutex
	for h := 0; h < 5; h++ {
		x = 0
		for i := 0; i < 1000; i++ {
			go inc(&x, &m)
		}
		time.Sleep(time.Millisecond)
		fmt.Println("x[", h, "]:", x)
	}
}
/*実行結果
x[ 0 ]: 1000
x[ 1 ]: 1000
x[ 2 ]: 1000
x[ 3 ]: 1000
x[ 4 ]: 1000
*/

WaitGroup

A Tour of Goでは扱っていませんが、よく使われるパッケージなので、sync.WaitGroupも紹介だけしておきます。これでcode:7-12のtime.Sleepで遅延を起こす問題は解決されます。細かい使い方は公式ドキュメントなどを読んでください。

// code:7-13
func inc(x *int, m *sync.Mutex, w *sync.WaitGroup) {
	m.Lock()
	*x++
	m.Unlock()
	w.Done()
}

func main() {
	var x int
	var m sync.Mutex
	var w sync.WaitGroup
	for h := 0; h < 5; h++ {
		x = 0
		for i := 0; i < 1000; i++ {
			w.Add(1)
			go inc(&x, &m, &w)
		}
		w.Wait()
		fmt.Println("x[", h, "]:", x)
	}
}
/*実行結果
x[ 0 ]: 1000
x[ 1 ]: 1000
x[ 2 ]: 1000
x[ 3 ]: 1000
x[ 4 ]: 1000
*/

Exercise: Web Crawler

最後に該当するエクササイズの概要と私の解答を載せておきます。参考までに。
私の解答

最後に

お疲れ様です。長かったですが、これでA Tour of Goの内容は終了しました。解りにくいところもあったかと思いますが、少しでも読んでくれた方のためになればと思います。
このあとは、ローカルでの環境構築とGoを使ったウェブ関係のことについてまとめていこうと思います。よかったらご一読ください。

Discussion