前言

当多个 goroutine 都需要创建同⼀种对象的时候，如果 goroutine 数量过多，导致对象的创建剧增，进⽽导致 GC 压⼒增大。形成下面的恶性循环：

并发⼤ --> 占⽤内存⼤ -->增加GC频率 -->限制程序执行效率 --> 处理并发能⼒降低 --> 并发更⼤

此时就需要有⼀个对象池，每个 goroutine 不⾃⼰调mallocgc从runtime分配对象，⽽是从对象池中获取⼀个对象，用完后再把对象放回池里。sync.Pool 就是这样一个对象池，可以减少GC频率和内存分配

• 减少GC频率：内存分配少了，出发GC的频率就会降低
• 减少内存分配：都不需要调malloc分配空间了，直接从池中拿对象

应用

在工程中 sync.Pool有广泛的应用，例如在 gin 中会给每个请求分配一个 Context 用于承载这次请求的上下文信息，从 pool 中获取 Context 对象，用完后还回去。每个请求都需要一个Context，因此在并发量高时，很适合用pool复用该对象

func (engine *Engine) ServeHTTP(w http.ResponseWriter, req *http.Request) {

    // 从pool冲获取ctx
    c := engine.pool.Get().(*Context)
	c.reset(w)
	c.Request = req
	c.reset()

    // 处理请求
	engine.handleHTTPRequest(c)

    // 归还ctx到pool中
	engine.pool.Put(c)
}

标准库fmt 也用了sync.Pool，用来复用pp对象，该对象承载执行一次print需要的上下文

// fmt.Println方法
func Println(a ...any) (n int, err error) {
	return Fprintln(os.Stdout, a...)
}

func Fprintln(w io.Writer, a ...any) (n int, err error) {
	p := newPrinter()
	p.doPrintln(a)
	n, err = w.Write(p.buf)
	p.free()
	return
}

func newPrinter() *pp {
    // 从pool获取一个pp
	p := ppFree.Get().(*pp)
	// ...
	return p
}

func (p *pp) free() {
	// 如果buf过大，不归还buf，只归还pp本身
	if cap(p.buf) > 64*1024 {
		p.buf = nil
	} else {
		p.buf = p.buf[:0]
	}

    // ...
    // 归还pp
	ppFree.Put(p)
}

需要特别注意free实现中的一个细节，如果pp的buffer经过了扩容，其容量大于某个阈值，则不把该pp放回pool，防止出现大内存不会被释放的情况

为啥大内存不会被释放？可以看看这个issue，如果buf被扩容得很大，不断经过放回pool，从pool中取出的循环，那么其占用的空间一直不会被释放

虽然可能后面使用该对象的请求，只使用buf中很少一部分，但由于对象pp持有大容量buf的引用，这片大内存就不会被回收，造成的后果就是内存泄露

怎么解决？对象大小超过一定阈值就不放回pool，交给GC进行内存回收

使用

• New：声明对象的构造方法，当池中没有对象时，会调改方法创建对象

• Get：从池中获取一个对象，返回any类型，需要调用方自行类型断言成需要的类型

  • 最佳实践：一个pool只放一种类型的对象

• Put：对象使用完毕后，将对象归还到池中

需要注意：

• sync.Pool不适用于线程池、DB连接池之类的存储，因为 GC 时会清除 sync.Pool 的数据，存储在 pool 中的对象在不被通知的情况下随时可能被清理掉

  • 为啥要在GC时清除？避免数据长时间占用内存，造成内存泄露

• 不应该对pool池中的对象做任何假定，不保证上次Put和下次Get的对象一定是同一个。因为sync.Pool的目的就是缓解GC压力，不应该牵扯到任何业务

  • 推荐做法：在调用 Pool.Put 前进行对象 Reset 操作，或在 Pool.Get 操作后进行 Reset 操作，防止读出脏数据

源码走读

sync.Pool的底层存储设计和GMP十分契合：

• 为每个处理器 P 分配了 private 和 shared 两部分存储空间

• private 是 P 私有的对象，当前 P 可以完全无锁化访问 private，性能极高

  • 为啥能无锁化？因为一个P同一时刻只能运行一个G，没有并发问题
  • 且Get 和 Put 时都优先用private，也就是优先用高性能的方式

• shared 是 P 下的共享象列表，当前 P 访问 shared 时也需要通过 CAS 操作保证并发安全，因为其他 P 也会来此尝试窃取对象，存在并发行为

数据结构

• sync.Pool中包含 local 和 victim，都指向poolLocal数组，其长度和P的数量一致
• 每个poolLocal包含一个私有空间private和公共链表shared
• 公共链表shared的每个节点poolChainElt包含 环形数组poolDequeue，对象实际存储在环形数组中

