目的

Many Go programs and packages try to reuse memory either for locality reasons or to reduce GC pressure。

缓解GC压力

GC（garbage collector）:

• 自动垃圾回收，减轻了程序员的压力
• 减轻压力的同时，也增加了运行时开销。

sync.pool应运而生，设计的目的是用来保存和复用临时对象，减小GC分配，降低GC压力。

Pool设计用意是在全局变量里维护的释放链表，尤其是被多个 goroutine 同时访问的全局变量。使用Pool代替自己写的释放链表，可以让程序运行的时候，在恰当的场景下从池里-重用-某项值。
sync.Pool的一种使用场景是，为临时缓冲区创建一个池，多个客户端使用这个缓冲区来共享全局资源。
另一方面，不恰当的使用例子，如果释放链表是某个对象的一部分，并由这个对象维护，而这个对象只由一个客户端使用，在这个客户端工作完成后释放链表，那么用Pool实现这个释放链表是不合适的。

由来讨论

Brad Fizpatrick曾建议在创建一个工友的Cache类型。这个建议引发了一长串的讨论。Go 语言应该在标准库里提供一个这个样子的类型，还是应当将这个类型作为私下的实现？这个实现应该真的释放内存么？如果释放，什么时候释放？这个类型应当叫做Cache，或者更应该叫做Pool
https://github.com/golang/go/issues/4720
https://my.oschina.net/u/115763/blog/282376

简单介绍

• A Pool is a set of temporary objects that may be individually saved and retrieved.

• 池是一组可以单独保存和检索的临时对象

• Any item stored in the Pool may be removed automatically at any time without notification. If the Pool holds the only reference when this happens, the item might be deallocated.

• 存储在池中的任何项目都可以随时自动删除，而无需通知。如果发生这种情况时池保存唯一的引用，则可能会释放该项

• A Pool is safe for use by multiple goroutines simultaneously

• 并发安全

上面三句是pool源码上的摘抄解释

pool特性

-没有大小限制，大小只受限与GC的临界值
-对象的最大缓存周期是GC周期，当GC调用时，没有被引用的对象的会被清理掉
-Get方法返回的都是池子中的任意一个对象，没有顺序，注意是没有顺序的；如果当期池子为空，会调用New方法创建一个对象，没有New方法则会返回nil

使用场景

高并发场景下，当多个goroutine都需要创建同⼀个临时对象的时候，因为对象是占内存的，进⽽导致的就是内存回收的GC压⼒增加。

造成 “并发⼤大-占⽤内存⼤大-GC缓慢-处理理并发能⼒力力降低-并发更更 ⼤大”这样的恶性循环。

vicious loop

业界使用

Echo：

使用了sync.pool来从用内存，实现了0动态内存分配

echo routing

https://echo.labstack.com/guide/routing

github.com/labstack/echo

Gin：

gin context

上面是gin使用pool作为context的缓存
https://github.com/gin-gonic/gin/blob/73ccfea3ba5a115e74177dbfbc1ea0fff88c13f4/gin.go

fmt：

原生的fmt包里，也包含了sync.pool的调用。

fmt 部分代码

源码分析

mpg.png

如上图所示：在go的M、P、G模型中，每个P绑定了一个poolLocalInternal，这结合了go的优势，使得当前P绑定等待队列中的任何G对poolLocalInternal的访问都不需要加锁。每个poolLocalInternal中包含private和shared。private为单个对象，为每个P单独维护，不具有共享特质，每次获取和添加都会首先设置private；shared为一系列的临时对象，为共享队列，各个P之间通过shard共享对象集，在go1.13之前，shard为数组，在1.13之后修改为使用环形数组，通过CAS实现了lock-free。

数据结构

从全局的角度来看，全局维护了一个统一的结构，如上图所示的红色的pool，pool维护每个产生的local，每个local指向每个P绑定的poolLocalInternal。

before Go1.13：

