GoのWorker pool：goroutineの制限、結果の取得、安全なキャンセル

1. 概要

Goのgoroutineは非常に軽量なため、タスクごとにgoを呼び出すことは一般的です。しかし、タスク数が数万に達し、各タスクがデータベースやHTTPを呼び出す場合、このアプローチはconnection poolを枯渇させたり、外部サービスのレート制限を超えたりしやすくなります。

この記事では、worker poolを構築する方法を案内します。これは固定数のgoroutineが1つのchannelからタスクを受け取る仕組みです。この記事を読み終わると、基本的なpoolの書き方、結果とエラーの取得、contextでのキャンセル、errgroup.SetLimitでの簡潔な実装、そしてpoolが本当に並行性を制限しているかをテストする方法を理解できるでしょう。

2. 問題：goroutineの無制限使用

注文リストを処理するコードを考えてみましょう：

for _, o := range orders {
	go chargeOrder(o) // 50,000個の注文 -> 50,000個のgoroutineが同時に決済APIを呼び出す
}

各goroutineは数KB程度のメモリしか使わないため、問題は通常RAMにはありません。問題はバックエンドのリソースにあります：

• データベースのconnection poolは通常20～50個の接続しかない。
• 決済ゲートウェイは1秒あたりのリクエスト数を制限している。
• プロセスのファイルディスクリプタ数に制限がある。

Worker poolはこの問題を2つの概念を分離することで解決します。タスク数は非常に大きくなることができますが、同時に実行するタスク数はworkerの数によって制限されます。

3. 基本的なWorker pool

このモデルは3つの部分から構成されています：

1. 実行するタスクを保持するjobs channel。
2. N個のworker goroutine。各workerはchannelに対してfor rangeでループを回します。
3. すべてのworkerの終了を待つためのsync.WaitGroup。

package main

import (
	"fmt"
	"sync"
	"time"
)

type Order struct {
	ID     int
	Amount float64
}

func worker(id int, orders <-chan Order, wg *sync.WaitGroup) {
	defer wg.Done()
	for o := range orders {
		time.Sleep(100 * time.Millisecond) // 決済APIの呼び出しをシミュレート
		fmt.Printf("worker %d が注文 %d (%.0f) を処理しました\n", id, o.ID, o.Amount)
	}
}

func main() {
	orders := make(chan Order)
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go worker(i, orders, &wg)
	}

	start := time.Now()
	for i := 1; i <= 9; i++ {
		orders <- Order{ID: i, Amount: float64(i) * 10}
	}
	close(orders)
	wg.Wait()
	fmt.Println("完了:", time.Since(start).Round(100*time.Millisecond))
}

このコードの動作は以下の通りです：

• orders <-chan Order はworker側で受信のみするchannelです。workerが誤って送信またはchannelを閉じようとした場合、compilerはそれを阻止します。
• for o := range orders はchannelがcloseされてすべての要素が取得されると自動的に終了します。
• close(orders) は「仕事終了」の信号です。この行を忘れると、workerは永遠に待機し、wg.Wait()がdeadlockを引き起こします。
• wg.Add(1) はgo worker(...)の前に呼び出す必要があり、goroutine内では呼び出してはいけません。

実行結果（行の順序は異なる場合があります）：

worker 3 が注文 1 (10) を処理しました
worker 1 が注文 2 (20) を処理しました
worker 2 が注文 3 (30) を処理しました
worker 2 が注文 4 (40) を処理しました
...
完了: 300ms

9個の注文を3個のworkerで処理し、各注文に100msかかるため、総処理時間は順序実行の900msではなく約300msになります。

注意： Go 1.25から、wg.Go(func() { worker(i, orders) })と書くことができます。この関数は自動的にAddとDoneを呼び出すため、どちらかを忘れるエラーを回避できます。

4. 結果とエラーの取得

実際には、どの注文が成功し、どの注文がエラーになったかを知る必要があります。このためにresults channelを追加します：

package main

import (
	"errors"
	"fmt"
	"sync"
)

type Order struct {
	ID     int
	Amount float64
}

type Result struct {
	OrderID int
	Fee     float64
	Err     error
}

func chargeOrder(o Order) (float64, error) {
	if o.Amount <= 0 {
		return 0, errors.New("不正な金額")
	}
	return o.Amount * 0.02, nil // 2%の手数料
}

