詳解Go語言Sync.Pool為何不加鎖也能夠?qū)崿F(xiàn)線程安全

更新時間：2023年04月23日 08:37:26 作者：starrySky

在這篇文章中，我們將剖析sync.Pool內(nèi)部實現(xiàn)中，介紹了sync.Pool比較巧妙的內(nèi)部設(shè)計思路以及其實現(xiàn)方式。在這個過程中，也間接介紹了為何不加鎖也能夠?qū)崿F(xiàn)線程安全，感興趣的可以學(xué)習(xí)一下

1. 簡介

我們在 Sync.Pool: 提高go語言程序性能的關(guān)鍵一步一文中，已經(jīng)了解了使用sync.Pool來實現(xiàn)對象的復(fù)用以減少對象的頻繁創(chuàng)建和銷毀，以及使用sync.Pool的一些常見注意事項。

在這篇文章中，我們將剖析sync.Pool內(nèi)部實現(xiàn)中，介紹了sync.Pool比較巧妙的內(nèi)部設(shè)計思路以及其實現(xiàn)方式。在這個過程中，也間接介紹了為何不加鎖也能夠?qū)崿F(xiàn)線程安全。

主要會涉及到Go語言中實現(xiàn)并發(fā)的GMP模型以及其基本的調(diào)度原理，以及本地緩存的設(shè)計，無鎖隊列的使用這幾個部分的內(nèi)容，綜上這幾個方面的內(nèi)容實現(xiàn)了不加鎖也能夠保證線程安全。

2. GMP之間的綁定關(guān)系

為了之能夠幫助我們后續(xù)更好得理解sync.Pool的設(shè)計與實現(xiàn)，這里需要對GMP模型進行簡單的介紹。GMP模型是Go語言中的一種協(xié)作式調(diào)度模型，其中G表示Goroutine，M可以理解為內(nèi)核線程，P為邏輯處理器，簡單理解其維護了一條Goroutine隊列。

2.1 M和P的關(guān)系

在GMP模型中，M和P可以動態(tài)綁定，一個M可以在運行時綁定到任意一個P上，而一個P也可以與任意一個M綁定。這種綁定方式是動態(tài)的，可以根據(jù)實際情況進行靈活調(diào)整，從而實現(xiàn)更加高效的協(xié)程調(diào)度。

盡管M和P可以動態(tài)綁定，但在特定時間點，一個M只會對應(yīng)一個P。這是因為M是操作系統(tǒng)線程，而P是Go語言的邏輯處理器，Go語言的邏輯處理器需要在某個操作系統(tǒng)線程中運行，并且是被該邏輯處理器(P)單獨占用的。

P的數(shù)量一般是和CPU核數(shù)保持一致，每個P占用一個CPU核心來執(zhí)行，可以通過runtime.GOMAXPROCS函數(shù)來修改。不過在大多數(shù)情況下，不需要手動修改，Go語言的調(diào)度器會根據(jù)實際情況自動進行調(diào)整。

2.2 P和G的關(guān)系

剛創(chuàng)建的Goroutine會被放入當(dāng)前線程對應(yīng)P的本地隊列中等待被執(zhí)行。如果本地隊列已滿，則會放入全局隊列中，供其他線程的P來搶占執(zhí)行。

當(dāng)P空閑時，會嘗試從全局隊列中獲取Goroutine來執(zhí)行。如果全局隊列中沒有Goroutine，則會從其他處理器的本地運行隊列中"偷取"一些Goroutine來執(zhí)行。

如果協(xié)程執(zhí)行過程中遇到阻塞操作（比如等待I/O或者鎖），處理器（P）會立即將協(xié)程移出本地運行隊列，并執(zhí)行其他協(xié)程，直到被阻塞的協(xié)程可以繼續(xù)執(zhí)行為止。被阻塞的協(xié)程會被放到相應(yīng)的等待隊列中等待事件發(fā)生后再次被喚醒并加入到運行隊列中，但不一定還是放回原來的處理器(P)的等待隊列中。

從上述過程可以看出，G和P的綁定關(guān)系是動態(tài)綁定的，在不同的時間點，同一個G可能在不同的P上執(zhí)行，同時，在不同的時間點，P也會調(diào)度執(zhí)行不同的G。

