Day 031 — 並行処理パターン③（Pipeline）

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

Pipeline（パイプライン）パターンは、「あるステージの出力channelを次のステージの入力channelにそのまま繋ぐ」ことで、複数の処理段階をストリームとして連結するパターンです。Unixのcmd1 | cmd2 | cmd3のパイプと発想は同じで、各ステージが独立したgoroutineとして動き、channelを通じてデータが流れていきます。

Day 029のWorker Poolが「同じ処理を複数goroutineで並列実行して負荷分散する」パターン、Day 030のFan-in/Fan-outが「1ステージを複数goroutineに分散・集約する」パターンだったのに対し、Pipelineは「異なる処理を複数ステージに分割し、ストリームとして流す」パターンです。Fan-in/Fan-outはPipelineの1ステージを内部で並列化する技法として組み合わせて使われることもあります。

📋 コピーstage3 := filterEven(ctx, square(ctx, generate(ctx, n))) // 3段のステージを直接連結

Pipelineを正しく実装する上で最重要なのが「送信側が必ず自分の出力channelをcloseする」という規律です。Goのchannelは受信側からcloseすることはできず、送信側の責務としてcloseします。これを怠ると、下流のfor v := range chが永久にブロックし、goroutineリークが発生します。

また、上流でキャンセルが発生した際に下流まで正しく伝播させるには、context.Contextをパイプライン全体で共有し、各ステージの送信処理をselectでctx.Done()とも競合させる必要があります。これにより「受信側がもう受け取らない（例: 早期終了した）」ケースでも送信側がchannel送信でブロックしたまま残り続ける、いわゆるgoroutineリークを防げます。

📝 問題

以下の3段パイプラインを実装してください。

1. generate ステージ: 1からnまでの整数を1つずつ出力channelに送る
2. square ステージ: 受け取った整数を2乗して出力channelに送る
3. filterEven ステージ: 受け取った値が偶数であれば出力channelに送り、奇数なら捨てる

各ステージはfunc(ctx context.Context, in <-chan int) <-chan int（generateのみfunc(ctx context.Context, n int) <-chan int）のシグネチャを持ち、内部で新しいgoroutineを起動して即座に出力channelを返す（＝呼び出し側をブロックしない）ノンブロッキング関数として実装してください。

さらに、main関数で3ステージを連結し、最終出力をfor v := rangeで受け取ってすべて合計し、標準出力に表示してください。加えて、context.WithCancelを使い、合計値が500を超えた時点でcancel()を呼んで残りのパイプライン処理を打ち切る動作も実装してください（打ち切り後にgoroutineがリークしていないことを意識すること）。

要件

• 各ステージの送信はselect { case out <- v: case <-ctx.Done(): return }のパターンで、キャンセル時に必ずreturnして出力channelをcloseすること（defer close(out)を使う）
• n = 100として実行し、合計が500を超えた時点で打ち切られることを確認できるログを出力すること
• go run main.goでそのまま実行できること

🔍 ヒント（段階的開示）

📋 コピーselect {
case out <- v:
case <-ctx.Done():
    return
}

📋 コピーfunc generate(ctx context.Context, n int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := 1; i <= n; i++ {
			select {
			case out <- i:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func square(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for v := range in {
			select {
			case out <- v * v:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}
// filterEven も同様の形

✅ 模範解答

ファイル: main.go

📋 コピーpackage main

import (
	"context"
	"fmt"
)

// generate は 1..n の整数を出力channelに流すパイプラインの起点ステージ。
func generate(ctx context.Context, n int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := 1; i <= n; i++ {
			select {
			case out <- i:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// square は受け取った整数を2乗して次段に流す中間ステージ。
func square(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for v := range in {
			select {
			case out <- v * v:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// filterEven は偶数のみを次段に通す中間ステージ。
func filterEven(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for v := range in {
			if v%2 != 0 {
				continue
			}
			select {
			case out <- v:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel() // main終了時に念のため全ステージへキャンセルを伝播させる

	const n = 100
	pipeline := filterEven(ctx, square(ctx, generate(ctx, n)))

	sum := 0
	for v := range pipeline {
		sum += v
		fmt.Printf("received=%d sum=%d\n", v, sum)
		if sum > 500 {
			fmt.Println("合計が500を超えたためパイプラインを打ち切ります")
			cancel()
			// cancel後もgenerateが送信中の値をfor rangeが消費し続けるため、
			// 全ステージがctx.Done()経由で終了し出力channelがcloseされるまでループを回す。
		}
	}
	fmt.Printf("最終合計: %d\n", sum)
}

$ go run main.go
received=4 sum=4
received=16 sum=20
received=36 sum=56
received=64 sum=120
received=100 sum=220
received=144 sum=364
received=196 sum=560
合計が500を超えたためパイプラインを打ち切ります
最終合計: 560

go run main.goでそのまま動作する単一ファイルの完全なコードとして検証済みです。

🪜 Step-by-Step 解説

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

🌐 他言語との比較

🏆 実務での使いどころ

• ログ処理パイプライン: ログ行を読み込む → パースする → フィルタする → 集計してメトリクス化する、という多段処理を各ステージ独立のgoroutineで動かし、スループットを段階ごとに調整できる
• 画像/動画処理バッチ: デコード → リサイズ → エンコード → アップロード、のようなCPU/IOバウンドが混在する処理をステージ分割し、IOバウンドなステージだけFan-outで並列数を増やすといった最適化がしやすい
• ETLジョブ: 抽出（DB/APIから読む）→ 変換（正規化・バリデーション）→ ロード（書き込み）の3段をchannelで繋ぐことで、メモリに全件を溜め込まずストリーミングで処理できる
• APIゲートウェイのリクエスト処理: 認証 → レート制限チェック → バリデーション → ハンドラ呼び出し、という単一リクエストの処理段階をミドルウェアチェーンとして表現する際、Pipelineの発想（各段が独立しctxのキャンセルが全段に伝播する）が直接活きる

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: filterEvenステージをFan-outし、複数goroutineで並列にフィルタ処理させた上でFan-inで1本のchannelに集約する「Pipeline + Fan-out/Fan-in」のハイブリッド構成を実装してみましょう
• 次回予告: errgroup パッケージ — 複数goroutineのエラーを集約し、最初のエラーで全体をキャンセルする仕組みを学びます（Pipelineの各ステージでエラーが発生した場合の伝播をより堅牢に書けるようになります）

🎯 自己評価

自分の回答

気づき・メモ