提示

本文代码基于 go1.17.13，src/sync/pool.go、src/sync/poolqueue.go

一、简述

• Pool 是一组可以单独保存和检索的临时对象的集合，可以缓存已分配但未使用的对象供以后重用，从而减轻垃圾回收器的压力；
• Pool 中有对象时，直接取出使用，Pool 中没有对象时，立即构建新的对象，从而使对象的分配开销得以摊销，并且是多线程安全的；
• fmt 包是使用 Pool 的一个很好示例，它维护一个动态大小的临时输出缓冲区存储，当许多 goroutine 打印时缓冲区变大，静止时变小；
• Pool 首次使用后不得被复制；

二、基本原理

• 根据当前 P 的个数，即默认是 CPU 核数或是用户自定义的核数，Pool 为每个 P 分配一个私有对象和共享对象的结构体，下图中 poolLocalInternal，多个poolLocalInternal结构体组成一个数组；
  • 私有对象可以直接存储该 Pool 需要存储的临时对象，每个 P 都只能对自己的私有对象进行存取，无权访问其他 P 的私有对象；
  • 共享对象是一个双向链表，链表的节点是一个个长度为 2 的幂的环，且后一个是前一个的两倍长，环中存储的也是 Pool 需要存储的临时对象，如下图中右侧的环；
  • 对于共享对象的双向链表，生产者可以从头部的环存取临时对象，消费者只能从尾部的环取临时对象，即每个 P 可以对自己的共享对象的双向链表头部进行读取，对其他 P 的共享对象的双向链表的尾部进行取；
  • 每个 P 都是自己共享对象的生产者，都是其他 P 共享对象的消费者，如下图中 P0 能够从自己的 share 所指向的链表的 head 处进行读写数据，P1、P2、P3 只能从 P0 的 share 所指向的链表的 tail 处读数据；
• 根据 GMP 模型，每个 P 上可以运行多个 G，在 G 需要去 Pool 中存取临时对象时，会将 G 与 P 绑定，并按照 P 的序号去数组中取其对应的结构体对象，这样避免了多线程之间的竞争；

三、基本用法

1，应用场景

• 减少内存分配和垃圾回收压力：对于某些需要频繁创建和销毁对象的业务场景，使用 sync.Pool 可以有效减少内存分配和垃圾回收的压力，避免重复创建和销毁，提高系统的响应速度和性能稳定性；

2，简单示例

点击查看

package main

import (
	"fmt"
	"sync"
)

func main() {
	// 创建一个 Pool
	var pool = &sync.Pool{
		New: func() interface{} {
			fmt.Println("Creating a new object.")
			return struct{}{}
		},
	}

	// 获取一个对象
	obj := pool.Get()

	// 使用对象
	// ...

	// 将对象放回 Pool 中
	pool.Put(obj)

	// 再次获取对象，这次将重用之前放回的对象
	obj2 := pool.Get()

	// 使用对象
	// ...

	// 将对象放回 Pool 中
	pool.Put(obj2)
}

• 在创建Pool对象时，需要给New元素赋一个函数，该函数能够创建Pool中需要存储的对象，用于从Pool中取不到临时对象时调用；
• 从Pool中取出临时对象后，使用完毕时，仍然可以放回去，供其他 P 使用；

四、源码解读

（一）双向链表

1，poolDequeue

(1) poolDequeue 结构

点击查看

type poolDequeue struct {
	headTail uint64

	vals []eface
}

type eface struct {
	typ, val unsafe.Pointer
}
const dequeueBits = 32
const dequeueLimit = (1 << dequeueBits) / 4

// dequeueNil 在 poolDequeue 中表示 interface{}(nil).
// 由于我们使用 nil 来表示空插槽，因此我们需要一个哨兵值来表示 nil，以区分空槽与空值。
type dequeueNil *struct{}