2.3 總結(jié)

每個P在某個時刻只能綁定一個M，而每個G在某個時刻也只存在于某個P的等待隊列中，等待被調(diào)度執(zhí)行。這是GMP模型的基本調(diào)度原理，也是Go語言高效調(diào)度的核心所在。通過動態(tài)綁定和靈活調(diào)度，可以充分利用多核處理器的計算能力，從而實現(xiàn)高并發(fā)、高效率的協(xié)程調(diào)度。

通過對GMP模型的基本了解，能夠幫助我們后續(xù)更好得理解sync.Pool的設(shè)計與實現(xiàn)。

3.Sync.Pool與GMP模型

3.1 sync.Pool性能問題

這里我們回到sync.Pool, 可以簡單使用切片，存儲可復(fù)用的對象，在需要時從中取出對象，用完之后再重新放回池子中，實現(xiàn)對象的重復(fù)使用。

當(dāng)多個協(xié)程同時從 sync.Pool 中取對象時，會存在并發(fā)問題，因此需要實現(xiàn)并發(fā)安全。一種簡單的實現(xiàn)方式是加鎖，每個協(xié)程在取數(shù)據(jù)前先加鎖，然后獲取數(shù)據(jù)，再解鎖，實現(xiàn)串行讀取的效果。但是這種方式在并發(fā)比較大的場景下容易導(dǎo)致大量協(xié)程進入阻塞狀態(tài)，從而進一步降低性能。

因此，為了提高程序的性能，我們需要尋找一種減少并發(fā)沖突的方式。有什么方式能夠減少并發(fā)沖突呢?

3.2 基于GMP模型的改進

回到GMP模型，從第二節(jié)對GMP模型的介紹中，我們知道協(xié)程(G)需要在邏輯處理器(P)上執(zhí)行，而邏輯處理器的數(shù)量是有限的，一般與CPU核心數(shù)相同。而之前的sync.Pool實現(xiàn)方式是所有P競爭同一份數(shù)據(jù)，容易導(dǎo)致大量協(xié)程進入阻塞狀態(tài)，影響程序性能。

那我們這里，是不是能夠?qū)?sync.Pool 分成多個小的存儲池，每個P都用擁有一個小的存儲池呢? 在每個小存儲池中分別使用獨立的鎖進行并發(fā)控制。這樣可以避免多個協(xié)程同時競爭同一個全局鎖的情況，降低鎖的粒度，從而減少并發(fā)沖突。

協(xié)程運行時都需要綁定一個邏輯處理器(P)，此時每個P都有自己的數(shù)據(jù)緩存，需要對象時從綁定的P的緩存中獲取，用完后重新放回。這種實現(xiàn)方式減少了協(xié)程競爭同一份數(shù)據(jù)的情況，只有在同一個邏輯處理器上的協(xié)程才存在競爭，從而減少并發(fā)沖突，提升性能。

3.3 能不能完全不加鎖

在上面的實現(xiàn)中，處于不同的P上的協(xié)程都是操作不同的數(shù)據(jù)，此時并不會出現(xiàn)并發(fā)問題。唯一可能出現(xiàn)并發(fā)問題的地方，為協(xié)程在獲取緩存對象時，邏輯處理器中途調(diào)度其他協(xié)程來執(zhí)行，此時才可能導(dǎo)致的并發(fā)問題。那這里能不能避免并發(fā)呢?

那如果能夠?qū)f(xié)程固定到邏輯處理器P上，并且不允許被搶占，也就是該P上永遠都是執(zhí)行某一個協(xié)程，直到成功獲取緩存對象后，才允許邏輯處理器去調(diào)度執(zhí)行其他協(xié)程，那么就可以完全避免并發(fā)沖突的問題了。

因此，如果我們能夠做到協(xié)程在讀取緩沖池中的數(shù)據(jù)時，能夠完全占用邏輯處理器P，不會被搶占，此時就不會出現(xiàn)并發(fā)了，也不需要加鎖了。

