golang sync.Pool在1.14中的优化

分手后的思念是犯贱 2023-03-14 05:49 128阅读 0赞

### golang sync.Pool在1.14中的优化 ###

*  sync.Pool在1.12中实现的原理简述
 *  1.14 Pool 数据结构
 *   *  poolChain
     *   *  poolChain.popHead()
         *  poolChain.pushHead()
         *  poolChain.popTail()
     *  poolDequeue
     *   *  数据结构
         *  pack/unpack
         *  poolDequeue.pushHead
         *  poolDequeue.popHead
         *  poolDequeue.popTail
     *  数据结构总结
 *  Pool.Put
 *  Pool.Get
 *  victim cache优化与GC
 *  sync.Pool 总结

> 本文基于golang 1.14 对sync.Pool进行分析;

# sync.Pool在1.12中实现的原理简述 #

参考 [Golang 的 sync.Pool设计思路与原理][Golang _ sync.Pool]，这篇文章基于1.12版本对golang的sync.Pool实现原理进行了分析。关于sync.Pool的使用场景、基本概念的理解可以参考前面的文章。

在1.12中sync.Pool的设计思路简单总结一下：

1.  将 G 和 P 绑定，设置与P绑定的M禁止抢占以防止 G 被抢占。在绑定期间，GC 无法清理缓存的对象。
2.  每个p都有独享的缓存队列，当g进行sync.pool操作时，先找到所属p的private，如果没有对象可用，加锁从 shared切片里获取数据。如果还没有拿到缓存对象，那么到其他P的poolLocal进行偷数据，如果偷不到，那么就创建新对象。

1.12 sync.pool的源码，可以发现sync.pool里会有各种的锁逻辑，从自己的shared拿数据加锁。getSlow()偷其他P缓存，也是需要给每个p加锁。put归还缓存的时候，还是会mutex加一次锁。

go mutex锁的实现原理简单说，他开始也是atomic cas自旋，默认是4次尝试，当还没有拿到锁的时候进行waitqueue gopack休眠调度处理，等待其他协程释放锁时进行goready调度唤醒。

Go 1.13之后，Go 团队对sync.Pool的锁竞争这块进行了很多优化，这里还改变了shared的数据结构，以前的版本用切片做缓存，现在换成了poolChain双端链表。这个双端链表的设计很有意思，你看sync.pool源代码会发现跟redis quicklist相似，都是链表加数组的设计。

# 1.14 Pool 数据结构 #

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{ }
    }
    
    // Local per-P Pool appendix.
    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 poolLocal struct { 
    	poolLocalInternal
    
    	// Prevents false sharing on widespread platforms with
    	// 128 mod (cache line size) = 0 .
    	pad [128 - unsafe.Sizeof(poolLocalInternal{ })%128]byte
    }

`Pool.local` 实际上是一个类型 `[P]poolLocal` 数组，数组长度是调度器中P的数量，也就是说每一个P有自己独立的poolLocal。通过P.id来获取每个P自己独立的poolLocal。在poolLocal中有一个poolChain。

这里我们先忽略其余的机构，重点关注`poolLocalInternal.shared` 这个字段。poolChain是一个双端队列链，缓存对象。 1.12版本中对于这个字段的并发安全访问是通过mutex加锁实现的；1.14优化后通过poolChain(无锁化)实现的。

这里我们先重点分析一下poolChain 是怎么实现并发无锁编程的。

## poolChain ##

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
    }

`poolChain`是一个动态大小的双向链接列表的双端队列。每个出站队列的大小是前一个队列的两倍。`也就是说poolChain里面每个元素poolChainElt都是一个双端队列`。

head指向的poolChainElt，是用于Producer去Push元素的，不需要做同步处理。  
tail指向的poolChainElt，是用于Consumer从tail去pop元素的，这里的读写需要保证原子性。

简单来说，`poolChain`是一个单Producer，多Consumer并发访问的双端队列链。

对于`poolChain`中的每一个双端队列 poolChainElt，包含了双端队列实体`poolDequeue` 一起前后链接的指针。

`poolChain` 主要方法有：

