一、基础概念

1.1、Pool 的概念

Pool's purpose is to cache allocated but unused items for later reuse, relieving pressure on the garbage collector. That is, it makes it easy to build efficient, thread-safe free lists. However, it is not suitable for all free lists. A Pool is safe for use by multiple goroutines simultaneously.

sync.Pool 是一个并发安全的缓存池，能够并发且安全地存储、获取元素/对象。常用于对象实例创建会占用较多资源的场景。但是它不具有严格的缓存作用，因为 Pool 中的元素/对象的释放时机是随机的。 作为缓存的一种姿势，sync.Pool 能够避免元素/对象的申请内存操作和初始化过程，以提高性能。当然，这里有个 trick，释放元素/对象的操作是直接将元素/对象放回池子，从而免去了真正释放的操作。 另外，不考虑内存浪费和初始化消耗的情况下，“使用 sync.Pool 管理多个对象”和“直接 New 多个对象”两者的区别在于后者会创建出更多的对象，并发高时会给 GC 带来非常大的负担，进而影响整体程序的性能。因为 Go 申请内存是程序员触发的，而回收却是 Go 内部 runtime GC 回收器来执行的。即，使用 sync.Pool 还可以减少 GC 次数。

总结一下：sync.Pool 是以缓存池的形式，在创建对象等场景下，减少 GC 次数、提高程序稳定性。后文将称之为 Pool 池，池子中的元素统一称为对象。

1.2、接口使用

1、创建一个 Pool 实例 Pool 池的模式是通用型的(存储对象的类型为interface{})，所有的类型的对象都可以进行使用。注意的是，作为使用方不能对 Pool 里面的对象个数做假定，同时也无法获取 Pool 池中对象个数。

pool := &sync.Pool{}

2、Put 和 Get 接口 (1) Put(interface{})：将一个对象加入到 Pool 池中

• 注意的是这仅仅是把对象放入池子，池子中的对象真正释放的时机是不受外部控制的。 (2) Get() interface{}：返回 Pool 池中存在的对象
• 注意的是 Get() 方法是随机取出对象，无法保证以固定的顺序获取 Pool 池中存储的对象。

3、为 Pool 实例配置 New 方法 没有配置 New 方法时，如果 Get 操作多于 Put 操作，继续 Get 会得到一个 nil interface{} 对象，所以需要代码进行兼容。 配置 New 方法后，Get 获取不到对象时（Pool 池中已经没有对象了），会调用自定义的 New 方法创建一个对象并返回。

pool := &sync.Pool {
    New: func() interface {} {
        return struct{}{}
    }
}

注意的是，sync.Pool 本身数据结构是并发安全的，但是 Pool.New 函数（用户自定义的）不一定是线程安全的。并且 Pool.New 函数可能会被并发调用，如果 New 函数里面的实现逻辑是非并发安全的，那就会有问题。

4、关于 sync.Pool 的性能优势，可以试一下 go/src/sync/pool_test.go 中的几个压测函数。

二、使用 & 实践

2.1、使用场景

1、增加临时对象的重用率（sync.Pool 本质用途）。在高并发业务场景下出现 GC 问题时，可以使用 sync.Pool 减少 GC 负担。

2、不适合存储带状态的对象，因为获取对象是随机的（Get 到的对象可能是刚创建的，也可能是之前创建并 cache 住的），并且缓存对象的释放策略完全由 runtime 内部管理。像 socket 长连接、数据库连接，有人说适合有人说不适合。

3、不适合需要控制缓存元素个数的场景，因为无法知道 Pool 池里面的对象个数，以及对象的释放时机也是未知的。

4、典型的应用场景之一是在网络包收取发送的时候，用 sync.Pool 会有奇效，可以大大降低 GC 压力。

2.2、实践

1、使用规范：作为对象生成器 (a) 初始化 sync.Pool 实例，并需要配置并发安全的 New 方法； (b) 创建对象的地方，通过 Pool.Get() 获取； (c) 在 Get 后马上进行 defer Pool.Put(x)；

2、使用建议：不要对 Pool 池中的对象做任何假定 即使用前需要清理缓存对象，有两种方案：在调用 Pool.Put 前进行 memset 对象操作；在 Pool.Get 操作后对对象进行 memset 操作。