• poolDequeue 是一个无锁、固定大小的单生产者、多消费者队列，生产者可以从头部读写，消费者从尾部读；它有一个附加功能，即剔除未使用的插槽，以避免不必要的对象保留；
• headTail 将一个 32 位头部索引和一个 32 位尾部索引打包在一起进行存储；
  • tail 队列中最老数据的索引，永远在前面；
  • head 将要填充的下一个槽的索引，永远在后面；
  • 槽的范围是 [tail, head) ；
  • tail 和 head 像是一个双指针一样，同时往后移动，它们之间的距离永远在 0 和 len(vals) 之间；
  • 队列为空时，head == tail；队列满了时，tail + len(vals) == head，dequeueLimit 保证其不会越界；
  • head，tail 不会在超过队列长度时被赋值为其求余后的值，而是在要需要通过其取值时，用其对 len(vals) 求余后的余值作为索引，然后取数组对应位置的数据；
• vals 是一个存储 interface{} 的数组，它的长度必须是 2 的幂，与索引 head、tail 一起形成一个环形队列；
• eface 为切片实际存储的元素，包含了指向元素类型和元素值的两个指针；
• dequeueLimit 表示切片的最大长度，这样能够保证 head、tail 不会超过机器的限制；
• dequeueNil 表示 eface 切片的一个空值，以与整个 eface 切片为空时的 nil 区分开；

(2) unpack

点击查看

func (d *poolDequeue) unpack(ptrs uint64) (head, tail uint32) {
	const mask = 1<<dequeueBits - 1
	head = uint32((ptrs >> dequeueBits) & mask) // 右移 32 位后与 32 个 1 与运算，并 32 位截断
	tail = uint32(ptrs & mask)                  // 与 32 个 1 与运算后，32 位截断
	return
}

• 从 64 位 ptrs 中解出 32 位 head, tail 值；

(3) pack

点击查看

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

• 将 32 位 head, tail 值打包为 64 位值；

(4) pushHead

点击查看

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 {
		// 队列已满
		return false
	}
	slot := &d.vals[head&uint32(len(d.vals)-1)]

	// 检查 popTail 是否释放了头插槽
	typ := atomic.LoadPointer(&slot.typ)
	if typ != nil {
		// 当前槽不为空，插入后会形成覆盖，说明另一个 goroutine 仍在清理尾部，因此队列实际上仍然已满。
		return false
	}

	// head 索引处的插槽为空，可以插入数据
	if val == nil {
		// 待插入的数据为 nil 时，设置为 dequeueNil，以区分空槽与空值
		val = dequeueNil(nil)
	}
	*(*interface{})(unsafe.Pointer(slot)) = val

	// 增加 head，这会将插槽的所有权传递给 popTail，并充当写入插槽的存储屏障。
	// 因为可能会与 popTail 有竞争，所以此处需要原子操作
	atomic.AddUint64(&d.headTail, 1<<dequeueBits)
	return true
}

• pushHead 在队列头部添加 val，如果队列已满，则返回 false，必须被单生产者调用；
• 在 push 时，需要判断队列是否满了，队列为满的判断尾部索引 tail 落后头部索引 head 一整个队列的长度，head 是指向下一个要写入的位置；
• 当 push 的元素为空时，需要设为 dequeueNil，以区分是整个插槽为 nil 还是插槽中的元素为 nil；
• 最后需要以原子操作的方式修改 headTail，因为可能会和 popTail 的其他 goroutine 有竞争；

(5) popHead

点击查看

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--
		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
	}
	// 将插槽归零，避免在 pushHead 时发现不为 nil。
	// 与 popTail 不同的是，这不是与 pushHead 竞争，所以我们在这里不需要小心。
	*slot = eface{}
	return val, true
}

• popHead 移除并返回队列首部的元素，如果队列为空，则返回 false，必须由但生产者调用；
• 需要判断队列是否为空，即 tail 和 head 是否相等；
• 取出元素后，需要将插槽归零，避免在 pushHead 时发现不为 nil，无法插入；
• 同一个时刻只会有单个生产者将该插槽归零，不存在竞争，所以无需原子操作；

(6) popTail

点击查看

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) {
			// 成功拥有尾部的插槽
			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
	}

	// 告诉pushHead，我们已经用完了这个插槽。将槽置零也很重要，这样我们就不会留下可能使该对象存活时间超过必要时间的引用。
	// 我们首先写入 val，然后通过原子写入 typ 来发布我们已经完成了这个插槽。
	slot.val = nil
	atomic.StorePointer(&slot.typ, nil)

	return val, true
}