popHead() (interface{ }, bool);
    pushHead(val interface{ }) 
    
    popTail() (interface{ }, bool)

`popHead`和`pushHead`函数是给Producer调用的；`popTail`是给Consumer并发调用的。

### poolChain.popHead() ###

前面我们说了，`poolChain`的head 指针的操作是单Producer的。

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
    }

poolChain要求，`popHead`函数只能被Producer调用。看一下逻辑：

1.  获取头结点 head;
2.  如果头结点非空就从头节点所代表的双端队列`poolDequeue`中调用`popHead`函数。注意这里`poolDequeue`的`popHead`函数和`poolChain`的`popHead`函数并不一样。`poolDequeue`是一个固定size的ring buffer。
3.  如果从head中拿到了value，就直接返回；
4.  如果从head中拿不到value，就从head.prev再次尝试获取；
5.  最后都获取不到，就返回nil。

### poolChain.pushHead() ###

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)
    }

poolChain要求，`pushHead`函数同样只能被Producer调用。看一下逻辑：

1.  首先还是获取头结点 head;
2.  如果头结点为空，需要初始化chain
    
    1.  创建`poolChainElt` 节点，作为head, 当然也是tail。
    2.  `poolChainElt` 其实也是固定size的双端队列`poolDequeue`，size必须是2的n次幂。
3.  调用`poolDequeue`的`pushHead`函数将 val push进head的双端队列`poolDequeue`。
4.  如果push失败了，说明双端队列满了，需要重新创建一个双端队列d2，新的双端队列的size是前一个双端队列size的2倍；
5.  更新poolChain的head指向最新的双端队列，并且建立双链关系；
6.  然后将val push到最新的双端队列。

这里需要注意一点的是head其实是指向最后chain中最后一个结点(poolDequeue)，chain执行push操作是往最后一个节点push。 所以这里的head的语义不是针对链表结构，而是针对队列结构。

### poolChain.popTail() ###

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
    	}
    }

poolChain要求，`popTail`函数能被任何P调用，也就是所有的P都是Consumer。这里总结下，当前G所对应的P在Pool里面是Producer角色，任何P都是Consumer角色。

`popTail`函数是并发调用的，所以需要特别注意。

1.  首先需要原子的load chain的tail指向的双端队列d（`poolDequeue`）；
2.  如果d为空，pool还是空，所以直接return nil
3.  下面就是典型的无锁原子化编程：进入一个for循环
    
    1.  首先就是获取tail的next结点d2。这里需要强调一下为什么需要在tail执行popTail之前先load tail 的next结点。
        
        1.  tail有可能存在短暂性为空的场景。比如head和tail实际指向同一个结点(双端队列)时候，可能tail为空只是暂时的，因为存在有线程往head push数据的情况。
        2.  如果因为tail 执行popTail()时因为tail为空而失败了，然后再load tail.next，发现 tail.next非空，再将tail原子切换到tail.next，这个时候就会出现错误了。假设tail和head指向同一个结点，在判断tail是空之后，head往里面插入了很多个数据，直接将tail结点打满，然后head指向下一个结点了。这时候tail.next也非空了。然后就将tail更新到tail.next，就会导致丢数据了。
        3.  所以必须在：1）tail执行popTail之前tail.next是非空的，2）tail执行popTail时发现tail是空的。满足这两个条件才能说明tail是永久性是空的。也就是需要提前load tail.next指针。
    2.  如果从tail里面pop数据成功，就直接返回val。
    3.  如果从tail里面pop数据失败，并且d2也是空，说明当前chain里面只有一个结点，并且是空。直接返回nil
    4.  如果从tail里面pop数据失败并且d2非空，说明tail已经被drain干净了，原子的tail到tail.next，并清除双向链表关系。
    5.  从d2开始新的一轮for循环。

上面的流程是典型的的无锁并发编程。

## poolDequeue ##

poolChain中每一个结点都是一个双端队列`poolDequeue`。

`poolDequeue`是一个无锁的、固定size的、单Producer、多Consumer的deque。只有一个Producer可以从head去push或则pop；多个Consumer可以从tail去pop。

### 数据结构 ###

