Day 029 — 並行処理パターン①（Worker Pool）

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

Worker Poolは「固定数のgoroutineでジョブを処理し、無制限にgoroutineを増やさない」ための並行処理パターンです。Phase 2で学んだgoroutine・channel・WaitGroupの基本要素を、実務で最も頻繁に登場する形に組み合わせたものだと言えます。

典型的な構成要素は3つです。

📋 コピーjobs := make(chan Job, N)      // 仕事を流し込むchannel
results := make(chan Result, N) // 結果を受け取るchannel
var wg sync.WaitGroup           // 全workerの完了を待つ

• なぜgoroutineを無制限に増やさないのか: 1リクエストごとにgo handle(req)していると、リクエストが殺到した瞬間にgoroutine数が青天井に増え、メモリとスケジューラの両方を圧迫します。Worker Poolは「同時に処理するジョブ数の上限」をnumWorkersという1つの数値でコード上に明示する設計です。
• channelの所有権（ownership）: Goのchannelには「誰が閉じるか」という暗黙のルールがあります。送信する側（sender）だけがcloseしてよく、受信する側（receiver）は絶対にcloseしないという原則です。複数のgoroutineが同じchannelに書き込んでいる場合、closeしてよいのは「全員が書き込み終わったことを検知できる立場」の1箇所だけです。今日の問題は、このownershipの原則が破られたときに何が起きるかを、実際のバグとして体験します。

Worker Poolの正しい終了シーケンスは次の順序です。

1. ジョブ送信側が全ジョブを送り終えたらjobsをcloseする
2. 各workerはfor job := range jobsでジョブを受け取り続け、jobsがcloseされて空になったらループを抜けてwg.Done()
3. wg.Wait()を待つ専用のgoroutineが全worker完了を検知したらresultsをclose
4. メインgoroutineはresultsをfor rangeで読み切る（③のcloseによってループが自然に終了する）

このうち1つでも順序を間違えると、デッドロック・パニック・結果の欠落のいずれかが起こります。

📝 問題

以下は「複数の注文の消費税込み金額をWorker Poolで並行計算する」ことを意図したコードです。一見動きそうに見えますが、実行すると必ずfatal error: all goroutines are asleep - deadlock!で異常終了します。

📋 コピーpackage main

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

type Order struct {
	ID        int
	AmountYen int
}

type Result struct {
	OrderID  int
	TaxedYen int
}

func worker(id int, jobs <-chan Order, results chan<- Result, wg *sync.WaitGroup) {
	for job := range jobs {
		time.Sleep(10 * time.Millisecond) // 税計算などの重い処理を模擬
		results <- Result{OrderID: job.ID, TaxedYen: int(float64(job.AmountYen) * 1.1)}
	}
	wg.Done()
	close(results)
}

func main() {
	orders := []Order{
		{ID: 1, AmountYen: 1000},
		{ID: 2, AmountYen: 2000},
		{ID: 3, AmountYen: 3000},
		{ID: 4, AmountYen: 4000},
		{ID: 5, AmountYen: 5000},
	}

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

	const numWorkers = 3
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go worker(i, jobs, results, &wg)
	}

	for _, o := range orders {
		jobs <- o
	}

	wg.Wait()

	for r := range results {
		fmt.Printf("order=%d taxed=%d\n", r.OrderID, r.TaxedYen)
	}
}

課題

1. このコードには独立した欠陥が3つ含まれています。それぞれを指摘し、「なぜその欠陥が問題を引き起こすのか」を実行時に何が起きるかのレベルで説明してください（単に「ここがおかしい」だけでなく、goroutineが何を待って止まっているかまで言葉にすること）。
2. 3つの欠陥をすべて修正した完全なコードを書いてください。runWorkerPool(orders []Order, numWorkers int) []Resultという関数に処理をまとめ、mainから呼び出す形にリファクタリングして構いません。
3. 修正版に対して、以下を検証するテーブル駆動テストを書いてください。
   • 5件の注文を渡すと、5件すべての結果が欠落なく返る（受け取った件数が注文件数と一致する。Worker Poolは並行実行のため到着順は保証されないので、OrderIDでソートしてから期待値と比較すること）
   • 注文が0件のときは空の結果が返る（パニックしない）
   • numWorkersが注文件数より多い場合でも正しく全件処理される
4. go test -raceで検証し、データ競合が検出されないことを確認してください。

