Go sync.WaitGroup と調整の問題

この投稿は、Go での同時実行性の処理に関するシリーズの一部です:

• 同期に進みます。Mutex: 通常モードと飢餓モード
• sync.WaitGroup と調整の問題 (ここにあります) に進みます
• 同期プールとその背後にある仕組み
• 最も見落とされている同期メカニズムである sync.Cond を実行してください
• Go sync.Map: 適切な仕事に適したツール
• Go Singleflight は DB ではなくコードに溶け込みます

WaitGroup は基本的に、複数のゴルーチンが作業を完了するのを待つ方法です。

各同期プリミティブには独自の問題があり、これも例外ではありません。ここでは、WaitGroup の調整の問題に焦点を当てます。そのため、内部構造がバージョンごとに変更されています。

この記事は Go 1.23 に基づいています。今後何か変化があった場合は、X(@func25) を通じてお気軽にお知らせください。

sync.WaitGroupとは何ですか?

すでに sync.WaitGroup に精通している場合は、そのまま読み飛ばしてください。

まず問題に飛び込んでみましょう。大きな仕事を抱えているので、それを相互に依存せずに同時に実行できる小さなタスクに分割することにしたとします。

これを処理するには、ゴルーチンを使用します。ゴルーチンを使用すると、これらの小さなタスクを同時に実行できるためです。

func main() {
    for i := 0; i < 10; i++ {
        go func(i int) {
            fmt.Println("Task", i)
        }(i)
    }

    fmt.Println("Done")
}

// Output:
// Done

しかし、ここが問題です。他のゴルーチンが作業を完了する前に、メインのゴルーチンが終了して終了する可能性が十分にあります。

多くのゴルーチンをスピンオフしてそれぞれの処理を実行するとき、メインのゴルーチンが終了して他のゴルーチンが終了する前に終了しないように、それらを追跡したいと考えています。ここで WaitGroup が登場します。ゴルーチンの 1 つがそのタスクを終了するたびに、WaitGroup に通知します。

すべてのゴルーチンが「完了」としてチェックインされると、メインのゴルーチンは安全に終了できることがわかり、すべてがきれいに終了します。

func main() {
    var wg sync.WaitGroup

    wg.Add(10)
    for i := 0; i < 10; i++ {
        go func(i int) {
            defer wg.Done()
            fmt.Println("Task", i)
        }(i)
    }

    wg.Wait()
    fmt.Println("Done")
}

// Output:
// Task 0
// Task 1
// Task 2
// Task 3
// Task 4
// Task 5
// Task 6
// Task 7
// Task 8
// Task 9
// Done

一般的には次のようになります:

• ゴルーチンの追加: ゴルーチンを開始する前に、予想されるゴルーチンの数を WaitGroup に伝えます。これは、WaitGroup.Add(n) を使用して行います。n は、実行する予定のゴルーチンの数です。
• Goroutines running: 各 goroutine が起動し、それぞれの処理を実行します。それが完了したら、WaitGroup.Done() を呼び出してカウンターを 1 つ減らすことで、WaitGroup に通知する必要があります。
• すべてのゴルーチンを待機: メインのゴルーチン (重い処理を行わないゴルーチン) で、WaitGroup.Wait() を呼び出します。これにより、WaitGroup 内のカウンターがゼロになるまで、メインの goroutine が一時停止されます。わかりやすく言えば、他のすべてのゴルーチンが終了し、完了の合図が行われるまで待機します。

通常、ゴルーチンを起動するときに WaitGroup.Add(1) が使用されていることがわかります。

for i := 0; i < 10; i++ {   
    wg.Add(1)
    go func() {
        defer wg.Done()
        ...
    }()
}

どちらの方法も技術的には問題ありませんが、wg.Add(1) を使用するとパフォーマンスに若干の影響があります。それでも、wg.Add(n).

を使用する場合に比べて、エラーが発生する可能性は低くなります。

