golang sync.Pool分析

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{})%128]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
}

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
}

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

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
}

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
}

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

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
}

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

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
}

// 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() // 解除禁止抢占
}

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

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
}

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
}

func poolCleanup() {
	for _, p := range allPools {
	    // 用victim接管pool
		p.victim = p.local
		p.victimSize = p.localSize
		p.local = nil
		p.localSize = 0
	}
}