3、demo：重用一个缓冲区，用于文件写入数据的存储

func writeFile(pool *sync.Pool, filename string) error {
    buf := pool.Get().(*bytes.Buffer)  // 如果是第一个调用，则创建一个缓冲区
    defer pool.Put(buf) // 将缓冲区放回 sync.Pool中

    buf.Reset()  // Reset 缓存区，不然会连接上次调用时保存在缓存区里的内容
    buf.WriteString("Demo")
    return ioutil.WriteFile(filename, buf.Bytes(), 0644)
}

4、几个真实的使用场景

• gin 框架在 gin.go:L144/L346 中，通过 sync.Pool 对 Context 做了缓存；
• fmt.Printf 中通过 sync.Pool 缓存了 pp struct（newPrinter 函数中）；
• logrus 框架中缓存了 logger entry 对象。

2.3、思考

1、先 Put，或者主动 Put 会造成什么？ 首先，Put 任意类型的元素都不会报错，因为存储的是 interface{} 对象，且内部没有进行类型的判断和断言。如果 Put 在业务上不做限制，那么 Pool 池中就可能存在各种类型的数据，就导致在 Get 后的代码会非常繁琐（需要进行类型判断，否则有 panic 隐患）。另外，Get 得到的对象是随机的，缓存对象的回收也是随机的，所以先 Put 一个对象根本就没有实际作用。

综上，sync.Pool 本质上可以是一个杂货铺，支持存放任何类型，所以 Get 出来和 Put 进去的对象类型要业务自己把控。

2、如果只调用 Get 不调用 Put 会怎么样？ 首先就算不调用 Pool.Put，GC 也会去释放 Get 获取的对象（当没有人去再用它时）。但是只进行 Get 操作的话，就相当于一直在生成新的对象，Pool 池也失去了它最本质的功能。

三、源码分析

基于 go 1.14 version

2.1、Pool 的数据结构

type Pool struct {  
   // 用于检测 Pool 池是否被 copy，因为 Pool 不希望被 copy。用这个字段可以在 go vet 工具中检测出被 copy（在编译期间就发现问题） 
   noCopy noCopy  // A Pool must not be copied after first use.

   // 实际指向 []poolLocal，数组大小等于 P 的数量；每个 P 一一对应一个 poolLocal
   local     unsafe.Pointer 
   localSize uintptr      // []poolLocal 的大小

   // GC 时，victim 和 victimSize 会分别接管 local 和 localSize；
   // victim 的目的是为了减少 GC 后冷启动导致的性能抖动，让分配对象更平滑；
   victim     unsafe.Pointer 
   victimSize uintptr       

   // 对象初始化构造方法，使用方定义
   New func() interface{}
}

1、noCopy 字段是为了检测 Pool 池的 copy 行为。但无法阻止编译（用户进行 copy 行为也能成功运行程序），只能通过 go vet 检查出用户的 copy 行为。noCopy 是 go1.7 引入的一个静态检查机制，对用户代码有效。

2、local 和 localSize 这两个字段实现了一个数组，数组元素为 poolLocal 结构体。每个 Processor(P) 都会对应数组中的一个元素，访问时，P 的 id 就是 [P]poolLocal 下标索引。通过这样的设计，多个 goroutine 使用同一个 Pool 时，减少了数据竞争。

3、victim 和 victimSize 在 (GC)poolCleanup 流程里赋值为 local 和 localSize。victim 机制是把 Pool 池的清理由一轮 GC 改成两轮 GC，进而提高对象的复用率，减少抖动。

4、函数变量 New 对外暴露，使用者定义对象初始化构造行为。todo，没找到 New 变量初始化的地方。

2.2、poolLocal 的数据结构

// Pool.local 指向的数组元素类型
type poolLocal struct {
   poolLocalInternal
   pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}

// Local per-P Pool appendix.
type poolLocalInternal struct {
   private interface{} // Can be used only by the respective P.
   shared  poolChain   // 双链表结构，用于挂接 cache 元素
}

1、Pool.local 指针指向的就是 poolLocal 数组。

2、poolLocal struct 中真实有用的只有 poolLocalInternal struct。其中的 pad 字段是用于内存填充，对齐 cache line，防止伪共享(false sharing)的性能问题。