// A Pool must not be copied after first use.type Pool struct {   noCopy noCopy   local     unsafe.Pointer // local fixed-size per-P pool, actual type is [P]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() interface{}}// Local per-P Pool appendix.// 1.13之前type poolLocalInternal struct {   private interface{}   // Can be used only by the respective P.   shared  []interface{} // Can be used by any P.   Mutex                 // Protects shared.}

上面定义了一个Pool结构体，其中声明了noCopy；poolLocalInternal是每个P的一个附件，其中包含一个private的私有对象，只能当前P访问，在获取和设置的时候会优先从改私有对象中获取和一个shared的数组，可以被任意的P访问。

// Put adds x to the pool.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()   }   l := p.pin()   if l.private == nil {      l.private = x      x = nil   }   runtime_procUnpin()   if x != nil {      l.Lock()      l.shared = append(l.shared, x)      l.Unlock()   }   if race.Enabled {      race.Enable()   }}

Put函数为sync.pool的主要函数，用于添加对象。调用了p.pin()获取当前P的绑定附件，runtime_procUnpin解除绑定关系，并且设计设置禁止关系（不禁止强占可能造成并发问题），通过P先判断是否可以放进private对象中，否则放进shard数组中。

// Get selects an arbitrary item from the Pool, removes it from the// Pool, and returns it to the caller.// Get may choose to ignore the pool and treat it as empty.// Callers should not assume any relation between values passed to Put and// the values returned by Get.//// If Get would otherwise return nil and p.New is non-nil, Get returns// the result of calling p.New.func (p *Pool) Get() interface{} {   if race.Enabled {      race.Disable()   }   l := p.pin()   x := l.private   l.private = nil   runtime_procUnpin()   if x == nil {      l.Lock()      last := len(l.shared) - 1      if last >= 0 {         x = l.shared[last]         l.shared = l.shared[:last]      }      l.Unlock()      if x == nil {         x = p.getSlow()      }   }   if race.Enabled {      race.Enable()      if x != nil {         race.Acquire(poolRaceAddr(x))      }   }   if x == nil && p.New != nil {      x = p.New()   }   return x}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   // Try to steal one element from other procs.   pid := runtime_procPin()   runtime_procUnpin()   for i := 0; i < int(size); i++ {      l := indexLocal(local, (pid+i+1)%int(size))      l.Lock()      last := len(l.shared) - 1      if last >= 0 {         x = l.shared[last]         l.shared = l.shared[:last]         l.Unlock()         break      }      l.Unlock()   }   return x}

Get函数和Put函数一致，通过pin()获取当前P绑定的附件。先从private中获取，再冲shard中获取，获取失败再调用getslow函数，在getslow函数中，通过遍历获取其余P的shared资源，会偷取最后一个，最后再偷取失败才会使用出事化函数New()

Get执行流程：private->shard->getslow()->New()

// pin pins the current goroutine to P, disables preemption and returns poolLocal pool for the P.// Caller must call runtime_procUnpin() when done with the pool.func (p *Pool) pin() *poolLocal {   pid := runtime_procPin()   // In pinSlow we store to localSize and then to local, 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)   }   return p.pinSlow()}func (p *Pool) pinSlow() *poolLocal {   // Retry under the mutex.   // Can not lock the mutex while pinned.   runtime_procUnpin()   allPoolsMu.Lock()   defer allPoolsMu.Unlock()   pid := runtime_procPin()   // poolCleanup won't be called while we are pinned.   s := p.localSize   l := p.local   if uintptr(pid) < s {      return indexLocal(l, pid)   }   if p.local == nil {      allPools = append(allPools, p)   }   // If GOMAXPROCS changes between GCs, we re-allocate the array and lose the old one.   size := runtime.GOMAXPROCS(0)   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]}

pin函数索引当前G对应的绑定的P，通过runtime_procPin设置禁止强占，返回当前P拥有的poolLocal，获取不到时调用pinslow进行第二次获取。第二次调用会先使用runtime_procUnpin()进行强占解除，对全局锁加锁，这是如果local为空（第一次创建），则加入全局队列中。