func main() {
	orders := []Order{{1, 100}, {2, 0}, {3, 250}, {4, 80}}

	jobs := make(chan Order)
	results := make(chan Result)
	var wg sync.WaitGroup

	for range 3 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for o := range jobs {
				fee, err := chargeOrder(o)
				results <- Result{OrderID: o.ID, Fee: fee, Err: err}
			}
		}()
	}

	go func() {
		for _, o := range orders {
			jobs <- o
		}
		close(jobs)
	}()

	go func() {
		wg.Wait()
		close(results)
	}()

	for r := range results {
		if r.Err != nil {
			fmt.Printf("注文 %d エラー: %v\n", r.OrderID, r.Err)
			continue
		}
		fmt.Printf("注文 %d 手数料 %.2f\n", r.OrderID, r.Fee)
	}
}

出力は以下のようになります：

注文 1 手数料 2.00
注文 3 手数料 5.00
注文 2 エラー: 不正な金額
注文 4 手数料 1.60

注意すべき3つのポイントがあります：

• jobの送信は別のgoroutineで実行されます。mainが同じgoroutine上でjobを送信し結果を読み込むと、workerはresults <-で、mainはjobs <-でそれぞれブロックされ、deadlockが発生します。
• 3番目のgoroutineはwg.Wait()を待ってからclose(results)します。すべてのworkerが終了するまで、誰もresultsに送信する可能性があるため、その時点でのみchannelを安全に閉じられます。一般的なルール：channelを閉じるのは送信側だけです。
• エラーはResultにラップされ、pool全体を停止させません。1つの注文のエラーが他の注文をブロックしてはいけません。

結果の順序は入力の順序と一致しません。順序を保持する必要がある場合は、jobに添付されたインデックス（index）を持ち運ぶか、セクション6の方法を使用してください。

5. contextによるpoolのキャンセル

HTTPリクエストがキャンセルされたりタイムアウトしたりした場合、poolは決済APIの呼び出しを続けるのではなく停止する必要があります。これはworkerにctx.Done()をリッスンさせることで実現します：

func worker(ctx context.Context, jobs <-chan Order, results chan<- Result) {
	for {
		select {
		case <-ctx.Done():
			return
		case o, ok := <-jobs:
			if !ok {
				return
			}
			fee, err := chargeOrder(o)
			select {
			case results <- Result{OrderID: o.ID, Fee: fee, Err: err}:
			case <-ctx.Done():
				return
			}
		}
	}
}

producerも同様にselectを使用する必要があります：

go func() {
	defer close(jobs)
	for _, o := range orders {
		select {
		case jobs <- o:
		case <-ctx.Done():
			return
		}
	}
}()

そして、呼び出し元でタイムアウト付きのcontextを作成します：

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

両方向での送信でselectが必要な理由：workerがjobを受け取るときだけctxをチェックする場合、読み込み側が去った後、results <-で永遠に立ち往生する可能性があります。これはgoroutine leak（ゴルーチンリーク）の非常に一般的な原因です。

注意： selectはworkerがjob間で終了するのを助けるだけです。chargeOrderがHTTPを呼び出す場合、ctxをそれに渡す必要があります（http.NewRequestWithContext）。実行中のリクエスト全体をキャンセルできます。

6. errgroup.SetLimitで簡潔に

手動でchannel、WaitGroup、contextを書くと誤りやすいです。errgroupパッケージにはSetLimitが組み込まれており、わずか数行で制限付きのpoolを作成できます。まず、依存関係を追加します：

go get golang.org/x/sync/errgroup

以下のgeneric関数は、最大limit個のgoroutineでsliceを処理し、結果の順序を保持し、最初のエラーで停止します：

package pool

import (
	"context"

	"golang.org/x/sync/errgroup"
)