3、并不是一个缓存对象就对应了一个 poolLocal 数组中的一个元素（一个 poolLocalInternal struct 对象），而是有多少个 P(Processor) 就有多少个 poolLocalInternal 对象。可以把 poolLocalInternal 对象理解成一个数据桶，每个 P 下面挂载了一个数据桶。

4、poolLocalInternal struct 中的 shared poolChain 成员就是数据桶的作用；private 成员算是一个 vip 缓存位，每次都最先被访问。poolChain struct 由头尾链表指针构成，其实就是一个双向链表。

type poolChain struct {
   head *poolChainElt  // 头指针
   tail *poolChainElt  // 尾指针
}
type poolChainElt struct {
   poolDequeue   // 本质是个数组内存空间，管理成 ringbuffer 的模式；
   next, prev *poolChainElt  // 前向、后向指针
}

2.3、数据桶(poolChain + poolDequeue)

双链表的节点 poolChainElt struct 对象，仍然不是 Pool 池中的对象。poolChainElt struct 中的 poolDequeue struct 是一段数组空间，类似于 ringbuffer，是一个无锁环形队列。Pool 池管理的对象存储在 poolDequeue 的 vals[] 数组里，即缓存对象存储在环形队列 poolDequeue 中。

poolDequeue 是单生产者多消费者的固定大小的无锁环形队列。生产者可以从 head 写/插入、从 head 读/删除，而消费者仅可从 tail 读/删除。headTail 变量指向了队列的头和尾，通过位运算区分 head 和 tail 的值（高32位为 head 的索引下标，低32位为 tail 的索引下标）。利用 atomic 和 CAS 对 headTail 值进行原子操作，从而实现了无锁机制。

type poolDequeue struct {
   headTail uint64
   vals []eface
}
type eface struct {
   typ, val unsafe.Pointer
}

1、入队 pushHead

• 如果双向链表 poolChain 为 nil 则先进行初始化；
• 将对象放入 head 位置上的环形队列；
• 如果 head 位置的环形队列 poolDequeue 满了，则新建一个双倍容量的链表节点（初始节点的队列长度为 8）。环形队列最大容量为 (1<<32)/4 =1073741824，注意的是，达到上限后，再生成的队列容量都将是 (1<<32)/4。
• 新建节点并移动 head 后，将对象放入新 head 位置上的环形队列 poolDequeue 中。

func (c *poolChain) pushHead(val interface{}) {
   d := c.head

   // Initialize the chain.
   if d == nil {
      const initSize = 8  // 初始长度为8
      d = new(poolChainElt)
      d.vals = make([]eface, initSize)
      c.head = d
      storePoolChainElt(&c.tail, d)
   }

   if d.pushHead(val) {
      return
   }

   // The current dequeue is full. Allocate a new one of twice the size.
   newSize := len(d.vals) * 2
   if newSize >= dequeueLimit {  // Can't make it any bigger.
      newSize = dequeueLimit
   }

   // 拼接链表
   d2 := &poolChainElt{prev: d}
   d2.vals = make([]eface, newSize)
   c.head = d2
   storePoolChainElt(&d.next, d2)
   d2.pushHead(val)
}

func (d *poolDequeue) pushHead(val interface{}) bool {
   ptrs := atomic.LoadUint64(&d.headTail)
   head, tail := d.unpack(ptrs)
  
   // Ring式队列，头尾相等则队列已满
   if (tail+uint32(len(d.vals)))&(1<<dequeueBits-1) == head {
      return false
   }

   slot := &d.vals[head&uint32(len(d.vals)-1)]

   // Check if the head slot has been released by popTail.
   typ := atomic.LoadPointer(&slot.typ)
   if typ != nil {
      // Another goroutine is still cleaning up the tail, so
      // the queue is actually still full.
      return false
   }

   // The head slot is free, so we own it.
   if val == nil {
      val = dequeueNil(nil)
   }
   *(*interface{})(unsafe.Pointer(slot)) = val

   // Increment head. This passes ownership of slot to popTail
   // and acts as a store barrier for writing the slot.
   atomic.AddUint64(&d.headTail, 1<<dequeueBits)
   return true
}

2、出队 popHead