type Pool struct {
	noCopy noCopy

    // 真实类型为[P]poolLocal
	local     unsafe.Pointer 
    // local的长度，与P个数相同
	localSize uintptr        

    // 上一轮的local
	victim     unsafe.Pointer 
    // 上一轮的localSize
	victimSize uintptr        

	// 构造方法
	New func() any
}

poolLocal可以看作和poolLocalInternal等价

type poolLocal struct {
	poolLocalInternal

	pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte
}

type poolLocalInternal struct {
    // 每个P的私有对象存储空间
	private any
    // 和其他P共享的链表
	shared  poolChain 
}

poolChain是个双向链表，链表中每个节点的类型为poolChainElt，poolChain持有头尾节点的指针

type poolChain struct {
	head *poolChainElt
	tail *poolChainElt
}

每个节点poolChainElt存储了在链表中的前后指针，以及真正存储数据的地方poolDequeue：

type poolChainElt struct {
	poolDequeue

	// 前后指针
	next, prev *poolChainElt
}

poolDequeue包含一个环形数组val，以及数组的头尾索引：

type poolDequeue struct {
	
	headTail atomic.Uint64

	vals []eface
}

poolDequeue是一个基于CAS访问的无锁化环形队列

• 当前P的g通过pushHead添加对象，通过popHead获取对象
• 其他P的g通过popTail获取对象
  
  

Get获取对象

func (p *Pool) Get() any {
	// ...
    // 将当前G和P绑定到一起，禁止抢占，并返回P对应的local容器
	l, pid := p.pin()

    // 尝试获取private对象
	x := l.private
	l.private = nil
	if x == nil {
		// 尝试从自己的shared获取，从头部开始尝试
		x, _ = l.shared.popHead()
		if x == nil {
			x = p.getSlow(pid)
		}
	}
	runtime_procUnpin()

    // ...
  
    // pool中没获取到，调New构造一个新对象
	if x == nil && p.New != nil {
		x = p.New()
	}
	return x
}

popHead从shared的头节点开始尝试获取对象，调每个节点的popHead方法获取，一直尝试到尾节点

为啥要将当前G和P绑定到一起？

保证同一时刻只能有一个G在操作一个P对应的poolLocal。假设不绑定，当前G获取到P的id后，通过该id找到对应的poolLocal，然后P被抢占，运行新的G，此时新的G也根据该P的id找到相同的poolLocal，就会出现并发操作poolLocal.private，和并发调pushHead的情况

为啥从头节点pop呢？

• 基于局部性原理：当前P可能刚往头节点的环形数组push了数据，该数据还在cpu cache中，因此从头部获取效率更高
• 当前P操作该poolLocal从头节点开始pop，而其他P操作该poolLocal从尾节点pop，这样减少因操作同一个节点产生并发冲突的的概率

func (c *poolChain) popHead() (any, bool) {
	d := c.head
	for d != nil {
        // 从当前节点的环形数组中popHead
		if val, ok := d.popHead(); ok {
			return val, ok
		}

        // 检查下一个节点
		d = loadPoolChainElt(&d.prev)
	}
	return nil, false
}

如果本地shared没获取到，再调 getSlow 方法获取，具体流程为：

1. 尝试从各个 P 对应 local 的 shared 中窃取对象，从当前 P 的下一个 P 开始
2. 尝试获取当前 P 对应 victim 的 private 对象
3. 尝试从各个 P 对应 victim 的 shared 中窃取对象对象，从当前 P 开始
4. 如果至此仍未获得对象，说明为 victim 是空的，将 victimSize 置为 0，这样后续流程无需遍历victim

func (p *Pool) getSlow(pid int) any {
	// See the comment in pin regarding ordering of the loads.
	size := runtime_LoadAcquintptr(&p.localSize) // load-acquire
	
    // 获取pool的local
    locals := p.local                            
	// 尝试从其他P的shared中偷对象
	for i := 0; i < int(size); i++ {
        // 从当前p的下一个开始
		l := indexLocal(locals, (pid+i+1)%int(size))
		if x, _ := l.shared.popTail(); x != nil {
			return x
		}
	}

	size = atomic.LoadUintptr(&p.victimSize)
	if uintptr(pid) >= size {
		return nil
	}
	locals = p.victim
	l := indexLocal(locals, pid)
    // 尝试当前P的victim.private
	if x := l.private; x != nil {
		l.private = nil
		return x
	}

    // 尝试窃取victim中各个shared中的对象，从当前P开始
	for i := 0; i < int(size); i++ {
		l := indexLocal(locals, (pid+i)%int(size))
		if x, _ := l.shared.popTail(); x != nil {
			return x
		}
	}

	// victim没有了，将其标记为空，这样下次Get就不用搜索victim了
	atomic.StoreUintptr(&p.victimSize, 0)

	return nil
}

