深入浅出Golang中的sync.Pool

学习到的内容：

1.一个64位的int类型值，充分利用高32位和低32位，进行相关加减以及从一个64位中拆出高32位和低32位.

扩展：如何自己实现一个无锁队列.

• 如何判断队列是否满.

• 如何实现无锁化.

• 优化方面需要思考的东西.

2.内存相关操作以及优化

• 内存对齐

• CPU Cache Line

• 直接操作内存.

一、原理分析

1.1 结构依赖关系图

下面是相关源代码，不过是已经删减了对本次分析没有用的代码.

type Pool struct {
   // GMP中，每一个P(协程调度器)会有一个数组，数组大小位localSize.
local     unsafe.Pointer
// p 数组大小.
localSize uintptr
New func() any
}

// poolLocal 每个P(协程调度器)的本地pool.
type poolLocal struct {
poolLocalInternal
   // 保证一个poolLocal占用一个缓存行
pad [128 - unsafe.Sizeof(poolLocalInternal{})％128]byte
}

type poolLocalInternal struct {
private any       // Can be used only by the respective P. 16
shared  poolChain // Local P can pushHead/popHead; any P can popTail. 8
}

type poolChain struct {
head *poolChainElt
tail *poolChainElt
}

type poolChainElt struct {
poolDequeue
next, prev *poolChainElt
}

type poolDequeue struct {
// head 高32位，tail低32位.
headTail uint64
vals []eface
}

// 存储具体的value.
type eface struct {
typ, val unsafe.Pointer
}

1.2 用图让代码说话

1.3 Put过程分析

Put 过程分析比较重要，因为这里会包含pool所有依赖相关分析.

总的分析学习过程可以分为下面几个步骤：

1.获取P对应的poolLocal

2.val如何进入poolLocal下面的poolDequeue队列中的.

3.如果当前协程获取到当前P对应的poolLocal之后进行put前，协程让出CPU使用权，再次调度过来之后，会发生什么？

4.读写内存优化.

数组直接操作内存，而不经过Golang

充分利用uint64值的特性，将head和tail用一个值来进行表示，减少CPU访问内存次数.

获取P对应的poolLocal

sync.Pool.local其实是一个指针，并且通过变量+结构体大小来划分内存空间，从而将这片内存直接划分为数组. Go 在Put之前会先对当前Goroutine绑定到当前P中，然后通过pid获取其在local内存地址中的歧视指针，在获取时是会进行内存分配的. 具体如下：

func (p *Pool) pin() (*poolLocal, int) {
// 返回运行当前协程的P(协程调度器),并且设置禁止抢占.
pid := runtime_procPin()
s := runtime_LoadAcquintptr(&p.localSize) // load-acquire
l := p.local                              // load-consume
// pid < 核心数. 默认走该逻辑.
if uintptr(pid) < s {
 return indexLocal(l, pid), pid
}
// 设置的P大于本机CPU核心数.
return p.pinSlow()
}

// indexLocal 获取当前P的poolLocal指针.
func indexLocal(l unsafe.Pointer, i int) *poolLocal {
// l p.local指针开始位置.
// 我猜测这里如果l为空，编译阶段会进行优化.
lp := unsafe.Pointer(uintptr(l) + uintptr(i)*unsafe.Sizeof(poolLocal{}))
// uintptr真实的指针.
// unsafe.Pointer Go对指针的封装: 用于指针和结构体互相转化.
return (*poolLocal)(lp)
}

从上面代码我们可以看到，Go通过runtime_procPin来设置当前Goroutine独占P，并且直接通过头指针+偏移量(数组结构体大小)来进行对内存划分为数组.

Put 进入poolDequeue队列：

Go在Push时，会通过headtail来获取当前队列内元素个数，如果满了，则会重新构建一个环型队列poolChainElt，并且设置为poolChain.head，并且赋值next以及prev.

通过下面代码，我们可以看到，Go通过逻辑运算判断队列是否满的设计时非常巧妙的，如果后续我们去开发组件，也是可以这么进行设计的。

func (c *poolChain) pushHead(val any) {
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
 // 将新构建的d赋值给tail.
 storePoolChainElt(&c.tail, d)
}
// 入队.
if d.pushHead(val) {
 return
}
// 队列满了.
newSize := len(d.vals) * 2
if newSize >= dequeueLimit {
       // 队列大小默认为2的30次方.
 newSize = dequeueLimit
}

// 赋值链表前后节点关系.
// prev.
// d2.prev=d1.
// d1.next=d2.
d2 := &poolChainElt{prev: d}
d2.vals = make([]eface, newSize)
c.head = d2
// next .
storePoolChainElt(&d.next, d2)
d2.pushHead(val)
}

// 入队poolDequeue
func (d *poolDequeue) pushHead(val any) bool {
ptrs := atomic.LoadUint64(&d.headTail)
head, tail := d.unpack(ptrs)
// head 表示当前有多少元素.
if (tail+uint32(len(d.vals)))&(1<<dequeueBits-1) == head {
 return false
}
// 环型队列. head&uint32(len(d.vals)-1) 表示当前元素落的位置一定在队列上.
slot := &d.vals[head&uint32(len(d.vals)-1)]

typ := atomic.LoadPointer(&slot.typ)
if typ != nil {
 return false
}

// The head slot is free, so we own it.
if val == nil {
 val = dequeueNil(nil)
}
   // 向slot写入指针类型为*any，并且值为val.
*(*any)(unsafe.Pointer(slot)) = val
   // headTail高32位++
atomic.AddUint64(&d.headTail, 1<<dequeueBits)
return true
}

Get实现逻辑：

其实我们看了Put相关逻辑之后，我们可能很自然的就想到了Get的逻辑，无非就是遍历链表，并且如果队列中最后一个元素不为空，则会将该元素返回，并且将该插槽赋值为空值.

二、学习收获

如何自己实现一个无锁队列. 本文未实现，后续文章会进行实现.

2.1 如何自己实现一个无锁队列

横向思考，并未进行实现，后续会进行实现&ldquo;

• 存储直接使用指针来进行存储，充分利用uintptr和unsafe.Pointer和结构体指针之间的依赖关系来提升性能.

• 状态存储要考虑CPU Cache Line、内存对齐以及减少访问内存次数等相关问题.

• 充分利用Go中的原子操作包来进行实现，通过atomic.CompareAndSwapPointer来设计自旋来达到无锁化.

来源：https://juejin.cn/post/7209625823580520504