5.6 緩存池 · golan基礎

# 5.6 緩存池 `sync.Pool`是一個臨時對象池。一句話來概括，`sync.Pool`管理了一組臨時對象，當需要時從池中獲取，使用完畢后從再放回池中，以供他人使用。其公共方法與成員包括： ``` type Pool struct { New func() interface{} ... } // Get 從 Pool 中選擇一個任意的對象，將其移出 Pool, 并返回給調用方。 // Get 可能會返回一個非零值對象（被其他人使用過），因此調用方不應假設 // 返回的對象具有任何形式的狀態。 func (p *Pool) Get() interface{} { ... } func (p *Pool) Put(x interface{}) { ... } ``` 使用`sync.Pool`只需要指定`sync.Pool`對象的創建方法`New`，則在使用`sync.Pool.Get`失敗的情況下，會池的內部會選擇性的創建一個新的值。因此獲取到的對象可能是剛被使用完畢放回池中的對象、亦或者是由`New`創建的新對象。 ## 底層結構 `sync.Pool`未公開的字段包括： ``` type Pool struct { local unsafe.Pointer // local 固定大小 per-P 數組, 實際類型為 [P]poolLocal localSize uintptr // local array 的大小 victim unsafe.Pointer // 來自前一個周期的 local victimSize uintptr // victim 數組的大小 ... } ``` 其內部本質上保存了一個`poolLocal`元素的數組，即`local`，每個`poolLocal`都只被一個 P 擁有，而`victim`則緩存了上一個垃圾回收周期的`local`。而`poolLocal`則由`private`和`shared`兩個字段組成： ``` type poolLocalInternal struct { private interface{} shared poolChain } type poolLocal struct { poolLocalInternal pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte } ``` 從前面結構體的字段不難猜測，`private`是一個僅用于當前 P 進行讀寫的字段（即沒有并發讀寫的問題），而 shared 則遵循字面意思，可以在多個 P 之間進行共享讀寫，是一個`poolChain`鏈式隊列結構，我們先記住這個結構在局部 P 上可以進行`pushHead`和`popHead`操作（隊頭讀寫），在所有 P 上都可以進行`popTail`（隊尾出隊）操作，之后再來詳細看它的實現細節。 ## Get 當從池中獲取對象時，會先從 per-P 的`poolLocal`slice 中選取一個`poolLocal`，選擇策略遵循： 1. 優先從 private 中選擇對象 2. 若取不到，則嘗試從`shared`隊列的隊頭進行讀取 3. 若取不到，則嘗試從其他的 P 中進行偷取`getSlow` 4. 若還是取不到，則使用 New 方法新建 ``` func (p *Pool) Get() interface{} { ... // 獲取一個 poolLocal l, pid := p.pin() // 先從 private 獲取對象 x := l.private l.private = nil if x == nil { // 嘗試從 localPool 的 shared 隊列隊頭讀取， // 因為隊頭的內存局部性比隊尾更好。 x, _ = l.shared.popHead() // 如果取不到，則獲取新的緩存對象 if x == nil { x = p.getSlow(pid) } } runtime_procUnpin() ... // 如果 getSlow 還是獲取不到，則 New 一個 if x == nil && p.New != nil { x = p.New() } return x } ``` 其實我們不難看出： 1. `private`只保存了一個對象; 2. 第一次從`shared`中取對象時，還未涉及跨 P 讀寫，因此`popHead`是可用的； 3. 當`shared`讀取不到對象時，說明當前局部 P 所持有的`localPool`不包含任何對象，這時嘗試從其他的`localPool`進行偷取。 4. 實在是偷不到，才考慮新創建一個對象。 ## Put `Put`的過程則相對簡單，只需要將對象放回到池中。與`Get`取出一樣，放回遵循策略： 1. 優先放入`private` 2. 如果 private 已經有值，即不能放入，則嘗試放入`shared` ``` // Put 將 x 放回到池中 func (p *Pool) Put(x interface{}) { if x == nil { return } ... // 獲得一個 localPool l, _ := p.pin() // 優先放入 private if l.private == nil { l.private = x x = nil } // 如果不能放入 private 則放入 shared if x != nil { l.shared.pushHead(x) } runtime_procUnpin() ... } ``` ## 偷取細節上面已經介紹了`Get/Put`的具體策略。我們還有一些細節需要處理。 ### `pin()`與`pinSlow()` `pin()`用于取當前 P 中的`poolLocal`。我們來仔細看一下它的實現細節。 ``` // pin 會將當前的 goroutine 固定到 P 上，禁用搶占，并返回 localPool 池以及當前 P 的 pid。 func (p *Pool) pin() (*poolLocal, int) { // 返回當前 P.id pid := runtime_procPin() // 在 pinSlow 中會存儲 localSize 后再存儲 local，因此這里反過來讀取 // 因為我們已經禁用了搶占，這時不會發生 GC // 因此，我們必須觀察 local 和 localSize 是否對應 // 觀察到一個全新或很大的的 local 是正常行為 s := atomic.LoadUintptr(&p.localSize) // load-acquire l := p.local // load-consume // 因為可能存在動態的 P（運行時調整 P 的個數）procresize/GOMAXPROCS // 如果 P.id 沒有越界，則直接返回 if uintptr(pid) < s { return indexLocal(l, pid) } // 沒有結果時，涉及全局加鎖 // 例如重新分配數組內存，添加到全局列表 return p.pinSlow() } ``` `pin()`首先會調用運行時實現獲得當前 P 的 id，將 P 設置為禁止搶占，達到固定當前 goroutine 的目的。然后檢查`pid`與`p.localSize`的值來確保從`p.local`中取值不會發生越界。如果不會發生，則調用`indexLocal()`完成取值。否則還需要繼續調用`pinSlow()`。 ``` func indexLocal(l unsafe.Pointer, i int) *poolLocal { // 簡單的通過 p.local 的頭指針與索引來第 i 個 pooLocal lp := unsafe.Pointer(uintptr(l) + uintptr(i)*unsafe.Sizeof(poolLocal{})) return (*poolLocal)(lp) } ``` 在這個過程中我們可以看到在運行時調整 P 的大小的代價。如果此時 P 被調大，而沒有對應的`poolLocal`時，必須在取之前創建好，從而必須依賴全局加鎖，這對于以性能著稱的池化概念是比較致命的。既然需要對全局進行加鎖，`pinSlow()`會首先取消 P 的禁止搶占，這是因為使用 mutex 時 P 必須為可搶占的狀態。然后使用`allPoolsMu`進行加鎖。當完成加鎖后，再重新固定 P ，取其 pid。注意，因為中途可能已經被其他的線程調用，因此這時候需要再次對 pid 進行檢查。如果 pid 在 p.local 大小范圍內，則不再此時創建，直接返回。如果`p.local`為空，則將 p 扔給`allPools`并在垃圾回收階段回收所有 Pool 實例。最后再完成對`p.local`的創建（徹底丟棄舊數組）： ``` var ( allPoolsMu Mutex // allPools 是一組 pool 的集合，具有非空主緩存。 // 有兩種形式來保護它的讀寫：1. allPoolsMu 鎖; 2. STW. allPools []*Pool ) func (p *Pool) pinSlow() (*poolLocal, int) { // 這時取消 P 的禁止搶占，因為使用 mutex 時候 P 必須可搶占 runtime_procUnpin() // 加鎖 allPoolsMu.Lock() defer allPoolsMu.Unlock() // 當鎖住后，再次固定 P 取其 id pid := runtime_procPin() // 并再次檢查是否符合條件，因為可能中途已被其他線程調用 // 當再次固定 P 時 poolCleanup 不會被調用 s := p.localSize l := p.local if uintptr(pid) < s { return indexLocal(l, pid), pid } // 如果數組為空，新建 // 將其添加到 allPools，垃圾回收器從這里獲取所有 Pool 實例 if p.local == nil { allPools = append(allPools, p) } // 根據 P 數量創建 slice，如果 GOMAXPROCS 在 GC 間發生變化 // 我們重新分配此數組并丟棄舊的 size := runtime.GOMAXPROCS(0) local := make([]poolLocal, size) // 將底層數組起始指針保存到 p.local，并設置 p.localSize atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-release atomic.StoreUintptr(&p.localSize, uintptr(size)) // store-release // 返回所需的 pollLocal return &local[pid], pid } ``` ### `getSlow()` 終于，我們獲取到了`poolLocal`，現在回到我們`Get`的取值過程。在取對象的過程中，我們仍然會面臨既不能從`private`取、也不能從`shared`中取得尷尬境地。這時候就來到了`getSlow()`。試想，如果我們在本地的 P 中取不到值，是不是可以考慮從別人那里偷一點過來？總會比創建一個新的要快。因此，我們再次固定 P，并取得當前的 P.id 來從其他 P 中偷值，那么我們需要先獲取到其他 P 對應的`poolLocal`。假設`size`為數組的大小，`local`為`p.local`，那么嘗試遍歷其他所有 P： ``` for i := 0; i < int(size); i++ { // 獲取目標 poolLocal, 引入 pid 保證不是自身 l := indexLocal(local, (pid+i+1)%int(size)) ``` 我們來證明一下此處確實不會發生取到自身的情況，不妨設：`pid = (pid+i+1)%size`則`pid+i+1 = a*size+pid`。即：`a*size = i+1`，其中 a 為整數。由于`i<size`，于是`a*size = i+1 < size+1`，則：`(a-1)*size < 1`\==>`size < 1 / (a-1)`，由于`size`為非負整數，這是不可能的。因此當取到其他`poolLocal`時，便能從 shared 中取對象了。 ``` func (p *Pool) getSlow(pid int) (x interface{}) { size := atomic.LoadUintptr(&p.localSize) // load-acquire local := p.local // load-consume for i := 0; i < int(size); i++ { // 獲取目標 poolLocal, 引入 pid 保證不是自身 l := indexLocal(local, (pid+i+1)%int(size)) // 從其他的 P 中固定的 localPool 的 share 隊列的隊尾偷一個緩存對象 if x, _ := l.shared.popTail(); x != nil { return x } } // 當 local 失敗后，嘗試再嘗試從上一個垃圾回收周期遺留下來的 victim。 // 如果 pid 比 victim 遺留的 localPool 還大，則說明從根據此 pid 從 // victim 獲取 localPool 會發生越界（同時也表明此時 P 的數量已經發生變化） // 這時無法繼續讀取，直接返回 nil size = atomic.LoadUintptr(&p.victimSize) if uintptr(pid) >= size { return nil } // 獲取 localPool，并優先讀取 private locals = p.victim l := indexLocal(locals, pid) if x := l.private; x != nil { l.private = nil return x } for i := 0; i < int(size); i++ { l := indexLocal(locals, (pid+i)%int(size)) // 從其他的 P 中固定的 localPool 的 share 隊列的隊尾偷一個緩存對象 if x, _ := l.shared.popTail(); x != nil { return x } } // 將 victim 緩存置空，從而確保之后的 get 操作不再讀取此處的值 atomic.StoreUintptr(&p.victimSize, 0) return nil } ``` ## 緩存的回收 `sync.Pool`的垃圾回收發生在運行時 GC 開始之前。在`src/sync/pool.go`中: ``` // 將緩存清理函數注冊到運行時 GC 時間段 func init() { runtime_registerPoolCleanup(poolCleanup) } // 由運行時實現 func runtime_registerPoolCleanup(cleanup func()) ``` 在`src/runtime/mgc.go`中: ``` // 開始 GC func gcStart(trigger gcTrigger) { ... clearpools() ... } // 實現緩存清理 func clearpools() { // clear sync.Pools if poolcleanup != nil { poolcleanup() } ... } var poolcleanup func() // 利用編譯器標志將 sync 包中的清理注冊到運行時 //go:linkname sync_runtime_registerPoolCleanup sync.runtime_registerPoolCleanup func sync_runtime_registerPoolCleanup(f func()) { poolcleanup = f } ``` 再來看實際的清理函數： ``` // oldPools 是一組 pool 的集合，具有非空 victim 緩存。由 STW 保護 var oldPools []*Pool func poolCleanup() { // 該函數會注冊到運行時 GC 階段(前)，此時為 STW 狀態，不需要加鎖 // 它必須不處理分配且不調用任何運行時函數。 // 由于此時是 STW，不存在用戶態代碼能嘗試讀取 localPool，進而所有的 P 都已固定（與 goroutine 綁定） // 從所有的 oldPols 中刪除 victim for _, p := range oldPools { p.victim = nil p.victimSize = 0 } // 將主緩存移動到 victim 緩存 for _, p := range allPools { p.victim = p.local p.victimSize = p.localSize p.local = nil p.localSize = 0 } // 具有非空主緩存的池現在具有非空的 victim 緩存，并且沒有任何 pool 具有主緩存。 oldPools, allPools = allPools, nil } ``` 注意，即便是最后`p.local`已經被置換到`oldPools`的`p.victim`，其中的緩存對象仍然有可能被偷取放回到`allPools`中，從而延緩了`victim`中緩存對象被回收的速度。 ## `poolChain` 前面已經看到 poolChain 的功能了：一個隊首非并發安全、隊尾并發安全的鏈式隊列（變長）。它的結構包含隊頭和隊尾的兩個`poolChainElt`指針： ``` type poolChain struct { head *poolChainElt tail *poolChainElt } ``` 而從`poolChainElt`的結構我們可以看出，這是一個雙向隊列，包含`next`和`prev`指針： ``` type poolChainElt struct { poolDequeue next, prev *poolChainElt } ``` 其中的`poolDequeue`是一個單生產者、多消費者的固定長度的環狀隊列，其中 headTail 字段的前 32 位表示了下一個需要被填充的對象槽的索引，而后 32 位則表示了隊列中最先被插入的數據的索引，`eface`數組存儲了實際的對象，其 eface 依賴運行時對`interface{}`的實現，即一個`interface{}`由`typ`和`val`兩段數據組成： ``` type poolDequeue struct { headTail uint64 vals []eface } type eface struct { typ, val unsafe.Pointer } ``` 因此`poolChain`本質上串聯了若干個`poolDequeue`。 ### `poolChain`的`popHead`、`pushHead`和`popTail` `poolChain`實際上是多個生產者消費者模型的鏈表。對于一個局部 P 而言，充當了多個隊頭的單一生產者，它可以安全的在整個鏈表中所串聯的隊列的隊頭進行操作。而其他的多個 P 而言，則充當了多個隊尾的消費者，可以在所串聯的隊列的隊尾進行消費（偷取）。 `popHead`操作發生在從本地 shared 隊列中消費并獲取對象（消費者）。`pushHead`操作發生在向本地 shared 隊列中放置對象（生產者）。`popTail`操作則發生在從其他 P 的 shared 隊列中偷取的過程。 ``` const ( dequeueBits = 32 dequeueLimit = (1 << dequeueBits) / 4 ) func (c *poolChain) popHead() (interface{}, bool) { d := c.head // d 是一個 poolDequeue，如果 d.popHead 是并發安全的， // 那么這里取 val 也是并發安全的。若 d.popHead 失敗，則 // 說明需要重新嘗試。這個過程會持續到整個鏈表為空。 for d != nil { if val, ok := d.popHead(); ok { return val, ok } d = loadPoolChainElt(&d.prev) } return nil, false } func (c *poolChain) pushHead(val interface{}) { d := c.head // 如果鏈表空，則創建一個新的鏈表 if d == nil { const initSize = 8 // 固定長度為 8，必須為 2 的指數 d = new(poolChainElt) d.vals = make([]eface, initSize) c.head = d storePoolChainElt(&c.tail, d) } // 如果向隊列中存值失敗，則檢查是否當前隊列已滿 if d.pushHead(val) { return } newSize := len(d.vals) * 2 if newSize >= dequeueLimit { newSize = dequeueLimit } // 如果已滿，則創建一個新的 poolDequeue // 由于是新創建的，則 push 一定會成功 d2 := &poolChainElt{prev: d} d2.vals = make([]eface, newSize) c.head = d2 storePoolChainElt(&d.next, d2) d2.pushHead(val) } func (c *poolChain) popTail() (interface{}, bool) { d := loadPoolChainElt(&c.tail) if d == nil { return nil, false } // 普通的 CAS 操作 for { d2 := loadPoolChainElt(&d.next) if val, ok := d.popTail(); ok { return val, ok } if d2 == nil { return nil, false } if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) { storePoolChainElt(&d2.prev, nil) } d = d2 } } ``` ### `poolDequeue`的`popHead`、`pushHead`和`popTail` 正如前面所說`poolDequeue`是一個單生產者、多消費者的固定長度的環狀隊列，`popHead`、`pushHead`由局部的 P 操作隊首，而`popTail`由其他并行的 P 操作隊尾。其中`headTail`字段的前 32 位表示了下一個需要被填充的對象槽的索引，而后 32 位則表示了隊列中最先被插入的數據的索引。通過`pack`/`unpack`方法來實現對`head`和`tail` 的讀寫： ``` // 將 head 和 tail 指針從 d.headTail 中分離開來 func (d *poolDequeue) unpack(ptrs uint64) (head, tail uint32) { const mask = 1<<dequeueBits - 1 head = uint32((ptrs >> dequeueBits) & mask) tail = uint32(ptrs & mask) return } // 將 head 和 tail 指針打包到 d.headTail 一個 64bit 的變量中 func (d *poolDequeue) pack(head, tail uint32) uint64 { const mask = 1<<dequeueBits - 1 return (uint64(head) << dequeueBits) | uint64(tail&mask) } ``` 從`poolChain`的實現中我們可以看到，每個`poolDequeue`的`vals`長度為 8。但由于是循環隊列，實現中并不關心隊列的長度，只要收尾元素的索引相等，則說明隊列已滿。因此通過 CAS 原語實現單一生產者的對隊頭的讀`popHead`和寫`pushHead`： ``` func (d *poolDequeue) popHead() (interface{}, bool) { var slot *eface for { ptrs := atomic.LoadUint64(&d.headTail) head, tail := d.unpack(ptrs) if tail == head { return nil, false // 隊列滿 } head-- ptrs2 := d.pack(head, tail) if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) { slot = &d.vals[head&uint32(len(d.vals)-1)] break } } val := *(*interface{})(unsafe.Pointer(slot)) if val == dequeueNil(nil) { val = nil } *slot = eface{} return val, true } func (d *poolDequeue) pushHead(val interface{}) bool { ptrs := atomic.LoadUint64(&d.headTail) head, tail := d.unpack(ptrs) if (tail+uint32(len(d.vals)))&(1<<dequeueBits-1) == head { return false // 隊列滿 } slot := &d.vals[head&uint32(len(d.vals)-1)] // 此處可能與 popTail 發生競爭，參見 popTail typ := atomic.LoadPointer(&slot.typ) if typ != nil { return false } if val == nil { val = dequeueNil(nil) } *(*interface{})(unsafe.Pointer(slot)) = val atomic.AddUint64(&d.headTail, 1<<dequeueBits) return true } ``` 以及多個消費者讀的處理手段非常巧妙，通過`interface{}`的 typ 和 val 兩段式結構的讀寫先后順序，在`popTail`和`pushHead`之間消除了競爭： ``` func (d *poolDequeue) popTail() (interface{}, bool) { var slot *eface for { ptrs := atomic.LoadUint64(&d.headTail) head, tail := d.unpack(ptrs) if tail == head { return nil, false // 隊列滿 } ptrs2 := d.pack(head, tail+1) if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) { slot = &d.vals[tail&uint32(len(d.vals)-1)] break } } val := *(*interface{})(unsafe.Pointer(slot)) if val == dequeueNil(nil) { val = nil } // 注意：此處可能與 pushHead 發生競爭，解決方案是： // 1. 讓 pushHead 先讀取 typ 的值，如果 typ 值不為 nil，則說明 popTail 尚未清理完 slot // 2. 讓 popTail 先清理掉 val 中的內容，在清理掉 typ，從而確保不會與 pushHead 對 slot 的寫行為發生競爭 slot.val = nil atomic.StorePointer(&slot.typ, nil) return val, true } ``` ## 小結至此，我們完整分析了 sync.Pool 的所有代碼。總結： ~~~ goroutine goroutine goroutine | | | P P P | | | private private private | | | [ poolLocal poolLocal poolLocal ] sync.Pool | | | shared shared shared ~~~ 一個 goroutine 固定在 P 上，從當前 P 對應的`private`取值， shared 字段作為一個優化過的鏈式無鎖變長隊列，當在`private`取不到值的情況下，從對應的`shared`隊列的隊首取，若還是取不到，則嘗試從其他 P 的`shared`隊列隊尾中偷取。若偷不到，則嘗試從上一個 GC 周期遺留到`victim`緩存中取，否則調用`New`創建一個新的對象。對于回收而言，池中所有臨時對象在一次 GC 后會被放入`victim`緩存中，而前一個周期被放入`victim`的緩存則會被清理掉。對于調用方而言，當 Get 到臨時對象后，便脫離了池本身不受控制。用方有責任將使用完的對象放回池中。本文中介紹的`sync.Pool`實現為 Go 1.13 優化過后的版本，相較于之前的版本，主要有以下幾點優化： 1. 引入了`victim`（二級）緩存，每次 GC 周期不再清理所有的緩存對象，而是將`locals`中的對象暫時放入`victim`，從而延遲到下一個 GC 周期進行回收； 2. 在下一個周期到來前，`victim`中的緩存對象可能會被偷取，在`Put`操作后又重新回到`locals`中，這個過程發生在從其他 P 的`shared`隊列中偷取不到、以及`New`一個新對象之前，進而是在犧牲了`New`新對象的速度的情況下換取的； 3. `poolLocal`不再使用`Mutex`這類昂貴的鎖來保證并發安全，取而代之的是使用了 CAS 算法優化實現的`poolChain`變長無鎖雙向鏈式隊列。這種兩級緩存的優化的優勢在于： 1. 顯著降低了 GC 發生前清理當前周期中產生的大量緩存對象的影響：因為回收被推遲到了下個 GC 周期； 2. 顯著降低了 GC 發生后 New 對象的成本：因為密集的緩存對象讀寫可能從上個周期中未清理的對象中偷取。