今回は Go 言語の goroutine と channel を使って、簡単な並行プログラミングに挑戦してみましょう。
「並行 (concurrent) プログラミング」は複数のプログラムを並行に実行しますが、このとき複数の CPU で同時に動かす場合と、ひとつの CPU で複数のプログラムを動かす場合があります。一般的には、前者を「並列 (parallel) プログラミング」といい、複数のハードウェアを並列に実行することによる処理速度の向上が主な目的となります。
後者の場合、一定時間毎に実行するプログラムを切り替えることで、複数のプログラムを並列に実行しているかのように見せることができます。この処理を「時分割 (time sharing)」もしくは「タイム・スライス (time slice)」といいます。一般に、タイム・スライスは OS でサポートされている機能です。OS が実行するプログラムのことを「プロセス (process)」または「タスク (task)」といいます。
並列的に動作するプログラムをひとつのプロセスだけで作るのはけっこう大変です。そこで、プロセス内では逐次的な処理にとどめ、複数のプロセス間で情報交換を行うことにより、全体で並列的な動作を実現することを考えます。このほうが自然にプログラムを記述できる場合があるのです。これが後者の主な目的となります。
プロセスは互いに独立したプログラムですが、OS にはプロセス間でデータをやり取りする機能 (プロセス間通信) が用意されています。たとえば、UNIX ライクな OS では「パイプ (pipe)」を使って複数のプログラム (コマンド) を連結することができます。この場合、パイプを通してデータがプログラムに送られ、各プログラムは並行に動作することになります。入出力処理で待たされるコマンドがあったとしても、その間に他のコマンドが実行されるので、各コマンドを逐次的に実行するよりも、効率的に処理することが可能です。
最近では、ひとつのプログラムの中で独立した複数の処理を記述できるようになりました。この機能を「スレッド (thread)」とか「マルチスレッド」いいます。スレッドは「縫い糸」という意味ですが、プログラムでは「制御の流れ」という意味で使われています。並列的な動作をプログラムする場合、逐次的な処理をひとつのスレッドに割り当て、複数のスレッドを並行に動作させることにより、全体で並列的な動作を実現するわけです。
一般に、スレッドは一定時間毎に実行するスレッドを強制的に切り替えます。このとき、スレッドのスケジューリングは処理系が行います。これを「プリエンプティブ (preemptive)」といいます。これに対し、Ruby のファイバーや Lua のコルーチンは、プログラムの実行を一定時間毎に切り替えるものではありません。他のプログラムが実行できるよう自主的に処理を中断する、といった協調的な動作を行わせることで、複数のプログラムを並行に動作させています。これを「ノンプリエンプティブ (nonpreemptive)」といいます。
スレッドは同じプロセス内に存在するので、メモリ空間を共有することができます。これを「共有メモリ」といいます。スレッド間の通信は共有メモリを使って簡単に行うことができますが、プリエンプティブなスレッドの場合、共有メモリのアクセス時に発生する「競合」が問題になります。このため、プリエンプティブなマルチスレッドをサポートしているプログラミング言語では、競合を回避するための仕組みが用意されています。
Go 言語の goroutine はプリエンプティブなマルチスレッドに近いもので、goroutine は OS ではなく Go 言語が管理します。Go 言語の goroutine は共有メモリを使って通信することができますが、それは推薦されていないようです。そのかわり「チャネル (channel)」という通信路が用意されいて、それを使って goroutine 間でデータの送受信を行うことができます。他のプログラミング言語では、関数型言語 Erlang が同様な方法を採用しています。Erlang のプロセスと同じように、Go 言語の goroutine は軽量で高速に動作するといわれています。
Go 言語はマルチコア CPU に対応していて、複数の goroutine を複数のコアで並列に実行させることができます。使用するコア数は runtime パッケージの関数 GOMAXPROCS や環境変数 GOMAXPROCS で設定することができます。
func GOMAXPROCS(n int) int
引数 n は同時に使用可能な最大コア数で、返り値は前に設定されていたコア数です。n < 1 の場合、コア数を変更せずに現在設定されているコア数を返します。Go 言語 ver 1.4 までは GOMAXPROCS が 1 に設定されていましたが、Go 言語 ver 1.5 からは適切な値に設定されるようになりました。
利用できる CPU のコア数は runtime パッケージの関数 NumCPU で取得できます。
func NumCPU() int
関数 NumGoroutine は現存している goroutine の数を返します。
func NumGoroutine() int
M.Hiroi のパソコン (CPU, Intel Core i5-6200U 2.30GHz) は物理コア数が 2 で、1 コアにつきハイパースレッディングで 2 分割できるので、NumCPU の値は 2 * 2 = 4 になります。これを「論理コア数」と呼ぶことがあります。それでは、実際に確かめてみましょう。
リスト : コア数を表示する (sample1200.go)
package main
import (
"fmt"
"runtime"
)
func main() {
fmt.Println(runtime.NumCPU())
fmt.Println(runtime.GOMAXPROCS(0))
fmt.Println(runtime.NumGoroutine())
}
$ go run sample1200.go 4 4 1
NumCPU の返り値は 4 になりました。GOMAXPROCS は NumCPU と同じ値 (4) に設定されています。なお、本稿は並行プログラミングが主旨なので、GOMAXPROCS の値はあえて 1 に設定することとします。並列プログラミングについては次回以降に取り上げます。
それでは goroutin を起動してみましょう。Go 言語の場合、関数 (またはメソッド) の呼び出しの前に go を付けると、異なる goroutine で関数を実行することができます。
go 関数名(引数, ...)
go は新しい goroutine を生成して、その goroutine 内で指定された関数を実行します。新しい goroutine は並行に動作するので、関数の実行終了を待つことなく、go のあとに記述されているプログラムを実行します。go で呼び出された関数の実行が終了すると、その goroutine の実行も終了します。このとき、関数の返り値は破棄されます。結果を受け取りたい場合はチャネルを使うことになります。
簡単な例を示しましょう。0.5 秒間隔で name を n 回表示するプログラムを作ります。
リスト : 0.5 秒間隔で Name を N 回表示する (sample1201.go)
package main
import (
"fmt"
"time"
"rumtime"
)
func test(n int, name string) {
for i := 1; i <= n; i++ {
fmt.Println(i, name)
time.Sleep(500 * time.Millisecond)
}
}
func main() {
runtime.GOMAXPROCS(1)
test(5, "foo")
test(5, "bar")
}
パッケージ time の関数 Sleep はプログラムの実行を指定した時間だけ休止します。
func Sleep(d Duration)
Duration は int64 の別名で時間 (期間) を表します。単位はナノ秒です。Millisecond は 1 ミリ秒を表す定数です。整数値を Duration に変換するには関数 time.Duration を使ってください。test を通常の関数呼び出しすると、次のようになります。
$ go run sample1201.go 1 foo 2 foo 3 foo 4 foo 5 foo 1 bar 2 bar 3 bar 4 bar 5 bar
test が順番に呼び出されるので、最初に foo が表示されてから、次に bar が表示されます。ここで go を用いると、foo と bar の表示が並行に行われます。それでは試してみましょう。
リスト : test を並行に実行する (sample1202.go)
func main() {
go test(5, "foo")
test(5, "bar")
}
$ go run sample1202.go 1 bar 1 foo 2 bar 2 foo 3 bar 3 foo 4 bar 4 foo 5 bar 5 foo
go で test(5, "foo") を呼び出すと、新しい goroutine が生成されて、その goroutine で test が実行されます。そのあと、すぐに test(5, "bar") が実行されるので、bar と foo が交互に表示されます。逐次処理とは違って、foo と bar を表示する処理が並行に行われていることがわかります。
ただし、次のように 2 つの test を go で呼び出すと、画面には何も表示されません。
リスト : 無表示になるプログラム
func main() {
go test(5, "foo")
go test(5, "bar")
}
新しい goroutine を 2 つ生成して関数 test を実行しますが、そのあと関数 main の処理は何もないので、main の実行はすぐに終了してしまいます。main が終了するとプログラム全体の処理も終了するので、画面には何も表示されないのです。この場合、main は goroutine が終了するまで待たなければいけません。これはチャネルを使うと簡単に実現できます。
チャネルは Go 言語に標準で用意されているデータの一つで、スライスと同じタイプのデータ構造です。スライスと同様に、チャネルは関数の引数に渡したり、関数の中でチャネルを生成して、それを返すこともできます。
チャネルは組み込み関数 make で生成します。
ch := make(chan T, バッファサイズ)
送受信するデータの型を T とすると、チャンネルの型は chan T となります。バッファサイズはデータを格納するバッファの大きさで、省略するとバッファサイズは 0 になります。なお、関数の引数や変数の型指定で chan の前に <- を付けると受信専用に、後ろに <- を付けると送信専用になります。
バッファリングを行わない場合、通信を行う goroutine の実行は「同期」されます。つまり、データを書き込む側と読む出す側が揃うまで、どちらか一方の実行はブロック (休止) されます。バッファリングを行う場合、チャネルのバッファは「キュー」として動作します。この場合、通信を行う goroutine の実行は「非同期」になります。
たとえば、バッファにデータがある場合、読み出す側はデータを読み込んでプログラムの実行を続けます。読み出す側がブロックされるのはバッファにデータがないときだけです。逆に、書き込む側はバッファに空きがあれば、データを書き込んでプログラムの実行を続けます。バッファが満杯のときだけブロックされます。
データの送受信はチャネルオペレータ <- を使います。
ch <- v // データ v を ch へ送信 v := <- ch // ch からデータを受信して v にセット
チャネルは組み込み関数 close でクローズすることができます。クローズしたチャネルからの受信は必ず成功してゼロ値が返ってきます。チャネルがクローズされているかチェックするには、チャネルオペレータの返り値を多重代入で受け取ります。
v, ok := <- ch
正常な通信でデータ v を受信した場合、ok には true がセットされます。ch がクローズされている場合、v はゼロ値が返され、ok には false がセットされます。チャネルはファイルと違ってクローズしなくてもいいのですが、for ループの range でチャネルからデータを取り出す場合はクローズする必要があります。
それでは簡単な例を示しましょう。0.5 秒間隔で name を n 回表示する関数にチャネルを渡すよう修正します。
リスト : 0.5 秒間隔で Name を N 回表示する (sample1203.go)
package main
import (
"fmt"
"time"
"runtime"
)
func test(n int, name string, c chan<- string) {
for i := 1; i <= n; i++ {
fmt.Println(i, name)
time.Sleep(500 * time.Millisecond)
}
c <- name
}
func main() {
runtime.GOMAXPROCS(1)
c := make(chan string)
go test(6, "foo", c)
go test(4, "bar", c)
go test(8, "baz", c)
for i := 0; i < 3; i++ {
fmt.Println(<- c)
}
}
main の中でチャネルを生成して変数 c にセットします。チャネルの型は chan string としました。関数 test で name を n 回表示したら、最後に name をチャネル c に送信します。c の型は chan<- string で送信専用に設定しています。main 側では go で 3 つの test を呼び出して、for ループの中で c からデータを受け取ります。ここで main 側の処理がブロックされます。3 つのデータを受け取ったら for ループを終了します。
実行結果は次のようになります。
$ go run sample1203.go 1 foo 1 bar 1 baz 2 foo 2 bar 2 baz 3 foo 3 bar 3 baz 4 foo 4 bar 4 baz 5 foo bar 5 baz 6 foo 6 baz foo 7 baz 8 baz baz
このように、goroutine 側から main 側へ終了を通知することで、3 つの goroutine を最後まで実行することができます。
チャネルのほかに、パッケージ sync の WaitGroup を使って goroutine の終了を待つ方法があります。WaitGroup は構造体で、次に示すメソッドが用意されています。
WaitGroup は待ち合わせを行う goroutine の個数を管理します。Add は待ち合わせを行う goroutine の数を delta だけ増やします。Done は待ち合わせを行っている goroutine の数を -1 します。Wait は WaitGrout で管理している goroutine の数が 0 になるまで goroutine の実行をブロックします。
WaitGroup を使って sample1203.go を書き直すと次のようになります。
リスト : 0.5 秒間隔で Name を N 回表示する (WaitGroup バージョン)
package main
import (
"fmt"
"time"
"runtime"
"sync"
)
func test(n int, name string, wg *sync.WaitGroup) {
for i := 1; i <= n; i++ {
fmt.Println(i, name)
time.Sleep(500 * time.Millisecond)
}
wg.Done()
}
func main() {
runtime.GOMAXPROCS(1)
var wg sync.WaitGroup
wg.Add(3)
go test(6, "foo", &wg)
go test(4, "bar", &wg)
go test(8, "baz", &wg)
wg.Wait()
}
最初に WaitGroup の変数 wg を用意し、wg.Add(3) で実行する goroutine の個数を追加します。関数 test は引数に wg へのポインタを受け取り、for ループを終了したらメソッド Done を実行します。これで WaitGroup のカウンタを -1 することができます。あとは、go で test を実行して、メソッド Wait で起動した 3 つの goroutine の終了を待つだけです。
チャンネルを使って同期をとることで、複数の goroutine を協調的に動作させることができます。たとえば、1 文字を表示する goroutine を複数個作成し、"hey! " とういう文字列を複数回画面に表示するプログラムを作ってみましょう。次のリストを見てください。
リスト : goroutine の同期 (sample1204.go)
package main
import (
"fmt"
"time"
"runtime"
)
func makeRoutine(code string, in <-chan int) chan int {
out := make(chan int)
go func(){
for {
<- in
fmt.Print(code)
time.Sleep(50 * time.Millisecond)
out <- 0
}
}()
return out
}
func main(){
runtime.GOMAXPROCS(1)
ch1 := make(chan int)
ch2 := makeRoutine("h", ch1)
ch3 := makeRoutine("e", ch2)
ch4 := makeRoutine("y", ch3)
ch5 := makeRoutine("!", ch4)
ch6 := makeRoutine(" ", ch5)
for i := 0; i < 10; i++ {
ch1 <- 0
<- ch6
}
}
$ go run sampl1204.go hey! hey! hey! hey! hey! hey! hey! hey! hey! hey!
makeRoutine は引数の文字 code を表示する goroutine を起動します。最初に、文字を表示したことを通知するチャネル out を生成し、匿名関数を go で起動してから return でチャネル out を返します。匿名関数は無限ループになっていて、チャネル in からデータを受信するのを待ちます。次に、code を表示してから 50 msec 後に out へデータを送信します。たとえば、goroutine 1 の送信チャネルを goroutine 2 の受信チャネルに設定すると、goroutine 1 で文字が表示されたあと、goroutine 2 で文字が表示されることになります。
main 関数では、最初にチャネル ch1 を生成し、それを makeRoutine に渡して返り値を変数 ch2 にセットします。これで ch1 にデータを送信すると、"h" が表示されて ch2 にデータが送信されます。次に、ch2 を makeRoutine に渡して、返り値を ch3 にセットします。ch2 にデータを送信すると、"e" が表示されて ch3 にデータが送信されます。あとは、表示する文字だけ makeRoutine で goroutine を生成します。
最後に、for ループで ch1 にデータを送信して、画面に文字列を表示します。"hey! " が表示されたあと、ch6 にデータが送信されるので、それを受信するのを待ちます。あとは、これを 10 回繰り返すだけです。
for ループの range はチャネルにも適用することができます。
for v := range チャネル { 処理 }
チャネルの場合、スライスやマップと違って返り値は多値ではありません。また、データの送信が終了したらチャネルをクローズする必要があります。
簡単な例を示しましょう。拙作のページ 二分探索木 で作成した二分木の要素を for ループで順番に取り出すことを考えます。この処理は goroutine を使うと簡単に実現できます。次のリストを見てください。
リスト : 二分木の要素を順番に取り出す
// 要素を取り出してチャネルに送信
func (t *Tree) each() chan Item {
ch := make(chan Item)
go func(){
t.foreachTree(func(x Item) { ch <- x })
close(ch)
}()
return ch
}
// 簡単なテスト
func main() {
runtime.GOMAXPROCES(1)
a := newTree()
b := []int{5,6,4,3,7,8,2,1,9,0}
for _, x := range b {
a.insertTree(Int(x))
}
for x := range a.each() {
fmt.Println(x)
}
}
メソッド each の中でチャネル ch を生成し、go で匿名関数を呼び出します。この中で高階関数 foreachTree を呼び出して二分木を巡回して、要素 x をチャネル ch へ送信します。二分木の要素の型はインターフェース Item なので、チャネルの型は chan Item になります。巡回が終わったら close でチャネル ch をクローズします。main 側では a.each() の返り値 (チャネル) を for ループの range に渡すだけです。
それでは実行してみましょう。
$ go run tree.go 0 1 2 3 4 5 6 7 8 9
goroutine とチャネルを使って高階関数をジェネレータに変換することもできます。次のリストを見てください。
リスト : ジェネレータの生成
func (t *Tree) makeGen() func() Item {
ch := make(chan Item)
go func(){
t.foreachTree(func(x Item){ ch <- x })
close(ch)
}()
return func() Item { return <- ch }
}
func main() {
runtime.GOMAXPROCES(1)
a := newTree()
b := []int{5,6,4,3,7,8,2,1,9,0}
for _, x := range b {
a.insertTree(Int(x))
}
resume := a.makeGen()
for i := 0; i < 11; i++ {
fmt.Println(resume())
}
}
メソッド makeGen は each とほとんど同じですが、返り値は匿名関数になります。この中でチャネル ch からデータを受信して return で値を返します。main 側では、makeGen で生成したジェネレータを変数 resume にセットします。あとは resume を呼び出せば二分木からデータを順番に取り出すことができます。この場合、チャネルの終了は nil (ゼロ値) でチェックすることができます。
それでは、実際に実行してみましょう。
$ go run tree2.go 0 1 2 3 4 5 6 7 8 9 <nil>
9 を取り出したあとチャネルはクローズされるので、チャネルからデータを受信してもゼロ値 (nil) が返ってきます。
もうひとつ、簡単な例を示します。今度はお互いにデータを交換する処理を考えてみましょう。次のリストを見てください。
リスト : データの送受信
package main
import (
"fmt"
"time"
"runtime"
)
// リクエストの定義
type Req struct {
Color string
Reply chan<- int
}
// リクエストの生成
func newReq(color string, ch chan int) *Req {
req := new(Req)
req.Color = color
req.Reply = ch
return req
}
// color の送信
func sendColor(n int, color string, ch chan<- *Req) {
in := make(chan int)
v := newReq(color, in)
for ; n > 0; n-- {
ch <- v
<- in
time.Sleep(100 * time.Millisecond)
}
ch <- nil
}
// color の受信
func receiveColor(n int, ch <-chan *Req) {
for n > 0 {
req := <- ch
if req == nil {
n--
} else {
fmt.Println(req.Color)
req.Reply <- 0
}
}
}
func main() {
runtime.GOMAXPROCS(1)
ch := make(chan *Req)
go sendColor(8, "red", ch)
go sendColor(7, "blue", ch)
go sendColor(6, "green", ch)
receiveColor(3, ch)
}
関数 sendColor はチャネル ch に色データ color を n 回送信します。ch の型は chan<- *Req になります。送信したあと、送信先からのメッセージを受信するまで待ちます。このように返信が必要な場合、メッセージを送信するときに受信用のチャネルもいっしょに送ります。このため、色データと受信用のチャネルを格納する構造体 Req を用意します。Req のチャネル Reply は受け取った側が書き込むチャネルになるので、型を送信専用に設定します。
sendColor は最初に受信用のチャネルを make で生成し、送信するデータを newReq で生成して変数 v にセットします。次の for ループで、v を ch へ送信したあと、in からデータがくるまで待ちます。これを n 回繰り返したあと、最後に nil を送信します。
receiveColor の引数 n は起動した goroutine の個数、ch は goroutine からのデータを受け取るチャネルです。受信データ req が nil ならば、goroutine が一つ終了したので n の値を -1 します。そうでなければ、色データを画面に出力して、チャネル req.Reply へ 0 を送信します。これで相手方に返信することができます。
それでは実行してみましょう。
$ go run sample1205.go green red blue blue green red red blue green green red blue blue green red red blue green red blue red
正常に動作していますね。
複数のチャネルを扱う場合、select を使うとデータの送受信が可能なチャネルを選んで処理することができます。同時に複数のチャネルが処理可能な場合、処理するチャネルはランダムに選択されます。
簡単な例を示しましょう。
リスト : select の使用例 (sample1206.go)
package main
import (
"fmt"
"strconv"
"time"
"runtime"
)
func test1(n int, ch, quit chan<- int) {
for ; n > 0; n-- {
ch <- n
time.Sleep(500 * time.Millisecond)
}
quit <- 0
}
func test2(n int, ch chan<- float64, quit chan<- int) {
for ; n > 0; n-- {
ch <- float64(n) / 10.0
time.Sleep(250 * time.Millisecond)
}
quit <- 0
}
func test3(n int, ch chan<- string, quit chan<- int) {
for ; n > 0; n-- {
ch <- strconv.Itoa(n * 10)
time.Sleep(750 * time.Millisecond)
}
quit <- 0
}
func main() {
runtime.GOMAXPROCS(1)
ch1 := make(chan int)
ch2 := make(chan float64)
ch3 := make(chan string)
quit := make(chan int)
go test1(6, ch1, quit)
go test2(8, ch2, quit)
go test3(4, ch3, quit)
for n := 3; n > 0; {
select {
case c := <- ch1: fmt.Println(c)
case c := <- ch2: fmt.Println(c)
case c := <- ch3: fmt.Println(c)
case <- quit: n--
default:
fmt.Println("None")
time.Sleep(250 * time.Millisecond)
}
}
}
test1 は int を、test2 は float64 を、test3 は string を送信します。そして、どの関数も終了を通知するチャネル quit にデータを送信します。go でそれぞれの関数を実行し、for ループの中の select でチャネルを監視します。どのチャネルも送受信できない場合は default 節が実行されます。ここで None を画面に表示して 250 msec だけ待機します。ch1, ch2, ch3 からデータを受信した場合は、そのデータを画面へ出力します。quit からデータを受信した場合は goroutine が一つ終了したので n の値を -1 します。
それでは実行してみましょう。
$ go run sample1206.go None 40 0.8 6 None 0.7 None 5 0.6 None 30 0.5 None 0.4 4 None 0.3 None 20 0.2 3 None 0.1 None 2 None 10 None 1 None None
正常に動作していますね。
select と time パッケージの関数 After を組み合わせると、待ち時間を設定することができます。
func After(d Duration) <-chan Time
After は一定時間経過したらデータを送信するチャネルを返します。select で After が返すチャネルを監視することで、タイムアウトの処理を行うことができます。
簡単な例を示しましょう。
リスト : タイムアウト処理 (sample1207.go)
package main
import (
"fmt"
"time"
"runtime"
)
func fibo(n int) int {
if n < 2 {
return 1
} else {
return fibo(n - 2) + fibo(n - 1)
}
}
func main(){
runtime.GOMAXPROCS(1)
ch := make(chan int, 5)
for _, n := range []int{41, 41, 39, 35, 36} {
go func(x int){
ch <- fibo(x)
}(n)
}
for i := 5; i > 0; {
select {
case n := <- ch:
fmt.Println(n)
i--
case <- time.After(time.Second):
fmt.Println("Timeout")
i = 0
}
}
}
関数 fibo はフィボナッチ数列を計算します。この関数は二重再帰になっているので、実行時間はとても遅いです。この関数で 41, 41, 39, 35, 36 の値を並行に求めますが、1 秒以内に計算できない場合は実行を中断することにします。
最初の for ループで、匿名関数を go で実行します。go のあと for ループの変数 n はすぐに書き換えられるので、n の値は匿名関数の引数 x に渡していることに注意してください。次の for ループで、データを 5 つ受信します。このとき、select で After が返すチャネルも同時に監視しています。1 秒経過したら、このチャネルからデータを受信するので、Timeout を表示してから for ループを脱出して処理を終了します。
それでは実行してみましょう。
$ go run sample1207.go 14930352 24157817 102334155 Timeout
1 秒以内にすべての値を計算することができませんでした。マルチ CPU で並列に実行すると、タイムアウトせずに実行できるかもしれません。これは並列プログラミングを説明するときに試してみましょう。
今回はここまでです。次回は「哲学者の食事」という並行プログラミングで有名な問題を取り上げる予定です。