go run . / go test -race -v ./...でそのまま動作する完全なコードとして提示すること。

🔍 ヒント（段階的開示）

📋 コピーfunc runWorkerPool(orders []Order, numWorkers int) []Result {
	jobs := make(chan Order, len(orders))
	results := make(chan Result, len(orders))
	var wg sync.WaitGroup

	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go worker(i, jobs, results, &wg) // worker側は defer wg.Done() のみ。close(results)はしない
	}

	for _, o := range orders {
		jobs <- o
	}
	close(jobs) // 欠陥①の修正: 送信側が送り終えたらjobsをclose

	go func() {
		wg.Wait()
		close(results) // 欠陥②の修正: closeするのはここ1箇所だけ
	}()

	var out []Result
	for r := range results { // 欠陥③の修正: wg.Wait()を待たずに即座に受信開始
		out = append(out, r)
	}
	return out
}

✅ 模範解答

ファイル: main.go

📋 コピーpackage main

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

// Order は消費税計算対象の注文。
type Order struct {
	ID        int
	AmountYen int
}

// Result はWorker Poolによる計算結果。
type Result struct {
	OrderID  int
	TaxedYen int
}

// worker はjobsを受け取り続け、closeされて空になったらdefer wg.Done()で完了を通知する。
// results のcloseはここでは行わない（closeの責務は呼び出し元の1箇所に集約する）。
func worker(id int, jobs <-chan Order, results chan<- Result, wg *sync.WaitGroup) {
	defer wg.Done()
	for job := range jobs {
		time.Sleep(10 * time.Millisecond) // 税計算などの重い処理を模擬
		results <- Result{OrderID: job.ID, TaxedYen: int(float64(job.AmountYen) * 1.1)}
	}
}

// runWorkerPool はnumWorkers個のworkerでordersを並行処理し、全結果を返す。
// 結果の順序は並行実行のため保証されない（呼び出し側がソート等で必要に応じて整える）。
func runWorkerPool(orders []Order, numWorkers int) []Result {
	jobs := make(chan Order, len(orders))
	results := make(chan Result, len(orders))
	var wg sync.WaitGroup

	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go worker(i, jobs, results, &wg)
	}

	for _, o := range orders {
		jobs <- o
	}
	close(jobs) // 送信側（唯一の送り手であるmain）がjobsを閉じる

	// wg.Wait()は専用goroutineで行う。メインgoroutineをここでブロックすると、
	// resultsの受信が始まらずworkerが送信できずデッドロックする。
	go func() {
		wg.Wait()
		close(results) // 全worker完了を検知できる唯一の場所でresultsを閉じる
	}()

	var out []Result
	for r := range results {
		out = append(out, r)
	}
	return out
}

func main() {
	orders := []Order{
		{ID: 1, AmountYen: 1000},
		{ID: 2, AmountYen: 2000},
		{ID: 3, AmountYen: 3000},
		{ID: 4, AmountYen: 4000},
		{ID: 5, AmountYen: 5000},
	}
	results := runWorkerPool(orders, 3)
	for _, r := range results {
		fmt.Printf("order=%d taxed=%d\n", r.OrderID, r.TaxedYen)
	}
}

ファイル: main_test.go

📋 コピーpackage main

import (
	"sort"
	"testing"
)

func TestRunWorkerPool(t *testing.T) {
	orders := []Order{
		{ID: 1, AmountYen: 1000},
		{ID: 2, AmountYen: 2000},
		{ID: 3, AmountYen: 3000},
		{ID: 4, AmountYen: 4000},
		{ID: 5, AmountYen: 5000},
	}

	results := runWorkerPool(orders, 3)

	if len(results) != len(orders) {
		t.Fatalf("len(results) = %d, want %d", len(results), len(orders))
	}

	// Worker Poolは並行実行のため到着順は保証されない。OrderIDでソートしてから比較する。
	sort.Slice(results, func(i, j int) bool { return results[i].OrderID < results[j].OrderID })

	for i, r := range results {
		want := int(float64(orders[i].AmountYen) * 1.1)
		if r.OrderID != orders[i].ID {
			t.Fatalf("results[%d].OrderID = %d, want %d", i, r.OrderID, orders[i].ID)
		}
		if r.TaxedYen != want {
			t.Fatalf("results[%d].TaxedYen = %d, want %d", i, r.TaxedYen, want)
		}
	}
}