• popTail 移除并返回队列尾部的元素，如果队列为空，则返回 false，可以被任意数量的消费者调用；
• 需要判断队列是否为空，即 tail 和 head 是否相等；
• 取出元素后，需要将插槽归零，避免在 pushHead 时发现不为 nil，无法插入；
• 同一个时刻可能会有多个消费者将该插槽归零，存在竞争，所以需要原子操作；

2，poolChainElt

点击查看

type poolChainElt struct {
	poolDequeue

	// next、prev 链接到 poolChain 相邻的 poolChainElts
	next, prev *poolChainElt
}

• poolChainElt 是一个双向链表的节点，该节点包含一个环形队列 poolDequeue 的元素，以及指向前后节点的指针，每个 poolDequeue 队列的长度都是 2 幂，并且是前一个节点中队列长度的两倍，结合上图可知该结构；
• next 指向的方向为 poolChain 中的 head，prev 指向的方向为 poolChain 中的 tail；
• 当前 P 上的 G 对于该 P 上的 poolDequeue 是生产者，从 next 上写和读，对于其他 P 上的 poolDequeue 是消费者，只能从 prev 读，不能写；

3，poolChain

(1) poolChain 结构

点击查看

type poolChain struct {
	head *poolChainElt
	tail *poolChainElt
}

• poolChain 是一个动态长度的链表，其节点是 poolChainElt，其中 head 元素指向链表的最新节点，tail 元素指向链表的最老节点；
• head 指向的是最新创建的节点，也是最大的队列，从头部读写元素时，即是从 head 指向的 poolChainElt 节点中的 poolDequeue 队列的头部进行读写，这只能由生产者访问，不需要同步；
• tail 指向的是最早创建的节点，也是最小的队列，从尾部读元素时，即是从 tail 指向的 poolChainElt 节点中的 poolDequeue 队列的进行读，这可以由多个消费者访问，需要进行同步；
• 当从头部写元素时，当 head 指向的 poolChainElt 节点中的 poolDequeue 环形队列中满了时，会重新创建一个新的节点，该节点中的环形队列的长度是当前 head 中环形队列长度的两倍，并将该节点加入到链表中，更新 head 指向该最新节点，继续往该节点写入；
• 当从尾部读元素时，当 tail 指向的 poolChainElt 节点中的 poolDequeue 环形队列中空了时，会从该链表中删除该节点，并更新 tail 指向其下一个节点；

(2) pushHead

点击查看

// 以原子的方式进行存储
func storePoolChainElt(pp **poolChainElt, v *poolChainElt) {
	atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(pp)), unsafe.Pointer(v))
}
// 以原子的方式进行加载
func loadPoolChainElt(pp **poolChainElt) *poolChainElt {
	return (*poolChainElt)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(pp))))
}
func (c *poolChain) pushHead(val interface{}) {
	d := c.head
	if d == nil {
		// 初始化 chain.
		const initSize = 8 // 必须为 2 的幂
		d = new(poolChainElt)
		d.vals = make([]eface, initSize)
		c.head = d
		storePoolChainElt(&c.tail, d) // tail 的写入必须是原子方式的
	}

	if d.pushHead(val) {
		return
	}

	// 当前队列已满，新分配的队列长度是当前的两倍
	newSize := len(d.vals) * 2
	if newSize >= dequeueLimit {
		newSize = dequeueLimit
	}

	d2 := &poolChainElt{prev: d}     // 新创建的 poolChainElt 的前一个指向当前的 poolChainElt
	d2.vals = make([]eface, newSize) // 新创建的 poolChainElt 尺寸翻倍
	c.head = d2                      // head 指向新创建的 poolChainElt
	storePoolChainElt(&d.next, d2)   // 当前的 poolChainElt 的下一个指向新创建的 poolChainElt
	d2.pushHead(val)                 // 将 val 插入新创建的 poolChainElt 的头部
}

• 从链表的头部写入元素，如果链表头部为空，则说明整个链表为空，则直接创建一个 poolChainElt 节点，其环形队列的长度为 8，即初始环形队列的长度为 8，并将链表的头部和尾部同时指向该节点，可能会有多个消费者同时修改 tail，所以尾部的写入必须是原子的；
• 如果链表的头部不为空，则直接写入头部节点的环形队列，写入失败则说明该头部的的环形队列已满，需要重新创建环形队列是当前两倍大小的新节点，更新头部节点并写入数据；

