Go並行処理の静かな殺し屋：Mutex、セマフォ、およびGoroutineのリーク

Goは並行処理を簡単に見せます.

あなたはこう書きます.

go func() {
    // do something concurrently
}()

そして突然、あなたのコードが別のgoroutineで実行されます.

そのシンプルさは、私がGoをとても好きな理由の一つです。しかし、バックエンドシステム、通知パイプライン、高トラフィックAPI、実際の負荷での生産サービスを扱ってから、私は重要なことを学びました：

Goにおける多くの並行処理の問題は、並行処理を使用しないから生じるものではありません。
それらは、実際のボトルネックがどこにあるかを理解せずに並行処理を使うことから来ています。

時々問題はロックが欠けている場合があります。

しかし、特にプロダクションのGoサービスでは、問題は逆の場合が多いです：

ロックが多すぎます
長時間ロックが保持されています
クリティカルセクション内のネットワークI/O
終了しないgoroutines
無制限なgoroutineの作成
値渡しでコピーされたWaitGroups
キャンセル戦略なしで使用されるチャネル

本記事では、実際のシステムで見た並行性の問題、メルクセスとセマフォについての私の考察方法、そしてこれらの問題が生産環境のインシデントになる前に通常どのようにデバッグするかを説明したいと思います.

本当の問題：意図せず順次処理になる並行性

サービスは外部から並行的に見えるけれども、内部では単一スレッドのアプリケーションとして振る舞うことがあります。

これは通常、リクエストフローの大部分が一つの共有ロックの背後に隠されている時に起こります

このようなパターンは多くの開発者が認めないほど一般的です

mu.Lock()
user.Name = "Test User"
sendEmail(user)
callDatabase(user)
mu.Unlock()

最初は安全に見えるかもしれません

開発者は共有状態を保護しようとしていた。その部分は合理的だ。しかし、今のロックは共有メモリだけでなくもっと多くを保護している。それは全てのフローを保護している：

フィールドを更新
メールを送信
データベースを呼び出す
おそらくネットワークI/Oの待機
おそらく再試行
おそらく長い時間で他のgoroutinesをブロック

それはもはや互換キーではありません.

それは交通渋滞です.

同じロックが必要なすべてのゴルーチンは、全体のフローが終わるまで待つ必要があります。したがって、あなたのサービスが数百や数千のゴルーチンを持っていても、システムの大部分が順次処理になります。

危険なのは、CPU使用率がまだ正常に見えたり、低く見えたりすることです。メモリも正常に見えるかもしれません。しかし、遅延が増加し、スループットが低下し、p95/p99の応答時間が不安定になります。

これが、ロック競合が基本的なインフラメトリクスだけでは時々見つからにくい理由です。

プロダクションスタイルの例：ミューテックス内のメール

ユーザーステートを更新し、通知を送信するサービスがあると想像してみてください.

type Service struct {
    mu    sync.Mutex
    state map[int]string
}

func (s *Service) ProcessUsers(users []User) {
    s.mu.Lock()
    defer s.mu.Unlock()

    for _, user := range users {
        s.state[user.ID] = "processed"
        sendEmail(user) // slow network I/O inside the lock
    }
}

このコードはデータレースの観点から安全です.

しかし、パフォーマンスの観点からは危険です.

ミューテックスは可能な限り小さな共有メモリ操作を保護すべきです。遅い外部作業（例えば：）を保護すべきではありません。

メールの送信
別のマイクロサービスを呼び出している
データベースクエリ
HTTPリクエスト
ファイルアップロード
遅い外部シンクにログを記録
第三者APIの待機

メモリの更新はナノ秒またはマイクロ秒かかる可能性があります。メールの呼び出しはミリ秒または秒かかる可能性があります

その違いは重要です

ロックが保持されている間にsendEmailが実行されると、s.muが必要な他のgoroutineはネットワーク呼び出しの後にブロックされます

より良いバージョンでは、共有状態の変更を遅い作業から分離します

func (s *Service) ProcessUsers(users []User) {
    emails := make([]User, 0, len(users))

    s.mu.Lock()
    for _, user := range users {
        s.state[user.ID] = "processed"
        emails = append(emails, user)
    }
    s.mu.Unlock()

    for _, user := range emails {
        sendEmail(user)
    }
}

これも改善されていますなぜならロックは共有マップのみを保護するからです

