Question

用来在垃圾标记时保存灰色对象，同样是本地和全局两级结构。

本地

持有数据的是 workbuf 结构，以数组存储多个指针形成数据包。在本地和全局间转移时，也以 workbuf 为单位，实现批量操作。

type workbuf struct {
    workbufhdr
    obj [(_WorkbufSize - unsafe.Sizeof(workbufhdr{})) / goarch.PtrSize]uintptr
}

type workbufhdr struct {
    node lfnode // must be first
    nobj int
}

本地队列 gcWork 持有主从两个 workbuf 缓冲。目的是在本地和全局间平衡，同时避免转移复制。

// runtime2.go

type p struct {
    
    // gcw is this P's GC work buffer cache. The work buffer is
    // filled by write barriers, drained by mutator assists, and
    // disposed on certain GC state transitions.
    gcw gcWork
    
}

// mgcwork.go

// Garbage collector work pool abstraction.
//
// This implements a producer/consumer model for pointers to grey
// objects. A grey object is one that is marked and on a work
// queue. A black object is marked and not on a work queue.
//
// Write barriers, root discovery, stack scanning, and object scanning
// produce pointers to grey objects. Scanning consumes pointers to
// grey objects, thus blackening them, and then scans them,
// potentially producing new pointers to grey objects.

// A gcWork provides the interface to produce and consume work for the
// garbage collector.

type gcWork struct {
    
    // wbuf1 and wbuf2 are the primary and secondary work buffers.
    wbuf1, wbuf2 *workbuf
    
    // Bytes marked (blackened) on this gcWork. This is aggregated
    // into work.bytesMarked by dispose.
    bytesMarked uint64
    
    // Heap scan work performed on this gcWork. 
    heapScanWork int64
    
    // flushedWork indicates that a non-empty work buffer was
    // flushed to the global work list since the last gcMarkDone
    // termination check. Specifically, this indicates that this
    // gcWork may have communicated work to another gcWork.
    flushedWork bool
}

看一下具体操作，就可了解设计细节。如主缓冲区满，主从对调。而等新换上的主缓冲也满了，就将其整个上交给全局队列，再挂上空白缓冲区。如此，等于平衡一半任务到全局队列。

// mgcwork.go

// put enqueues a pointer for the garbage collector to trace.
// obj must point to the beginning of a heap object or an oblet.

func (w *gcWork) put(obj uintptr) {
    flushed := false
    
    // 默认使用主缓冲。
    wbuf := w.wbuf1
    
    if wbuf == nil {
        
        // 新缓冲区。
        w.init()
        wbuf = w.wbuf1
        // wbuf is empty at this point.
        
    } else if wbuf.nobj == len(wbuf.obj) {
        
        // 缓冲区已满，将主从对调。
        w.wbuf1, w.wbuf2 = w.wbuf2, w.wbuf1
        wbuf = w.wbuf1
        
        // 如果新换主缓冲区是否也满了，
        if wbuf.nobj == len(wbuf.obj) {
            
            // 直接将该缓冲区交到全局。
            putfull(wbuf)
            w.flushedWork = true
            
            // 取一个空缓冲区挂上。
            wbuf = getempty()
            w.wbuf1 = wbuf
            flushed = true
        }
    }
    
    // 存储。
    wbuf.obj[wbuf.nobj] = obj
    wbuf.nobj++
    
    // 通知控制器让更多工人上班。
    if flushed && gcphase == _GCmark {
        gcController.enlistWorker()
    }
}

快慢两个版本获取方法。

// 快：仅尝试本地主缓冲区。

func (w *gcWork) tryGetFast() uintptr {
    wbuf := w.wbuf1
    if wbuf == nil {
        return 0
    }
    if wbuf.nobj == 0 {
        return 0
    }
    
    wbuf.nobj--
    return wbuf.obj[wbuf.nobj]
}