幸運的是，runtime包中提供了runtime_procPin調(diào)用，可以將當(dāng)前協(xié)程固定到協(xié)程所在邏輯處理器P上，并且不允許被搶占，也就是邏輯處理器P一直都被當(dāng)前協(xié)程所獨享。在獲取緩存對象時，我們可以使用runtime_procPin將當(dāng)前協(xié)程固定到邏輯處理器P上，然后從該邏輯處理器P的緩存中獲取對象。這樣做不僅可以避免并發(fā)沖突，還可以避免上下文切換和鎖競爭等性能問題。

4. sync.Pool初步實現(xiàn)

下面來看看當(dāng)前sync.Pool的部分代碼，其原理便是上述所提到的方式。具體來說，每個邏輯處理器P保存一份數(shù)據(jù)，并利用runtime_procPin來避免同一邏輯處理器P中的協(xié)程發(fā)生并發(fā)沖突。

需要注意的是，下面所展示的代碼只是部分代碼，并不包含完整的實現(xiàn)。但是這些代碼涵蓋了前面所提到的實現(xiàn)方式。同時，為了講述方便，也修改部分實現(xiàn)，后文會說明當(dāng)前sync.Pool當(dāng)前真正的實現(xiàn)。

4.1 sync.Pool結(jié)構(gòu)體定義

type Pool struct {
   // 指向 poolLocal 結(jié)構(gòu)體切片的地址，長度與cpu核心數(shù)保持一致
   local     unsafe.Pointer // local fixed-size per-P pool, actual type is [P]poolLocal
   // 記錄當(dāng)前 poolLocal 切片的長度
   localSize uintptr        // size of the local array
   // New optionally specifies a function to generate
   // a value when Get would otherwise return nil.
   // It may not be changed concurrently with calls to Get.
   New func() any
}

type poolLocal struct {
   // 上文所說的小緩沖池的實現(xiàn)
   private []any       // Can be used only by the respective P.
}

其中，New 函數(shù)用于創(chuàng)建一個新的對象，以便向池中添加對象。當(dāng)池中沒有可用對象時，會調(diào)用該函數(shù)。local 是指向 poolLocal 切片的指針，poolLocal 即為上文提到的小緩沖池，每個邏輯處理器都有一個。

4.2 Put方法

func (p *Pool) Put(x any) {
   if x == nil {
      return
   }
   // 這里調(diào)用pin方法,獲取到poolLocal
   l, _ := p.pin()
   // 將對象重新放入邏輯處理的小緩沖池當(dāng)中
   l.private = l.private.append(x)
   // 這個為解除 Proccssor的固定，Processor能夠調(diào)度其他協(xié)程去執(zhí)行
   runtime_procUnpin()
}

Put方法是用于將對象重新放入緩沖池，首先調(diào)用pin方法獲取到poolLocal,然后將對象放入到poolLocal當(dāng)中，然后再通過runtime_procUnpin調(diào)用，解除對當(dāng)前P的綁定。

可以發(fā)現(xiàn)，其中比較重要的邏輯，是調(diào)用pin方法獲取到P對應(yīng)的poolLocal,下面我們來看pin方法的實現(xiàn)。

func (p *Pool) pin() (*poolLocal, int) {
    // 調(diào)用runtime_procPin,占用Processor,不會被搶占
   pid := runtime_procPin()
   // 獲取localSize字段的值
   s := runtime_LoadAcquintptr(&p.localSize) // load-acquire
   // 獲取poolLocal切片
   l := p.local                              // load-consume
   // pid為協(xié)程編號，如果pid < localSize的值，說明屬于該processor的緩沖池已經(jīng)創(chuàng)建好了
   if uintptr(pid) < s {
      // 根據(jù)pid獲取對應(yīng)的緩沖池
      return indexLocal(l, pid), pid
   }
   // 否則走下面邏輯
   return p.pinSlow()
}
func indexLocal(l unsafe.Pointer, i int) *poolLocal {
   // 直接通過Processor的編號，計算出偏移量獲取到對應(yīng)的poolLocal
   lp := unsafe.Pointer(uintptr(l) + uintptr(i)*unsafe.Sizeof(poolLocal{}))
   return (*poolLocal)(lp)
}

在函數(shù)開始時，pin方法通過調(diào)用runtime_procPin方法，占用當(dāng)前goroutine所在的P，不允許其他goroutine搶占該P，這里能夠避免處于同一等待隊列中的協(xié)程出現(xiàn)并發(fā)讀取poolLocal的數(shù)據(jù)的問題。