実際の生産環境では、私は通常遅い作業をキューや制限付きワーカープールにプッシュすることを好みます：

func (s *Service) ProcessUsers(users []User, jobs chan<- EmailJob) {
    s.mu.Lock()
    for _, user := range users {
        s.state[user.ID] = "processed"
    }
    s.mu.Unlock()

    for _, user := range users {
        jobs <- EmailJob{UserID: user.ID, Email: user.Email}
    }
}

今ではリクエストパスが直接メールプロバイダーの遅延に依存しません

これが本当の修正です

「ゴルーチンを使う」だけではありません

共有メモリ、外部I/O、バックプレッシャーとの境界を設計することで解決策が設計されています

ミューテックスは悪くありません。大きなクリティカルセクションは悪いです。

私は時々開発者がミューテックスを恐れるのを見ます

それは間違った教訓です

sync.Mutexは正しく使えば単純で速く、完璧です。問題はミューテックスではなく、クリティカルセクションのサイズです。

これを心に留めておくべきです：

mu.Lock()
// only touch shared memory here
mu.Unlock()

これではありません：

mu.Lock()
// shared memory
// database call
// HTTP call
// email call
// JSON encoding
// logging
// metrics push
mu.Unlock()

良いクリティカルセクションは退屈であるべきです

通常、これらのいずれかを行うべきです

共有状態を読み取る
共有状態を更新する
ローカル変数に共有状態をコピーします
ポインタを交換します
カウンターをインクリメントします
保護されたスライス/マップに追加します

その後、ロックを解除します

その後、ロック外で高コストの作業を行います

ドライブ下：ミューテックスがあなたに提供するもの

高レベルでは、互斥ロックは排他性を提供します：一度に一つのゴルーチンだけが保護されたセクションに入ることができます

しかし、メモリ順序の保証も提供します。

Goのメモリモデルでは、ロック操作の前に同じmutexの後のロック操作でアンロック操作が同期します。実用的には、これは一つのgoroutineが共有データを更新してアンロックした場合、後で同じmutexをロックする別のgoroutineがその更新を安全に観察できることを意味します。

それが多くの開発者が忘れる部分です。

互排 locks は「他の goroutines をブロックすること」だけでなく、「goroutines の間の安全な可視性境界を作る」ことも意味します。

その境界がないと、異なる goroutines が同じ時刻に同じメモリを読み書きし、それでデータ競合が発生します。データ競合が発生すると、あなたのプログラムは自信を持って考えることができないものになります。

それが why I do not like “clever” lock-free code unless there is a very strong reason for it.

Most backend services do not need clever concurrency.

They need clear concurrency.

セマフォ：キャパシティを制御する、所有権ではない

互換メカニズムは通常、共有メモリの所有権についてです.

セマフォはキャパシティについてです。

例えば、10,000人のユーザーを処理したいが、同時に10,000件のメールを送信したくないとしよう。

単純なバージョンでは、こんなことをするかもしれない：

for _, user := range users {
    go sendEmail(user)
}

これは危険なため、無制限の並行性が生じる。

usersには10,000のアイテムがあり、あなたは10,000のgoroutineを作成します。各goroutineがネットワークI/Oを実行し、接続を開き、メモリを割り当て、外部プロバイダーに待機すると、メールプロバイダーを過負荷にする前にあなた自身のサービスを過負荷にすることができます。

シンプルなセマフォパターンでこれを修正できます：

sem := make(chan struct{}, 20) // allow only 20 concurrent email sends
var wg sync.WaitGroup

for _, user := range users {
    user := user

    sem <- struct{}{}
    wg.Add(1)

    go func() {
        defer wg.Done()
        defer func() { <-sem }()

        sendEmail(user)
    }()
}

wg.Wait()

現在コードはまだ並行処理を使用していますが、並行処理は制限されています.

その詳細は本番環境では非常に重要です.

無制限の並行処理は拡張性がありません.

それは遅延された失敗です.

本番環境コードのためのより良いワーカープール

セマフォパターンは便利ですが、継続的に実行されるサービスの場合、私はよくワーカープールを好みます.

type EmailJob struct {
    UserID int
    Email  string
}