• 从 head 位置获取对象，如果该环形队列中还有数据则会返回 true；
• 如果 head 位置的环形队列空了，会定位到 prev 节点继续尝试获取对象；

func (c *poolChain) popHead() (interface{}, bool) {
   d := c.head
   for 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
}

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--  // head位置是队头的前一个位置，所以此处要先退一位
      ptrs2 := d.pack(head, tail)  // 新的 headTail 值
      
      // 通过 CAS 判断当前没有并发修改就拿到数据
      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，typ和val均为nil
   *slot = eface{}
   return val, true
}

3、从队尾获取元素 popTail 假如在链表长度为3的情况下。尾部节点指向的无锁队列里缓存对象被偷光了。那么尾部节点会沿着next指针前移，把旧的无锁队列内存释放掉。此时链表长度变为2，这就是链表的收缩策略。最小时也剩下一个节点，不会收缩成空链表。

func (c *poolChain) popTail() (interface{}, bool) {
   d := loadPoolChainElt(&c.tail)
   if d == nil {
      return nil, false
   }

   for {
      // It's important that we load the next pointer
      // *before* popping the tail. In general, d may be
      // transiently empty, but if next is non-nil before
      // the pop and the pop fails, then d is permanently
      // empty, which is the only condition under which it's
      // safe to drop d from the chain.
      d2 := loadPoolChainElt(&d.next)

      if val, ok := d.popTail(); ok {
         return val, ok
      }

      if d2 == nil {
         // This is the only dequeue. It's empty right
         // now, but could be pushed to in the future.
         return nil, false
      }

      // The tail of the chain has been drained, so move on
      // to the next dequeue. Try to drop it from the chain
      // so the next pop doesn't have to look at the empty
      // dequeue again.
      if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) {
         // We won the race. Clear the prev pointer so
         // the garbage collector can collect the empty
         // dequeue and so popHead doesn't back up
         // further than necessary.
         storePoolChainElt(&d2.prev, nil)
      }
      d = d2
   }
}

2.4、Get() interface{}

Get 是从 Pool 池里获取一个对象，它会进行逐步尝试：先尝试从本 P 对应的 poolLocal 中的 private 区获取；再尝试从本 P 的 shared 队列中获取元素；然后尝试从其他 P 的 shared 队列中获取元素；接着尝试从 victim cache 里获取元素；最后初始化得到一个新的元素。（下图来自码农桃花源）

func (p *Pool) Get() interface{} {
   if race.Enabled {
      race.Disable()
   }
   
   //1. Pool 池中有元素
   l, pid := p.pin()  // l是一个 poolLocal 对象  // pin()的逻辑看后文
   x := l.private  // 从 private 区里获取
   l.private = nil
   if x == nil {
      // 从 shared 队列里获取
      x, _ = l.shared.popHead()
      
      // 尝试从获取其他 P 的队列里取元素，或者尝试从 victim cache 里取元素
      if x == nil {
         x = p.getSlow(pid)
      }
   }
  
   // G-M 锁定解除；上锁的逻辑在 pin() 中
   runtime_procUnpin()

   if race.Enabled {
      race.Enable()
      if x != nil {
         race.Acquire(poolRaceAddr(x))
      }
   }
   
   //2. Pool 池中没有元素，需要当场初始化得到一个
   if x == nil && p.New != nil {
      x = p.New()
   }
   return x
}

func (p *Pool) pin() (*poolLocal, int) {
   pid := runtime_procPin()

   // In pinSlow we store to local and then to localSize, here we load in opposite order.
   // Since we've disabled preemption, GC cannot happen in between.
   // Thus here we must observe local at least as large localSize.
   // We can observe a newer/larger local, it is fine (we must observe its zero-initialized-ness).
   s := atomic.LoadUintptr(&p.localSize) // load-acquire
   l := p.local                          // load-consume
   if uintptr(pid) < s {
      return indexLocal(l, pid), pid
   }
   return p.pinSlow()
}

1、Pool.pin() 函数的工作是：

• 锁住当前的 P；
• pinSlow 中先存储 local 再存储 localSize，这里是反顺加载两者，因为锁住 P 后就不会再发生 GC，保证 local 的大小至少跟 localSize 一样即可；
• Pool 实例如果是第一次执行 Get() 操作，那么初始化 Pool.local 数组，即执行 pinSlow()；（这句话，划重点）
• 返回当前 P 所对应的 poolLocal(poolLocalInternal) 对象，和 Processor.ID；