「wg.Add(n) はエラーが発生しやすいと考えられるのはなぜですか?」

重要なのは、誰かが特定の反復をスキップする continue ステートメントを追加した場合など、ループのロジックが途中で変更された場合、状況が混乱する可能性があるということです。

wg.Add(10)
for i := 0; i < 10; i++ {
    if someCondition(i) {
        continue  
    }

    go func() {
        defer wg.Done()
        ...
    }()
}

この例では、ループが常に正確に n 個のゴルーチンを開始すると仮定して、ループの前に wg.Add(n) を使用しています。

しかし、一部の反復がスキップされた場合など、その仮定が当てはまらない場合、プログラムは開始されていないゴルーチンを待機している状態でスタックする可能性があります。正直に言うと、これは追跡するのが非常に困難な種類のバグです。

この場合、wg.Add(1) の方が適しています。パフォーマンスのオーバーヘッドがわずかに発生する可能性がありますが、人的エラーのオーバーヘッドに対処するよりもはるかに優れています。

sync.WaitGroup を使用するときによくある間違いもあります。

for i := 0; i < 10; i++ {
    go func() {
        wg.Add(1)  
        defer wg.Done()
        ...
    }()
}

結局のところ、wg.Add(1) はゴルーチンの内部で呼び出されています。メインの goroutine が wg.Wait() を呼び出した後で goroutine が実行を開始する可能性があるため、これが問題になる可能性があります。

これにより、あらゆる種類のタイミングの問題が発生する可能性があります。また、上記のすべての例では wg.Done() で defer を使用していることに気づいたでしょう。実際、複数のリターン パスやパニック リカバリの問題を回避するには、defer とともに使用して、常に呼び出され、呼び出し元を無期限にブロックしないようにする必要があります。

これで基本はすべてカバーできるはずです。

sync.WaitGroup はどのように見えるか?

まず、sync.WaitGroup のソース コードをチェックアウトしましょう。 sync.Mutex にも同様のパターンがあることがわかります。

繰り返しになりますが、ミューテックスの仕組みに詳しくない場合は、まずこの記事「Go Sync Mutex: Normal & Starvation Mode」を参照することを強くお勧めします。

type WaitGroup struct {
    noCopy noCopy

    state atomic.Uint64 
    sema  uint32
}

type noCopy struct{}

func (*noCopy) Lock()   {}
func (*noCopy) Unlock() {}

Go では、構造体を別の変数に代入するだけで簡単にコピーできます。ただし、WaitGroup などの一部の構造体は実際にはコピーすべきではありません。

Copying a WaitGroup can mess things up because the internal state that tracks the goroutines and their synchronization can get out of sync between the copies. If you've read the mutex post, you'll get the idea, imagine what could go wrong if we copied the internal state of a mutex.

The same kind of issues can happen with WaitGroup.

noCopy

The noCopy struct is included in WaitGroup as a way to help prevent copying mistakes, not by throwing errors, but by serving as a warning. It was contributed by Aliaksandr Valialkin, CTO of VictoriaMetrics, and was introduced in change #22015.

The noCopy struct doesn't actually affect how your program runs. Instead, it acts as a marker that tools like go vet can pick up on to detect when a struct has been copied in a way that it shouldn't be.

type noCopy struct{}

func (*noCopy) Lock()   {}
func (*noCopy) Unlock() {}

Its structure is super simple:

1. It has no fields, so it doesn't take up any meaningful space in memory.
2. It has two methods, Lock and Unlock, which do nothing (no-op). These methods are there just to work with the -copylocks checker in the go vet tool.

When you run go vet on your code, it checks to see if any structs with a noCopy field, like WaitGroup, have been copied in a way that could cause issues.

It will throw an error to let you know there might be a problem. This gives you a heads-up to fix it before it turns into a bug:

func main() {
    var a sync.WaitGroup
    b := a

    fmt.Println(a, b)
}

