Day 033 — errgroup パッケージ（SetLimitによる同時実行数制御）

📚 背景知識（読んでから問題へ）

Day 032ではerrgroup.Groupを使い、複数goroutineの「エラー集約」と「早期打ち切り」を同時に実現しました。しかし、そこで書いたfetchAllにはもう1つの弱点が残っています。idsが5件ではなく1000件だったらどうなるでしょうか——forループが1000回まわり、g.Goが1000個のgoroutineを一気に起動してしまいます。

実務で外部APIを呼ぶ場合、多くのサービスにはレートリミット（例: 同時10リクエストまで）が設定されています。1000件を一度に投げつければ、レートリミットエラーの連発か、最悪サービス側から一時的にブロックされます。この「同時に走らせてよい数の上限」を制御する仕組みがerrgroup.Group.SetLimit(n)です。

SetLimit(n)を呼ぶと、そのグループで同時にアクティブになれるgoroutineの数が最大nに制限されます。nを超えるタイミングで呼ばれたg.Go(f)は、内部でブロックし、既存のgoroutineのどれかが終了して枠が空くまで新しいgoroutineを起動しません。つまりg.Goは「必ず即座にgoroutineを起動する」わけではなく、「上限内に収まるまで待ってから起動する」という、Worker Poolに似た挙動に変わります。

重要なのは、SetLimitが実現しているのは同時実行数の制御だけだという点です。エラー集約・早期打ち切り（ctxのキャンセル伝播）は、Day 032で学んだerrgroup本来の機能がそのまま生きています。つまりerrgroupは「待ち合わせ」「エラー集約」「キャンセル伝播」「同時実行数制御」という4つの関心事を、1つの小さな型に段階的に積み上げて提供しています。

📝 問題

Day 032のfetchAllを拡張し、以下の要件を満たすfetchAllLimitedを実装してください。

要件

1. IDのリスト（100件程度を想定）を受け取り、fetchPrice(ctx, id)を並列に呼び出して結果を集める
2. errgroup.Group.SetLimit(limit)を使い、同時にアクティブなgoroutine数がlimitを超えないようにすること
3. Day 032と同様、どれか1つが失敗したら残りは早期打ち切りされること（ctxのキャンセル伝播を維持する）
4. 結果はID順のスライスとして返す（インデックス固定書き込みで、Day 032同様データ競合がないこと）
5. 同時実行数が実際にlimitを超えていないことを、コード自身で検証できるようにすること（atomicパッケージでアクティブ数をカウントし、最大値を記録・出力する）

制約・ヒント条件

• limitに0を渡すとどうなるか、その挙動をコメントで説明すること（実際に試す必要はなく、ドキュメントの記述から挙動を導く）
• SetLimitを呼ぶタイミング（g.Goを1回でも呼んだ後にSetLimitを呼ぶとどうなるか）についても、模範解答のコード上で正しい順序を守ること

🔍 ヒント（段階的開示）

📋 コピーg, ctx := errgroup.WithContext(context.Background())
g.SetLimit(limit) // g.Go を呼ぶ前に必ず設定する

var active atomic.Int32
var maxActive atomic.Int32

results := make([]fetchResult, len(ids))
for i, id := range ids {
	i, id := i, id
	g.Go(func() error {
		cur := active.Add(1)
		defer active.Add(-1)

		// 最大値の更新（CompareAndSwapで競合なく更新する）
		for {
			m := maxActive.Load()
			if cur <= m || maxActive.CompareAndSwap(m, cur) {
				break
			}
		}

		r, err := fetchPrice(ctx, id)
		if err != nil {
			return err
		}
		results[i] = r
		return nil
	})
}

err := g.Wait()
fmt.Println("max concurrent:", maxActive.Load()) // limit以下であることを確認

✅ 模範解答

ファイル構成: day033/go.mod + day033/main.go

📋 コピー// go.mod
module example.com/day033

go 1.22

require golang.org/x/sync v0.8.0

📋 コピーpackage main

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"

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

type fetchResult struct {
	name  string
	price int
}