func poolCleanup() {   // This function is called with the world stopped, at the beginning of a garbage collection.   // It must not allocate and probably should not call any runtime functions.   // Defensively zero out everything, 2 reasons:   // 1. To prevent false retention of whole Pools.   // 2. If GC happens while a goroutine works with l.shared in Put/Get,   //    it will retain whole Pool. So next cycle memory consumption would be doubled.   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{}}var (   allPoolsMu Mutex   allPools   []*Pool)func init() {   runtime_registerPoolCleanup(poolCleanup)}

poolCleanup为运行时的注册函数，在GC开始时调用，逻辑很暴力，三层for循环赋空！

这个版本有啥缺点

• 对全局shared加锁读写，性能较低
• 三层for循环赋空很暴力，容易造成GC的尖峰
• 每次GC对全量清空，造成的缓存命中率下降

After Go1.13

在GO1.13之后，优化了以上的问题：

• 对全局的shard加锁，使用了CAS实现了lock-free
• 对GC造成的尖峰问题，引入了受害者缓存。延长了缓存的声明周期，增加了缓存的命中效率

after go1.13

可以很清楚的发现，和之前的数据结构相比，1.13之后的版本增加了黄色的poolDequene，那这这和黄色部分又是何方神圣呢？

// 1.13之后// Local per-P Pool appendix.type Pool struct {   noCopy noCopy   local     unsafe.Pointer // local fixed-size per-P pool, actual type is [P]poolLocal   localSize uintptr        // size of the local array   victim     unsafe.Pointer // local from previous cycle   victimSize uintptr        // size of victims 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() interface{}}type poolLocalInternal struct {   private interface{} // Can be used only by the respective P.   shared  poolChain   // Local P can pushHead/popHead; any P can popTail.}type poolChain struct {   // head is the poolDequeue to push to. This is only accessed   // by the producer, so doesn't need to be synchronized.   head *poolChainElt   // tail is the poolDequeue to popTail from. This is accessed   // by consumers, so reads and writes must be atomic.   tail *poolChainElt}type poolChainElt struct {   poolDequeue   // next and prev link to the adjacent poolChainElts in this   // poolChain.   //   // next is written atomically by the producer and read   // atomically by the consumer. It only transitions from nil to   // non-nil.   //   // prev is written atomically by the consumer and read   // atomically by the producer. It only transitions from   // non-nil to nil.   next, prev *poolChainElt}// poolDequeue is a lock-free fixed-size single-producer,// multi-consumer queue. The single producer can both push and pop// from the head, and consumers can pop from the tail.//// It has the added feature that it nils out unused slots to avoid// unnecessary retention of objects. This is important for sync.Pool,// but not typically a property considered in the literature.type poolDequeue struct {   // headTail packs together a 32-bit head index and a 32-bit   // tail index. Both are indexes into vals modulo len(vals)-1.   //   // tail = index of oldest data in queue   // head = index of next slot to fill   //   // Slots in the range [tail, head) are owned by consumers.   // A consumer continues to own a slot outside this range until   // it nils the slot, at which point ownership passes to the   // producer.   //   // The head index is stored in the most-significant bits so   // that we can atomically add to it and the overflow is   // harmless.   headTail uint64   // vals is a ring buffer of interface{} values stored in this   // dequeue. The size of this must be a power of 2.   //   // vals[i].typ is nil if the slot is empty and non-nil   // otherwise. A slot is still in use until *both* the tail   // index has moved beyond it and typ has been set to nil. This   // is set to nil atomically by the consumer and read   // atomically by the producer.   vals []eface}

对锁的优化：
Go在1.13之后增加了poolDequene：

• lock-free
• 生产者可以进行pushHead和popTail
• 消费者只能进行popTail

// Put adds x to the pool.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()   }   l, _ := p.pin()   if l.private == nil {      l.private = x      x = nil   }   if x != nil {      l.shared.pushHead(x)   }   runtime_procUnpin()   if race.Enabled {      race.Enable()   }}func (c *poolChain) pushHead(val interface{}) {   d := c.head   if d == nil {      // Initialize the chain.      const initSize = 8 // Must be a power of 2      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)}// pushHead adds val at the head of the queue. It returns false if the// queue is full. It must only be called by a single producer.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 {      // Queue is full.      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}

新版本使用l.shared.pushHead(x)，进行头添加，删除了锁的使用。

func (p *Pool) Get() interface{} {   if race.Enabled {      race.Disable()   }   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 race.Enabled {      race.Enable()      if x != nil {         race.Acquire(poolRaceAddr(x))      }   }   if x == nil && p.New != nil {      x = p.New()   }   return x}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}// popHead removes and returns the element at the head of the queue.// It returns false if the queue is empty. It must only be called by a// single producer.func (d *poolDequeue) popHead() (interface{}, bool) {   var slot *eface   for {      ptrs := atomic.LoadUint64(&d.headTail)      head, tail := d.unpack(ptrs)      if tail == head {         // Queue is empty.         return nil, false      }      // Confirm tail and decrement head. We do this before      // reading the value to take back ownership of this      // slot.      head--      ptrs2 := d.pack(head, tail)      if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) {         // We successfully took back slot.         slot = &d.vals[head&uint32(len(d.vals)-1)]         break      }   }   val := *(*interface{})(unsafe.Pointer(slot))   if val == dequeueNil(nil) {      val = nil   }   // Zero the slot. Unlike popTail, this isn't racing with   // pushHead, so we don't need to be careful here.   *slot = eface{}   return val, true}

