GoSudaGo 並行処理スターターパック
概要
簡単な紹介
Go言語には多くの並行性管理のためのツールがあります。このアーティクルでは、その中のいくつかとテクニックを紹介いたします。
ゴルーチン?
goroutineはGo言語でサポートされる新しい形式の並行性モデルです。一般的に、プログラムは同時に複数の作業を行うために、OSからOSスレッドを受け取り、コア数だけ並列に作業を実行します。そして、より小さな単位の並行性を実行するためには、ユーザーランドでグリーンスレッドを生成し、1つのOSスレッド内で複数のグリーンスレッドが交互に作業を実行するようにします。しかし、goroutineの場合は、このような形態のグリーンスレッドをより小さく効率的に作りました。このようなgoroutineはスレッドより少ないメモリを使用し、スレッドより速く生成および切り替えることができます。
goroutineを使用するためには、単にgoキーワードを使用するだけで済みます。これにより、プログラムを作成する過程で直感的に同期コードを非同期コードとして実行できます。
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch := make(chan struct{})
10 go func() {
11 defer close(ch)
12 time.Sleep(1 * time.Second)
13 fmt.Println("Hello, World!")
14 }()
15
16 fmt.Println("Waiting for goroutine...")
17 for range ch {}
18}
このコードは、簡単に1秒間待ってからHello, World!を出力する同期的なコードを、非同期の流れに変更します。現在の例は単純ですが、少し複雑なコードを同期コードから非同期コードへ変更すると、コードの可読性、視認性、理解度が従来のasync awaitやpromiseのような方式よりもさらに向上します。
ただし、多くの場合、このような同期コードを単純に非同期で呼び出す流れや、fork & joinのような流れ（分割統治に似た流れ）を理解しないまま、不適切なgoroutineコードが作られることもあります。このような場合に備えるためのいくつかの方法と技法を、このアーティクルで紹介いたします。
並行性管理
context
最初の管理技法としてcontextが登場するのは意外に思われるかもしれません。しかしGo言語においてcontextは単純なキャンセル機能を超えて、作業ツリー全体を管理するうえで優れた役割を果たします。もしご存じない方のために、簡単にこのパッケージを説明いたします。
1package main
2
3func main() {
4 ctx, cancel := context.WithCancel(context.Background())
5 defer cancel()
6
7 go func() {
8 <-ctx.Done()
9 fmt.Println("Context is done!")
10 }()
11
12 time.Sleep(1 * time.Second)
13
14 cancel()
15
16 time.Sleep(1 * time.Second)
17}
上記のコードはcontextを使用して、1秒後にContext is done!を出力するコードです。contextはDone()メソッドを通じてキャンセルの有無を確認でき、WithCancel、WithTimeout、WithDeadline、WithValueなどのメソッドを通じて多様なキャンセル方法を提供します。
簡単な例を作ってみましょう。もしみなさんが、何らかのデータを取得するためにaggregatorパターンを使用して、user、post、commentを取得するコードを書くと仮定しましょう。そして、すべてのリクエストが2秒以内に行われなければならない場合、次のように記述できます。
1package main
2
3func main() {
4 ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second)
5 defer cancel()
6
7 ch := make(chan struct{})
8 go func() {
9 defer close(ch)
10 user := getUser(ctx)
11 post := getPost(ctx)
12 comment := getComment(ctx)
13
14 fmt.Println(user, post, comment)
15 }()
16
17 select {
18 case <-ctx.Done():
19 fmt.Println("Timeout!")
20 case <-ch:
21 fmt.Println("All data is fetched!")
22 }
23}
上記のコードは、2秒以内にすべてのデータを取得できなければTimeout!を出力し、すべてのデータを取得するとAll data is fetched!を出力します。このようにcontextを使用すると、複数のgoroutineが動作するコードでもキャンセルとタイムアウトを容易に管理できます。
context関連のさまざまな関数とメソッドはgodoc contextで確認できます。簡単なものは学習して、気軽に利用できるようになっていただければと思います。
channel
unbuffered channel
channelはゴルーチン間の通信のためのツールです。channelはmake(chan T)で生成できます。この時、Tは該当channelが伝達するデータの型です。channelは<-でデータを送受信でき、closeでchannelを閉じることができます。
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
上記のコードはchannelを使用して1と2を出力するコードです。このコードでは、単純にchannelに値を送って受け取ることのみを示しています。しかしchannelはこれよりも多くの機能を提供します。まずbuffered channelとunbuffered channelについて見てみましょう。始める前に、上に書かれた例はunbuffered channelであり、チャネルにデータを送る動作とデータを受け取る動作が同時に行われなければなりません。もしこれらの動作が同時に行われなければ、デッドロックが発生する可能性があります。
buffered channel
もし上のコードが単純な出力ではなく、重い作業を行う2つのプロセスだとしたらどうでしょうか？2番目のプロセスが読み取って処理を行う途中で長時間かかると、1番目のプロセスもその間停止してしまうでしょう。私たちはこのような状況を防ぐためにbuffered channelを使用できます。
1package main
2
3func main() {
4 ch := make(chan int, 2)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
上記のコードはbuffered channelを使用して1と2を出力するコードです。このコードではbuffered channelを使用して、チャネルにデータを送る動作とデータを受け取る動作が同時に行われなくてもよいようにしました。このようにチャネルにバッファを設けると、その長さだけ余裕が生まれ、後順位の作業の影響によって生じる作業の遅延を防ぐことができます。
select
複数のチャネルを扱う際、select構文を使用するとfan-in構造を簡単に実装できます。
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch1 := make(chan int, 10)
10 ch2 := make(chan int, 10)
11 ch3 := make(chan int, 10)
12
13 go func() {
14 for {
15 ch1 <- 1
16 time.Sleep(1 * time.Second)
17 }
18 }()
19 go func() {
20 for {
21 ch2 <- 2
22 time.Sleep(2 * time.Second)
23 }
24 }()
25 go func() {
26 for {
27 ch3 <- 3
28 time.Sleep(3 * time.Second)
29 }
30 }()
31
32 for i := 0; i < 3; i++ {
33 select {
34 case v := <-ch1:
35 fmt.Println(v)
36 case v := <-ch2:
37 fmt.Println(v)
38 case v := <-ch3:
39 fmt.Println(v)
40 }
41 }
42}
上記のコードは、周期的に1、2、3を伝達する3つのチャネルを作り、selectを使用してチャネルから値を受け取って出力するコードです。このようにselectを使用すると、複数のチャネルから同時にデータを受け取りながら、チャネルから値を受け取った順に処理できます。
for range
channelはfor rangeを使用して簡単にデータを受け取ることができます。for rangeをチャネルに使用すると、該当チャネルにデータが追加されるたびに動作し、チャネルが閉じられるとループを終了します。
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
上記のコードはchannelを使用して1と2を出力するコードです。このコードではfor rangeを使用して、チャネルにデータが追加されるたびにデータを受け取って出力します。そして、チャネルが閉じられるとループを終了します。
上で何度か書いたとおり、この構文は単純な同期手段としても使用できます。
1package main
2
3func main() {
4 ch := make(chan struct{})
5 go func() {
6 defer close(ch)
7 time.Sleep(1 * time.Second)
8 fmt.Println("Hello, World!")
9 }()
10
11 fmt.Println("Waiting for goroutine...")
12 for range ch {}
13}
上記のコードは、1秒間待ってからHello, World!を出力するコードです。このコードではchannelを使用して同期的なコードを非同期的なコードに変更しました。このようにchannelを使用すると、同期的なコードを非同期的なコードへ容易に変更でき、join地点を設定することができます。
etc
- nil channelにデータを送信または受信すると、無限ループに陥り、デッドロックが発生する可能性があります。
- チャネルを閉じた後にデータを送信すると、panicが発生します。
- チャネルをあえて閉じなくても、GCが回収する際にチャネルを閉じます。
mutex
spinlock
spinlockは反復文を回りながら継続してロックを試みる同期方法です。Go言語ではポインタを使用して簡単にスピンロックを実装してみることができます。
1package spinlock
2
3import (
4 "runtime"
5 "sync/atomic"
6)
7
8type SpinLock struct {
9 lock uintptr
10}
11
12func (s *SpinLock) Lock() {
13 for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) {
14 runtime.Gosched()
15 }
16}
17
18func (s *SpinLock) Unlock() {
19 atomic.StoreUintptr(&s.lock, 0)
20}
21
22func NewSpinLock() *SpinLock {
23 return &SpinLock{}
24}
上記のコードはspinlockパッケージを実装したコードです。このコードではsync/atomicパッケージを使用してSpinLockを実装しました。Lockメソッドではatomic.CompareAndSwapUintptrを使用してロックを試み、Unlockメソッドではatomic.StoreUintptrを使用してロックを解除します。この方式は休まずにロックを試みるため、ロックを取得するまで継続してCPUを使用することになり、無限ループに陥る可能性があります。したがって、spinlockは単純な同期に使用するか、短い時間の間だけ使用するのが良いでしょう。
sync.Mutex
mutexはゴルーチン間の同期のためのツールです。syncパッケージが提供するmutexはLock、Unlock、RLock、RUnlockなどのメソッドを提供します。mutexはsync.Mutexで生成でき、sync.RWMutexを用いて読み取り/書き込みロックを使用することもできます。
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 var mu sync.Mutex
9 var count int
10
11 go func() {
12 mu.Lock()
13 count++
14 mu.Unlock()
15 }()
16
17 mu.Lock()
18 count++
19 mu.Unlock()
20
21 println(count)
22}
上記のコードでは、ほぼ同時に2つのゴルーチンが同じcount変数にアクセスします。この時、mutexを使用してcount変数にアクセスするコードをクリティカルセクションにすると、count変数への同時アクセスを防ぐことができます。そうすると、このコードは何度実行しても同じく2を出力します。
sync.RWMutex
sync.RWMutexは読み取りロックと書き込みロックを区別して使用できるmutexです。RLock、RUnlockメソッドを使用して読み取りロックを取得および解放できます。
1package cmap
2
3import (
4 "sync"
5)
6
7type ConcurrentMap[K comparable, V any] struct {
8 sync.RWMutex
9 data map[K]V
10}
11
12func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) {
13 m.RLock()
14 defer m.RUnlock()
15
16 value, ok := m.data[key]
17 return value, ok
18}
19
20func (m *ConcurrentMap[K, V]) Set(key K, value V) {
21 m.Lock()
22 defer m.Unlock()
23
24 m.data[key] = value
25}
上記のコードはsync.RWMutexを使用してConcurrentMapを実装したコードです。このコードでは、Getメソッドで読み取りロックを取得し、Setメソッドで書き込みロックを取得して、dataマップに安全にアクセスして修正できます。読み取りロックが必要な理由は、単純な読み取り作業が多い場合に、書き込みロックをかけず読み取りロックのみをかけて、複数のゴルーチンが同時に読み取り作業を行えるようにするためです。これにより、状態の変更がないためにわざわざ書き込みロックをかける必要がない場合には、読み取りロックのみをかけて性能を向上させることができます。
fakelock
fakelockはsync.Lockerを実装する簡単なトリックです。この構造体はsync.Mutexと同じメソッドを提供しますが、実際の動作はしません。
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
上記のコードはfakelockパッケージを実装したコードです。このパッケージはsync.Lockerを実装してLock、Unlockメソッドを提供しますが、実際には何の動作もしません。なぜこのようなコードが必要なのかは、機会があれば説明いたします。
waitgroup
sync.WaitGroup
sync.WaitGroupはゴルーチンの作業がすべて終わるまで待つツールです。Add、Done、Waitメソッドを提供し、Addメソッドでゴルーチンの数を追加し、Doneメソッドでゴルーチンの作業が終わったことを知らせます。そしてWaitメソッドで、すべてのゴルーチンの作業が終わるまで待機します。
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
上記のコードはsync.WaitGroupを使用して、100個のゴルーチンが同時にc変数に値を加算するコードです。このコードではsync.WaitGroupを使用して、すべてのゴルーチンが終わるまで待った後、c変数に加算した値を出力します。単純にいくつかの作業をfork & joinする場合はチャネルのみを利用しても十分ですが、多量の作業をfork & joinする場合はsync.WaitGroupを使用することも良い選択肢です。
with slice
スライスと一緒に使われる場合、waitgroupはロックなしで優れた並行実行作業を管理するツールになることができます。
1package main
2
3import (
4 "fmt"
5 "sync"
6 "rand"
7)
8
9func main() {
10 var wg sync.WaitGroup
11 arr := [10]int{}
12
13 for i := 0; i < 10; i++ {
14 wg.Add(1)
15 go func(id int) {
16 defer wg.Done()
17
18 arr[id] = rand.Intn(100)
19 }(i)
20 }
21
22 wg.Wait()
23 fmt.Println("Done")
24
25 for i, v := range arr {
26 fmt.Printf("arr[%d] = %d\n", i, v)
27 }
28}
上記のコードはwaitgroupのみを使用して、各ゴルーチンが同時に10個のランダムな整数を生成し、割り当てられたインデックスに保存するコードです。このコードではwaitgroupを使用して、すべてのゴルーチンが終わるまで待った後、Doneを出力します。このようにwaitgroupを使用すると、複数のゴルーチンが同時に作業を行い、すべてのゴルーチンが終わるまでロックなしでデータを保存し、作業終了後に一括して後処理を行うことができます。
golang.org/x/sync/errgroup.ErrGroup
errgroupはsync.WaitGroupを拡張したパッケージです。errgroupはsync.WaitGroupと異なり、ゴルーチンの作業のうち一つでもエラーが発生すると、すべてのゴルーチンをキャンセルしてエラーを返します。
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/errgroup"
7)
8
9func main() {
10 g, ctx := errgroup.WithContext(context.Background())
11 _ = ctx
12
13 for i := 0; i < 10; i++ {
14 i := i
15 g.Go(func() error {
16 if i == 5 {
17 return fmt.Errorf("error")
18 }
19 return nil
20 })
21 }
22
23 if err := g.Wait(); err != nil {
24 fmt.Println(err)
25 }
26}
上記のコードはerrgroupを使用して10個のゴルーチンを生成し、5番目のゴルーチンでエラーを発生させるコードです。意図的に5番目のゴルーチンでエラーを発生させ、エラーが発生する場合を示しました。ただし実際に使用する際には、errgroupを使用してゴルーチンを生成し、各ゴルーチンでエラーが発生した場合にさまざまな後処理を行う方式で使用すれば良いです。
once
一度だけ実行されるべきコードを実行するツールです。以下のコンストラクタを通じて関連コードを実行できます。
1func OnceFunc(f func()) func()
2func OnceValue[T any](f func() T) func() T
3func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)
OnceFunc
OnceFuncは単純に該当関数が全体にわたってただ一度だけ実行されるようにします。
1package main
2
3import "sync"
4
5func main() {
6 once := sync.OnceFunc(func() {
7 println("Hello, World!")
8 })
9
10 once()
11 once()
12 once()
13 once()
14 once()
15}
上記のコードはsync.OnceFuncを使用してHello, World!を出力するコードです。このコードではsync.OnceFuncを使用してonce関数を生成し、once関数を何度呼び出してもHello, World!は一度だけ出力されます。
OnceValue
OnceValueは単純に該当関数が全体にわたってただ一度だけ実行されるのではなく、該当関数の戻り値を保存して、再び呼び出される際に保存された値を返します。
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValue(func() int {
8 c += 1
9 return c
10 })
11
12 println(once())
13 println(once())
14 println(once())
15 println(once())
16 println(once())
17}
上記のコードはsync.OnceValueを使用してc変数を1ずつ増加させるコードです。このコードではsync.OnceValueを使用してonce関数を生成し、once関数を何度呼び出してもc変数は一度だけ増加した1を返します。
OnceValues
OnceValuesはOnceValueと同様に動作しますが、複数の値を返すことができます。
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValues(func() (int, int) {
8 c += 1
9 return c, c
10 })
11
12 a, b := once()
13 println(a, b)
14 a, b = once()
15 println(a, b)
16 a, b = once()
17 println(a, b)
18 a, b = once()
19 println(a, b)
20 a, b = once()
21 println(a, b)
22}
上記のコードはsync.OnceValuesを使用してc変数を1ずつ増加させるコードです。このコードではsync.OnceValuesを使用してonce関数を生成し、once関数を何度呼び出してもc変数は一度だけ増加した1を返します。
atomic
atomicパッケージは原子的演算を提供するパッケージです。atomicパッケージはAdd、CompareAndSwap、Load、Store、Swapなどのメソッドを提供しますが、最近ではInt64、Uint64、Pointerなどの型の使用が推奨されています。
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
先ほど使用した例です。atomic.Int64型を使用してc変数を原子的に増加させるコードです。AddメソッドとLoadメソッドで変数を原子的に増加させ、変数を読み取ることができます。また、Storeメソッドで値を保存し、Swapメソッドで値を交換し、CompareAndSwapメソッドで値を比較して適合すれば交換することができます。
cond
sync.Cond
condパッケージは条件変数を提供するパッケージです。condパッケージはsync.Condで生成でき、Wait、Signal、Broadcastメソッドを提供します。
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 c := sync.NewCond(&sync.Mutex{})
9 ready := false
10
11 go func() {
12 c.L.Lock()
13 ready = true
14 c.Signal()
15 c.L.Unlock()
16 }()
17
18 c.L.Lock()
19 for !ready {
20 c.Wait()
21 }
22 c.L.Unlock()
23
24 println("Ready!")
25}
上記のコードはsync.Condを使用して、ready変数がtrueになるまで待つコードです。このコードではsync.Condを使用してready変数がtrueになるまで待った後、Ready!を出力します。このようにsync.Condを使用すると、複数のゴルーチンが同時に特定の条件を満たすまで待機させることができます。
これを活用して、簡単なqueueを実装できます。
1package queue
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8type Node[T any] struct {
9 Value T
10 Next *Node[T]
11}
12
13type Queue[T any] struct {
14 sync.Mutex
15 Cond *sync.Cond
16 Head *Node[T]
17 Tail *Node[T]
18 Len int
19}
20
21func New[T any]() *Queue[T] {
22 q := &Queue[T]{}
23 q.Cond = sync.NewCond(&q.Mutex)
24 return q
25}
26
27func (q *Queue[T]) Push(value T) {
28 q.Lock()
29 defer q.Unlock()
30
31 node := &Node[T]{Value: value}
32 if q.Len == 0 {
33 q.Head = node
34 q.Tail = node
35 } else {
36 q.Tail.Next = node
37 q.Tail = node
38 }
39 q.Len++
40 q.Cond.Signal()
41}
42
43func (q *Queue[T]) Pop() T {
44 q.Lock()
45 defer q.Unlock()
46
47 for q.Len == 0 {
48 q.Cond.Wait()
49 }
50
51 node := q.Head
52 q.Head = q.Head.Next
53 q.Len--
54 return node.Value
55}
このようにsync.Condを活用すると、spin-lockで多くのCPU使用量を消費する代わりに、効率的に待機し、条件が満たされると再び動作できます。
semaphore
golang.org/x/sync/semaphore.Semaphore
semaphoreパッケージはセマフォを提供するパッケージです。semaphoreパッケージはgolang.org/x/sync/semaphore.Semaphoreで生成でき、Acquire、Release、TryAcquireメソッドを提供します。
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/semaphore"
7)
8
9func main() {
10 s := semaphore.NewWeighted(1)
11
12 if s.TryAcquire(1) {
13 fmt.Println("Acquired!")
14 } else {
15 fmt.Println("Not Acquired!")
16 }
17
18 s.Release(1)
19}
上記のコードはsemaphoreを使用してセマフォを生成し、Acquireメソッドでセマフォを取得して、Releaseメソッドでセマフォを解放するコードです。このコードではsemaphoreを使用してセマフォを取得して解放する方法を示しました。
おわりに
基本的な内容はここまであれば十分だと思われます。このアーティクルの内容をもとに、みなさんがゴルーチンを使用して並行性を管理する方法を理解し、実際に使用できるようになっていただければ幸いです。このアーティクルがみなさんのお役に立てば嬉しいです。ありがとうございました。