（a）runtime_procPin() 就是封装了procPin，禁止当前执行 G 所在的 MP 被抢占调度

func procPin() int {
    _g_ := getg()
    mp := _g_.m
    mp.locks++
    return int(mp.p.ptr().id)  // 并返回当前所在 P 的 ID
}

procPin 函数的目的是把当前 G 锁住在当前 M（声明当前 M 不能被抢占，也可以认为是 G 绑定到 P）。具体的操作就是执行 mp.locks++。因为在 newstack 里会对此条件做判断：

if preempt {
    // 已经打了抢占标识了，但是还需要判断条件满足才能让出执行权；
    if thisg.m.locks != 0 || thisg.m.mallocing != 0 || thisg.m.preemptoff != "" || thisg.m.p.ptr().status != _Prunning {
        gp.stackguard0 = gp.stack.lo + _StackGuard
        gogo(&gp.sched) // never return
    }
}

相对的，后面的 runtime_procUnpin() 操作就是执行了 mp.locks-- 操作。

（b）锁住 PG 后为什么会无法触发 GC？ 触发 GC 都会调用 gcStart(trigger gcTrigger) 函数，其大致逻辑是：先调用 preemptall() 尝试抢占所有的 P，然后停掉当前 P，遍历所有的 P，如果 P 处于系统调用则直接 stop 掉；然后处理空闲的 P；最后检查是否存在需要等待处理的P，如果有则循环等待，并尝试调用preemptall()。 preemptall() 函数中会调用 preemptone(p)，该逻辑中也会检查 m.locks 标识符。

GMP、GC 相关的详细介绍可以参考：mp.weixin.qq.com/s/vR44Y6AGK…

（c）Pool.pin() 函数中如果 Processor.ID 小于 Pool.localSize（如果大于，其实就是localSize=0、Pool 实例未初始化），就返回 Pool.local 数组中的第 Processor.ID 个元素和 Processor.ID。 而 Pool.local 数组大小就是 P 的数量，详细的见 local 初始化过程。

（d）Pool.pinSlow() 函数：初始化 Pool.local 数组 Pool 实例第一次调用 Get 的时候才会进入该逻辑（注意，是每个 P 都有第一次 Get 调用的概念，但是只有一个 P 上的 G 才能执行这个操作，因为有 allPoolsMu 锁互斥）。

func (p *Pool) pinSlow() (*poolLocal, int) {
   // G-M 先解锁
   runtime_procUnpin()
   
   // 全局锁 allPoolsMu，控制了只有一个 PG 能够执行初始化逻辑
   allPoolsMu.Lock()
   defer allPoolsMu.Unlock()
  
   // 再加锁，因为后面还有解锁操作
   pid := runtime_procPin()
  
   // runtime_procPin() 后不会发生 GC，所以获取当前的localSize和local值
   s := p.localSize
   l := p.local
   if uintptr(pid) < s {  // 已经被初始化
      return indexLocal(l, pid), pid
   }
   
   // 第一次时进行注册
   if p.local == nil {
      allPools = append(allPools, p)
   }
   
   size := runtime.GOMAXPROCS(0)  // P 的个数
   local := make([]poolLocal, size)
   atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-release
   atomic.StoreUintptr(&p.localSize, uintptr(size))         // store-release
   return &local[pid], pid
}

• Pool 把自己注册进 allPools 数组；
• Pool.local 数组按照 runtime.GOMAXPROCS(0) 的大小进行分配。如果是默认的，那么 P 个数就是 CPU 的个数；

2、shared.popHead() 函数是从无锁队列 poolDequeue.Head 处获取数据

• 当前 P 上的 G 取缓存对象时，只从头部链表节点指向的无锁队列里取。取不到，沿着 prev 指针到下一个无锁队列上重复操作，也没有的话。就到别的 P 上窃取。
• 盗窃者 G 在偷缓存对象时，只从尾部链表节点指向的无锁队列里取。取不到，沿着 next 指针到下一个无锁队列上重复操作，也没有的话。就到别的 P 那继续尝试偷，直到都偷不着，就调用 New 方法。

