Goのgoroutineで並列処理した結果を順番を維持したまま配列（スライス）に格納する

# 注意
掲載したコードを実行したとき，デッドロックが発生する可能性があります．

## 目標
以下の関数`longProcess`の結果をスライスに格納する．
```go
func longProcess(n int) string {
	rndValue := rand.Intn(1000)
	sleepTime := time.Millisecond * time.Duration(rndValue)
	time.Sleep(sleepTime)
	msg := fmt.Sprintf("%d^2=%d [%dms]", n, n*n, rndValue)
	fmt.Println(msg)
	return msg
}
```
![process.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3644068/18983dc5-53a4-31e3-9130-103c19229d8c.png)

### longProcessの概要
- `longProcess`関数の処理は2乗の値の計算．戻り値として計算式と処理時間を文字列として返す．
- `randValue`として0~999の乱数を取得し，`randValue`[ms]の間sleepすることでランダムな実行時間にする．

例えばこのコードで10回実行して実行順に格納するとき，結果例は以下のようになる．
<details><summary>10回実行するコード例</summary>

```go
package main

import (
	"fmt"
	"math/rand"
	"time"
)

func main() {
	size := 10
	problem := make([]int, size)
	for i := 0; i < size; i++ {
		problem[i] = rand.Intn(100)
	}

	answer := make([]string, size)
	for i, v := range problem {
		answer[i] = longProcess(v)
	}

	for i, v := range answer {
		fmt.Printf("%d: %s\n", i, v)
	}
}

func longProcess(n int) string {
	rndValue := rand.Intn(1000)
	sleepTime := time.Millisecond * time.Duration(rndValue)
	time.Sleep(sleepTime)
	msg := fmt.Sprintf("%d^2=%d [%dms]", n, n*n, rndValue)
	fmt.Println(msg)
	return msg
}
```
</details>

```
0: 5^2=25 [842ms]
1: 68^2=4624 [542ms]
2: 89^2=7921 [383ms]
3: 75^2=5625 [871ms]
4: 28^2=784 [129ms]
5: 57^2=3249 [128ms]
6: 91^2=8281 [370ms]
7: 37^2=1369 [540ms]
8: 74^2=5476 [177ms]
9: 24^2=576 [92ms]
```

この例では`problem`にランダムな問題として0〜99の乱数を格納しておき，`problem`の順に`longProcess`を実行し，その結果をスライス`answer`に格納する．
実行時間は問題数`size`とランダムな処理時間に依存する．実行時間の期待値は
$$\sum_{k=0}^{size-1}\frac{1}{1000}\sum_{l=0}^{1000-1}l \;[ms] \fallingdotseq 500size[ms]$$
なので，`size=10`だと約5秒，`size=100`だと約50秒になり，かなり遅くなる．そこで，並列処理で高速化する．

## 並列化
### goroutineの使用
並列化のためにgoroutine (ゴルーチン)と呼ばれる軽量なスレッドをつくる．実は普段実行しているプログラムもgoroutineの上で動作している（メイン goroutineと呼ばれる）．
新しいgoroutineは関数呼び出しの頭にgoをつけるだけで作ることができる．
```go
for i, v := range problem {
    go longProcess(v)
}
```

### Channel
goroutineは戻り値をもたないので`ans`への代入はできない．そこでgoroutine間の通信にはChannel（チャネル）型を使う．Channnelはスライスのようにして生成する．

```go
ch := make(chan 型)
```
今回は複数（size）個のデータを通信するので，バッファサイズも指定する．
```go
ch := make(chan string, size)
```
送り手はこれ以上の送信する値がないことを示すためチャネルをcloseできる．
```go
close(ch)
```

戻り値をChannelで渡せるようにするために`longProcess`を書き換える．引数に`ch chan string`を追加する．戻り値の型は削除してvoid型とし，かわりに`ch`に代入する．
```go
func longProcess(n int, ch chan string) {
	rndValue := rand.Intn(1000)
	sleepTime := time.Millisecond * time.Duration(rndValue)
	time.Sleep(sleepTime)
	msg := fmt.Sprintf("%d^2=%d [%dms]", n, n*n, rndValue)
	fmt.Println(msg)
	ch <- msg
}
```