// go vet:
// assignment copies lock value to b: sync.WaitGroup contains sync.noCopy
// call of fmt.Println copies lock value: sync.WaitGroup contains sync.noCopy
// call of fmt.Println copies lock value: sync.WaitGroup contains sync.noCopy

In this case, go vet will warn you about 3 different spots where the copying happens. You can try it yourself at: Go Playground.

Note that it's purely a safeguard for when we're writing and testing our code, we can still run it like normal.

Internal State

The state of a WaitGroup is stored in an atomic.Uint64 variable. You might have guessed this if you've read the mutex post, there are several things packed into this single value.

WaitGroup structure

Here's how it breaks down:

• Counter (high 32 bits): This part keeps track of the number of goroutines the WaitGroup is waiting for. When you call wg.Add() with a positive value, it bumps up this counter, and when you call wg.Done(), it decreases the counter by one.
• Waiter (low 32 bits): This tracks the number of goroutines currently waiting for that counter (the high 32 bits) to hit zero. Every time you call wg.Wait(), it increases this "waiter" count. Once the counter reaches zero, it releases all the goroutines that were waiting.

Then there's the final field, sema uint32, which is an internal semaphore managed by the Go runtime.

when a goroutine calls wg.Wait() and the counter isn't zero, it increases the waiter count and then blocks by calling runtime_Semacquire(&wg.sema). This function call puts the goroutine to sleep until it gets woken up by a corresponding runtime_Semrelease(&wg.sema) call.

We'll dive deeper into this in another article, but for now, I want to focus on the alignment issues.

Alignment Problem

I know, talking about history might seem dull, especially when you just want to get to the point. But trust me, knowing the past is the best way to understand where we are now.

Let's take a quick look at how WaitGroup has evolved over several Go versions:

sync.WaitGroup in different Go versions

I can tell you, the core of WaitGroup (the counter, waiter, and semaphore) hasn't really changed across different Go versions. However, the way these elements are structured has been modified many times.

When we talk about alignment, we're referring to the need for data types to be stored at specific memory addresses to allow for efficient access.

For example, on a 64-bit system, a 64-bit value like uint64 should ideally be stored at a memory address that's a multiple of 8 bytes. The reason is, the CPU can grab aligned data in one go, but if the data isn't aligned, it might take multiple operations to access it.

Alignment issues

Now, here's where things get tricky:

On 32-bit architectures, the compiler doesn't guarantee that 64-bit values will be aligned on an 8-byte boundary. Instead, they might only be aligned on a 4-byte boundary.

This becomes a problem when we use the atomic package to perform operations on the state variable. The atomic package specifically notes:

"On ARM, 386, and 32-bit MIPS, it is the caller's responsibility to arrange for 64-bit alignment of 64-bit words accessed atomically via the primitive atomic functions." - atomic package note

What this means is that if we don't align the state uint64 variable to an 8-byte boundary on these 32-bit architectures, it could cause the program to crash.

So, what's the fix? Let's take a look at how this has been handled across different versions.

Go 1.5: state1 [12]byte

I'd recommend taking a moment to guess the underlying logic of this solution as you read the code below, then we'll walk through it together.

type WaitGroup struct {
    state1 [12]byte
    sema   uint32
}

func (wg *WaitGroup) state() *uint64 {
    if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 {
        return (*uint64)(unsafe.Pointer(&wg.state1))
    } else {
        return (*uint64)(unsafe.Pointer(&wg.state1[4]))
    }
}

Instead of directly using a uint64 for state, WaitGroup sets aside 12 bytes in an array (state1 [12]byte). This might seem like more than you'd need, but there's a reason behind it.

WaitGroup in Go 1.5

The purpose of using 12 bytes is to ensure there's enough room to find an 8-byte segment that's properly aligned.

The full post is available here: https://victoriametrics.com/blog/go-sync-waitgroup/

以上がGo sync.WaitGroup と調整の問題の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。