type poolDequeue struct { 
    	// 用高32位和低32位分别表示head和tail
    	// head是下一个fill的slot的index;
    	// tail是deque中最老的一个元素的index
    	// 队列中有效元素是[tail, head)
    	headTail uint64
    
    	vals []eface
    }
    
    type eface struct { 
    	typ, val unsafe.Pointer
    }

这里通过一个字段 `headTail` 来表示head和tail的index。`headTail`是8个字节64位。

1.  高32位表示head;
2.  低32位表示tail。
3.  head和tail自加溢出时是安全的。

`vals`是一个固定size的slice，其实也就是一个 `ring buffer`，size必须是2的次幂(为了做位运算)；

### pack/unpack ###

一个字段 `headTail` 来表示head和tail的index，所以需要有具体的pack和unpack逻辑：

const dequeueBits = 32
    
    func (d *poolDequeue) unpack(ptrs uint64) (head, tail uint32) { 
    	const mask = 1<<dequeueBits - 1
    	head = uint32((ptrs >> dequeueBits) & mask)
    	tail = uint32(ptrs & mask)
    	return
    }
    
    func (d *poolDequeue) pack(head, tail uint32) uint64 { 
    	const mask = 1<<dequeueBits - 1
    	return (uint64(head) << dequeueBits) |
    		uint64(tail&mask)
    }

pack:

1.  首先拿到mask，这里实际上就是 0xffffffff（2^32-1）
2.  head左移32位 | tail&0xffffffff 就可以得到head和tail pack之后的值。

unpack:

1.  首先拿到mask，这里实际上就是 0xffffffff（2^32-1）
2.  ptrs右移32位拿到高32位然后 & mask 就可以得到head；
3.  ptrs直接 & mask 就可以得到低32位，也就是tail。

### poolDequeue.pushHead ###

`pushHead` 将val push到head指向的位置，如果deque满了，就返回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 { 
    		// 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
    }

主要逻辑：

1.  原子load head和tail，
2.  如果tail + len(vals) == head 说明deque已经满了。
3.  拿到head在vals中index的slot
4.  如果slot的type非空，说明该slot还没有被popTail release，实际上deque还是满的；所以直接return false;
5.  更新val到slot的指针指向的值。
6.  原子的自加head

需要注意的是，pushHead不是并发安全的，只能有一个Producer去执行；只有slot的的type指针为空时候slot才是空。

### poolDequeue.popHead ###

`popHead` 将head指向的前一个位置弹出，如果deque是空，就返回false。

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
    }

主要逻辑：

1.  由于从head前一个位置pop元素，可能会与tail位置pop冲突，所以不可避免的需要cas操作。所以最开始进入就是一个for循环；
2.  原子load `poolDequeue.headTail`然后unpack拿到head和tail
3.  如果head == tail，表示deque是空，直接return nil.
4.  head –
5.  根据新的head和老的tail， 重新pack出ptrs2;
6.  原子cas更新`poolDequeue.headTail`，`atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2)`，
7.  如果更新成功，就拿到head执行的slot，并获取到实际的value，并return；
8.  如果原子更新失败了，重新进入for循环再次执行。

### poolDequeue.popTail ###

这个函数是可以被Consumer并发访问的。

func (d *poolDequeue) popTail() (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 head and tail (for our speculative check
    		// above) and increment tail. If this succeeds, then
    		// we own the slot at tail.
    		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
    	}
    
    	// Tell pushHead that we're done with this slot. Zeroing the
    	// slot is also important so we don't leave behind references
    	// that could keep this object live longer than necessary.
    	//
    	// We write to val first and then publish that we're done with
    	// this slot by atomically writing to typ.
    	slot.val = nil
    	atomic.StorePointer(&slot.typ, nil)
    	// At this point pushHead owns the slot.
    
    	return val, true
    }

主要逻辑：

1.  并发访问，所以与cas相关的for循环不可少；
2.  原子load，拿到head和tail值；
3.  将(tail+1)和head重新pack成ptrs2；
4.  CAS：`atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2)`; 如果更新成功，就拿到vals\[tail\]t的指针。如果失败就再次返回1的for循环。
5.  拿到slot对应的val。
6.  将slot的val和type都清为nil, 告诉pushHead, slot我们已经使用完了，pushHead可以往里面填充数据了。

