Go语言之sync.Pool

什么是sync.Pool

常使用 sync.Pool 来缓存对象

对于很多需要重复分配、回收内存的地方，sync.Pool 是一个很好的选择。频繁地分配、回收内存会给 GC 带来一定的负担，严重的时候会引起 CPU 的毛刺

而 sync.Pool 可以将暂时不用的对象缓存起来，待下次需要的时候直接使用，不用再次经过内存分配，复用对象的内存，减轻 GC 的压力，提升系统的性能。

当多个 goroutine 都需要创建同⼀个对象的时候，如果 goroutine 数过多，导致对象的创建数⽬剧增，进⽽导致 GC 压⼒增大。形成 “并发⼤－占⽤内存⼤－GC 缓慢－处理并发能⼒降低－并发更⼤”这样的恶性循环
在这个时候，需要有⼀个对象池，每个 goroutine 不再⾃⼰单独创建对象，⽽是从对象池中获取出⼀个对象（如果池中已经有的话）

关键思想：对象的复用，避免重复创建、销毁

举个简单例子：

package main
import ("fmt""sync"
)var pool *sync.Pooltype Person struct {Name string
}func initPool() {pool = &sync.Pool {New: func()interface{} {fmt.Println("Creating a new Person")return new(Person)},}
}func main() {initPool()p := pool.Get().(*Person)fmt.Println("首次从 pool 里获取：", p)p.Name = "first"fmt.Printf("设置 p.Name = %s\n", p.Name)pool.Put(p)fmt.Println("Pool 里已有一个对象：&{first}，调用 Get: ", pool.Get().(*Person))fmt.Println("Pool 没有对象了，调用 Get: ", pool.Get().(*Person))
}

其实在标准库 encoding/json和fmt中也都用到了 sync.Pool 来提升性能。还有 gin 框架，对 context 取用也到了 sync.Pool。

接下来我们来看看sync.Pool是如何实现的？

Pool结构体

首先来看 Pool 的结构体：

type Pool struct {// 防止该结构体被复制，只能被引用赋值noCopy noCopy// 每个 P 的本地队列，实际类型为 [P]poolLocallocal unsafe.Pointer // local fixed-size per-P pool, actual type is [P]poolLocal// [P]poolLocal的实际大小localSize uintptr        // size of the local array// local from previous cycle//  Go 1.13 优化后添加的，相当于二级缓存，缓存上一次GC中local变量中存储的数据victim     unsafe.Pointer // size of victims arrayvictimSize uintptr        // 自定义的对象创建回调函数，当 pool 中无可用对象时会调用此函数New func() interface{}
}// noCopy 用于嵌入一个结构体中来保证其第一次使用后不会被复制
type noCopy struct{}// Lock 是一个空操作用来给 `go vet` 的 -copylocks 静态分析
// Error: copies lock value
func (*noCopy) Lock()   {}
func (*noCopy) Unlock() {}

因为 Pool 不希望被复制，所以结构体里有一个 noCopy 的字段，使用 go vet 工具可以检测到用户代码是否复制了 Pool。

noCopy 是 go1.7 开始引入的一个静态检查机制。它不仅仅工作在运行时或标准库，同时也对用户代码有效，用户只需实现这样的不消耗内存、仅用于静态分析的结构，来保证一个对象在第一次使用后不会发生复制

字段的具体解释：

local 字段存储指向 [P]poolLocal 数组（严格来说，它是一个切片）的指针，localSize 则表示 local 数组的大小。访问时，P 的 id 对应 [P]poolLocal 下标索引。通过这样的设计，多个 goroutine 使用同一个 Pool 时，减少了竞争，提升了性能。

在一轮 GC 到来时，victim 和 victimSize 会分别“接管” local 和 localSize。victim 的机制用于减少 GC 后冷启动导致的性能抖动，让分配对象更平滑。

引入了 victim（二级）缓存，每次 GC 周期不再清理所有的缓存对象，而是将 local 中的对象暂时放入 victim ，从而延迟到下一个 GC 周期进行回收；

在下一个周期到来前，victim 中的缓存对象可能会被偷取，在 Put 操作后又重新回到 local 中，这个过程发生在从其他 P 的 shared 队列中偷取不到、以及 New 一个新对象之前，进而是在牺牲了 New 新对象的速度的情况下换取的；