`longProcess`関数の変更に合わせて`main`関数も書き換える．
```go
func main() {
	size := 100
	problem := make([]int, size)
	ch := make(chan string, size)
	for i := 0; i < size; i++ {
		problem[i] = rand.Intn(100)
	}

	for _, v := range problem {
		go longProcess(v, ch)
	}
	close(ch)

	for v := range ch {
		fmt.Printf("%s\n", v)
	}
}
```

この状態でコードを実行すると，何も表示されずにプログラムが終了する．これは，メインgoroutineが終了したとき，そこから呼び出されたgoroutineも終了するためだ．

![wait.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3644068/cc7c3b0a-fe76-244c-64da-85c2eefcd320.png)


<details><summary>Channelで値を渡すことができるコード</summary>

```go
package main

import (
	"fmt"
	"math/rand"
	"time"
)

type Result struct {
	index int
	value string
}

func main() {
	size := 100
	problem := make([]int, size)
	ch := make(chan string, size)
	for i := 0; i < size; i++ {
		problem[i] = rand.Intn(100)
	}

	for _, v := range problem {
		go longProcess(v, ch)
	}
	close(ch)

	for v := range ch {
		fmt.Printf("%s\n", v)
	}
}

func longProcess(n int, ch chan string) {
	rndValue := rand.Intn(1000)
	sleepTime := time.Millisecond * time.Duration(rndValue)
	time.Sleep(sleepTime)
	msg := fmt.Sprintf("%d^2=%d [%dms]", n, n*n, rndValue)
	fmt.Println(msg)
	ch <- msg
}
```

</details>

### WaitGroup
`longProcess`が途中終了しないようにするためには，`longProcess`の終了までメインgoroutineを待機させる．処理を待機させるために`sync.WaitGroup`を使う．
```go
var wg sync.WaitGroup
```

WaitGroupの内部実装は以下のようになっている．
```go
type WaitGroup struct {
	noCopy noCopy
	state atomic.Uint64
	sema  uint32
}
```
`sema`がカウンタになっており，待機する処理を表す．カウンタは`Add`で更新する．
```go
wg.Add(size)
```
処理が1つ終わった場合は`Done`によりカウンタを減らす．
```go
wg.Done()
```

`longProcess`でWaitGroupを操作できるようにする．引数としてWaitGroupの参照を受け取り，処理の終了後に`Done()`を呼び出す．
```go
func longProcess(n int, ch chan string, wg *sync.WaitGroup) {
	rndValue := rand.Intn(1000)
	sleepTime := time.Millisecond * time.Duration(rndValue)
	time.Sleep(sleepTime)
	msg := fmt.Sprintf("%d^2=%d [%dms]", n, n*n, rndValue)
	fmt.Println(msg)
	ch <- msg
	wg.Done()
}
```
`longProcess`の呼び出しでは`wg`の参照を渡す．
```go
for _, v := range problem {
    go longProcess(v, ch, &wg)
}
```
メインgoroutineでは，他のgoroutineを待つために`Wait()`を呼び出す．
```go
wg.Wait()
```
ここで，`Wait`の呼び出しは`close`より前にする．そうしないと，メインgoroutineは待機前にChannelをcloseしてしまい，panicが起きる．
```go
for _, v := range problem {
    go longProcess(v, ch, &wg)
}
wg.Wait()
close(ch)
```

これで`longProcess`を並列呼び出しできた．トータルの実行時間は大幅に短縮された．
```
66^2=4356 [1ms]
89^2=7921 [3ms]
58^2=3364 [7ms]
33^2=1089 [9ms]
1^2=1 [28ms]
︙
```

<details><summary>メインgoroutineを待機させることができるコード</summary>