从每个shared偷取时，调用popTail方法：

注意这里 next 代表前一个节点，prev 代表后一个，和一般的用法不一样，否则这一段源码很难理解

func (c *poolChain) popTail() (any, bool) {
	d := loadPoolChainElt(&c.tail)
	if d == nil {
		return nil, false
	}

	for {
		// 获取前一个节点
		d2 := loadPoolChainElt(&d.next)

        // 尝试从当前节点获取对象
		if val, ok := d.popTail(); ok {
			return val, ok
		}

        // 没有前一个了，当前节点也没有，返回空
		if d2 == nil {
			// 最后一个节点为空时，不删除，未来Push时可以直接用
			return nil, false
		}

        // d没有对象了，将其从链表中删除，这样下次请求再来时没必要再检查一次这个节点
        // 怎么删除d呢？将链表的tail设为d2，也就是上一个
		if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) {
			storePoolChainElt(&d2.prev, nil)
		}
		d = d2
	}
}

Put归还对象

Put流程为：

1. 尝试将对象x放到当前P对应poolLocal的private
2. 将对象从头部推到当前P对应poolLocal的shared中

func (p *Pool) Put(x any) {
	// ...
  
	l, _ := p.pin()
    // 放到当前P的private
	if l.private == nil {
		l.private = x
	} else {
        // pushHead到当前P的shared
		l.shared.pushHead(x)
	}
	runtime_procUnpin()

    // ...	
}

pushHead将val推到shared中

1. 如果第一次Put，初始化头结点head，初始容量为8
2. 如果head没满，往其插入val
3. 否则，构造一个双倍容量的新头结点newHead
4. 将newHead插到链表头部
5. 往newHead插入val

func (c *poolChain) pushHead(val any) {
	d := c.head
	if d == nil {
		// 如果头结点为空，初始化
		const initSize = 8
		d = new(poolChainElt)
		d.vals = make([]eface, initSize)
		c.head = d
		storePoolChainElt(&c.tail, d)
	}

    // 放入头结点
	if d.pushHead(val) {
		return
	}

	// 当前节点满了，构造一个双倍容量的新头结点
	newSize := len(d.vals) * 2
	if newSize >= dequeueLimit {
		newSize = dequeueLimit
	}

	d2 := &poolChainElt{prev: d}
	d2.vals = make([]eface, newSize)
	c.head = d2
    // 之前节点d的前向指针设为新的头结点d2
	storePoolChainElt(&d.next, d2)
    // 往新的头结点pushHead
	d2.pushHead(val)
}

poolDeque分析

上面Get和Put最终会poolDeque，这是一个单生产者，多消费者的环形数组，生产者可以从head插入，head弹出，而消费者仅可从tail弹出

其除了有装数据的 底层数组vals，在逻辑上还有两个指针：

• tail：最老对象的索引
• head：指向下一个要插入数组的位置

head和tail在实现上被包装成了一个字段 headTail，低32位为tail，高32位为head。为啥要包装呢？方便用一个变量的CAS操作判断有没有其他请求修改过poolDeque

接下来看pushHead，popHead，popTail这3个方法怎么实现的

pushHead往头部插入val，流程为：

1. 判断队列是否满了，如果满了返回

   1. 怎么判断？如果head - tail == 队列容量，就表示满了

2. 如果没满，但是head指向的slot.typ不为nil，那一定是popTail中刚刚修改了headTail，但还没来得及清空slot。此时也当做队列满了

   1. 这里和popTail的实现有关，popTail中先修改了headTail的值，将tail++，再清空原来tail位置的数据
   2. 依据CAS操作的happens before原则：一旦slot.typ为空，那么该位置的数据就被清除干净了，也就可以push数据了

3. 然后就放心地将val放到head指向的slot中。为什么可以放心，因为不会有其他g往head位置放数据或拿数据

   1. 对于head右边来说，只会有别的g把对象从从tail弹出，只会离head越来越远，因此不会和head发生冲突
   2. 对于head左边来说，会从tail不断弹出，一直到head-1位置结束，也不会和head发生冲突

const dequeueBits = 32

func (d *poolDequeue) pushHead(val any) bool {
	ptrs := d.headTail.Load()
	head, tail := d.unpack(ptrs)

    // 队列满了，通过 head - tail == 队列容量 判断
	if (tail+uint32(len(d.vals)))&(1<<dequeueBits-1) == head {
		return false
	}

    // 准备往head指向的位置放val
	slot := &d.vals[head&uint32(len(d.vals)-1)]

	typ := atomic.LoadPointer(&slot.typ)
	if typ != nil {
        // 如果没满，且这里head指向的slot.typ不为nil，那一定是popTail中刚刚修改了headTail，但还没来得及清空slot
        // 这种情况也当作队列满了
		return false
	}

	if val == nil {
		val = dequeueNil(nil)
	}
    // val放到slot中
    // 这里可以直接放,因为不会有别的g调pushHead（只有持有当前P的g能调）
    // 其他g只会从tail弹出元素，也就是说此时head位置此时一定是空的，以后也一定是空的
	*(*any)(unsafe.Pointer(slot)) = val

    // head++
	d.headTail.Add(1 << dequeueBits)
	return true
}