// fetchPrice は外部APIをシミュレートする。idが13の倍数の場合だけ失敗させる。
// 待機はselectで実装し、ctxのキャンセルで即座に打ち切れるようにしている。
func fetchPrice(ctx context.Context, id int) (fetchResult, error) {
	select {
	case <-time.After(20 * time.Millisecond):
	case <-ctx.Done():
		return fetchResult{}, ctx.Err()
	}
	if id%13 == 0 {
		return fetchResult{}, fmt.Errorf("item %d: fetch failed", id)
	}
	return fetchResult{name: fmt.Sprintf("item-%d", id), price: id * 10}, nil
}

// fetchAllLimited はerrgroup.SetLimitで同時実行数をlimitに制限しつつ、
// エラー集約・早期打ち切り・インデックス固定書き込みによる競合回避を両立する。
//
// 補足: limitに0を渡すと「同時にアクティブになれるgoroutineが0個」という制約になり、
// g.Goがいつまでも枠を待ち続けて全体がデッドロックする(公式ドキュメントで明示されている挙動)。
// limitに負値を渡した場合は「無制限」を意味する(SetLimitを呼ばない場合と同じ)。
func fetchAllLimited(ids []int, limit int) ([]fetchResult, int32, error) {
	g, ctx := errgroup.WithContext(context.Background())
	g.SetLimit(limit) // g.Goを1回でも呼ぶ前に設定する(呼んだ後の変更は未定義動作)

	var active atomic.Int32
	var maxActive atomic.Int32
	results := make([]fetchResult, len(ids))

	for i, id := range ids {
		i, id := i, id
		g.Go(func() error {
			cur := active.Add(1)
			defer active.Add(-1)

			for {
				m := maxActive.Load()
				if cur <= m || maxActive.CompareAndSwap(m, cur) {
					break
				}
			}

			r, err := fetchPrice(ctx, id)
			if err != nil {
				return fmt.Errorf("id=%d: %w", id, err)
			}
			results[i] = r
			return nil
		})
	}

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

func main() {
	ids := make([]int, 100)
	for i := range ids {
		ids[i] = i + 1
	}

	const limit = 10
	start := time.Now()
	results, maxConcurrent, err := fetchAllLimited(ids, limit)
	elapsed := time.Since(start)

	fmt.Printf("limit=%d, max observed concurrency=%d, elapsed=%s\n", limit, maxConcurrent, elapsed)
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Printf("fetched %d results, first=%v, last=%v\n", len(results), results[0], results[len(results)-1])

	if maxConcurrent > limit {
		panic("BUG: concurrency limit was violated")
	}
}

mkdir day033 && cd day033
go mod init example.com/day033
go get golang.org/x/sync@v0.8.0
# main.go を上記の内容で保存
go run -race .

id=13, 26, 39, ... のいずれかで失敗するためerror: id=13: item 13: fetch failedのようなエラーが出力され、max observed concurrencyがlimit=10を超えていないことが確認できます。

🪜 Step-by-Step 解説

💡 設計思想・なぜこう書くのか

🌐 他言語との比較

🏆 実務での使いどころ

• レートリミットのある外部APIへの一括問い合わせ: 決済代行会社・SMS送信APIなど、同時接続数やRPS上限があるサービスへの一括処理で、上限を超えないように制御しつつ並列化の恩恵も受ける
• DBコネクションプールを使い切らないバッチ処理: 数千件のレコードを並列更新する際、コネクションプールのサイズを超えてgoroutineを起動すると接続待ちで詰まるため、SetLimitをプールサイズ以下に設定する
• 大量ファイルの並列変換・アップロード処理: S3などへの並列アップロードで、帯域やAPIのスロットリングを考慮し、同時アップロード数を制御する
• クローラー・スクレイパーの並列度制御: 対象サイトに過度な負荷をかけないよう、同時リクエスト数を制限しながら並列クロールする

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: fetchPriceが呼ばれるたびにHTTPリクエストのタイムアウトを個別に設定したい場合、context.WithTimeout(ctx, ...)を各goroutine内でさらに派生させる設計を試してみましょう（親のctxのキャンセルと子のctxのタイムアウトがどう合成されるかを確認する）
• 次回予告: race condition検出（go test -race）— これまで手動でデータ競合を見つけてきましたが、Day 034では体系的に「どんなコードパターンが-raceで検出されるか」を網羅的に学びます

🎯 自己評価

自分の回答

気づき・メモ