3、Pool.getSlow(pid) 从其他 P 中窃取对象，或者从本 P 的二级缓存 victim 中进行搜索。 （a）用 popTail() 从其他 P 那窃取对象 回顾一下：从当前 P 的 poolLocal 中取对象使用的是 popHead 方法，而从其他 P 的 poolLocal 中窃取对象时使用的是 popTail 方法，再回到上文中对 poolDequeue 的定义，可以知道，当前 P 对本地 poolLocal 是生产者，对其他 P 的 poolLocal 而言是消费者。 注意的是，popTail() 对成员变量的操作与 pushHead() 是反着来的，避免两个操作有交叉环节。

func (p *Pool) getSlow(pid int) interface{} {
   size := atomic.LoadUintptr(&p.localSize) // load-acquire
   locals := p.local                        // load-consume
  
   // 遍历其他 P 对应的数据桶（双向链表+环形队列），从队尾尝试获取对象
   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 中取对象
   size = atomic.LoadUintptr(&p.victimSize)
   if uintptr(pid) >= size {
      return nil
   }
   locals = p.victim
   // 先 private 区
   l := indexLocal(locals, pid)
   if x := l.private; x != nil {
      l.private = nil
      return x
   }
   // 后 shared 区
   for i := 0; i < int(size); i++ {
      l := indexLocal(locals, (pid+i)%int(size))
      if x, _ := l.shared.popTail(); x != nil {
         return x
      }
   }

   // 如果没有找到对象，则清空victim cache
   atomic.StoreUintptr(&p.victimSize, 0)
   return nil
}

（b）victim cache 翻译为“受害者缓存”

受害者缓存是由Norman Jouppi提出的一种提高缓存性能的硬件技术。如他的论文所述：Miss caching places a fully-associative cache between cache and its re-fill path. Misses in the cache that hit in the miss cache have a one cycle penalty, as opposed to a many cycle miss penalty without the miss cache. Victim Caching is an improvement to miss caching that loads the small fully-associative cache with victim of a miss and not the requested cache line.

其实可以理解为二级缓存，在一级缓存中搜索不到对象后，再下降到二级继续搜索。 注意的是，如果在二级缓存 victim 依旧没有找到元素的话，就把 victimSize 置 0，防止后来的“人”再到 victim 里找。

2.5、Put(x interface{})

Put 就是将缓存对象放入当前 P 对应的数据桶中，它也有渐进的逻辑：优先将变量放入 private 区中，对应的在 Get 时会先取这个位置上的对象；private 区如果已经被占用了，就放入 shared 双向链表中。

func (p *Pool) Put(x interface{}) {
   if x == nil {
      return
   }
   if race.Enabled {
      if fastrand()%4 == 0 { // Randomly drop x on floor.
         return
      }
      race.ReleaseMerge(poolRaceAddr(x))
      race.Disable()
   }
  
   // 初始化；禁用抢占； 获取对应的 poolLocal(poolLocalInternal) 对象
   l, _ := p.pin()
  
   // 尝试放到最快的位置，这个位置也跟 Get 请求的顺序是一一对应的；
   if l.private == nil {
      l.private = x
      x = nil
   }
   if x != nil {  // 复用 x 变量来控制逻辑
      l.shared.pushHead(x)  // 放到 shared 双向链表中
   }
   
   runtime_procUnpin()   // G-M 锁定解除，pin中有加锁
   
   if race.Enabled {
      race.Enable()
   }
}

注意的是，Put 操作也会调用 Pool.pin() ，所以上文的“第一次执行 Get() 操作时初始化 Pool.local 数组”是不严谨的（Pool.local 也可能会在这里创建）。

2.6、清理动作在 GC 时

1、三个全局变量

var (
   allPoolsMu Mutex  // 全局锁
   allPools []*Pool  // 存放程序运行期间所有的 Pool 实例地址
   oldPools []*Pool  // 配合 victim 机制用的
)

2、在 GC 开始的时候，gcStart() 函数中会调用 clearpools() -> poolCleanup() 函数。也就是说，每一轮 GC 都是对所有的 Pool 做一次清理

func init() {
   runtime_registerPoolCleanup(poolCleanup)
}