(3) popHead

点击查看

func (c *poolChain) popHead() (interface{}, bool) {
	d := c.head
	for d != nil {
		if val, ok := d.popHead(); ok {
			return val, ok
		}
		// 从 head 往 tail 的过程中加载 poolChainElt 节点，虽然只有一个生产者，但是可能会与其他消费者从 tail 往 head 的过程中加载到同一个节点，所以此处需要原子操作 
		d = loadPoolChainElt(&d.prev)
	}
	return nil, false
}

• 从链表的头部读取元素，尝试从 head 指向的节点的环形队列中开始读取元素，如果没有，则继续从前一个节点读取，即上图中 prev 指向的节点，直到读取到元素立即返回，当达到最后一个节点的 prev 时跳出循环，返回 nil，false；
• **从 head 往 tail 的过程中加载 poolChainElt 节点，虽然只有一个生产者，但是可能会与其他消费者从 tail 往 head 的过程中加载到同一个节点，所以此处需要原子操作；
• 在整个读取过程中，即使某个节点的环形队列为空，该节点也不会被删除；

(4) popTail

点击查看

// tail 侧的空队列会被删除
func (c *poolChain) popTail() (interface{}, bool) {
	d := loadPoolChainElt(&c.tail)
	if d == nil {
		return nil, false
	}

	for {
		// 从 tail 往 head 的过程中加载 poolChainElt 节点，可能会有多个消费者或某个生产者同时加载到同一个节点，所以此处需要原子操作
		d2 := loadPoolChainElt(&d.next)

		if val, ok := d.popTail(); ok {
			return val, ok
		}

		if d2 == nil {
			// 这是唯一的队列。它现在是空的，但将来可能会被推入数据。
			return nil, false
		}

		// 不同 P 读同一个 P 的尾部时，会有竞争，故用原子操作
		// 链条的尾部已弹空，尝试将 tail 指向当前队列的 next 队列，以从链表中删除当前队列，这样下一次弹出时就不必再次查看空队列
		if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) {
			// CAS 成功，清除 prev 指针，当前队列的 next 队列的 pre 是指向当前队列的，此时需要置为 nil，以便垃圾回收期可以回收当前这个空队列
			// 逐步删除短的队列，可以保证所有的元素都在一个或多个连续的队列中，而队列的长度和元素的长度是相近的，可以避免内存浪费
			storePoolChainElt(&d2.prev, nil)
		}
		d = d2
	}
}

• 从链表的头部读取元素，尝试从 tail 指向的节点的环形队列中开始读取元素，如果 tail 为空，则说明整个链表为空，直接返回；
• 提前获取 tail 的下一个节点，从当前节点读取数据，能读到则直接返回，读不到但下一个节点为空则整个链表也读不到直接返回，读不到但下一个节点不为空，则删除当前节点，并从下一个节点继续开始读取数据，直到 head 节点的 next 节点；
• 从 tail 往 head 的过程中加载 poolChainElt 节点，可能会有多个消费者或某个生产者同时加载到同一个节点，所以此处需要原子操作；

（二）缓存池

1，Pool

点击查看

type Pool struct {
	noCopy 		noCopy
	local 		unsafe.Pointer
	localSize 	uintptr
	victim     	unsafe.Pointer
	victimSize 	uintptr

	// 指定一个函数，用于在 Pool 中没有对象时创建新的对象
	New func() interface{}
}

• Pool 是一组可以单独保存和检索的临时对象的集合；
• noCopy 不消耗内存仅用于静态分析的结构，保证一个对象在第一次使用后不会发生复制；
• local 指向本地 [localSize]poolLocal 数组的指针，即该数组的第一个元素，每个 P 对应一个 poolLocal，多个 goroutine 对同一个 Pool 操作时，每个运行在 P 上的 goroutine 优先取该 P 上 poolLocal 中的元素，能够减少不同 goroutine 之间的竞争，提升性能；
• localSize 本地 [localSize]poolLocal 数组的大小，一般是系统核数，除非程序中自定义了运行核数，或运行中修改了运行核数；
• 在一轮 GC 到来时，victim 和 victimSize 会分别接管 local 和 localSize 当从 local 中未查询到时，会进一步在 victim 中查询；
• **在 GC 后冷启动时，local 中没有缓存对象，victim 中有，能够避免冷启时大量创建对象导致性能抖动，让分配对象更平滑；是一种一空间换时间的做法；

