Day 050 — gRPC/Protocol Buffers実装（続き）: Server streaming疑似実装とgoroutineリークのデバッグ

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

Day 049では、net/rpcにはストリーミングの概念がなく「継続的に更新を受け取りたい」という要件にはポーリングで近似するしかないこと、そして本物のgRPCのServer streaming RPCを選ぶと「双方でストリームのライフサイクル管理（切断検知・エラーハンドリング）が必要になり実装がやや複雑になる」と述べました。今日はこの「複雑さ」の正体を、実際に手を動かして体験します。

本物のgRPCでは、クライアントがストリームの受信を途中でやめる（contextをキャンセルする、コネクションを切るなど）と、サーバー側で実行中のstream.Send(...)はエラー（context.Canceled相当のgRPCステータスコード）を返すようになります。これはgRPCのランタイムが、ストリームをHTTP/2のストリームおよびcontext.Contextに紐付けて管理しているためで、「受信側がいなくなったこと」がサーバー側のSendの戻り値という形で自動的に伝わります。

一方、今日Unary RPC（net/rpcのclient.Call）をポーリングして疑似的にServer streamingを実現する場合、この「受信側がいなくなったことを送信側が知る仕組み」はGoの標準ライブラリが自動で用意してくれるものではありません。自分でchannelとcontextを正しく配線して初めて実現される、開発者自身の責任です。これを怠ると、一見エラーも出さずに正常終了しているように見えるプログラムの裏で、goroutineリークという静かなバグが発生します。今日はこのバグを実際に発生させ、観測し、修正します。

📝 問題

1. 概念確認: 本物のgRPCでは「クライアントがストリームの受信をやめたこと」がサーバー側のstream.Send(...)の戻り値というかたちで自動的に伝わります。これに対し、net/rpcベースの疑似ストリーミング実装ではこの仕組みが自動で得られない理由を、net/rpcが持つ機能の範囲（Day 048・049で扱った規約）に基づいて説明してください。
2. デバッグ（バグの発見）: 以下は、namesのリストを一定間隔でポーリングし、結果をchannel経由でストリームのように返すStreamGreetings関数です。mainはこの関数を使って最初の2件だけ受け取り、breakで早期に受信をやめます。このプログラムはエラーも出さず正常終了しているように見えますが、実は1つgoroutineがリークしています。どこで・なぜリークするか指摘してください。

   📋 コピーfunc StreamGreetings(client *rpc.Client, names []string, interval time.Duration) <-chan HelloReply {
   	results := make(chan HelloReply) // unbuffered
   	go func() {
   		for _, name := range names {
   			var reply HelloReply
   			if err := client.Call("Greeter.SayHello", &HelloRequest{Name: name}, &reply); err != nil {
   				return
   			}
   			results <- reply // ← ここに注目
   			time.Sleep(interval)
   		}
   	}()
   	return results
   }
   
   func main() {
   	// ...サーバー・クライアントのセットアップは省略...
   	names := []string{"Alice", "Bob", "Carol", "Dave", "Eve"}
   	stream := StreamGreetings(client, names, 10*time.Millisecond)
   
   	count := 0
   	for reply := range stream {
   		fmt.Println("received:", reply.Message)
   		count++
   		if count == 2 {
   			break // 2件で十分として早期に打ち切る
   		}
   	}
   }

3. 修正と定量的な検証: context.Contextを使って(2)のリークを修正したStreamGreetingsを実装してください。要件:
   • シグネチャにctx context.Contextを追加する
   • results <- replyをselectでctx.Done()と競争させ、呼び出し側が受信をやめた場合は送信を諦めてgoroutineを終了する
   • main側は2件受け取ったらcancel()を呼んでからbreakする
   • runtime.NumGoroutine()を使い、「早期break前後でgoroutine数が増えていないこと」を実際の数値で確認するコードを含める（修正前・修正後の両方で数値を比較できるようにする）
4. 設計・一般化: 「unbuffered channelへの送信」と「受信側がループを早期に抜ける可能性」という組み合わせは、今日のServer streamingポーリングに限らず、Fan-out的に複数のconsumerへ配信する処理やイベント通知全般で繰り返し登場する危険な組み合わせです。この組み合わせがなぜ一般的に危険か説明し、select + ctx.Done()以外に採りうる設計上の選択肢を2つ挙げ、それぞれのトレードオフを述べてください。

