如何使用就不讲了，网上很多文章
1. 结构
代码语言：javascript复制type Pool struct {
noCopy noCopy // 用于保证pool不会被复制
local unsafe.Pointer // 实际类型是 [P]poolLocal
localSize uintptr // local的size
victim unsafe.Pointer // 在新一轮GC来临时接管local，用于减少GC之后冷启动之后的性能抖动
victimSize uintptr // 在新一轮GC来临时接管localSize
New func() interface{} // 当pool中没有对象时会调用这个函数生成一个新的
}
type poolLocal struct {
poolLocalInternal
// 避免false sharing问题
pad [128 - unsafe.Sizeof(poolLocalInternal{})8]byte
}
type poolLocalInternal struct {
// P的私有缓存区，使用时无需加锁，Put对象时优先放到这里
private interface{}
// 公共缓存区，本地P可以pushHead/popHead，其他P只能popTail
shared poolChain
}
// 双端队列
type poolChain struct {
head *poolChainElt
tail *poolChainElt
}
type poolChainElt struct {
poolDequeue
next, prev *poolChainElt
}
// 环形队列
type poolDequeue struct {
headTail uint64 // 头尾指针，之所以用一个变量持有两个字段大概率是为了方便原子操作一次性修改两个值吧
vals []eface // 容量从8开始，依次x2，上限为2 ^ 30
}
type eface struct {
typ, val unsafe.Pointer
}
2. Get
2.1 主流程
代码语言：javascript复制func (p *Pool) Get() interface{} {
// 当G与P绑定禁止抢占，返回P对应的poolLocal以及P的id
l, pid := p.pin()
x := l.private
l.private = nil
if x == nil {
// 如果private为空则从shared头部pop出一个
x, _ = l.shared.popHead()
if x == nil {
// 如果shared中也没有则尝试从其他P的shared尾部偷一个
x = p.getSlow(pid)
}
}
// 解除非抢占
runtime_procUnpin()
if x == nil && p.New != nil {
// 如果上面的步骤都没有取到则New个出来
x = p.New()
}
return x
}其中涉及到了一些函数，我们再看下具体实现
2.2 pin
pin的作用是将当前G与P绑定，禁止被抢占。那么为什么要禁止被抢占呢？原因是G被抢占后再恢复执行之后再绑定的可能就不是被抢占之前的P了
代码语言：javascript复制func (p *Pool) pin() (*poolLocal, int) {
// 执行绑定并返回当前pid
pid := runtime_procPin()
s := atomic.LoadUintptr(&p.localSize)
l := p.local
if uintptr(pid) < s { // pid<localSize说明已经完成了poolLocal的创建，可以取
return indexLocal(l, pid), pid
}
// pid>=localSize说明poolLocal还没有创建或者用户通过runtime.GOMAXPROCS(X)增加了p的数量，需要先创建
return p.pinSlow()
}
// 返回local[i]即当前P的poolLocal
func indexLocal(l unsafe.Pointer, i int) *poolLocal {
lp := unsafe.Pointer(uintptr(l) uintptr(i)*unsafe.Sizeof(poolLocal{}))
return (*poolLocal)(lp)
}
// runtime/proc.go
func procPin() int {
_g_ := getg()
mp := _g_.m
// 完成禁止抢占
// 调度器执行抢占g之前会canPreemptM(mp *m)判断是否可以执行抢占，而canPreemptM有一个条件为m.locks==0
mp.locks
return int(mp.p.ptr().id)
}2.2.1 pinSlow
pinSlow主要用来在poolLocal还未创建时创建新poolLocal
代码语言：javascript复制func (p *Pool) pinSlow() (*poolLocal, int) {
// 解除禁止抢占
runtime_procUnpin()
// 上锁
allPoolsMu.Lock()
defer allPoolsMu.Unlock()
// 禁止抢占
pid := runtime_procPin()
s := p.localSize
l := p.local
if uintptr(pid) < s {
// 上锁之前可能其他线程已经进入到pinSlow了，所以再判断一下
return indexLocal(l, pid), pid
}
if p.local == nil {
// 说明local第一次初始化，需要将pool加到allPools中
allPools = append(allPools, p)
}
// 获取p的数量
size := runtime.GOMAXPROCS(0)
// 创建local
local := make([]poolLocal, size)
atomic.StorePointer(&p.local, unsafe.Pointer(&local[0]))
atomic.StoreUintptr(&p.localSize, uintptr(size))
return &local[pid], pid
}2.3 poolChain.popHead
我们再看下Get主流程中从shared中通过popHead从shared头部pop出一个对象的实现
代码语言：javascript复制func (c *poolChain) popHead() (interface{}, bool) {
d := c.head
for d != nil { // 从链表头开始遍历，d的type为*poolDequeue
if val, ok := d.popHead(); ok {
return val, ok
}
d = loadPoolChainElt(&d.prev)
}
return nil, false
}2.3.1 poolDequeue.popHead
poolDequeue.popHead用来从环形队列头部pop出一个缓存对象
代码语言：javascript复制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-1然后修改poolDequeue.headTail
head--
ptrs2 := d.pack(head, tail)
if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) {
// 取到head对应的槽
slot = &d.vals[head&uint32(len(d.vals)-1)]
break
}
}
val := *(*interface{})(unsafe.Pointer(slot))
if val == dequeueNil(nil) {
val = nil // 通过3.2.1 poolDequeue.pushHead分析，貌似val不可能为nil
}
// 清空槽
*slot = eface{}
return val, true
}
func (d *poolDequeue) unpack(ptrs uint64) (head, tail uint32) {
// dequeueBits=32
const mask = 1<<dequeueBits - 1
head = uint32((ptrs >> dequeueBits) & mask) // &mask为了将高32位清零
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)
}
type dequeueNil *struct{}2.4 getSlow
再看下主流程中的getSlow函数的实现，getSlow用于在当前P缓存中没有时从其他P的共享缓存区偷缓存对象
代码语言：javascript复制func (p *Pool) getSlow(pid int) interface{} {
size := atomic.LoadUintptr(&p.localSize)
locals := p.local
// 从其他P偷
for i := 0; i < int(size); i {
l := indexLocal(locals, (pid i 1)%int(size))
// 从其他P的共享区的尾部偷
if x, _ := l.shared.popTail(); x != nil {
return x
}
}
// 如果没有偷到，则尝试victim cache。我们将在尝试从所有主缓存中偷取之后这样做，因为我们想让victim cache中的对象尽可能的老化
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
}
}
// 走到这里说明victim cache中也没有对象
// 将victim cache标记为空，下次就不用尝试victim cache了
atomic.StoreUintptr(&p.victimSize, 0)
return nil
}2.4.1 poolChain.popTail
getSlow中会通过poolChain.popTail从双端队列尾部pop对象，看下具体是如何操作的
func (c *poolChain) popTail() (interface{}, bool) {
// 先获取双端队列的尾节点，因为这时被偷的P与当前P在并行，所以需要通过原子操作获取
d := loadPoolChainElt(&c.tail)
if d == nil {
// 尾节点为空说明被偷的P的双端队列为空直接返回即可
return nil, false
}
for {
// 在pop tail之前load下一个指针是非常重要的，通常，d可能暂时为空，但是如果
// 在pop之前next非nil并且pop失败，那么d永远为空，这是唯一可以安全的从链表
// 中删除d的方法
d2 := loadPoolChainElt(&d.next)
if val, ok := d.popTail(); ok {
return val, ok
}
if d2 == nil {
// next为空遍历终止
return nil, false
}
// 走到这里说明当前尾节点为空，并且有下一个节点
// 此时当前尾节点不可能有新对象被push进来了，可以删除掉了
// 尝试将环形队列的尾节点指针改成它的下一个节点
if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) {
// 走到这里说明赢得了race，清除prev指针以便gc能够收集空dequeue
// 因此popHead不会在必要时再次备份
storePoolChainElt(&d2.prev, nil)
}
d = d2
}
}要修改poolChain的空尾节点指针为尾节点的下一个节点必须同时满足下面两个条件（即删除当前尾节点） * 当前尾节点环形队列为空 * 当前尾节点必须有下一个节点
我们注意到golang中先获取了当前尾节点的next再popTail，这是为什么呢？如果先popTail再获取next有可能遇到这样的情况：
- d的队列为空popTail没有获取到数据
- 另外一个线程向d中push了n个对象，此时d不为空，并且生成了下一个节点
- 原子获取next，next不为空
- 误将还有缓存对象的d删除
2.4.1.1
代码语言：javascript复制func (d *poolDequeue) popTail() (interface{}, bool) {
var slot *eface
for {
ptrs := atomic.LoadUint64(&d.headTail)
head, tail := d.unpack(ptrs)
if tail == head {
// 说明队列为空
return nil, false
}
ptrs2 := d.pack(head, tail 1)
if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) {
// 修改成功，这个solt就被我们占有了
slot = &d.vals[tail&uint32(len(d.vals)-1)]
break
}
}
// 从槽中取值
val := *(*interface{})(unsafe.Pointer(slot))
if val == dequeueNil(nil) {
val = nil
}
slot.val = nil
atomic.StorePointer(&slot.typ, nil)
return val, true
}3. Put
3.1 主流程
代码语言：javascript复制// sync/pool.go
func (p *Pool) Put(x interface{}) {
if x == nil {
return
}
l, _ := p.pin() // 禁止G被抢占，并返回当前P对应的poolLocal
if l.private == nil {
// 如果私有缓存为空则直接放到私有缓存区
l.private = x
x = nil
}
if x != nil {
// 如果私有缓存已经被占了，则放到共享缓存区头
l.shared.pushHead(x)
}
runtime_procUnpin() // 解除禁止抢占
}接下来我们看下新元素具体是如何放到共享缓冲区头部的
3.2 poolChain.pushHead
代码语言：javascript复制func (c *poolChain) pushHead(val interface{}) {
d := c.head
if d == nil {
// 头结点为空则需要进行初始化
const initSize = 8 // 第一个节点的环形队列的长度为8
d = new(poolChainElt)
d.vals = make([]eface, initSize)
// 其他P可能会从尾部偷对象，所以poolChain的tail需要用atomic set，保证对其他P可见
c.head = d
storePoolChainElt(&c.tail, d)
}
// push到头结点双端队列的头部
if d.pushHead(val) {
return
}
// 走到这里说明当前头节点的环形队列已经满了，所以申请一个新的节点
// 新节点环形队列的长度为旧队列的两倍，但如果大于1 << 32 / 4则长度为1 << 32 / 4
newSize := len(d.vals) * 2
if newSize >= dequeueLimit {
newSize = dequeueLimit
}
d2 := &poolChainElt{prev: d}
d2.vals = make([]eface, newSize)
// 修改双端队列的头节点为新创建的节点
c.head = d2
// 其他P可能会用到next所以需要原子store
storePoolChainElt(&d.next, d2)
d2.pushHead(val)
}3.2.1 poolDequeue.pushHead
poolDequeue.pushHead用于将对象放到环形队列上
func (d *poolDequeue) pushHead(val interface{}) bool {
ptrs := atomic.LoadUint64(&d.headTail)
head, tail := d.unpack(ptrs)
// dequeueBits = 32
if (tail uint32(len(d.vals)))&(1<<dequeueBits-1) == head {
// 如果环形队列tail加上长度等于head，说明队列实际已经满了
return false
}
// 找到head对应的槽，slot的类型为*eface
slot := &d.vals[head&uint32(len(d.vals)-1)]
typ := atomic.LoadPointer(&slot.typ)
if typ != nil {
return false
}
if val == nil {
val = dequeueNil(nil) // 追了下代码调用链貌似val不可能为nil
}
// 将val放到槽上
*(*interface{})(unsafe.Pointer(slot)) = val
// 增加头指针
atomic.AddUint64(&d.headTail, 1<<dequeueBits)
return true
}4. GC
上面的流程中我们清楚了对象是如何被缓存已经如何被写入和获取的，但是缓存池容量不是无限的，何时清理呢？答案是GC时。
sync/pool.go中有个init函数，在这个函数中注册了GC时如何清理Pool的函数
func init() {
// 编译器会将poolCleanup赋值给runtime/mgc.go文件中`poolcleanup`变量
// 在runtime.clearpools()函数中会调用poolcleanup，而在gcStart函数中在开始标记之前会调用clearpools()
runtime_registerPoolCleanup(poolCleanup)
}
func poolCleanup() {
for _, p := range oldPools {
// 先清除所有旧pool中的victim
// 之后gc就能标记清理旧pool中缓存的对象了
p.victim = nil
p.victimSize = 0
}
for _, p := range allPools {
// 用victim接管pool
p.victim = p.local
p.victimSize = p.localSize
p.local = nil
p.localSize = 0
}
oldPools, allPools = allPools, nil
}当时看到这里时还有一点疑惑，poolCleanup为什么不写成下面这样：
代码语言：javascript复制func poolCleanup() {
for _, p := range allPools {
// 用victim接管pool
p.victim = p.local
p.victimSize = p.localSize
p.local = nil
p.localSize = 0
}
}看起来也能清理缓存队列。但实际有个非常浅而已见的坑，那就是allPools这个切片一直在增长，必须将allPools设置为nil清理下才行，所以就必须引入oldPools。
5. 总结
sync.Pool为每个P搞一个缓存队列，避免所有线程共用同一个队列引发的锁竞争问题。
5.1 Put流程
- push到双端队列的头部的环形队列头部，如果环形队列已满则创建一个新的环形队列
- 将环形队列作为双端队列的新头部
5.2 Get流程
- 先从当前P缓冲区的私有缓存取
- 如果私有缓存没有从共享缓存区的双端队列的环形队列的头部pop
- 还没获取到则从其他P的共享缓存区的双端队列的环形队列的尾部pop
- 还没获取到则从victim cache中取
5.3 总结
总的来说只要清楚了sync.Pool的数据结构基本都理解的大差不差了，还是很简单的。