func (w *gcWork) tryGet() uintptr {
    
    // 主缓冲区。
    wbuf := w.wbuf1
    if wbuf == nil {
        w.init()
        wbuf = w.wbuf1
        // wbuf is empty at this point.
    }
    
    // 如主缓冲区为空，主从对调。
    if wbuf.nobj == 0 {
        w.wbuf1, w.wbuf2 = w.wbuf2, w.wbuf1
        wbuf = w.wbuf1
        
        // 依旧为空，从全局获取。
        if wbuf.nobj == 0 {
            owbuf := wbuf
            wbuf = trygetfull()
            
            if wbuf == nil {
                return 0
            }
            
            // 释放新换上的空缓冲区。
            // 挂上对调前的。
            putempty(owbuf)
            w.wbuf1 = wbuf
        }
    }
    
    // 获取。
    wbuf.nobj--
    return wbuf.obj[wbuf.nobj]
}

全局

全局管理两个 workbuf 列表，分别代表有数据和空白待复用。

// mgc.go

var work struct {
    full  lfstack   // lock-free list of full blocks workbuf
    empty lfstack   // lock-free list of empty blocks workbuf
}

在 workbuf 头部有个 node 字段，用来连接 lfstack 结构。

type workbufhdr struct {
    node lfnode // must be first
    nobj int
}

// putfull puts the workbuf on the work.full list for the GC.
func putfull(b *workbuf) {
    work.full.push(&b.node)
}

// trygetfull tries to get a full or partially empty workbuffer.
func trygetfull() *workbuf {
	b := (*workbuf)(work.full.pop())
	if b != nil {
		return b
	}
	return b
}

无锁

全局队列使用了一种 Lock-Free Stack 结构。

// lfstack.go

// lfstack is the head of a lock-free stack.
// The zero value of lfstack is an empty list.

type lfstack uint64

// runtime2.go

// Lock-free stack node.
type lfnode struct {
	next    uint64
	pushcnt uintptr
}

用循环确保原子操作成功，常见的无锁算法。

// lfstack.go

func (head *lfstack) push(node *lfnode) {
    
    // 累加计数器，与地址共同生成唯一流水号。
	node.pushcnt++
	new := lfstackPack(node, node.pushcnt)
    
	for {
        
        // 将原有链表挂到新 node 上。
        // 但并未修改原链表，所以 CAS 保存不成功也不会影响。
        
        // 如果 CAS 失败，那么新一轮循环重新 Load 和 Cas。
        // 加上 CAS 会判断 old 是否相等，所以期间有其他并发操作也不影响。
        
		old := atomic.Load64((*uint64)(head))
		node.next = old
        
        // 修改成功就退出，否则一直重试。
		if atomic.Cas64((*uint64)(head), old, new) {
			break 
		}
	}
}

func (head *lfstack) pop() unsafe.Pointer {
	for {
        // 获取头部节点。
		old := atomic.Load64((*uint64)(head))
		if old == 0 {
			return nil
		}
        
        // 将 next 修改为新头部。
		node := lfstackUnpack(old)
		next := atomic.Load64(&node.next)
		if atomic.Cas64((*uint64)(head), old, next) {
			return unsafe.Pointer(node)
		}
	}
}

如果 CAS 判断的仅是 old 指针地址，而该地址又被意外重用，那就会造成错误结果，这就是所谓 ABA 问题。利用 地址 + 计数器 生成唯一流水号，实现 Double-CAS，就能避开。

// lfstack_64bit.go

func lfstackPack(node *lfnode, cnt uintptr) uint64 {
	return uint64(uintptr(unsafe.Pointer(node)))<<(64-addrBits) | uint64(cnt&(1<<cntBits-1))
}

func lfstackUnpack(val uint64) *lfnode {
	if GOARCH == "amd64" {
		return (*lfnode)(unsafe.Pointer(uintptr(int64(val) >> cntBits << 3)))
	}
	return (*lfnode)(unsafe.Pointer(uintptr(val >> cntBits << 3)))
}