func TestRunWorkerPool_EmptyOrders(t *testing.T) {
	results := runWorkerPool(nil, 3)
	if len(results) != 0 {
		t.Fatalf("len(results) = %d, want 0", len(results))
	}
}

func TestRunWorkerPool_MoreWorkersThanJobs(t *testing.T) {
	orders := []Order{{ID: 1, AmountYen: 100}}
	results := runWorkerPool(orders, 10)
	if len(results) != 1 {
		t.Fatalf("len(results) = %d, want 1", len(results))
	}
}

$ go vet ./... && go build ./...
（エラーなし）

$ go test -race -v ./...
=== RUN   TestRunWorkerPool
--- PASS: TestRunWorkerPool (0.02s)
=== RUN   TestRunWorkerPool_EmptyOrders
--- PASS: TestRunWorkerPool_EmptyOrders (0.00s)
=== RUN   TestRunWorkerPool_MoreWorkersThanJobs
--- PASS: TestRunWorkerPool_MoreWorkersThanJobs (0.01s)
PASS
ok  	example.com/day029	2.024s

$ go run .
order=2 taxed=2200
order=1 taxed=1100
order=3 taxed=3300
order=5 taxed=5500
order=4 taxed=4400

$ go run . # バグ入りコード（欠陥③が残った状態）
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [sync.WaitGroup.Wait]:
sync.(*WaitGroup).Wait(...)
main.main()
	.../main.go:51 +0x1dd

goroutine 18 [chan send]:
main.worker(...)
	.../main.go:22 +0x85
（goroutine 19, 20も同様に chan send で停止）
exit status 2

go.mod・main.go・main_test.goを1モジュール（module example.com/day029）として配置し、go vet / go build / go test -race -v ./... / go run .で実際に検証済みです。

🪜 Step-by-Step 解説

📋 コピーgoroutine 1 [sync.WaitGroup.Wait]:   // メインgoroutineがwg.Wait()で停止
goroutine 18 [chan send]:            // workerがresults<-で送信をブロックされている

📋 コピーgo func() {
	wg.Wait()
	close(results)
}()

for r := range results { // メインは即座にここへ来る

📋 コピーclose(jobs)    // mainが送信し終えた直後（送信側の責務）
close(results) // wg.Wait()の直後（全workerの完了を知っている唯一の場所の責務）

📋 コピーfunc worker(id int, jobs <-chan Order, results chan<- Result, wg *sync.WaitGroup) {
	defer wg.Done()
	for job := range jobs { ... }
}

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

🌐 他言語との比較

Goが「自分でchannelのcloseを設計させる」のは不便に見えますが、その分「誰が・いつ・何をcloseするか」を完全にコードで説明できる透明性があります。Java/Pythonのスレッドプールはこの部分をランタイム側が隠蔽してくれる代わりに、内部で何が起きているかをGoほど細かくは制御できません。

🏆 実務での使いどころ

• バッチ処理・大量データの並行変換: CSVの数万行を並行してパース・変換する処理はWorker Poolの典型的な使いどころです
• 外部API呼び出しの並行化: Day 028で学んだHTTPStockCheckerのようなHTTP呼び出しを、複数のSKUに対して同時に（かつnumWorkersで上限を切って）叩く場合にそのまま応用できます
• 画像・動画のリサイズ/エンコードパイプライン: CPU負荷の高い処理を「CPUコア数程度」のworkerに制限しながら並行実行する場面で使われます
• Fan-in/Fan-out・Pipelineパターンへの発展: Day 030以降で扱う予定のFan-in/Fan-outやPipelineパターンは、今日のWorker Poolを複数段つなげたものとして理解できます

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: resultsチャネルのバッファサイズを0（今回のバグ版と同じ）に戻した状態で修正版のロジック（wg.Wait()を専用goroutineに退避する形）を使うとどうなるか、実際に試してみましょう。バッファの有無が処理の速度やブロックの仕方にどう影響するかが体感できます
• 次回予告: Day 030 — 並行処理パターン②（Fan-in/Fan-out）（実装）。今日のWorker Poolを土台に、複数のソースからの結果を1つのchannelに集約する「Fan-in」と、1つの入力を複数の処理系統に分配する「Fan-out」を実装します

🎯 自己評価

自分の回答

気づき・メモ