## 数据结构总结 ##

用一张图完整描述sync.Pool的数据结构：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70]  
强调一点：

1.  head的操作只能是local P;
2.  tail的操作是任意P；

参考网上一张图来看更加清晰：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70 1]  
Pool 并没有直接使用 poolDequeue，因为它是fixed size的，而 Pool 的大小是没有限制的。因此，在 poolDequeue 之上包装了一下，变成了一个 poolChainElt 的双向链表，可以动态增长。

# Pool.Put #

func (p *Pool) Put(x interface{ }) { 
    	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()
    }
    
    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()
    }
    
    func (p *Pool) pinSlow() (*poolLocal, int) { 
    	// 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), 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], pid
    }

Put函数主要逻辑：

1.  先调用`p.pin()` 函数，这个函数会将当前 goroutine与P绑定，并设置当前g不可被抢占(也就不会出现多个协程并发读写当前P上绑定的数据)；  
    1. 在`p.pin()` 函数里面还会check per P的\[P\]poolLocal数组是否发生了扩容(P 扩张)。  
    2. 如果发生了扩容，需要调用`pinSlow()`来执行具体扩容。扩容获取一个调度器全局大锁`allPoolsMu`，然后根据当前最新的P的数量去执行新的扩容。这里的成本很高，所以尽可能避免手动增加P的数量。
2.  拿到per P的poolLocal后，优先将val put到private，如果private已经存在，就通过调用`shared.pushHead(x)` 塞到poolLocal里面的无锁双端队列的chain中。Put函数对于双端队列来说是作为一个Producer角色，所以这里的调用是无锁的。
3.  最后解除当前goroutine的禁止抢占。

# Pool.Get #

func (p *Pool) Get() interface{ } { 
    	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
    }
    
    func (p *Pool) getSlow(pid int) interface{ } { 
    	// See the comment in pin regarding ordering of the loads.
    	size := atomic.LoadUintptr(&p.localSize) // load-acquire
    	locals := p.local                        // load-consume
    	// Try to steal one element from other procs.
    	for i := 0; i < int(size); i++ { 
    		l := indexLocal(locals, (pid+i+1)%int(size))
    		if x, _ := l.shared.popTail(); x != nil { 
    			return x
    		}
    	}
    
    	// Try the victim cache. We do this after attempting to steal
    	// from all primary caches because we want objects in the
    	// victim cache to age out if at all possible.
    	size = atomic.LoadUintptr(&p.victimSize)
    	if uintptr(pid) >= size { 
    		return nil
    	}
    	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))
    		if x, _ := l.shared.popTail(); x != nil { 
    			return x
    		}
    	}
    
    	// Mark the victim cache as empty for future gets don't bother
    	// with it.
    	atomic.StoreUintptr(&p.victimSize, 0)
    
    	return nil
    }

Get函数主要逻辑：

1.  设置当前 goroutine 禁止抢占；
2.  从 poolLocal的private取，如果private不为空直接return；
3.  从 poolLocal.shared这个双端队列chain里面无锁调用去取，如果取得到也直接return；
4.  上面都去不到，调用`getSlow(pid)`去取
    
    1.  首先会通过 steal 算法，去别的P里面的poolLocal去取，这里的实现是无锁的cas。如果能够steal一个过来，就直接return；
    2.  如果steal不到，则从 victim 里找，和 poolLocal 的逻辑类似。最后，实在没找到，就把 victimSize 置 0，防止后来的“人”再到 victim 里找。
5.  最后还拿不到，就通过New函数来创建一个新的对象。

这里是一个很明显的多层级缓存优化 + GPM调度结合起来。

`private -> shared -> steal from other P -> victim cache -> New`

# victim cache优化与GC #

对于Pool来说并不能够无上限的扩展，否则对象占用内存太多了，会引起内存溢出。

> 几乎所有的池技术中，都会在某个时刻清空或清除部分缓存对象，那么在 Go 中何时清理未使用的对象呢？

这里是使用GC。在pool.go里面的`init函数` 会注册清理函数：

