ヒヨコとはじめる、やさしい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。
ねじねじおでした。