func startEmailWorkers(ctx context.Context, workerCount int, jobs <-chan EmailJob) {
    var wg sync.WaitGroup

    for i := 0; i < workerCount; i++ {
        wg.Add(1)

        go func(workerID int) {
            defer wg.Done()

            for {
                select {
                case <-ctx.Done():
                    return

                case job, ok := <-jobs:
                    if !ok {
                        return
                    }

                    if err := sendEmailJob(ctx, job); err != nil {
                        // In real systems: log, retry, dead-letter, or expose metrics.
                        fmt.Printf("worker=%d failed to send email user_id=%d err=%v\n", workerID, job.UserID, err)
                    }
                }
            }
        }(i)
    }

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

これにより、より優れた運用制御が可能になります：

固定された並行処理
簡単なメトリクス
簡単なシャットダウン
簡単なリトライ戦略
簡単なバックプレッシャー
簡単なレートリミッティング

「私はゴルーチンを使った」と「私は並行システムを設計した」という違いです

ゴルーチンリーク：すぐに爆発しないバグ

ゴルーチンリークはGoにおける最も一般的な生産問題の一つです

それは危険なため、サービスがすぐにクラッシュしない可能性があります。数時間や数日をかけて徐々に悪化する可能性があります

ここに典型的な例があります

func process() error {
    ch := make(chan result)

    go func() {
        ch <- heavyComputation()
    }()

    select {
    case res := <-ch:
        return handle(res)

    case <-time.After(1 * time.Second):
        return errors.New("timeout")
    }
}

問題は微妙です

chはバッファされていません

タイムアウトが最初に発生した場合、processが返されます。その後、chを待つ受信者がいません

heavyComputation()が終了すると、ゴルーチンはchに送信しようとし、永遠にブロックされます。

そのゴルーチンはリークしています.

一つのリークしたゴルーチンは問題ではありません.

数千のリークしたゴルーチンは問題です.

より安全なバージョンはバッファ付きチャネルを使用します.

func process() error {
    ch := make(chan result, 1)

    go func() {
        ch <- heavyComputation()
    }()

    select {
    case res := <-ch:
        return handle(res)

    case <-time.After(1 * time.Second):
        return errors.New("timeout")
    }
}

これはタイムアウト後にもゴルーチンが送信でブロックされるのを防ぎます.

実際のサービスでは、コンテキストベースのキャンセルを好みます：

func process(ctx context.Context) error {
    ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
    defer cancel()

    ch := make(chan result, 1)

    go func() {
        res := heavyComputation(ctx)

        select {
        case ch <- res:
        case <-ctx.Done():
        }
    }()

    select {
    case res := <-ch:
        return handle(res)

    case <-ctx.Done():
        return ctx.Err()
    }
}

重要な教訓：

各ゴルーチンには終了経路が必要です。

ゴルーチンがどのように停止するかを説明できない場合、おそらく漏れが発生する可能性があります。

WaitGroup by Value: 小さなミスが大きな影響を与える

コードレビューで非常に見落としやすいミスです：

func worker(wg sync.WaitGroup) { // wrong: copied by value
    defer wg.Done()

    // do work
}

sync.WaitGroupは、初めて使用した後にコピーしないでください。

値渡しすると、その内部状態がコピーされます。ワーカーはコピーに対してDone()を呼び出しますが、メインゴルーチンが待っている元のWaitGroupではありません。

これにより、デッドロックが発生する可能性があります。

正しいバージョン:

func worker(wg *sync.WaitGroup) {
    defer wg.Done()

    // do work
}

そして使用法:

var wg sync.WaitGroup

for i := 0; i < 10; i++ {
    wg.Add(1)
    go worker(&wg)
}

wg.Wait()

このルールは他の同期原語（例えばsync.Mutex）にも適用されます.

初めて使用した後はコピーしないでください.

ループ変数の罠

これはかつてGoの並行処理における最も有名なバグの一つでした：

for _, user := range users {
    go func() {
        sendEmail(user)
    }()
}

Goのバージョンや文脈によっては、ループ変数を誤ってキャプチャするとgoroutinesが間違った値を使用する可能性があります

防御的なパターンは依然としてシンプルで明確です：

for _, user := range users {
    user := user

    go func() {
        sendEmail(user)
    }()
}

新しいGoバージョンで改善されたとしても、本番コードでこのスタイルを好む理由は、変数の所有権を読者が明確に理解できるからです.

読みやすい並行処理は保守しやすい並行処理です.

Goでロックの競合をデバッグする方法

並行処理のボトルネックを疑うと、まず推測を始めません.

まず測定を始めます。

1. pprofを有効にする

import _ "net/http/pprof"

func main() {
    go func() {
        log.Println(http.ListenAndServe("localhost:6060", nil))
    }()

    // start application
}

次に、プロファイルを収集します：

go tool pprof http://localhost:6060/debug/pprof/profile?seconds=30

互斥ロックの競合に関しては、互斥ロックのプロファイリングを有効にします：

runtime.SetMutexProfileFraction(1)

次に、確認します

go tool pprof http://localhost:6060/debug/pprof/mutex

2. ゴルーチンの数を確認

ゴルーチンの数が増加することは、ブロックされたゴルーチンやリークのサインであることが多い

fmt.Println("goroutines:", runtime.NumGoroutine())

本番環境では、メトリクスとして公開する

prometheus.NewGaugeFunc(
    prometheus.GaugeOpts{
        Name: "go_goroutines_current",
        Help: "Current number of goroutines.",
    },
    func() float64 {
        return float64(runtime.NumGoroutine())
    },
)

3. ゴルーチンのスタックダンプ

サービスが固着している場合、ゴルーチンダンプは貴重です。

curl http://localhost:6060/debug/pprof/goroutine?debug=2

同じ行で多くのゴルーチンがブロックされているかを探す:

sync.(*Mutex).Lock
chan send
chan receive
net/http.(*Transport).RoundTrip

5,000個のgoroutineが同じロックやチャネルでブロックされている場合、ボトルネックを発見しました

4. テストでレース検出器を使用

go test -race ./...

レース検出器は無料ではありませんが、通常は本番環境で実行しませんが、CIおよびローカルデバッグでは非常に有用です。

私の生産的なGo並行処理のための実用的なルール

これらは、並行Goコードを書いたりレビューしたりする際に守ろうとするルールです：

1. ロックは小さく保ちなさい

保護が必要なデータのみをロックします.

リクエストのライフサイクル全体をロックしないでください.

2. ミューテックス内に遅いI/Oを置かないでください

クリティカルセクション内でデータベース呼び出し、HTTP呼び出し、メール送信、ファイルアップロード、サードパーティAPI呼び出しを避ける.

3. バウンドコンカレンシー

無制限のgoroutineを作成しない.

ワーカーポール、セマフォ、キュー、またはレートリミッターを使用する.

4. 各goroutineにはシャットダウン経路が必要

context.Context、チャネルクローズ、または明示的なキャンセルを使用する.

5. 同期化プリミティブをコピーしない

共有する際は、*sync.WaitGroup、*sync.Mutex、および類似のプリミティブをポインタで渡す

6. 最適化する前に測定する

pprof、実行時メトリクス、トレース、ログ、およびゴルーチンダンプを使用する

推測はデバッグではない

7. 嫌な並行性を好む

最良質な並行コードは通常賢いものではない

明確で測定可能で、停止しやすくなっている

最終的な考え

Goは私たちに強力な並行処理ツールを与えてくれますが、それが自動的に良い並行設計を与えてくれるわけではありません

ゴルーチンは安価ですが、無料ではありません

ミューテックスは速いですが、遅い作業の周りで保持しているとスループットを壊すことがあります

チャンネルは洗練されているが、誰も受信していない場合にゴルーチンがリークする可能性がある。

WaitGroupはシンプルだが、コピーすると全体的なフローが壊れる可能性がある。

私にとって、上級のGoエンジニアリングはすべての並行性原語を使うことではなく、それらを使わない時期、本当の境界線がどこか、そして負荷があるときにシステムがどのように振る舞うかを知ることである。

次回あなたがこれを書くとき

mu.Lock()

進む前に一つ質問する:

私は何を保護しているのか、そしてこのロックをどれくらいの速さで解放できるのか

その一つの質問で、あなたのサービスを静かな生産のボトルネックから救うことができる

参考文献

Go メモリモデル: https://go.dev/ref/mem
Go sync パッケージドキュメント: https://pkg.go.dev/sync
Go診断とプロファイリングツール: https://go.dev/doc/diagnostics
Goブログ: Goスケジューラとランタイムのノート: https://go.dev/blog/

おすすめ購読元

DEV Community