func init() { 
    	runtime_registerPoolCleanup(poolCleanup)
    }
    
    // mgc.go
    //go:linkname sync_runtime_registerPoolCleanup sync.runtime_registerPoolCleanup
    func sync_runtime_registerPoolCleanup(f func()) { 
    	poolcleanup = f
    }

编译器会把 `runtime_registerPoolCleanup` 函数调用链接到 mgc.go 里面的 `sync_runtime_registerPoolCleanup`函数调用，实际上注册到poolcleanup函数。整个调用链如下：

`gcStart() -> clearpools() -> poolcleanup()`

也就是每一轮GC开始都会执行pool的清除操作。

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
    }

poolCleanup 会在 STW 阶段被调用。整体看起来，比较简洁。主要是将 local 和 victim 作交换，这样也就不致于让 GC 把所有的 Pool 都清空了，有 victim 在“兜底”。

重点：如果 sync.Pool 的获取、释放速度稳定，那么就不会有新的池对象进行分配。如果获取的速度下降了，那么`对象可能会在两个 GC 周期内被释放，而不是以前的一个 GC 周期。`

在Go1.13之前的poolCleanup比较粗暴，直接清空了所有 Pool 的 p.local 和poolLocal.shared。

通过两者的对比发现，新版的实现相比 Go 1.13 之前，GC 的粒度拉大了，由于实际回收的时间线拉长，单位时间内 GC 的开销减小。

所以 p.victim 的作用其实就是次级缓存。

# sync.Pool 总结 #

1.  关键思想是对象的复用，避免重复创建、销毁。将暂时不用的对象缓存起来，待下次需要的时候直接使用，不用再次经过内存分配，复用对象的内存，减轻 GC 的压力。
2.  sync.Pool 是协程安全的，使用起来非常方便。设置好 New 函数后，调用 Get 获取，调用 Put 归还对象。
3.  不要对 Get 得到的对象有任何假设，默认Get到对象是一个空对象，Get之后手动初始化。
4.  好的实践是：Put操作执行前将对象“清空”，并且确保对象被Put进去之后不要有任何的指针引用再次使用，不然极大概率导致data race。
5.  第三和第四也就是考虑清楚复用对象的生命周期
6.  Pool 里对象的生命周期受 GC 影响，不适合于做连接池，因为连接池需要自己管理对象的生命周期。
7.  Pool 不可以指定⼤⼩，⼤⼩只受制于 GC 临界值。
8.  procPin 将 G 和 P 绑定，防止 G 被抢占。在绑定期间，GC 无法清理缓存的对象。
9.  在加入 victim 机制前，sync.Pool 里对象的最⼤缓存时间是一个 GC 周期，当 GC 开始时，没有被引⽤的对象都会被清理掉；加入 victim 机制后，最大缓存时间为两个 GC 周期。
10. Victim Cache 本来是计算机架构里面的一个概念，是 CPU 硬件处理缓存的一种技术，sync.Pool 引入的意图在于降低 GC 压力的同时提高命中率。
11. sync.Pool 的最底层使用切片加链表来实现双端队列，并将缓存的对象存储在切片中。
12. sync.Pool 的设计理念，包括：无锁、操作对象隔离、原子操作代替锁、行为隔离——链表、Victim Cache 降低 GC 开销。

参考文档：  
[理解Go 1.13中sync.Pool的设计与实现][Go 1.13_sync.Pool]  
[golang新版如何优化sync.pool锁竞争消耗？][golang_sync.pool]  
[深度解密 Go 语言之 sync.Pool][Go _ sync.Pool]

[Golang _ sync.Pool]: https://louyuting.blog.csdn.net/article/details/90647884
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70]: /images/20230312/dc30d0b0e84d4f85a8a7171833c2ce16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70 1]: /images/20230312/fd6dac7c8e584e6199e44b8369c68396.png
[Go 1.13_sync.Pool]: https://zhuanlan.zhihu.com/p/110140126
[golang_sync.pool]: http://xiaorui.cc/archives/5878
[Go _ sync.Pool]: https://www.cnblogs.com/qcrao-2018/p/12736031.html