提示

• Victim Cache（牺牲者缓存）是一种用于提高缓存性能的缓存内存类型，临时存储从主缓存中驱逐出来的数据，它通常位于主缓存和主存储器之间；
• 当主缓存发生缓存未命中时，在访问主存储器之前会检查牺牲者缓存；如果请求的数据在牺牲者缓存中找到，就认为是缓存命中，并将数据返回给处理器，而无需访问主存储器；
• 当主缓存需要用新数据替换一个缓存行时，它会将最近最少使用（LRU）的缓存行放入牺牲者缓存中，以防近期再次需要该数据；
• 牺牲者缓存通常比主缓存更小，关联度更低，目的是捕获那些可能在不久的将来再次访问的缓存行；
• 由于主缓存的大小限制而无法容纳，通过将这些被驱逐的缓存行保留在一个单独的缓存中，作为一种优化缓存的技术，可以减少系统对主存储器的访问次数，提高整体性能；

点击查看

type poolLocalInternal struct {
	private interface{} 
	shared  poolChain  
}

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

• poolLocalInternal 是每一个 P 所拥有的私有对象和共享对象的元素；
  • private 当前 P 私有的对象，只能由其所属的当前 P 存储和获取；
  • shared 当前 P 与其他 P 共有双向链表，链表中存储对象，当前 P 是生产者，能够 pushHead/popHead，其他 P 是消费者，只能 popTail；
• poolLocal 是 poolLocalInternal 按照内存对齐后的结构体；

:::info

• CPU 在访问数据是按照 64/128 字节作为一行一起加载的，如果某个变量不足一行，则会和其他变量同时加载进 CPU CacheLine，当一个变量失效时会导致该行其他变量也失效，这是一种伪共享现象；
• 第一、二层 CPU 缓存是每个 CPU 各自独有的，第三层 CPU 缓存是不同 CPU 之间共享的，CPU CacheLine 中有变量失效时，会导致整个 CPU CacheLine 都需要从主存中重新加载，对性能有影响；
• 如果没有 pad 字段，可能会导致一个 CPU CacheLine 中存在多个 poolLocal 对象，而这些对象又属于不同 CPU 上的 P，当某个 CPU 上的 P 修改了 CPU CacheLine 上的该 P 对应的 poolLocal 时，会导致其他 poolLocal 失效，那么该 poolLocal 对应的 P 所在的 CPU 就需要重新加载；
• 所以，pad 的目的是让专属于某个 P 的 poolLocal 独占一整个 CPU CacheLine，避免使得其他 poolLocal 在 CPU CacheLine 中失效，毕竟该 P 是优先访问自己的 poolLocal； :::

2，Get

点击查看

func (p *Pool) Put(x interface{}) {
	if x == nil {
		return
	}
	// 将当前 G 绑定到 P，并返回 P 的 poolLocal 和 id（CPU序号）
	l, _ := p.pin()
	if l.private == nil {
		// 如果 P 的 poolLocal 的私有对象为空，则直接将 x 赋给它
		l.private = x
		x = nil
	}
	if x != nil {
		// 说明 P 的 poolLocal 的私有对象不为空，则将 x push 到其附属的链表的头部，因为该 P 是其 poolLocal 的生产者
		l.shared.pushHead(x)
	}
	runtime_procUnpin() // 解除 G 与 P 的绑定
}

• Put 往池子中添加 x 对象；
• x 为 nil 则直接返回，不允许存储 nil 对象；
• 首先将当前 G 绑定到 P，并返回 P 的 poolLocal 和 id（CPU序号），如果 P 的 poolLocal 的私有对象为空，则直接将 x 赋给它，否则直接存储到其共享对象的头部，再解除 G 与 P 的绑定；
• 其中 pushHead 操作最终由 poolChain 链表中 poolChainElt 节点中的 poolDequue 环执行；

3，Put

点击查看