在获取临时对象的时候，会首先从private中获取，private为空会接着从shard变量中拉取，shared变量中也没有空闲，接着调用getSlow从其他P中偷取，偷取失败的时候，这时候会使用受害者缓存，这一步是新添加，接着才会调用New()。

Get执行流程：private->shard->getslow()->victim→New()

针对GC尖峰的优化：

func poolCleanup() {   // This function is called with the world stopped, at the beginning of a garbage collection.   // It must not allocate and probably should not call any runtime functions.   // Because the world is stopped, no pool user can be in a   // pinned section (in effect, this has all Ps pinned).   // Drop victim caches from all pools.   for _, p := range oldPools {      p.victim = nil      p.victimSize = 0   }   // Move primary cache to victim cache.   for _, p := range allPools {      p.victim = p.local      p.victimSize = p.localSize      p.local = nil      p.localSize = 0   }   // The pools with non-empty primary caches now have non-empty   // victim caches and no pools have primary caches.   oldPools, allPools = allPools, nil}

受害者缓存（Victim Cache）：是一个与直接匹配或低相联缓存并用的、容量很小的全相联缓存。当一个数据块被逐出缓存时，并不直接丢弃，而是暂先进入受害者缓存。如果受害者缓存已满，就替换掉其中一项。当进行缓存标签匹配时，在与索引指向标签匹配的同时，并行查看受害者缓存，如果在受害者缓存发现匹配，就将其此数据块与缓存中的不匹配数据块做交换，同时返回给处理器。

新版本的poolCleanup增加了victim，对于原来应该被GC的缓存，添加到了victim,销毁滞后到了下一轮，以此来解决缓存命中率低的问题。

基准测试

package mainimport (   "sync"   "testing")type info struct {   Val int}func BenchmarkNoPool(b *testing.B) {   b.ResetTimer()   var k *info   for i := 0; i < b.N; i++ {      k = &info{Val: 1}      k.Val += 1   }}var pInfo = sync.Pool{New: func() interface{} {   return new(info)}}func BenchmarkWithPool(b *testing.B) {   b.ResetTimer()   for i := 0; i < b.N; i++ {      k := pInfo.Get().(*info)      // 重置      k.Val = 0      k.Val += 1      pInfo.Put(k)   }}

测试结果

go test -bench=. -benchmem                                                                                                 goos: darwingoarch: amd64pkg: pool_testBenchmarkNoPool-4       78748666                13.7 ns/op             8 B/op          1 allocs/opBenchmarkWithPool-4     75934996                16.2 ns/op             0 B/op          0 allocs/opPASSok      pool_test       3.962s

给这篇文章的作者打赏

微信扫一扫打赏

支付宝扫一扫打赏

×

打赏 点赞(0)