func clearpools() {
   // clear sync.Pools
   if poolcleanup != nil {
      poolcleanup()
   }
....

3、poolCleanup() 批量清理 allPools 里的元素

func poolCleanup() {
   // 清理 oldPools 上的 victim 的元素
   for _, p := range oldPools {
      p.victim = nil
      p.victimSize = 0
   }

   // 把 local 数组里的元素迁移到 victim 上
   for _, p := range allPools {
      p.victim = p.local
      p.victimSize = p.localSize
      p.local = nil
      p.localSize = 0
   }

   // 清理掉 allPools 和 oldPools 中所有的 Pool 实例
   oldPools, allPools = allPools, nil
}

victim 将删除缓存的动作由一次操作变为了两次操作，每次清理的是上上次的缓存内容，本次只是将缓存内容转移至 victim 中。对应的在 Get 操作中，也会去 victim 成员中尝试获取元素。 这样的设计，不仅提高了缓存的时间、缓存对象的命中率，并且对 shared 区的 O(n) 复杂度的遍历，变成 O(1) 操作，避免了性能波动。 注意的是，清除只是解除对象的引用，真正的清理由 GC 完成。

4、从清理操作来看，“第一次执行 Get() 操作会初始化 Pool.local 数组”这句话还有另一个错误，正确的是在每次 GC 后 local 都会赋值为 nil，都需要重新初始化。

2.7、思考 & 总结

1、Pool 是如何实现并发安全的？ 某一时刻 P 只会调度一个 G，那么对于生产者而言，当前 P 操作的都是本 P 的 poolLocal。相当于是通过隔离避免了数据竞争，即调用 pushHead 和 popHead 并不需要加锁。

当消费者是其他 P（窃取时），会进行 popTail 操作，这时会和 pushHead/popHead 操作形成数据竞争。pushHead 的流程是先取 slot，再判断是否可插入，最后修改 headTail；而 popTail 的流程是先修改headTail，再取 slot，然后重置 slot。pushHead 修改 head 位置，popTail修改 tail 位置，并且都是用的公共字段 headTail，这样的话使用 CAS 原子操作就可以避免读写冲突。

2、pinSlow() 函数中为什么先执行了 runtime_procUnpin，随后又执行了 runtime_procPin？ 目的是避免获取不到全局锁 allPoolsMu 却一直占用当前 P 的情况。而 runtime_procUnpin 和 runtime_procPin 之间，本 G 可能已经切换到其他 P 上，而 []poolLocal 也可能已经被初始化。

3、为什么需要将 poolDequeue 串成链表？ 因为 poolDequeue 的实现是固定大小的。

4、为什么要禁止 copy sync.Pool 实例？ 因为 copy 后，对于同一个 Pool 实例中的 cache 对象，就有了两个指向来源。原 Pool 清空之后，copy 的 Pool 没有清理掉，那么里面的对象就全都泄露了。并且 Pool 的无锁设计的基础是多个 Goroutine 不会操作到同一个数据结构，Pool 拷贝之后则不能保证这点（因为存储的成员都是指针）。

四、版本迭代

当前 sync.Pool 的实现，是在 1.13 版本中进行大改动后的结果，其通过无锁机制和两轮回收机制大大提高了缓存对象的使用效率。优秀的实现都是在不断的迭代中慢慢进步的，而在 version1.12 及之前 sync.Pool 存在如下一些问题：

• 每次 GC 都回收所有缓存对象，如果缓存对象数量太大，会导致 STW1 阶段的耗时增加；并且会导致缓存对象命中率下降，New 方法的执行造成额外的内存分配消耗。
• Pool.Get 和 Pool.Put 方法都是通过 mutex 锁实现的并发安全，并发高的情况下效率低。如果本 P 数据桶中没有缓存会逐个去其他数据桶进行窃取，极端情况下，最多尝试 P 次抢锁并查询缓存。

为此，1.13 版本在并发和缓存时间上进行了优化。 相关代码（version1.12）如下，简单的看过就可以发现导致上述各个问题的原因：

1、数据结构

type Pool struct {
   noCopy noCopy
   // local fixed-size per-P pool, actual type is [P]poolLocal
   local     unsafe.Pointer 
   localSize uintptr      // size of the local array
   New func() interface{}
}
type poolLocal struct {
   poolLocalInternal
   pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}
type poolLocalInternal struct {
   // vip位置，即缓存的第一层
   private interface{}   
   // 数据桶 ，每个 P 都对应一个。因为别的 P 上的 G 可以进行窃取，所以要加锁。
   shared  []interface{}
   Mutex                
}

同样的，也是每个 P 对应了一个数据桶，local 成员就是 [P]poolLocal 数组，用 Processor.ID 区分各自的桶。数据桶里有 private 区和 shared 区（这个结构比 version1.13 的简单很多）。

private 区只能存放一个对象，作为缓存的第一层，每次优先 Get/Put private 区。因为每个 P 在任意时刻只运行一个 G，并且每个 P 只会操作自己的 private 区，所以在 private 区上写入和取出对象是不用加锁的。

shared 区是一个 slice，可以存放多个对象。进 shared 区就是 append 操作，出 shared 区就截取 slice[:last-1] 操作。shared 区上写入和取出对象要加锁，因为别的 PG 可能来窃取。

2、加锁 Get，并有轮询窃取的可能 查询缓存顺序为：先看当前 P 的 private 区是否有数据；再加锁，尝试从当前 P 的 shared 区获取数据；循环遍历其他 P 的 share 区并加锁，尝试获取数据；如果都获取不到数据就 New 一个

func (p *Pool) Get() interface{} {
   // 获取本 P 对应的数据桶
   l := p.pin()
   
   // 先看当前 P 的 private 区是否有数据
   x := l.private
   l.private = nil
   runtime_procUnpin()
   if x == nil {
      // 再加锁，尝试从当前 P 的 shared 区获取数据
      l.Lock()
      last := len(l.shared) - 1
      if last >= 0 {
         x = l.shared[last]
         l.shared = l.shared[:last]
      }
      l.Unlock()
      
      // 循环遍历其他 P 的 share 区并加锁，尝试获取数据
      if x == nil {
         x = p.getSlow()
      }
   }
.......
}

func (p *Pool) getSlow() (x interface{}) {
   // See the comment in pin regarding ordering of the loads.
   size := atomic.LoadUintptr(&p.localSize) // load-acquire
   local := p.local                         // load-consume
   pid := runtime_procPin()
   runtime_procUnpin()
  
   // 遍历一次其他 P 的 share 区，尝试进行窃取
   for i := 0; i < int(size); i++ {
      l := indexLocal(local, (pid+i+1)%int(size))  // 定位到某个P上的shared区
      l.Lock()
      last := len(l.shared) - 1
      // 如果有缓存对象，就返回，并解锁
      if last >= 0 {
         x = l.shared[last]
         l.shared = l.shared[:last]
         l.Unlock()
         break
      }
      l.Unlock()  // 没有缓存对象，解锁，继续遍历下一个P
   }
   return x
}

由于这里的各种加锁逻辑（其实 Put 操作里也有锁），就会导致并发效率问题。 为此，version1.13 引入了单生产者多消费者的双端无锁环形队列。

3、清理缓存 同样的，sync.Pool 在 init() 中向 runtime 注册了一个 cleanup 方法。它在 STW1 阶段被调用的，如果执行时间过久，就会硬生生延长 STW1 阶段耗时。 同样的，cleanup 就是遍历清空当前注册的全部 Pool 实例，很容易造成 Pool 池中没有数据。

func poolCleanup() {
   for i, p := range allPools {
      allPools[i] = nil
      for i := 0; i < int(p.localSize); i++ {
         l := indexLocal(p.local, i)
         l.private = nil
         for j := range l.shared {
            l.shared[j] = nil
         }
         l.shared = nil
      }
      p.local = nil
      p.localSize = 0
   }
   allPools = []*Pool{}
}

为了解决相应问题，version1.13 就是引入了 victim 成员，实现了两轮回收机制，将实际回收的时间线拉长，单位时间内 GC 的开销减小。

参考

深入浅出 sync.Pool ，围观最全的使用姿势，理解最深刻的原理 sync.Pool 源码级原理剖析 sync.Pool的缺点和优化之路 深入剖析sync.Pool（文中有各种引申的计算机原理、GMP调度、GC知识，值得好好学习） 深度解密Go语言之sync.pool