func (p *Pool) Get() interface{} {
	l, pid := p.pin()
	x := l.private
	l.private = nil
	if x == nil {
		// P 的 poolLocal 的私有对象为空，尝试从共享队列中的头部弹出对象
		// 对于重用的时间局部性，我们更喜欢头而不是尾。
		// 时间局部性是指处理器在短时间内多次访问相同的内存位置或附近的内存位置的倾向
		x, _ = l.shared.popHead() // 作为自己队列的生产者，可以从头部读
		if x == nil {
			// P 的 poolLocal 的共享队列为空，尝试从其他 P 的 poolLocal 的共享队列和受害者缓存中弹出
			x = p.getSlow(pid)
		}
	}
	runtime_procUnpin() // 解除 G 与 P 的绑定
	if x == nil && p.New != nil {
		// 如果弹出的对象为空，并且 New 函数不为空，则直接调用 New 函数创建一个新的对象
		x = p.New()
	}
	return x
}

• Get 从池中选择任意对象，并将其从池中删除，然后将其返回给调用方；
• 调用方不应该假定传递给 Put 的值与 Get 返回的值之间存在任何类型关系，因为每次存取的类型都是 interface{} 类型；
• 将当前 G 绑定到 P，并返回 P 的 poolLocal 和 id（CPU序号）；
• 优先从 P 的 poolLocal 的私有对象中获取，私有对象为空，则从 P 的 poolLocal 的共享对象获取，当 P 的 poolLocal 的共享对象也取不到时，则进入慢获取路径中；
• 在慢获取路径中，先从其他 P 的 poolLocal 的共享对象中获取，没有时再从受害者缓存中的其他 P 的 poolLocal 的共享对象中获取；
• 以上获取方式结束后，解除 G 与 P 的绑定，如果还没获取到则调用 New 方法创建一个对象并返回；

4，getSlow

点击查看

func (p *Pool) getSlow(pid int) interface{} {
	size := runtime_LoadAcquintptr(&p.localSize)
	locals := p.local
	// 尝试从其他 P 的
	for i := 0; i < int(size); i++ {
		// 依次获取其他 P 的 poolLocal
		// TODO 此处仍然会获取到当前 P 的 Local，并从其共享队列的尾部获取，不符合既定的逻辑？
		l := indexLocal(locals, (pid+i+1)%int(size))
		// 作为其他 P 的 poolLocal 的共享队列消费者，从其他 P 的 poolLocal 的共享队列的尾部获取对象
		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)
	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
		}
	}
	// 取不到则将 victimSize 置位 0，下次就不会再从 victim 中取了
	atomic.StoreUintptr(&p.victimSize, 0)

	return nil
}

// 返回第 i 个 poolLocal 对象，i 从 0 开始
func indexLocal(l unsafe.Pointer, i int) *poolLocal {
	lp := unsafe.Pointer(uintptr(l) + uintptr(i)*unsafe.Sizeof(poolLocal{}))
	return (*poolLocal)(lp)
}

• 尝试从其他 P 的 poolLocal 的共享队列中获取对象，获取不到时，再尝试从 victim 中获取；
• 作为其他 P 的 poolLocal 的共享队列消费者，从其他 P 的 poolLocal 的共享队列的尾部获取对象；
• 个人觉得，indexLocal(locals, (pid+i+1)%int(size)) 以指针偏移的方式获取其他 P 的 poolLocal 时仍然能够获取到自身的 poolLocal，如果获取成功会从自身的尾部获取元素，不符合既定的逻辑；
• 尝试从受害者缓存中获取对象时，与从主缓存中获取步骤一致，最后都取不到则将 victimSize 置位 0，下次就不会再从 victim 中取了；
• 以 runtime_LoadAcquintptr 的方式获取 p.localSize 的值，可以防止编译器和处理器对代码进行重排序，确保在获取 p.localSize 的值之后，后续的读操作都能看到最新的值；

提示

• 在并发编程中，为了避免出现数据竞争和不一致的情况，需要使用适当的同步机制来确保内存的一致性；
• 使用原子加载的方式获取 p.localSize 的值可以保证读取到的值是其他 Goroutine 写入的最新值，这样就可以避免出现数据访问的竞争条件。

5，pin

点击查看