Victim Cache 本来是计算机架构里面的一个概念，是 CPU 硬件处理缓存的一种技术，sync.Pool 引入的意图在于降低 GC 压力的同时提高命中率

poolLocal结构体

type poolLocal struct {poolLocalInternal// 将 poolLocal 补齐至两个缓存行的倍数，防止 false sharing,// 每个缓存行具有 64 bytes，即 512 bit// 伪共享，仅占位用，防止在 cache line 上分配多个 poolLocalInternalpad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}// Local per-P Pool appendix.
type poolLocalInternal struct {// P 的私有缓存区，使用时无需要加锁private interface{}// 公共缓存区。本地 P 可以 pushHead/popHead；其他 P 则只能 popTailshared  poolChain
}// 一个双端队列的实现（切片+链表）
// 可以看到 Pool 并没有直接使用 poolDequeue，原因是它的大小是固定的，
// 而 Pool 的大小是没有限制的。
// 因此，在 poolDequeue 之上包装了一下，变成了一个 poolChainElt 的双向链表，可以动态增长
// 并将缓存的对象存储在切片中
type poolChain struct {// 只有生产者会 push to，不用加锁head *poolChainElt// 读写需要原子控制。pop from， 其他P窃取的开始位置tail *poolChainElt
}type poolChainElt struct {poolDequeue// next 被 producer 写，consumer 读。所以只会从 nil 变成 non-nil// prev 被 consumer 写，producer 读。所以只会从 non-nil 变成 nilnext, prev *poolChainElt
}// poolDequeue 被实现为单生产者、多消费者的固定大小的无锁（atomic 实现） Ring 式队列（底层存储使用数组，使用两个指针标记 head、tail）。
// 生产者可以从 head 插入、head 删除，而消费者仅可从 tail 删除。
type poolDequeue struct {// The head index is stored in the most-significant bits so// that we can atomically add to it and the overflow is// harmless.// headTail 包含一个 32 位的 head 和一个 32 位的 tail 指针。// 这两个值都和 len(vals)-1 取模过。// tail 是队列中最老的数据，head 指向下一个将要填充的 slot// slots 的有效范围是 [tail, head)，由 consumers 持有。headTail uint64// vals 是一个存储 interface{} 的环形队列，它的 size 必须是 2 的幂// 如果 slot 为空，则 vals[i].typ 为空；否则，非空。// 一个 slot 在这时宣告无效：tail 不指向它了，vals[i].typ 为 nil// 由 consumer 设置成 nil，由 producer 读vals []eface
}

false sharing

现代 cpu 中，cache 都划分成以 cache line (cache block) 为单位，在 x86_64 体系下一般都是 64 字节，cache line 是操作的最小单元。
程序即使只想读内存中的 1 个字节数据，也要同时把附近 63 节字加载到 cache 中，如果读取超个 64 字节，那么就要加载到多个 cache line 中

简单来说，如果没有 pad 字段，那么当需要访问 0 号索引的 poolLocal 时，CPU 同时会把 0 号和 1 号索引同时加载到 cpu cache。在只修改 0 号索引的情况下，会让 1 号索引的 poolLocal 失效。

这样，当其他线程想要读取 1 号索引时，发生 cache miss，还得重新再加载，对性能有损。增加一个 pad，补齐缓存行，让相关的字段能独立地加载到缓存行就不会出现 false sharding 了

Pool 结构体结构示意图：

源码分析：

Get

func (p *Pool) Get() interface{} {// ......// 首先，调用 p.pin() 函数将当前的 goroutine 和 P 绑定，禁止被抢占，// 返回当前 P 对应的 poolLocal，以及 pid// 如果 G 被抢占，则 G 的状态从 running 变成 runnable，会被放回 P 的 localq 或 globaq，// 等待下一次调度。下次再执行时，就不一定是和现在的 P 相结合了。// 因为之后会用到 pid，如果被抢占了，有可能接下来使用的 pid 与所绑定的 P 并非同一个。l, pid := p.pin()// 直接取 l.private，赋值给 x，并置 l.private 为 nilx := l.privatel.private = nil // 判断 x 是否为空，若为空，则尝试从 l.shared 的头部 pop 一个对象出来，同时赋值给 xif x == nil {// 如果 x 仍然为空，则调用 getSlow 尝试从其他 P 的 shared 双端队列尾部“偷”一个对象出来x, _ = l.shared.popHead()if x == nil {x = p.getSlow(pid)}}// Pool 的相关操作做完了，调用 runtime_procUnpin() 解除非抢占runtime_procUnpin()// ......// 最后如果还是没有取到缓存的对象，那就直接调用预先设置好的 New 函数，创建一个出来if x == nil && p.New != nil {x = p.New()}return x
}// src/sync/pool.go
// 调用方必须在完成取值后调用 runtime_procUnpin() 来取消抢占。
func (p *Pool) pin() (*poolLocal, int) {pid := runtime_procPin()s := atomic.LoadUintptr(&p.localSize) // load-acquirel := p.local                          // load-consume// 因为可能存在动态的 P（运行时调整 P 的个数）if uintptr(pid) < s {return indexLocal(l, pid), pid}return p.pinSlow()
}func (p *Pool) pinSlow() (*poolLocal, int) {// Retry under the mutex.// Can not lock the mutex while pinned.// 解除非抢占，不然Lock可能持续很长时间获取不到，导致阻塞runtime_procUnpin()// 不过要想上锁的话，得先解除“绑定”，锁上之后，再执行“绑定”。// 原因是锁越大，被阻塞的概率就越大，如果还占着 P，那就浪费资源。allPoolsMu.Lock()defer allPoolsMu.Unlock()// 上锁成功后，在执行绑定pid := runtime_procPin()// poolCleanup won't be called while we are pinned.// 没有使用原子操作，因为已经加了全局锁了s := p.localSizel := p.local// 因为 pinSlow 中途可能已经被其他的线程调用，因此这时候需要再次对 pid 进行检查。// 如果 pid 在 p.local 大小范围内，则不用创建 poolLocal 切片，直接返回。if uintptr(pid) < s {return indexLocal(l, pid), pid}if p.local == nil {allPools = append(allPools, p)}// If GOMAXPROCS changes between GCs, we re-allocate the array and lose the old one.// 当前 P 的数量size := runtime.GOMAXPROCS(0)local := make([]poolLocal, size)// 旧的 local 会被回收atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-releaseatomic.StoreUintptr(&p.localSize, uintptr(size))         // store-releasereturn &local[pid], pid
}

popHead

Get 函数中，再来看另一个关键的函数：poolChain.popHead()

// 双端队列
func (c *poolChain) popHead() (interface{}, bool) {d := c.headfor d != nil {if val, ok := d.popHead(); ok {return val, ok}// There may still be unconsumed elements in the// previous dequeue, so try backing up.d = loadPoolChainElt(&d.prev)}return nil, false
}// 双端队列内部的环形队列
// /usr/local/go/src/sync/poolqueue.go
func (d *poolDequeue) popHead() (interface{}, bool) {var slot *efacefor {ptrs := atomic.LoadUint64(&d.headTail)head, tail := d.unpack(ptrs)// 判断队列是否为空if tail == head {// Queue is empty.return nil, false}// head 位置是队头的前一个位置，所以此处要先退一位。// 在读出 slot 的 value 之前就把 head 值减 1，取消对这个 slot 的控制head--ptrs2 := d.pack(head, tail)// 使用 atomic.CompareAndSwapUint64 比较 headTail 在这之间是否有变化，如果没变化，相当于获取到了这把锁，那就更新 headTail 的值。并且把 vals 相应索引处的元素赋值给 slotif atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) {// We successfully took back slot.slot = &d.vals[head&uint32(len(d.vals)-1)]break}}// 取出 valval := *(*interface{})(unsafe.Pointer(slot))if val == dequeueNil(nil) {val = nil}// 重置 slot，typ 和 val 均为 nil// 这里清空的方式与 popTail 不同，与 pushHead 没有竞争关系，所以不用太小心*slot = eface{}return val, true
}// src/sync/poolqueue.go
const dequeueBits = 32// pack
// mask 的低 31 位为全 1，其他位为 0，它和 tail 相与，就是只看 tail 的低 31 位。
// 而 head 向左移 32 位之后，低 32 位为全 0。
// 最后把两部分“或”起来，head 和 tail 就“绑定”在一起了
func (d *poolDequeue) pack(head, tail uint32) uint64 {const mask = 1<<dequeueBits - 1return (uint64(head) << dequeueBits) |uint64(tail&mask)
}// unpack
// 取出 head 指针的方法就是将 ptrs 右移 32 位，再与 mask 相与，同样只看 head 的低 31 位。
// 而 tail 实际上更简单，直接将 ptrs 与 mask 相与就可以了
func (d *poolDequeue) unpack(ptrs uint64) (head, tail uint32) {const mask = 1<<dequeueBits - 1head = uint32((ptrs >> dequeueBits) & mask)tail = uint32(ptrs & mask)return
}

getSlow

func (p *Pool) getSlow(pid int) interface{} {// See the comment in pin regarding ordering of the loads.size := atomic.LoadUintptr(&p.localSize) // load-acquirelocals := p.local                        // load-consume// Try to steal one element from other procs.// 从其他 P 中窃取对象// 从索引为 pid+1 的 poolLocal 处开始，尝试调用 shared.popTail() 获取缓存对象。// 如果没有拿到，则从 victim 里找和 poolLocal 的逻辑类似。for i := 0; i < int(size); i++ {l := indexLocal(locals, (pid+i+1)%int(size))if x, _ := l.shared.popTail(); x != nil {return x}}// 尝试从victim cache中取对象。这发生在尝试从其他 P 的 poolLocal 偷去失败后，// 因为这样可以使 victim 中的对象更容易被回收。size = atomic.LoadUintptr(&p.victimSize)if uintptr(pid) >= size {return nil}locals = p.victiml := indexLocal(locals, pid)if x := l.private; x != nil {l.private = nilreturn x}for i := 0; i < int(size); i++ {l := indexLocal(locals, (pid+i)%int(size))if x, _ := l.shared.popTail(); x != nil {return x}}// 最后，实在没找到，就把 victimSize 置 0，防止后来的“人”再到 victim 里找// 清空 victim cache。下次就不用再从这里找了atomic.StoreUintptr(&p.victimSize, 0)return nil
}

popTail

func (c *poolChain) popTail() (interface{}, bool) {d := loadPoolChainElt(&c.tail)if d == nil {return nil, false}for {d2 := loadPoolChainElt(&d.next)if val, ok := d.popTail(); ok {return val, ok}if d2 == nil {// 双向链表只有一个尾节点，现在为空return nil, false}// 双向链表的尾节点里的双端队列被“掏空”，所以继续看下一个节点。// 并且由于尾节点已经被“掏空”，所以要甩掉它。这样，下次 popHead 就不会再看它有没有缓存对象了。if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) {// 甩掉尾节点storePoolChainElt(&d2.prev, nil)}d = d2}
}// src/sync/poolqueue.go
func (d *poolDequeue) popTail() (interface{}, bool) {var slot *efacefor {ptrs := atomic.LoadUint64(&d.headTail)head, tail := d.unpack(ptrs)// 判断队列是否空if tail == head {// Queue is empty.return nil, false}// 先搞定 head 和 tail 指针位置。如果搞定，那么这个 slot 就归属我们了ptrs2 := d.pack(head, tail+1)if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) {// Success.slot = &d.vals[tail&uint32(len(d.vals)-1)]break}}// We now own slot.val := *(*interface{})(unsafe.Pointer(slot))if val == dequeueNil(nil) {val = nil}slot.val = nilatomic.StorePointer(&slot.typ, nil)// At this point pushHead owns the slot.return val, true
}

/ src/sync/pool.go
// Put 将对象添加到 Pool
func (p *Pool) Put(x interface{}) {if x == nil {return}// ……l, _ := p.pin()if l.private == nil {l.private = xx = nil}if x != nil {l.shared.pushHead(x)}runtime_procUnpin()//……
}// 来看 pushHead 的源码，比较清晰：
// src/sync/poolqueue.go
// 初始为 8，放满之后，再创建一个 poolChainElt 节点时，双端队列的长度就要翻倍。当然，有一个最大长度限制（2^30)
const dequeueBits = 32
const dequeueLimit = (1 << dequeueBits) / 4func (c *poolChain) pushHead(val interface{}) {d := c.headif d == nil {// poolDequeue 初始长度为8const initSize = 8 // Must be a power of 2d = new(poolChainElt)d.vals = make([]eface, initSize)c.head = dstorePoolChainElt(&c.tail, d)}if d.pushHead(val) {return}// 前一个 poolDequeue 长度的 2 倍newSize := len(d.vals) * 2if newSize >= dequeueLimit {// Can't make it any bigger.newSize = dequeueLimit}// 首尾相连，构成链表d2 := &poolChainElt{prev: d}d2.vals = make([]eface, newSize)c.head = d2storePoolChainElt(&d.next, d2)d2.pushHead(val)
}// 调用 poolDequeue.pushHead 尝试将对象放到 poolDeque 里去：
// src/sync/poolqueue.go
// 将 val 添加到双端队列头部。如果队列已满，则返回 false。此函数只能被一个生产者调用
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)]// 检测这个 slot 是否被 popTail 释放typ := atomic.LoadPointer(&slot.typ)if typ != nil {// 另一个 groutine 正在 popTail 这个 slot，说明队列仍然是满的// popTail 是先设置 val，再将 typ 设置为 nil。设置完 typ 之后，popHead 才可以操作这个 slotreturn false}// The head slot is free, so we own it.if val == nil {val = dequeueNil(nil)}// slot占位，将val存入vals中// slot 是 eface 类型，将 slot 转为 interface{} 类型，// 这样 val 能以 interface{} 赋值给 slot 让 slot.typ 和 slot.val 指向其内存块，// 于是 slot.typ 和 slot.val 均不为空。*(*interface{})(unsafe.Pointer(slot)) = val// head 增加 1atomic.AddUint64(&d.headTail, 1<<dequeueBits)return true
}

GC

对于 Pool 而言，并不能无限扩展，否则对象占用内存太多了，会引起内存溢出几乎所有的池技术中，都会在某个时刻清空或清除部分缓存对象，那么在 Go 中何时清理未使用的对象呢？

答案是 GC 发生时。

在 pool.go 文件的 init 函数里，注册了 GC 发生时，如何清理 Pool 的函数：

// src/sync/pool.gofunc init() {runtime_registerPoolCleanup(poolCleanup)
}// src/runtime/mgc.go// Hooks for other packagesvar poolcleanup func()// 利用编译器标志将 sync 包中的清理注册到运行时
//go:linkname sync_runtime_registerPoolCleanup sync.runtime_registerPoolCleanup
func sync_runtime_registerPoolCleanup(f func()) {poolcleanup = f
}// Go1.13优化版实现
// poolCleanup 会在 STW 阶段被调用
// 主要是将 local 和 victim 作交换，这样也就不致于让 GC 把所有的 Pool 都清空了，有 victim 在“兜底”。// 第 1 次GC STW 阶段，
// allPools 中所有 p.local 将值赋值给 victim 并置为 nil。
// allPools 赋值给 oldPools，最后 allPools 为 nil，// 第 2 次 GC STW 阶段，
// oldPools 中所有 p.victim 置 nil，前一次的 cache 在本次 GC 时被回收，
// allPools 所有 p.local 将值赋值给 victim 并置为nil，最后 allPools 为 nil
func poolCleanup() {for _, p := range oldPools {p.victim = nilp.victimSize = 0}// Move primary cache to victim cache.for _, p := range allPools {p.victim = p.localp.victimSize = p.localSizep.local = nilp.localSize = 0}oldPools, allPools = allPools, nil
}// 在 Go 1.13 之前的实现中，直接清空了所有 Pool 的 p.local 和 poolLocal.shared
func poolCleanup() {for i, p := range allPools {allPools[i] = nilfor i := 0; i < int(p.localSize); i++ {l := indexLocal(p.local, i)l.private = nilfor j := range l.shared {l.shared[j] = nil}l.shared = nil}p.local = nilp.localSize = 0}allPools = []*Pool{}
}

新版的实现相比 Go 1.13 之前：

GC 的粒度拉大了，由于实际回收的时间线拉长，单位时间内 GC 的开销减小。
p.victim 的作用。它的定位是次级缓存，GC 时将对象放入其中，下一次 GC 来临之前如果有 Get 调用则会从 p.victim 中取，直到再一次 GC 来临时回收。
同时，由于从 p.victim 中取出对象使用完毕之后并未放回 p.victim 中，在一定程度也减小了下一次 GC 的开销。原来 1 次 GC 的开销被拉长到 2 次且会有一定程度的开销减小，这就是 p.victim 引入的意图

END