popHead从头部弹出一个对象：

1. 如果队列为空，返回
2. 否则将 head--，通过CAS更新headTail
3. 将head位置清空

func (d *poolDequeue) popHead() (any, bool) {
	var slot *eface
	for {
		ptrs := d.headTail.Load()
		head, tail := d.unpack(ptrs)
        // 队列为空
		if tail == head {
			return nil, false
		}

		head--
		ptrs2 := d.pack(head, tail)
		if d.headTail.CompareAndSwap(ptrs, ptrs2) {
			// CAS成功，表示成功从头部弹出
			slot = &d.vals[head&uint32(len(d.vals)-1)]
			break
		}
	}

    // 此时slot为原来head指向的位置
    // 读取slot指针指向的值
	val := *(*any)(unsafe.Pointer(slot))
	if val == dequeueNil(nil) {
		val = nil
	}

    // 将弹出位置置空
	*slot = eface{}
	return val, true
}

popTail从尾部弹出一个对象：

1. 如果队列为空，返回
2. 否则将 tail++，并CAS更新headTail
3. 清空原来tail位置的slot

func (d *poolDequeue) popTail() (any, bool) {
	var slot *eface
	for {
		ptrs := d.headTail.Load()
		head, tail := d.unpack(ptrs)
        // 队列为空
		if tail == head {
			return nil, false
		}

		ptrs2 := d.pack(head, tail+1)
		if d.headTail.CompareAndSwap(ptrs, ptrs2) {
			// CAS成功，代表成功从尾部弹出一个
			slot = &d.vals[tail&uint32(len(d.vals)-1)]
			break
		}
	}

    // 此时slot为原来tail指向的位置
    // 读取slot指针指向的值
	val := *(*any)(unsafe.Pointer(slot))
	if val == dequeueNil(nil) {
		val = nil
	}

	// 将val置为nil，帮助GC	 
	slot.val = nil
    // 将typ置为nil，用原子指令发布，这样pushHead中一旦检测到typ是nil，那么这个slot就可用了
	atomic.StorePointer(&slot.typ, nil)
	// 到这一步，pushHead才能拥有这个slot

	return val, true
}

为啥popTail和popHead需要用CAS？因为这两个方法可能被多个g并发执行。假设此时只有1个元素，用CAS能保证只有一个请求能弹出这个唯一的元素，避免并发问题

一旦CAS修改headTail成功，就说明自己占有了这个位置

那为啥pushHead不先用CAS占有head位置呢？因为 一旦判定数组有空间后，就不会有别的g操作head了：

• 首先不会有别的g调pushHead，因为只有持有当前P的g能调，而那个g现在就是自己
• 其他g调popTail时，只会从尾部弹出对象，不会和head发生冲突，因此没有并发问题
  
  

GC时

在sync包的init方法中，将 poolCleanup 注册到GC的回调中，会在STW阶段执行

func init() {
	runtime_registerPoolCleanup(poolCleanup)
}

func poolCleanup() {
	
    // 清空所有的victim
	for _, p := range oldPools {
		p.victim = nil
		p.victimSize = 0
	}

	// 将local移动到victim，并清空local
	for _, p := range allPools {
		p.victim = p.local
		p.victimSize = p.localSize
		p.local = nil
		p.localSize = 0
	}

	oldPools, allPools = allPools, nil
}

这样如果pool中的对象没被用到，第一轮GC后会从local转移到victim中，第二轮GC后会清除掉，交给GC回收

为啥需要2轮才回收所有的对象？

• 这样能 提高缓存对象的命中率。如果每轮GC都清空所有对象，那势必会造成接下来的大量请求需要调New生成，给内存分配和GC造成压力
• 同时 减少了GC的开销，原来1次GC需要清理所有对象，现在分摊到2次GC中

总结

• 为啥需要sync.Pool？解决高并发时大量内存分配引起的恶性循环
• 使用时需要注意:
  • 大内存（例如：被扩容了的buf）不要往pool里放
  • Get后或Put前需要reset对象
• sync.Pool的巧妙设计：
  • 通过和GMP巧妙结合，实现了无锁化访问private
  • 在操作shared时，自己往头节点读写，其他g从尾节点读，从操作不同对象的方式减少锁冲突
  • 访问poolDequeue时，通过原子操作替代锁，提升并发性能
  • 使用victim提升缓存对象的命中率，以及减少GC的开销