同時runtime_procPin方法也會返回當(dāng)前P的編號，在系統(tǒng)內(nèi)部是唯一的，從0開始依次遞增的整數(shù)。其將能夠作為poolLocal的切片下標(biāo)，來讀取poolLocal。

接下來，通過原子操作runtime_LoadAcquintptr讀取localSize字段的值，該字段表示當(dāng)前poolLocal實例切片的長度。如果當(dāng)前的P編號小于localSize的值，則表示該P的poolLocal實例已經(jīng)被創(chuàng)建，可以直接獲取該P對應(yīng)的poolLocal實例并返回。

如果P編號大于等于localSize的值，此時說明該P對應(yīng)的poolLocal還沒創(chuàng)建，通過調(diào)用pinSlow()方法進行初始化。下面繼續(xù)來看pinSlow方法的具體實現(xiàn)。

func (p *Pool) pinSlow(pid int) (*poolLocal, int) {
   // 獲取processor數(shù)量
   size := runtime.GOMAXPROCS(0)
   // 創(chuàng)建一個新的poolLocal切片
   local := make([]poolLocal, size)
   // 將切片地址存儲到local字段當(dāng)中
   atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-release
   // 將切片數(shù)組長度 存儲到 localSize 當(dāng)中
   runtime_StoreReluintptr(&p.localSize, uintptr(size))     // store-release
   // 根據(jù)pid,獲取到對應(yīng)的poolLocal
   return &local[pid], pid
}

首先調(diào)用 runtime.GOMAXPROCS(0) 獲取當(dāng)前程序可用的 processor 數(shù)量，并基于這個數(shù)量創(chuàng)建了一個新的 poolLocal 切片 local，這里也印證了我們之前所說的，每一個Processor都有一個小緩沖池。

接著，使用原子操作 atomic.StorePointer 將指向 p.local 的指針修改為指向新創(chuàng)建的 local 切片的第一個元素的指針。然后使用 runtime_StoreReluintptr 原子操作將 p.localSize 修改為 size。到此為止，便完成了poolLocal切片的初始化操作。

最后返回當(dāng)前 processor 對應(yīng)的 poolLocal 指針和它的編號 pid。由于這里新創(chuàng)建的 local 切片是局部變量，在 pinSlow 函數(shù)返回后，它就無法被訪問了。但是，由于我們已經(jīng)將 p.local 修改為指向這個切片的第一個元素的指針，所以其他 processor在調(diào)用 pin 方法時，就能獲取到新創(chuàng)建的 poolLocal。

4.3 Get方法

func (p *Pool) Get() any {
   l, pid := p.pin()
   var x any
   if n := len(l.private); n > 0 {
       x = l.private[n-1]
       l.private[n-1] = nil // Just to be safe
       l.private = l.private[:n-1]
   }
   runtime_procUnpin()
   if x == nil && p.New != nil {
      x = p.New()
   }
   return x
}

首先調(diào)用pin()方法，獲得當(dāng)前協(xié)程綁定的緩存池local和協(xié)程編號pid。接著從local.private中嘗試取出對象，如果取出來是空，說明緩沖池中沒有對象，此時調(diào)用runtime_procUnpin()方法解綁協(xié)程與處理器的綁定。

如果沒有從緩沖池中成功獲取對象，并且Pool結(jié)構(gòu)體的New字段非空，則調(diào)用New字段所指向的函數(shù)創(chuàng)建一個新對象并返回。

4.4 總結(jié)

到此為止，sync.Pool已經(jīng)通過結(jié)合GMP模型的特點，給每一個P設(shè)置一份緩存數(shù)據(jù)，當(dāng)邏輯處理器上的協(xié)程需要從sync.Pool獲取可重用對象時，此時將從邏輯處理器P對應(yīng)的緩存中取出對象，避免了不同邏輯處理器的競爭。

此外，也通過調(diào)用runtime_procPin方法，讓協(xié)程能夠在某段時間獨占該邏輯處理器，避免了鎖的競爭和不必要的上下文切換的消耗，從而提升性能。

5. sync.Pool實現(xiàn)優(yōu)化

5.1 問題描述