func Process[T, R any](
	ctx context.Context,
	limit int,
	items []T,
	fn func(context.Context, T) (R, error),
) ([]R, error) {
	results := make([]R, len(items))
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(limit)

	for i, item := range items {
		g.Go(func() error {
			r, err := fn(ctx, item)
			if err != nil {
				return err
			}
			results[i] = r
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

このコードの動作は以下の通りです：

• g.SetLimit(limit)により、既にlimit個のgoroutineが実行中の場合、g.Goはブロックされます。効果的には、これはworker poolと同等です。
• errgroup.WithContextが返すctxは、任意の関数がエラーを返すと即座にキャンセルされます。ctxを使用している他のタスクはキャンセル信号を受け取ります。
• results[i] = rは各goroutineが固有のインデックスに書き込むため、並行実行時に安全です。この場合、mutexは不要です。
• Go 1.22から、ループ変数iとitemは各ループイテレーションで再作成されるため、closureは以前のバージョンのように共有変数を使用しなくなりました。

呼び出し方法：

fees, err := pool.Process(ctx, 5, orders, func(ctx context.Context, o Order) (float64, error) {
	return paymentService.Charge(ctx, o)
})

自分で書いたpoolとの違い：errgroupはN個のworkerを再利用するのではなく、各itemに対して新しいgoroutineを作成します。ほとんどのアプリケーションでは、この違いは無視できます。workerが独自の状態を維持する必要がある場合、たとえば各workerが1つのconnectionを持つ場合は、セクション3～5で自分で書いたpoolを使用してください。

7. Worker poolのテスト

2つの主要な動作をテストする必要があります。1つ目、poolは決してlimitを超えない。2つ目、エラーが正しく返される。並行性レベルを測定するためにatomic.Int32を使用します：

package pool

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func TestProcess_LimitsConcurrencyAndKeepsOrder(t *testing.T) {
	var running, maxSeen atomic.Int32
	items := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

	got, err := Process(context.Background(), 3, items, func(_ context.Context, n int) (int, error) {
		cur := running.Add(1)
		defer running.Add(-1)
		for {
			m := maxSeen.Load()
			if cur <= m || maxSeen.CompareAndSwap(m, cur) {
				break
			}
		}
		time.Sleep(10 * time.Millisecond)
		return n * 2, nil
	})

	if err != nil {
		t.Fatalf("エラーは予期されていません: %v", err)
	}
	if maxSeen.Load() > 3 {
		t.Errorf("並行実行数 %d、制限は 3", maxSeen.Load())
	}
	want := []int{2, 4, 6, 8, 10, 12, 14, 16, 18, 20}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("got[%d] = %d, want %d", i, got[i], want[i])
		}
	}
}

func TestProcess_ReturnsFirstError(t *testing.T) {
	errDeclined := errors.New("カードが拒否されました")

	_, err := Process(context.Background(), 2, []int{1, 2, 3}, func(_ context.Context, n int) (int, error) {
		if n == 2 {
			return 0, errDeclined
		}
		return n, nil
	})

	if !errors.Is(err, errDeclined) {
		t.Fatalf("errDeclinedを期待していますが、%vを受け取りました", err)
	}
}

race detectorを使用してテストを実行します：

go test -race ./...

ok  	example.com/shop/pool	0.061s

CompareAndSwapループはロックなしにmaxSeenを更新します。-raceフラグは並行コードに必須です。共有データアクセスを検出し、目視では見落としやすい競合状態を発見します。

8. よくあるエラーと使用すべきでない場合

よくあるエラー：

• close(jobs)を忘れる：workerは決して終了せず、プログラムがfatal error: all goroutines are asleep - deadlock!を報告します。
• 受信側でchannelを閉じる、または2回閉じる：panic: close of closed channelを引き起こします。
• goroutine内でwg.Addを呼び出す：WaitはgoroutineがAddを実行する前に戻る可能性があります。
• 同じgoroutineでjobを送信し結果を読み込む（非バッファchannelの場合）：セクション4で分析したようにdeadlockが発生します。
• 結果を送信するときにctx.Done()をselectしない：読み込み側が早期に終了した場合、goroutine leakが発生します。
• 任意にworkerの数を選択する：CPU-boundなタスクの場合はruntime.NumCPU()を使用します。I/O-boundなタスクの場合は、バックエンドリソースの制限、たとえばconnection poolのサイズに基づいてください。

Worker poolを使用すべきでない場合：

• タスク数が少なく事前に知られている場合。例えば3つのサービスを並行して呼び出す場合、errgroupでSetLimitなしで十分です。
• タスクが非常に短い場合、わずか数マイクロ秒。channelを通る費用は仕事自体より大きい可能性があるため、順序実行またはバッチ処理がより効率的です。
• 並行数ではなく速度（request/秒）で制限する必要がある場合。この場合、レート制限が必要です。例えばgolang.org/x/time/rateです。worker poolはレート制限を置き換えることはできません。

9. まとめ

Worker poolはGoで並行性レベルを制御するのに役立ちます。タスク数は非常に大きくなる可能性がありますが、バックエンドリソースは保護されます。この記事では以下のステップを段階的に説明しました：

• channelとsync.WaitGroupを使用した基本的なpool。
• results channelを通じた結果とエラーの取得、正しい場所でのchannelクローズ。
• 両方向の送信でのcontextによる安全なキャンセル。
• generic関数Processを使用したerrgroup.SetLimitによる簡潔な実装。
• go test -raceを使用した並行性制限とエラーのテスト。

ほとんどの場合、errgroup.SetLimitが最もシンプルで誤りの少ない選択肢です。workerが独自の状態を維持する必要がある場合、またはjobと結果の流れを細かく制御する必要がある場合にのみ、poolを自分で書くべきです。