🔍 ヒント（段階的開示）

📋 コピーfunc StreamGreetings(ctx context.Context, client *rpc.Client, names []string, interval time.Duration) <-chan HelloReply {
	results := make(chan HelloReply)
	go func() {
		defer close(results) // TODO: なぜcloseが必要か考える
		for _, name := range names {
			var reply HelloReply
			if err := client.Call("Greeter.SayHello", &HelloRequest{Name: name}, &reply); err != nil {
				return
			}
			select {
			case results <- reply:
			case <-ctx.Done():
				return // TODO: ここが今日のバグ修正の核心
			}
			// TODO: interval分の待機もctx.Done()と競争させる
		}
	}()
	return results
}

✅ 模範解答

(1) net/rpcベースの疑似ストリーミングで「自動で伝わる仕組み」が得られない理由:

Day 048で確認した通り、net/rpcが提供するのは「(args *ArgsType, reply *ReplyType) errorという規約を満たすメソッドを、クライアントがclient.Callで1回呼び出し、1回の応答を受け取る」というUnary RPCの骨格だけです。net/rpc自体はストリーム・コネクションのライフサイクルという概念を持たず、「クライアントが受信をやめた」というイベントをサーバー側やポーリングループ側に通知する仕組みは一切実装されていません。本物のgRPCでこの通知が自動で行われるのは、gRPCのランタイムがストリームをHTTP/2のストリームおよび呼び出し側のcontext.Contextに紐付けて管理し、コネクション断・キャンセルを検知してstream.Send()にエラーを注入する専用のコードを持っているからです。今日のStreamGreetingsのようにUnary RPCのポーリングでストリーミングを疑似的に実現する場合、この通知の配線は開発者がcontext.Contextとchannelを使って自分で作るしかなく、それを怠ると(2)のリークが発生します。

(2) バグの所在:

results <- replyが原因です。resultsはmake(chan HelloReply)で作られたunbuffered channelであり、送信は「対応する受信が実行されるまでブロックする」という性質を持ちます。main側が2件受け取った時点でbreakし、for reply := range streamのループを抜けると、以降誰もこのchannelから受信しなくなります。しかしStreamGreetings内のgoroutineはnamesのループを続けており、3件目（Carol）の結果をresults <- replyで送信しようとした瞬間、永久に受信されることのない送信文でブロックしたままになります。このgoroutineは二度とreturnに到達できず、プログラムが終了するまでメモリ上に残り続けます。main関数自体はエラーも出さず正常終了するため、意識的にruntime.NumGoroutine()で確認しない限り、このリークは気づかれません。

(3) 修正版の実装（完全なコード・実測結果付き）:

📋 コピーpackage main

import (
	"context"
	"fmt"
	"log"
	"net"
	"net/rpc"
	"runtime"
	"time"
)

type Greeter struct{}

type HelloRequest struct{ Name string }
type HelloReply struct{ Message string }

func (g *Greeter) SayHello(req *HelloRequest, reply *HelloReply) error {
	reply.Message = "Hello, " + req.Name + "!"
	return nil
}

// StreamGreetings はサーバー streaming RPC を net/rpc のポーリングで疑似的に実現する。
// 修正版: results への送信を select で ctx.Done() と競争させ、呼び出し側が
// 受信をやめた（= ctx がキャンセルされた）ことを検知したら送信を諦めて
// goroutine を終了させる。
func StreamGreetings(ctx context.Context, client *rpc.Client, names []string, interval time.Duration) <-chan HelloReply {
	results := make(chan HelloReply)
	go func() {
		defer close(results) // 呼び出し側がrangeでループを終了できるようにチャネルを閉じる
		for _, name := range names {
			var reply HelloReply
			if err := client.Call("Greeter.SayHello", &HelloRequest{Name: name}, &reply); err != nil {
				return
			}
			select {
			case results <- reply:
			case <-ctx.Done():
				return // 受信側がいなくなったので送信を諦めてgoroutineを終了する
			}
			select {
			case <-time.After(interval):
			case <-ctx.Done():
				return // 待機中にキャンセルされた場合もここで終了する
			}
		}
	}()
	return results
}