在sync.Pool的初步實現(xiàn)中，我們讓每一個邏輯處理器P都擁有一個小緩沖池，讓各個邏輯處理器P上的協(xié)程從sync.Pool獲取對象時不會競爭，從而提升性能。

現(xiàn)在可能存在的問題，在于每個 Processor 都保存一份緩存數(shù)據(jù)，那么當(dāng)某個 Processor 上的 goroutine 需要使用緩存時，可能會發(fā)現(xiàn)它所在的 Processor 上的緩存池為空的，而其他 Processor 上的緩存對象卻沒有被利用。這樣就浪費了其他 Processor 上的資源。

回到sync.Pool的設(shè)計初衷來看，首先是提升程序性能，減少重復(fù)創(chuàng)建和銷毀對象的開銷；其次是減少內(nèi)存壓力，通過對象復(fù)用，從而降低程序GC頻次。從這兩個方面來看，上面sync.Pool的初步實現(xiàn)其實存在一些優(yōu)化空間的。

這里就陷入了一個兩難的境地，如果多個Processor共享同一個緩沖池，會存在容易導(dǎo)致大量協(xié)程進入阻塞狀態(tài)，進一步降低性能。每個 Processor 都保存一份緩存數(shù)據(jù)的話，此時也容易陷入資源浪費的問題。那能怎么辦呢?

5.2 實現(xiàn)優(yōu)化

很多時候，可能并沒有十全十美的事情，我們往往需要折中。比如上面多個Processor共享同一個緩沖池，會降低性能；而每個 Processor 都保存一份緩存數(shù)據(jù)的話，容易陷入資源浪費的問題。

這個時候，我們可以折中一下，不采用完全共享的模式，也不采用完全獨占的模式。而采用部分獨有、部分共享的模式。每個 Processor 獨占一部分緩存，可以避免不同 Processor 之間的競爭，提高并發(fā)性能。同時，每個 Processor 也可以共享其他 Processor 上的緩存，避免了浪費。相對于完全共享和完全獨立的模式，這種設(shè)計方式是不是能夠更好地平衡并發(fā)性能和緩存利用效率。

同時，也可以基于部分獨有，部分共享的模式的基礎(chǔ)上，再對其進行優(yōu)化。對于共享部分的資源，可以使用多個緩沖池來存儲，是將其給了所有的Processor，每個Processor保留一部分共享數(shù)據(jù)。

當(dāng)Processor讀取數(shù)據(jù)時，此時先從自身的私有緩沖中讀取，讀取不到再到自身的共享緩存中讀取，讀取不到才到其他Processor讀取其共享部分。這樣子能夠避免了多個Processor同時競爭一個池導(dǎo)致的性能問題。同時，共享部分也可以被充分利用，避免了資源浪費。

6.Sync.Pool最終實現(xiàn)

6.1 sync.Pool結(jié)構(gòu)體定義

type Pool struct {
   noCopy noCopy
   // 1. 指向poolLocal切片的指針
   local     unsafe.Pointer // local fixed-size per-P pool, actual type is [P]poolLocal
   // 2. 對應(yīng)local切片的長度
   localSize uintptr        // size of the local array
   // 3. 緩存池中沒對象時，調(diào)用設(shè)置的New函數(shù)來創(chuàng)建對象
   New func() any
   // 部分與此次講述無關(guān)內(nèi)容,未被包含進來
   // ....
}

// 每個Processor都會對應(yīng)一個poolLocal
type poolLocal struct {
   // 存儲緩存對象的數(shù)據(jù)結(jié)構(gòu)
   poolLocalInternal
   // 用于內(nèi)存對齊
   pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}

// Local per-P Pool appendix.
type poolLocalInternal struct {
   // 存儲每個Processor獨享的對象
   private any       // Can be used only by the respective P.
   // 存儲Processor共享的對象，這個是一個無鎖隊列
   shared  poolChain // Local P can pushHead/popHead; any P can popTail.
}

首先說明poolLocal結(jié)構(gòu)體，可以認為是一個小緩沖池，每個Processor都會有對應(yīng)的poolLocal對象。poolLocal中對象的存儲通過poolLocalInternal來實現(xiàn)，至于poolLocal中的pad字段只是用于內(nèi)存對其。