```go
package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

type Result struct {
	index int
	value string
}

func main() {
	size := 100
	problem := make([]int, size)
	ch := make(chan string, size)
	var wg sync.WaitGroup
	wg.Add(size)
	for i := 0; i < size; i++ {
		problem[i] = rand.Intn(100)
	}

	for _, v := range problem {
		go longProcess(v, ch, &wg)
	}
	wg.Wait()
	close(ch)

	for v := range ch {
		fmt.Printf("%s\n", v)
	}
}

func longProcess(n int, ch chan string, wg *sync.WaitGroup) {
	rndValue := rand.Intn(1000)
	sleepTime := time.Millisecond * time.Duration(rndValue)
	time.Sleep(sleepTime)
	msg := fmt.Sprintf("%d^2=%d [%dms]", n, n*n, rndValue)
	fmt.Println(msg)
	ch <- msg
	wg.Done()
}

```

</details>

### 並び替え
プログラムの実行結果右側の処理時間（sleep時間）を見ればわかるが，結果の表示は処理の終了順になっている．本来は`problem`のインデックスと対応した並びになってほしい．処理時間に関係なく順番を維持するために，`longProcess`にスライスを直接渡してその要素を書き換える方法が考えられるが，安全性を考えるとMutexを使う必要がある．そこで，処理後の値を並び替える方法をとる．

インデックスと出力の文字列を保持するために，構造体`Result`を定義する．
```go
type Result struct {
	index int
	value string
}
```

これに合わせてChannnelの宣言を変更する．
```go
ch := make(chan Result, size)
```

`longProcess`の引数も変更する．また，インデックスの値も受け取ることができるようにする．そして，`ch`に`Result`型を生成して渡す．

```go
func longProcess(n int, index int, ch chan Result, wg *sync.WaitGroup) {
	rndValue := rand.Intn(1000)
	sleepTime := time.Millisecond * time.Duration(rndValue)
	time.Sleep(sleepTime)
	msg := fmt.Sprintf("%d^2=%d [%dms]", n, n*n, rndValue)
	fmt.Println(msg)
	ch <- Result{index: index, value: msg}
	wg.Done()
}
```
`longProcess`関数の定義変更に合わせて呼び出し側も変更する．
```go
for i, v := range problem {
    go longProcess(v, i, ch, &wg)
}
```

また，結果の出力を`string`から構造体に変更する．
```go
for v := range ch {
    fmt.Printf("%v\n", v)
}
```
プログラムを変更して実行すると以下のような出力が得られる．
```
{6 53^2=2809 [14ms]}
{76 28^2=784 [16ms]}
{78 36^2=1296 [41ms]}
{9 30^2=900 [50ms]}
{7 35^2=1225 [90ms]}
︙
```

最後に`Result`型の`index`の値を用いて，文字列を`answer`スライスの対応するインデックスに代入する．

```go
answer := make([]string, size)
for v := range ch {
    answer[v.index] = v.value
}
```
`answer`の出力は以下のようにすればできる．
```go
for i, v := range answer {
    fmt.Printf("%d: %s\n", i, v)
}
```

<details><summary>最終的なコード</summary>

```go
package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

type Result struct {
	index int
	value string
}

func main() {
	size := 100
	problem := make([]int, size)
	ch := make(chan Result, size)
	var wg sync.WaitGroup
	wg.Add(size)
	for i := 0; i < size; i++ {
		problem[i] = rand.Intn(100)
	}

	for i, v := range problem {
		go longProcess(v, i, ch, &wg)
	}
	wg.Wait()
	close(ch)

	answer := make([]string, size)
	for v := range ch {
		answer[v.index] = v.value
	}
	for i, v := range answer {
		fmt.Printf("%d: %s\n", i, v)
	}
}

func longProcess(n int, index int, ch chan Result, wg *sync.WaitGroup) {
	rndValue := rand.Intn(1000)
	sleepTime := time.Millisecond * time.Duration(rndValue)
	time.Sleep(sleepTime)
	msg := fmt.Sprintf("%d^2=%d [%dms]", n, n*n, rndValue)
	fmt.Println(msg)
	ch <- Result{index: index, value: msg}
	wg.Done()
}
```
</details>