func main() {
	rpc.Register(new(Greeter))
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		log.Fatal(err)
	}
	defer listener.Close()
	go func() {
		for {
			conn, err := listener.Accept()
			if err != nil {
				return
			}
			go rpc.ServeConn(conn)
		}
	}()

	client, err := rpc.Dial("tcp", listener.Addr().String())
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	time.Sleep(50 * time.Millisecond) // accept-loopのgoroutineが安定するのを待つ
	baseline := runtime.NumGoroutine()
	fmt.Printf("baseline goroutines = %d\n", baseline)

	ctx, cancel := context.WithCancel(context.Background())
	names := []string{"Alice", "Bob", "Carol", "Dave", "Eve"}
	stream := StreamGreetings(ctx, client, names, 10*time.Millisecond)

	count := 0
	for reply := range stream {
		fmt.Println("received:", reply.Message)
		count++
		if count == 2 {
			cancel() // 受信をやめる意思をctxで明示してからループを抜ける
			break
		}
	}

	time.Sleep(200 * time.Millisecond) // producer側が終了処理を終える猶予
	after := runtime.NumGoroutine()
	fmt.Printf("after early-break goroutines = %d (leaked = %d)\n", after, after-baseline)
}

実行結果（実測。go1.26.2 darwin/amd64で確認済み）:

修正前（問題(2)のバグありコード。contextなし、results <- replyのみ）:

📋 コピーbaseline goroutines = 4
received: Hello, Alice!
received: Hello, Bob!
after early-break goroutines = 5 (leaked = 1)

修正後（上記の完全なコード）:

📋 コピーbaseline goroutines = 4
received: Hello, Alice!
received: Hello, Bob!
after early-break goroutines = 4 (leaked = 0)

(4) 一般化とトレードオフ:

「unbuffered channelへの送信」と「受信側の早期離脱」の組み合わせが危険なのは、Goのchannelの送信操作が受信側の存在を前提にブロックするという性質を持つためです。送信側は「受信側がまだいるか」を送信操作自体からは知る術がなく、受信側が抜けた後もそれを知らずに送信を試み続けます。これはServer streamingのポーリングに限らず、以下のような場面で繰り返し登場します。

• Fan-outで複数のworkerに仕事を配る際、一部のworkerが早期に終了・パニックしても配信元がそれを知らない
• pub/subパターンでイベントを配信する際、subscriberが購読解除した後も配信元が送信を試み続ける
• HTTPハンドラがリクエスト処理中にクライアントが切断した場合、バックグラウンドgoroutineがレスポンス用channelへの送信で止まる

select + ctx.Done()以外に採りうる設計上の選択肢は以下の2つです。

🪜 Step-by-Step 解説

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

🌐 他言語との比較

🏆 実務での使いどころ

• ログ・メトリクスのストリーミングAPI: 「クライアントが購読を途中でやめる」ことは正常系の一部です。今日のような修正がなければ、クライアントが頻繁に接続・切断するたびにサーバー側でgoroutineが蓄積し、時間の経過とともにメモリ使用量が単調増加する障害（多くは深夜バッチやデプロイ直後の再起動でようやく解消する「じわじわ悪化する」タイプの障害）につながります。
• 本番テストでのリーク検出: 実務ではruntime.NumGoroutine()の手動比較よりも、go.uber.org/goleakをテストスイートのTestMainに組み込み、各テストの前後で残存goroutineを自動検査する運用が一般的です。今日のテクニックはその仕組みが内部で何をしているかを理解する土台になります。
• -raceと分業する: チームのCIに「go test -raceが通ればgoroutine関連のバグはない」という誤解が広がっていないか確認する価値があります。今日実測した通り、-raceはデータ競合しか検出しないため、goroutineリークを防ぐにはコードレビューでのselectパターンの確認や、goleakのような専用ツールの併用が必要です。

⚠️ よくある誤解・ミス

🚀 次のステップ

• 発展: 今日のStreamGreetingsを、namesが5件ではなく数万件あるケースに拡張し、受信側がcount == 2ではなく「ランダムなタイミングで購読解除する複数のクライアント」をシミュレートしてみましょう。sync.WaitGroupとruntime.NumGoroutine()を組み合わせ、全クライアントが購読解除した後にgoroutine数がベースラインに戻ることをテストコードとして自動化すると、goleakが内部でやっていることに近い体験になります。
• 次回予告: ミドルウェア設計パターン — http.Handlerのチェーン設計を扱います

🎯 自己評価

自分の回答

気づき・メモ