5.5 同步組 · golan基礎

# 5.5 同步組 sync.WaitGroup 可以達到并發 Goroutine 的執行屏障的效果，等待多個 Goroutine 執行完畢。 ## 結構我們首先來看 WaitGroup 的內部結構： ``` // WaitGroup 用于等待一組 Goroutine 執行完畢。 // 主 Goroutine 調用 Add 來設置需要等待的 Goroutine 的數量 // 然后每個 Goroutine 運行并調用 Done 來確認已經執行網完畢 // 同時，Wait 可以用于阻塞并等待所有 Goroutine 完成。 // // WaitGroup 在第一次使用后不能被復制 type WaitGroup struct { // 64 位值: 高 32 位用于計數，低 32 位用于等待計數 // 64 位的原子操作要求 64 位對齊，但 32 位編譯器無法保證這個要求 // 因此分配 12 字節然后將他們對齊，其中 8 字節作為狀態，其他 4 字節用于存儲原語 state1 [3]uint32 } ``` 可以看到，WaitGroup 內部僅僅只是一個 uint32 類型的數組。由于需要考慮 32 位機器的兼容性，這里采用了 uint32 結構的數組，保證在不同類型的機器上都是 12 個字節。通過`state()`函數來確定實際的存儲情況： ``` // state 返回 wg.state1 中存儲的狀態和原語字段 func (wg *WaitGroup) state() (statep *uint64, semap *uint32) { if uintptr(unsafe.Pointer(&wg.state1))%8 == 0 { return (*uint64)(unsafe.Pointer(&wg.state1)), &wg.state1[2] } return (*uint64)(unsafe.Pointer(&wg.state1[1])), &wg.state1[0] } ``` * 在 32 位機器上`state1[0]`和`state1[1]`分別用于計數和等待計數，而最后一個`state1[2]`用于存儲原語。 * 在 64 位機器上`state1[0]`作為存儲原語，而`state[1]`和`state[2]`用于計數和等待計數 ## `Add()`/`Done()` 先來看簡單的 Done 操作： ``` func (wg *WaitGroup) Done() { wg.Add(-1) } ``` 所以 Done 調用本質上還是 Add 操作，再來看 Add： ``` // Add 將 delta（可能為負）加到 WaitGroup 的計數器上 // 如果計數器歸零，則所有阻塞在 Wait 的 Goroutine 被釋放 // 如果計數器為負，則 panic // // 請注意，當計數器為 0 時發生的帶有正的 delta 的調用必須在 Wait 之前。 // 當計數器大于 0 時，帶有負 delta 的調用或帶有正 delta 調用可能在任何時候發生。 // 通常，這意味著 Add 調用必須發生在 Goroutine 創建之前或其他被等待事件之前。 // 如果一個 WaitGroup 被復用于等待幾個不同的獨立事件集合，必須在前一個 Wait 調用返回后才能調用 Add。 func (wg *WaitGroup) Add(delta int) { // 首先獲取狀態指針和存儲指針 statep, semap := wg.state() (...) // 將 delta 加到 statep 的前 32 位上，即加到計數器上 state := atomic.AddUint64(statep, uint64(delta)<<32) // 計數器的值 v := int32(state >> 32) // 等待器的值 w := uint32(state) (...) // 如果實際技術為負則直接 panic，因此是不允許計數為負值的 if v < 0 { panic("sync: negative WaitGroup counter") } // 如果等待器不為零，但 delta 是處于增加的狀態，而且存儲計數與 delta 的值相同，則立即 panic if w != 0 && delta > 0 && v == int32(delta) { panic("sync: WaitGroup misuse: Add called concurrently with Wait") } // 如果計數器 > 0 或者等待器為 0 則一切都很好，直接返回 if v > 0 || w == 0 { return } // 這時 Goroutine 已經將計數器清零，且等待器大于零（并發調用導致） // 這時不允許出現并發使用導致的狀態突變，否則就應該 panic // - Add 不能與 Wait 并發調用 // - Wait 在計數器已經歸零的情況下，不能再繼續增加等待器了 // 仍然檢查來保證 WaitGroup 不會被濫用 if *statep != state { panic("sync: WaitGroup misuse: Add called concurrently with Wait") } // 結束后將等待器清零 *statep = 0 // 等待器大于零，減少 runtime_Semrelease 產生的阻塞 for ; w != 0; w-- { runtime_Semrelease(semap, false, 0) } } ``` Add 將 statep 的值作為兩段來處理，前 32 位處理為計數器，后 32 位處理為等待器。 * 在初始階段，等待器為 0 ，計數器隨著 Add 正數的調用而增加。 * 如果 Add 使用錯誤導致計數器為負，則會立即 panic * 由于并發的效果，計數器和等待器的值是分開操作的，因此可能出現計數器已經為零（說明當前 Add 的操作為負，即 Done），但等待器為正的情況，依次調用存儲原語釋放產生的阻塞（本質上為加 1 操作）我們來考慮一個使用場景，首先剛創建的 WaitGroup 所有值為零： ~~~ statep 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 ~~~ 這時候調用 Add(1)： > 注意，有符號數為補碼表示，最高位為符號位 ~~~ int64(delta) 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0001 int64(delta)<<32 0000 0000 0000 0000 0000 0000 0000 0001 0000 0000 0000 0000 0000 0000 0000 0000 statep 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 state 0000 0000 0000 0000 0000 0000 0000 0001 0000 0000 0000 0000 0000 0000 0000 0000 ~~~ 那么這時候的`v`（計數器）為 1，而`w`等待器為 0。再來執行一遍減一的操作。再減一之前： ~~~ statep 0000 0000 0000 0000 0000 0000 0000 0001 0000 0000 0000 0000 0000 0000 0000 0000 ~~~ 減一： > 注意，有符號數為補碼表示，最高位為符號位 ~~~ int64(delta) 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 1111 int64(delta)<<32 1111 1111 1111 1111 1111 1111 1111 1111 0000 0000 0000 0000 0000 0000 0000 0000 statep 0000 0000 0000 0000 0000 0000 0000 0001 0000 0000 0000 0000 0000 0000 0000 0000 state 10000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 0000 ~~~ 即計數器歸零。 ## `Wait()` 當 Add 和 Done 都被合理的設置后，我們希望等待所有的 Goroutine 結束，Wait 提供了這樣的機制： ``` // Wait 會保持阻塞直到 WaitGroup 計數器歸零 func (wg *WaitGroup) Wait() { // 先獲得計數器和存儲原語 statep, semap := wg.state() (...) // 一個簡單的死循環，只有當計數器歸零才會結束 for { // 原子讀 state := atomic.LoadUint64(statep) // 計數 v := int32(state >> 32) // 無符號計數 w := uint32(state) // 如果計數器已經歸零，則直接退出循環 if v == 0 { (...) return } // 增加等待計數，此處的原語會比較 statep 和 state 的值，如果相同則等待計數加 1 if atomic.CompareAndSwapUint64(statep, state, state+1) { (...) // 會阻塞到存儲原語是否 > 0（即睡眠），如果 *semap > 0 則會減 1，因此最終的 semap 理論為 0 runtime_Semacquire(semap) // 在這種情況下，如果 *semap 不等于 0 ，則說明使用失誤，直接 panic if *statep != 0 { panic("sync: WaitGroup is reused before previous Wait has returned") } (...) return } } } ``` 可以看到 Wait 使用的是一個簡單的死循環來進行操作。在循環體中，每次先讀取計數器和等待器的值。然后增加等待計數，如果增加成功，會調用`runtime_Semacquire`來阻塞當前的死循環，直到存儲原語的值被`runtime_Semrelease`減少后才會解除阻塞狀態進入下一個循環。我們來完成考慮一下整個流程： ``` wg := sync.WaitGroup{} wg.Add(1) go func() { wg.Done() }() wg.Wait() ``` 在 wg 創建之初，計數器、等待器、存儲原語的值均初始化為零值。不妨假設調用`wg.Add(1)`，則計數器加 1 等待器、存儲原語保持不變，均為 0。 `wg.Done()`和`wg.Wait()`的調用順序可能成兩種情況： **情況 1**：先調用`wg.Done()`再調用`wg.Wait()`。這時候`wg.Done()`使計數器減 1 ，這時計數器、等待器、存儲原語均為 0，由于等待器為 0 則`runtime_Semrelease`不會被調用。于是當`wg.Wait()`開始調用時，讀取到計數器已經為 0，循環退出，`wg.Wait()`調用完畢。 **情況 2**：先調用`wg.Wait()`再調用`wg.Done()`。這時候`wg.Wait()`開始調用時，讀取到計數器為 1，則為等待器加 1，并調用`runtime_Semacquire`開始阻塞在存儲原語為 0 的狀態。在阻塞的過程中，Goroutine 被調度器調度，開始執行`wg.Done()`，于是計數器清零，但由于等待器為 1 大于零。這時將等待器也清零，并調用與等待器技術相同次數（此處為 1 次）的`runtime_Semrelease`，這導致存儲原語的值變為 1，計數器和等待器均為零。這時，`runtime_Semacquire`在存儲原語大于零后被喚醒，這時檢查計數器和等待器是否為零（如果不為零則說明 Add 與 Wait 產生并發調用，直接 panic），這時他們為 0，因此進入下一個循環，當再次讀取計數器時，發現計數器已經清理，于是退出`wg.Wait()`調用，結束阻塞。