手摸手 Go 深入剖析 sync.Pool

package main
import "sync"
type Person struct {	Age int}
// 初始化poolvar personPool = sync.Pool{	New: func() interface{} {		return new(Person)	},}
func main() {	// 获取一个实例	newPerson := personPool.Get().(*Person)	// 回收对象 以备其他协程使用	defer personPool.Put(newPerson)
	newPerson.Age = 25}

import (	"testing")
func BenchmarkWithoutPool(b *testing.B) {	var p *Person	b.ReportAllocs()	b.ResetTimer()	for i := 0; i < b.N; i++ {		for j := 0; j < 10000; j++ {			p = new(Person)			p.Age = 30		}	}}
func BenchmarkWithPool(b *testing.B) {	var p *Person	b.ReportAllocs()	b.ResetTimer()	for i := 0; i < b.N; i++ {		for j := 0; j < 10000; j++ {			p = personPool.Get().(*Person)			p.Age = 30			personPool.Put(p)		}	}}

BenchmarkWithoutPoolBenchmarkWithoutPool-8   	    7630	    135523 ns/op	   80000 B/op	   10000 allocs/opBenchmarkWithPoolBenchmarkWithPool-8   	    9865	    126072 ns/op	       0 B/op	       0 allocs/op

type Pool struct {	noCopy noCopy	// 实际指向[]poolLocal 每个P对应一个poolLocal 数组大小取决于P的数量 runtime.GOMAXPROCS(0)	local     unsafe.Pointer 	localSize uintptr        // []poolLocal的大小
	victim     unsafe.Pointer // local from previous cycle	victimSize uintptr        // size of victims array    //当缓存池无对应对象时调用	New func() interface{}}

type poolLocal struct {	poolLocalInternal	// Prevents false sharing on widespread platforms with	// 128 mod (cache line size) = 0 .	pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte}
// Local per-P Pool appendix.type poolLocalInternal struct {	private interface{} // 只能被对应的P使用	shared  poolChain   // 本地的P可以从Head 进行pushHead/popHead 其他的P可以popTail.}

func (d *poolDequeue) unpack(ptrs uint64) (head, tail uint32) {	const mask = 1<<dequeueBits - 1	head = uint32((ptrs >> dequeueBits) & mask)	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)}

func (p *Pool) Get() interface{} {  // 将当前goroutine与P进行绑定 runtime_procPin禁用抢占  // 返回poolLocal与P的id	l, pid := p.pin()	x := l.private //尝试直接从私有空间拿	l.private = nil	if x == nil {    //从共享区域头部拿		x, _ = l.shared.popHead()		if x == nil {      //直接实在没有 尝试去别人那边看看能不能偷个			x = p.getSlow(pid)		}	}  // 解除抢占禁用	runtime_procUnpin()  // 都没有 那只好自己New一个	if x == nil && p.New != nil {		x = p.New()	}	return x}

func (p *Pool) pin() (*poolLocal, int) {	pid := runtime_procPin()  // pinSlow中我们先存储local再存储localSize,这里我们以相反顺序加载  // 因为我们已经禁用了抢占 GC这期间不会发生 因此我们需要观察local的大小至少跟localSize一样	s := atomic.LoadUintptr(&p.localSize) // load-acquire	l := p.local                          // load-consume	if uintptr(pid) < s {		return indexLocal(l, pid), pid	}  // 运行过程中可能会存在调整P的情况 或者GC了	return p.pinSlow()}

func runtime_procPin() int//go:linkname sync_runtime_procPin sync.runtime_procPin//go:nosplitfunc sync_runtime_procPin() int {	return procPin()}//go:nosplitfunc procPin() int {	_g_ := getg()	mp := _g_.m
	mp.locks++	return int(mp.p.ptr().id)}

//go:nowritebarrierrecfunc sysmon() {  ...	// retake P's blocked in syscalls		// and preempt long running G's		if retake(now) != 0 {			idle = 0		} else {			idle++		}  ...}func retake(now int64) uint32 { ... if s == _Prunning || s == _Psyscall {			// Preempt G if it's running for too long.			t := int64(_p_.schedtick)			if int64(pd.schedtick) != t {				pd.schedtick = uint32(t)				pd.schedwhen = now			} else if pd.schedwhen+forcePreemptNS <= now {//G运行时间超过forcePreemptNS				preemptone(_p_)				// In case of syscall, preemptone() doesn't				// work, because there is no M wired to P.				sysretake = true			}  ...}

func preemptone(_p_ *p) bool {	mp := _p_.m.ptr()	if mp == nil || mp == getg().m {		return false	}	gp := mp.curg	if gp == nil || gp == mp.g0 {		return false	}
	gp.preempt = true
	// Every call in a go routine checks for stack overflow by	// comparing the current stack pointer to gp->stackguard0.	// Setting gp->stackguard0 to StackPreempt folds	// preemption into the normal stack overflow check.	gp.stackguard0 = stackPreempt
	// Request an async preemption of this P.	if preemptMSupported && debug.asyncpreemptoff == 0 {		_p_.preempt = true		preemptM(mp)	}
	return true}

//以asm_amd64.s为例TEXT runtime·morestack(SB),NOSPLIT,$0-0	... ...	// Call newstack on m->g0's stack.	MOVQ	m_g0(BX), BX	MOVQ	BX, g(CX)	MOVQ	(g_sched+gobuf_sp)(BX), SP	CALL	runtime·newstack(SB)	CALL	runtime·abort(SB)	// crash if newstack returns	RET

//go:nowritebarrierrecfunc newstack() {  ... ...	if preempt {		if !canPreemptM(thisg.m) {			// Let the goroutine keep running for now.			// gp->preempt is set, so it will be preempted next time.			gp.stackguard0 = gp.stack.lo + _StackGuard			gogo(&gp.sched) // never return		}	} ... ... }//go:nosplitfunc canPreemptM(mp *m) bool {	return mp.locks == 0 && mp.mallocing == 0 && mp.preemptoff == "" && mp.p.ptr().status == _Prunning}

func stopTheWorldWithSema() {	_g_ := getg()
	// If we hold a lock, then we won't be able to stop another M	// that is blocked trying to acquire the lock.	if _g_.m.locks > 0 {		throw("stopTheWorld: holding locks")	}	lock(&sched.lock)	sched.stopwait = gomaxprocs	atomic.Store(&sched.gcwaiting, 1)	preemptall()	// stop current P	_g_.m.p.ptr().status = _Pgcstop // Pgcstop is only diagnostic.	sched.stopwait--	// try to retake all P's in Psyscall status	for _, p := range allp {		s := p.status		if s == _Psyscall && atomic.Cas(&p.status, s, _Pgcstop) {			if trace.enabled {				traceGoSysBlock(p)				traceProcStop(p)			}			p.syscalltick++			sched.stopwait--		}	}	// stop idle P's	for {		p := pidleget()		if p == nil {			break		}		p.status = _Pgcstop		sched.stopwait--	}	wait := sched.stopwait > 0	unlock(&sched.lock)
	// wait for remaining P's to stop voluntarily	if wait {		for {			// wait for 100us, then try to re-preempt in case of any races			if notetsleep(&sched.stopnote, 100*1000) {				noteclear(&sched.stopnote)				break			}			preemptall()		}	}
	// sanity checks	bad := ""	if sched.stopwait != 0 {		bad = "stopTheWorld: not stopped (stopwait != 0)"	} else {		for _, p := range allp {			if p.status != _Pgcstop {				bad = "stopTheWorld: not stopped (status != _Pgcstop)"			}		}	}	if atomic.Load(&freezing) != 0 {		// Some other thread is panicking. This can cause the		// sanity checks above to fail if the panic happens in		// the signal handler on a stopped thread. Either way,		// we should halt this thread.		lock(&deadlock)		lock(&deadlock)	}	if bad != "" {		throw(bad)	}}

func preemptall() bool {	res := false	for _, _p_ := range allp {		if _p_.status != _Prunning {			continue		}		if preemptone(_p_) {			res = true		}	}	return res}

func indexLocal(l unsafe.Pointer, i int) *poolLocal {	lp := unsafe.Pointer(uintptr(l) + uintptr(i)*unsafe.Sizeof(poolLocal{}))	return (*poolLocal)(lp)}

func (p *Pool) pinSlow() (*poolLocal, int) {	// Retry under the mutex.	// Can not lock the mutex while pinned.	runtime_procUnpin()	allPoolsMu.Lock()	defer allPoolsMu.Unlock()	pid := runtime_procPin()	// poolCleanup won't be called while we are pinned.	s := p.localSize	l := p.local	if uintptr(pid) < s {		return indexLocal(l, pid), pid	}	if p.local == nil {		allPools = append(allPools, p)	}	// If GOMAXPROCS changes between GCs, we re-allocate the array and lose the old one.	size := runtime.GOMAXPROCS(0)	local := make([]poolLocal, size)	atomic.StorePointer(&p.local, unsafe.Pointer(&local[0])) // store-release	atomic.StoreUintptr(&p.localSize, uintptr(size))         // store-release	return &local[pid], pid}

func (c *poolChain) popHead() (interface{}, bool) {	d := c.head	for d != nil {		if val, ok := d.popHead(); ok {			return val, ok		}		// There may still be unconsumed elements in the		// previous dequeue, so try backing up.		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 {			// Queue is empty.			return nil, false		}
		// Confirm tail and decrement head. We do this before		// reading the value to take back ownership of this		// slot.		head--		ptrs2 := d.pack(head, tail)		if atomic.CompareAndSwapUint64(&d.headTail, ptrs, ptrs2) {			// We successfully took back slot.			slot = &d.vals[head&uint32(len(d.vals)-1)]			break		}	}
	val := *(*interface{})(unsafe.Pointer(slot))	if val == dequeueNil(nil) {		val = nil	}	// Zero the slot. Unlike popTail, this isn't racing with	// pushHead, so we don't need to be careful here.	*slot = eface{}	return val, true}

func (p *Pool) getSlow(pid int) interface{} {	// See the comment in pin regarding ordering of the loads.	size := atomic.LoadUintptr(&p.localSize) // load-acquire	locals := p.local                        // load-consume	// Try to steal one element from other procs.	for i := 0; i < int(size); i++ {		l := indexLocal(locals, (pid+i+1)%int(size))		if x, _ := l.shared.popTail(); x != nil {			return x		}	}
	// Try the victim cache. We do this after attempting to steal	// from all primary caches because we want objects in the	// victim cache to age out if at all possible.	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		}	}
	// Mark the victim cache as empty for future gets don't bother	// with it.	atomic.StoreUintptr(&p.victimSize, 0)
	return nil}

func (c *poolChain) popTail() (interface{}, bool) {	d := loadPoolChainElt(&c.tail)	if d == nil {		return nil, false	}
	for {		// It's important that we load the next pointer		// *before* popping the tail. In general, d may be		// transiently empty, but if next is non-nil before		// the pop and the pop fails, then d is permanently		// empty, which is the only condition under which it's		// safe to drop d from the chain.		d2 := loadPoolChainElt(&d.next)
		if val, ok := d.popTail(); ok {			return val, ok		}
		if d2 == nil {			// This is the only dequeue. It's empty right			// now, but could be pushed to in the future.			return nil, false		}
		// The tail of the chain has been drained, so move on		// to the next dequeue. Try to drop it from the chain		// so the next pop doesn't have to look at the empty		// dequeue again.		if atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&c.tail)), unsafe.Pointer(d), unsafe.Pointer(d2)) {			// We won the race. Clear the prev pointer so			// the garbage collector can collect the empty			// dequeue and so popHead doesn't back up			// further than necessary.			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 {			// Queue is empty.			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		}	}	val := *(*interface{})(unsafe.Pointer(slot))	if val == dequeueNil(nil) {		val = nil	}	slot.val = nil	atomic.StorePointer(&slot.typ, nil)	return val, true}

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		}	}
	// Mark the victim cache as empty for future gets don't bother	// with it.	atomic.StoreUintptr(&p.victimSize, 0)

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		}	}
	// Mark the victim cache as empty for future gets don't bother	// with it.	atomic.StoreUintptr(&p.victimSize, 0)

if x == nil && p.New != nil {		x = p.New()	}

func (p *Pool) Put(x interface{}) {	if x == nil {		return	}  // 将goroutine与P绑定 runtime_procPin禁用抢占 返回poolLocal	l, _ := p.pin()	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 {		// Initialize the chain.		const initSize = 8 // Must be a power of 2		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 {		// Can't make it any bigger.		newSize = dequeueLimit	}
	d2 := &poolChainElt{prev: d}	d2.vals = make([]eface, newSize)	c.head = d2	storePoolChainElt(&d.next, d2)	d2.pushHead(val)}

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 {		// Queue is full.		return false	}	slot := &d.vals[head&uint32(len(d.vals)-1)]	typ := atomic.LoadPointer(&slot.typ)	if typ != nil {// popTail可能还没处理完		return false	}
	// The head slot is free, so we own it.	if val == nil {		val = dequeueNil(nil)	}	*(*interface{})(unsafe.Pointer(slot)) = val	atomic.AddUint64(&d.headTail, 1<<dequeueBits)	return true}

func init() {	runtime_registerPoolCleanup(poolCleanup)}//go:linkname sync_runtime_registerPoolCleanup sync.runtime_registerPoolCleanupfunc sync_runtime_registerPoolCleanup(f func()) {	poolcleanup = f}func poolCleanup() {	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, allPools = allPools, nil}

type noCopy struct{}
// Lock is a no-op used by -copylocks checker from `go vet`.func (*noCopy) Lock()   {}func (*noCopy) Unlock() {}type People struct {	noCopy noCopy}
func say(p People) {
}
func main() {	var p People	say(p)}

go vet demo.go

# command-line-arguments./demo.go:12:12: say passes lock by value: command-line-arguments.People contains command-line-arguments.noCopy./demo.go:18:6: call of say copies lock value: command-line-arguments.People contains command-line-arguments.noCopy

type poolLocal struct {	poolLocalInternal
	// Prevents false sharing on widespread platforms with	// 128 mod (cache line size) = 0 .	pad [128 - unsafe.Sizeof(poolLocalInternal{})%128]byte}