poolLocalInternal其中包含private字段和shared字段，private字段保存了上文所說的Processor獨占的緩存對象，而shared字段，也就是我們上文所說的共享緩沖池組成的一部分，是允許Processor之間相互讀取的。shared字段的類型為poolChain，是一個無鎖隊列，調(diào)用pushHead能夠?qū)?shù)據(jù)放入共享緩沖池，調(diào)用popHead能夠從緩沖池中取出數(shù)據(jù)，無需加鎖也是并發(fā)安全的，這個并非今日的重點，在此簡單描述一下。

Pool結(jié)構(gòu)體中local字段，指向了poolLocal結(jié)構(gòu)體切片的地址，而localSize字段的值，為前面poolLocal切片的長度。

6.2 Get方法

func (p *Pool) Get() any {
   l, pid := p.pin()
   x := l.private
   l.private = nil
   if x == nil {
      // Try to pop the head of the local shard. We prefer
      // the head over the tail for temporal locality of
      // reuse.
      x, _ = l.shared.popHead()
      if x == nil {
         x = p.getSlow(pid)
      }
   }
   runtime_procUnpin()
   if x == nil && p.New != nil {
      x = p.New()
   }
   return x
}

首先調(diào)用pin()方法，獲取當(dāng)前Processor對應(yīng)的poolLocal對象和協(xié)程編號pid，同時占用該Processor。

開始嘗試獲取對象，首先從poolLocal對象中獲取私有緩存private。如果私有緩存為空，則嘗試從共享緩存shared的頭部彈出一個元素x，并賦值給x。如果共享緩存也為空，則調(diào)用getSlow()方法從其他Processor的共享緩存或New方法中獲取元素x。釋放當(dāng)前Processor的占用。如果元素x不為空，則返回x，否則如果New方法不為空，則調(diào)用New方法生成一個新的元素x并返回，否則返回nil。

可以看出來，在Get方法的外層，主要是嘗試從Proessor對應(yīng)的poolLocal中獲取數(shù)據(jù)，讀取不到，則調(diào)用getSlow方法，嘗試從其他Processor的共享數(shù)據(jù)中獲取。下面來看getSlow方法的邏輯:

func (p *Pool) getSlow(pid int) any {
   // See the comment in pin regarding ordering of the loads.
   size := runtime_LoadAcquintptr(&p.localSize) // load-acquire
   locals := p.local                            // load-consume
   // Try to steal one element from other procs.
   for i := 0; i < int(size); i++ {
      // 獲取poolLocal
      l := indexLocal(locals, (pid+i+1)%int(size))
      // poolLocal中的shared是一個無鎖隊列，無需加鎖，也能夠保證線程安全
      if x, _ := l.shared.popTail(); x != nil {
         return x
      }
   }
   // 與sync.Pool對象回收的相關(guān)邏輯先刪除,與此次講述并無太大關(guān)系
   // ....

   return nil
}

getSlow方法實現(xiàn)較為簡單，首先讀取Pool結(jié)構(gòu)體中localSize字段的值，得知當(dāng)前有多少個poolLocal。然后對所有的poolLocal進行遍歷，嘗試從其他poolLocal的共享緩存中獲取數(shù)據(jù)，成功獲取則直接返回。

6.3 Put方法

func (p *Pool) Put(x any) {
   if x == nil {
      return
   }
   l, _ := p.pin()
   if l.private == nil {
      l.private = x
      x = nil
   }
   if x != nil {
      l.shared.pushHead(x)
   }
   runtime_procUnpin()
}

首先調(diào)用pin方法，獲取當(dāng)前Processor對應(yīng)的poolLocal，然后將x放到該poolLocal的private字段中，也就是放到當(dāng)前Processor的私有緩存中。如果private字段不為空，說明已經(jīng)有對象放到private中了，那么x則會放到poolLocal的shared字段中，通過無鎖隊列的方式加入到共享資源池中。

6.4 總結(jié)

到此為止，在sync.Pool原本的實現(xiàn)上，對緩存數(shù)據(jù)的設(shè)計進行了優(yōu)化，將緩存數(shù)據(jù)中區(qū)分為私有緩存部分和共享部分。此時在一定程度上避免不同 Processor 之間的競爭，提高并發(fā)性能。同時，每個 Processor 也可以共享其他 Processor 上的緩存，避免了內(nèi)存的浪費。