func (p *Pool) pin() (*poolLocal, int) {
	// 将当前 goroutine 固定到 P
	pid := runtime_procPin()
	// 在 pinSlow 中，先存储到 local，然后存储到 localSize，这里以相反的顺序加载
	// 由于我们禁用了抢占，因此 GC 不会在两者之间发生，因此 local 至少和 localSize 一样大
	// 可以保证读取到的值是其他 Goroutine 写入的最新值，确保并发情况下的内存一致性和可见性
	s := runtime_LoadAcquintptr(&p.localSize)
	l := p.local
	if uintptr(pid) < s {
		// P 的索引小于 local 数组的长度时，直接取索引处的 poolLocal 返回
		return indexLocal(l, pid), pid
	}
	return p.pinSlow() // 可能是 GOMAXPROCS 在 gc 的时候发生了改变
}

func (p *Pool) pinSlow() (*poolLocal, int) {
	// 在互斥锁下重试。G 被固定时无法锁定互斥锁。
	runtime_procUnpin()
	allPoolsMu.Lock() // 保护 oldPools，避免在 poolCleanup 与 pinSlow 时有竞争
	defer allPoolsMu.Unlock()
	pid := runtime_procPin()
	// 当 G 被固定到 P 时，poolCleanup 不会被调用
	s := p.localSize
	l := p.local
	if uintptr(pid) < s {
		return indexLocal(l, pid), pid
	}
	// 说明 P 的最大数量发生改变，原先 Pool 的 local 数组小了，需要重新分配，并将旧的 Pool 在 gc 来临时置空
	if p.local == nil {
		allPools = append(allPools, p)
	}
	// 如果 GOMAXPROCS 在 gc 期间发生了改变，需要重新分配 local 数组并丢弃旧的数据
	size := runtime.GOMAXPROCS(0) // 只获取先前设置的最大并发数，不实际改变其值
	local := make([]poolLocal, size)
	atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-release
	runtime_StoreReluintptr(&p.localSize, uintptr(size))     // store-release
	return &local[pid], pid
}

• 将当前 goroutine 绑定到 P，禁用抢占并返回 P 的 poolLocal 和 P 的 ID，调用方必须在处理完池后调用 runtime_procUnpin()；
• P 的索引大于或等于 local 数组的长度时，则进入慢固定路径；
• 在慢固定路径中，需要先解除绑定，因为 G 被固定时无法锁定互斥锁，再进行加锁，以保护 oldPools，避免在 poolCleanup 与 pinSlow 时有竞争；
• P 的索引小于 local 数组的长度时，直接取索引处的 poolLocal 返回，大于则说明 GOMAXPROCS 在 gc 期间发生了改变，需要重新分配 local 数组并丢弃旧的数据，并返回新的 poolLocal 和 pid；

6，poolCleanup

点击查看

// 在垃圾回收开始时，STW 的情况下调用此函数，它不能分配，也可能不应该调用任何运行时函数
func poolCleanup() {
	// 清除所有 Pool 中的受害者缓存
	for _, p := range oldPools {
		p.victim = nil
		p.victimSize = 0
	}

	// 将主缓存中的数据移交给受害者缓存
	for _, p := range allPools {
		p.victim = p.local
		p.victimSize = p.localSize
		p.local = nil
		p.localSize = 0
	}

	// oldPools 具有非空的受害者缓存，并且没有主缓存
	oldPools, allPools = allPools, nil
}

var (
	allPoolsMu Mutex   // 保护 oldPools，避免在 poolCleanup 与 pinSlow 时有竞争
	allPools   []*Pool // allPools 是具有非空主缓存的一组池，需要清除掉。受 1) allPoolsMu and pinning or 2) STW 保护
	oldPools   []*Pool // oldPools 是具有非空 victim 缓存的一组池。受 STW 保护
)

func init() {
	// 包初始化时，将 poolCleanup 函数注册到运行时的池子中，gc 时会调用该函数
	runtime_registerPoolCleanup(poolCleanup)
}

• 包初始化时，将 poolCleanup 函数注册到运行时的池子中，gc 时会调用该函数；
• 该函数用于清除所有 Pool 中的受害者缓存，以及将需要清除的主缓存放入到受害者缓存中；

参考文章
[1] sync.Pool 高性能设计之集大成者 (opens new window)
[2] 深度解密 Go 语言之 sync.Pool (opens new window)