golang gc的内部优化
今天讲一个常见的gc compiler（也就是官方版本的go编译器和runtime）在垃圾回收的扫描标记阶段做的优化。我对这个优化的描述印象最深的是在bigcache的解释里，大致内容是假如map的键值都不包含指针，那么gc扫描的时间不管这个map多多数不会深入扫描map内部存储的数据，只查抄map自己是否必要回收。
这么做的好处显然是可以让gc的扫描速度大大增加，从而减少gc对性能的损耗。
减少指针数目自己就是常见的优化手段，但让我感到好奇的是解释里说的“跳过”。跳过的依据究竟是什么，以及只有map存在这种跳过吗？
于是我进行了全面的搜索，效果除了复读bigcache里那段话的，没什么有效的发现。
于是这篇文章诞生了。
跳过扫描指的是什么
前置知识少不得。
简单的说，gc在查抄对象是否存活的时间，除了对象自己，还要查抄对象的子对象是否引用了其他对象，具体来说：
[*]数组和slice的话指存储在里面的每一个元素是否存活，这里被存储的元素是数组/slice的子对象
[*]map的子对象就是里面存的键和值了
[*]struct的子对象是它的每一个字段
为了查抄这些子对象是否引用了其他对象（关系到这些被引用的对象是否能被回收），gc必要深入扫描这些子对象。子对象越多必要扫描的东西就越多。而且这个过程是递归的，因为子对象也会有子对象，想象一下嵌套的数组大概map。
跳过扫描自然就是指跳过这些子对象的扫描，只必要查抄对象自己即可的操作。
什么样的对象是可以跳过扫描的
这也是我的第一个疑问。跳过或不跳过的依据是什么，又大概是什么东西在控制这一过程。
bigcache告诉我们存有不包含指针的键值对的map是可以跳过的，那么具体情况是怎么样的呢？
找不到有效的资料，那只能看代码了，代码以Go 1.22.1为准。
起首应该想到的应该是从gc的代码开始看，于是很快就有了劳绩:
// runtime/mgcmark.go
// 负责gc扫描的函数，还有个它的兄弟gcDrainN，代码差不多就不放了
func gcDrain(gcw *gcWork, flags gcDrainFlags) {
...
// 先标记所有root对象，检查对象是否存活就是从这开始的
if work.markrootNext < work.markrootJobs {
for !(gp.preempt && (preemptible || sched.gcwaiting.Load() || pp.runSafePointFn != 0)) {
markroot(gcw, job, flushBgCredit)
// 检查自己是否需要被中断，需要的场合函数会直接跳到收尾工作然后返回
}
}
// 从工作队列里拿需要扫描的对象进行处理
for !(gp.preempt && (preemptible || sched.gcwaiting.Load() || pp.runSafePointFn != 0)) {
b := gcw.tryGetFast() // 从工作队列拿对象
scanobject(b, gcw)
...
}
...
}流程不考虑中断、数据统计和校验的话还是很简单的，就是先标记扫描的出发点，然后从gcw这个工作队列里拿东西出来处置惩罚，直到工作队列里再也没数据了为止。
markroot也很简单，根据root对象的种类，它会调用scanblock大概markrootSpans。此中scanblock会调用greyobject来标记待处置惩罚的对象。因此稍微看看markrootSpans即可。
markrootSpans是用来处置惩罚那些存放设置了终结器的对象的内存的：
// runtime/mgcmark.go
func markrootSpans(gcw *gcWork, shard int) {
...
for i := range specialsbits {
...
for j := uint(0); j < 8; j++ {
// 找到要处理的span（go内存使用的单位，你就当是“一块内存空间”就行）
s := ha.spans
...
lock(&s.speciallock)
for sp := s.specials; sp != nil; sp = sp.next {
if sp.kind != _KindSpecialFinalizer {
continue
}
// don't mark finalized object, but scan it so we
// retain everything it points to.
// spf是终结器本身
spf := (*specialfinalizer)(unsafe.Pointer(sp))
// A finalizer can be set for an inner byte of an object, find object beginning.
p := s.base() + uintptr(spf.special.offset)/s.elemsize*s.elemsize
// p是设置了终结器的对象
// 这里检查这个对象占用的内存上是否设置了跳过扫描的标记
// 设置了的话就不要继续扫描对象自己的子对象了
if !s.spanclass.noscan() {
scanobject(p, gcw)
}
// 这个span本身就是root对象，所以剩下的直接用scanblock处理
scanblock(uintptr(unsafe.Pointer(&spf.fn)), goarch.PtrSize, &oneptrmask, gcw, nil)
}
unlock(&s.speciallock)
}
}
}实在很简单，依旧是找到所有的对象，然后进行处置惩罚。然而我们看到了有意思的东西：s.spanclass.noscan()。
看起来这和是否跳过扫描有关。
但我们先不深入这个方法，为什么？因为终结器是被特殊处置惩罚的，没看完scanobject和greyobject之前我们不能断言这个方法是否控制着对对象的扫描。（实在解释上我已经告诉你就是这个东西控制的了，但假如你自己跟踪代码的话头一次看到这段代码的时间是不知道的）
所以我们接着看scanobject，这个函数是扫描对象的子对象的：
// runtime/mgcmark.go
func scanobject(b uintptr, gcw *gcWork) {
// 先拿到还没扫描过的内存
s := spanOfUnchecked(b)
n := s.elemsize
// n 表示mspan里有几个对象，在被这个函数检查的时候肯定不能是0
if n == 0 {
throw("scanobject n == 0")
}
if s.spanclass.noscan() {
// 如果内存设置了noscan标志，就报错
throw("scanobject of a noscan object")
}
var tp typePointers
if n > maxObletBytes {
// 大内存分割成不同的块放进工作队列，这样能被并行处理
if b == s.base() {
// 分割后入队
for oblet := b + maxObletBytes; oblet < s.base()+s.elemsize; oblet += maxObletBytes {
if !gcw.putFast(oblet) {
gcw.put(oblet)
}
}
}
// 获取类型信息
} else {
// 这里不重要
}
var scanSize uintptr
for {
var addr uintptr
// 获取子对象
// 整个循环的退出条件就是next不再返回子对象的时候（没东西可继续扫描了）
if tp, addr = tp.nextFast(); addr == 0 {
if tp, addr = tp.next(); addr == 0 {
break
}
}
// 拿到要处理的对象
scanSize = addr - b + goarch.PtrSize
obj := *(*uintptr)(unsafe.Pointer(addr))
// 排除nil和指向当前对象自身的指针
// 后者属于可以被回收的循环引用，当前对象能不能回收不受这个指针影响
// 因为如果当前对象不可访问了，那么它的字段自然也是不可能被访问到的，两者均从root不可达
// 而如果这个指针是可达的，那么当前对象的字段被引用，当前对象也是不需要回收的
// 所以指向当前对象本身的指针字段不需要处理
if obj != 0 && obj-b >= n {
if obj, span, objIndex := findObject(obj, b, addr-b); obj != 0 {
greyobject(obj, b, addr-b, span, gcw, objIndex)
}
}
}
...
}这个函数长归长，条理还是清晰的：
[*]起首看看对象是否太大要把对象的内存分割成小块交给工作队列里的其他协程并行处置惩罚
[*]接着扫描所有子对象，用greyobject标记这些对象
因为这个函数自己已经是在扫描了，所以不太会有“跳过”的相关的逻辑，而且你也看到了把这个函数放在不必要扫描子对象的对象上调用时会触发throw，throw会导致程序报错并退出实行。
所以机密就在greyobject里了。看看代码：
// runtime/mgcmark.go
func greyobject(obj, base, off uintptr, span *mspan, gcw *gcWork, objIndex uintptr) {
...
if useCheckmark {
if setCheckmark(obj, base, off, mbits) {
// Already marked.
return
}
} else {
...
// If marked we have nothing to do.
if mbits.isMarked() {
return
}
mbits.setMarked()
...
// 如果内存被标记为不需要进一步扫描，则会跳过后续的流程（内存会被放进gc扫描的工作队列里等着被取出来扫描）
if span.spanclass.noscan() {
...
return
}
}
// 对象被放进工作队列等待扫描
}这个函数会先查抄对象是否已经被处置惩罚过，然后标记对象，接着查抄span上的noscan标志，设置了的话就返回调用，没有设置说明必要被进一步扫描，于是被放进工作队列，等着gcDrain大概它兄弟来处置惩罚。
现在我们可以得出结论了，会不会跳过扫描，全部由内存上是否设置noscan标志来控制，设置了就可以跳过。
至于在这块内存上的是map还是slice还是struct，没关系。
跳过扫描的具体流程
看了上面的代码，我想信你一定是懵的，跳过具体发生的流程是什么样的呢？
没关系，我们看两个例子就知道了。
第一个例子是一个顶层的全局的可跳过扫描的对象A，介于我们还没说noscan会在什么情况下被设置，所以我们先忽略A的具体类型，只要知道它可以跳过扫描即可。
A的扫描流程是如许的：
[*]gc开始运行，先标记root对象
[*]A就是root之一，所以它要么被scanblock处置惩罚要么被markrootSpan处置惩罚
[*]假设A设置了终结器，又因为A是可跳过扫描子对象的，因此markrootSpan会直接调用scanblock
[*]scanblock会调用greyobject处置惩罚内存里的对象
[*]因为A可跳过扫描，所以greyobject做完标记就返回了，A不会进入工作队列
[*]A的扫描结束，整个流程上不会有scanobject的调用
A的例子相对简单，现在我们假设有个不是root对象的对象B，B自己不可跳过扫描，B有一个子对象C可以跳过扫描。我们来看看C的扫描流程：
[*]因为B并不是root对象，且不可跳过扫描，所以它作为某个root对象的子对象，现在肯定在gc工作队列里
[*]gcDrain从队列里拿到了B，于是交给了scanobject处置惩罚
[*]我们假设B不是很大因此不会被分割（反正分割了也一样）
[*]scanobject把每个B的子对象都用greyobject处置惩罚，C也不破例
[*]因为C可跳过扫描，所以greyobject做完标记就返回了，C不会进入工作队列
[*]C的扫描结束，整个流程上不会有对C的scanobject的调用
如许基本涵盖了所有的情况，一些我没单独说的比如“可跳过对象E是不可跳过root对象D的子对象”如许的情况，实际上和情况2没什么区别。
现在对象的子对象扫描是这么跳过的我们也知道了，只剩一个疑问了：noscan标志是怎么设置的？
noscan标志是怎么设置的
在深入之前，我们先来简单看下go的怎么分配内存的。完备讲解恐怕5篇长文也兜不住，所以我做些概念上的精简。
在go里，mspan是内存分配的基础单元，一个mspan上可以分配多个大小类似可以被归为一类的对象（比如13字节和14字节的对象都是一类，可以被分配到答应最大存储16字节对象的mspan上）。这个“类型”就叫mpan的sizeclass。一个简单的心智模型是把mspan当成一个能存大小相近的对象的列表。
为了加快内存分配，go会给每个线程预分配一块内存，然后按sizeclass分成多份，每份对应一个sizeclass的mspan。这个布局叫mcache。
当然了，总有对象的大小会超过所有mcache的sizeclass规定的范围，这个时间go就会像体系申请一大块内存，然后把内存交给mspan。
存储了span信息的比如sizeclass和noscan的布局叫spanClass。这个布局会作为字段存储在mspan的控制布局里。
知道了这些之后，我们就能看懂s.spanclass.noscan()了，它的意思就是查抄mspan的spanclass信息是否设置了不必要扫描子对象的标志。
而创建spanclass只能用makeSpanClass这个函数：
// runtime/mheap.go
type spanClass uint8
func makeSpanClass(sizeclass uint8, noscan bool) spanClass {
return spanClass(sizeclass<<1) | spanClass(bool2int(noscan))
}slice中的元素的类型信息被传给了mallocgc。假如slice的元素不包含指针，那么slice是可以跳过扫描的。
map比力特殊，跳过扫描的是它的bucket，而bucket外界是看不到的：
// runtime/malloc.go
func mallocgc(size uintptr, typ *_type, needzero bool) unsafe.Pointer {
...
// size是指类型的大小
// typ是需要创建的对象的类型信息，如果只是分配内存，typ要传nil
// typ是否是空的或者typ是否包含有指针
noscan := typ == nil || !typ.Pointers()
if 如果size足够小可以从mspan上分配 {
if size满足要求可以用tinyallocator分配 {
} else {
// 计算sizeclass（size对应到哪一类span）
spc := makeSpanClass(sizeclass, noscan) // noscan是这里传进去的
span = c.alloc // 从mcache拿mspan
v := nextFreeFast(span) // 从mspan真正拿到可用的内存
// 后面是把内存内容清零和维护gc信息等代码
}
} else {
// 大对象分配
// mcache.allocLarge也调用makeSpanClass(0, noscan)，然后用mheap.alloc根据span的信息从系统申请内存
span = c.allocLarge(size, noscan) // noscan是这里传进去的
// 后面是把内存内容清零和维护gc信息等代码
}
}可以看到要是键值对里都不包含指针的话，map就可以被跳过。
所以总结下，只要创建的对象不包含指针（例如数组/切片成员都是不包含指针的类型，map的键值对都不包含指针，布局体所有字段不包含指针）大概只是单纯分配块内存（makeslicecopy里分配一块内存然后再把数据copy进去的时间会判断element里包不包含指针，不包含的时间会传nil给mallocgc），noscan就会被设置。
现在所有的疑问都解决了：noscan是内存分配时根据类型信息来设置的；能跳过扫描的不只是map，符合条件的类型不管是slice、map还是struct都可以。
优化带来的提升
说了这么多，这个优化带来的提升有多少呢？
看个例子：
// runtime/slice.go
func makeslice(et *_type, len, cap int) unsafe.Pointer {
mem, overflow := math.MulUintptr(et.Size_, uintptr(cap))
if overflow || mem > maxAlloc || len < 0 || len > cap {
// NOTE: Produce a 'len out of range' error instead of a
// 'cap out of range' error when someone does make([]T, bignumber).
// 'cap out of range' is true too, but since the cap is only being
// supplied implicitly, saying len is clearer.
// See golang.org/issue/4085.
mem, overflow := math.MulUintptr(et.Size_, uintptr(len))
if overflow || mem > maxAlloc || len < 0 {
panicmakeslicelen()
}
panicmakeslicecap()
}
return mallocgc(mem, et, true)
}我们分别创建20个包含10000个int64大概*int64的slice（两个类型在x64体系上都是8字节大小），然后手动触发一次GC。为了让效果更正确，我们还在测试开始前禁用了自动触发的gc，而且我们创建的slice的长度和slice里元素的大小都是一样的，所以总体来说效果应该比力接近真实的gc回收内存时的性能。
这是效果：
// runtime/map.go
// 调用链：makemap -> makeBucketArray -> newarray -> mallocgc
func makeBucketArray(t *maptype, b uint8, dirtyalloc unsafe.Pointer) (buckets unsafe.Pointer, nextOverflow *bmap) {
base := bucketShift(b)
nbuckets := base
// For small b, overflow buckets are unlikely.
// Avoid the overhead of the calculation.
if b >= 4 {
// Add on the estimated number of overflow buckets
// required to insert the median number of elements
// used with this value of b.
nbuckets += bucketShift(b - 4)
sz := t.Bucket.Size_ * nbuckets
up := roundupsize(sz, !t.Bucket.Pointers())
if up != sz {
nbuckets = up / t.Bucket.Size_
}
}
if dirtyalloc == nil {
// t.Bucket.Pointers() 返回键值对中是否包含指针
buckets = newarray(t.Bucket, int(nbuckets))
} else {
// dirtyalloc was previously generated by
// the above newarray(t.Bucket, int(nbuckets))
// but may not be empty.
buckets = dirtyalloc
size := t.Bucket.Size_ * nbuckets
if t.Bucket.Pointers() {
memclrHasPointers(buckets, size)
} else {
memclrNoHeapPointers(buckets, size)
}
}
if base != nbuckets {
// We preallocated some overflow buckets.
// To keep the overhead of tracking these overflow buckets to a minimum,
// we use the convention that if a preallocated overflow bucket's overflow
// pointer is nil, then there are more available by bumping the pointer.
// We need a safe non-nil pointer for the last overflow bucket; just use buckets.
nextOverflow = (*bmap)(add(buckets, base*uintptr(t.BucketSize)))
last := (*bmap)(add(buckets, (nbuckets-1)*uintptr(t.BucketSize)))
last.setoverflow(t, (*bmap)(buckets))
}
return buckets, nextOverflow
}
func newarray(typ *_type, n int) unsafe.Pointer {
if n == 1 {
return mallocgc(typ.Size_, typ, true)
}
mem, overflow := math.MulUintptr(typ.Size_, uintptr(n))
if overflow || mem > maxAlloc || n < 0 {
panic(plainError("runtime: allocation size out of range"))
}
return mallocgc(mem, typ, true)
}内存用量大家都一样，但存指针的时间速度慢了五分之一。slice越大差距也会越大。可见跳过扫描带来的提升还是很大的。
别的少用指针还有助于增加数据的局部性，不但仅是惠及gc扫描。
如何使用这一优化
最后我们看看如何使用这一优化。
少用指针可以减轻gc压力大家都知道，但有一些“不得不用”指针的时间。
以一个本地cache为例：
var a int64 = 1000
func generateIntSlice(n int64) []int64 {
ret := make([]int64, 0, n)
for i := int64(0); i < n; i++ {
ret = append(ret, a)
}
return ret
}
func generatePtrSlice(n int64) []*int64 {
ret := make([]*int64, 0, n)
for i := int64(0); i < n; i++ {
ret = append(ret, &a)
}
return ret
}
func BenchmarkGCScan1(b *testing.B) {
defer debug.SetGCPercent(debug.SetGCPercent(-1)) // 测试期间禁止自动gc
for i := 0; i < b.N; i++ {
for j := 0; j < 20; j++ {
generatePtrSlice(10000)
}
runtime.GC()
}
}
func BenchmarkGCScan2(b *testing.B) {
defer debug.SetGCPercent(debug.SetGCPercent(-1))
for i := 0; i < b.N; i++ {
for j := 0; j < 20; j++ {
generateIntSlice(10000)
}
runtime.GC()
}
}值必要用指针是有两个原因，一是map的元素不能取地址，假如我们想要cache里的数据可以自由使用的话那就不得不用临时变量加复制，如许假如我们想更新值的时间就会很麻烦；二是假如值很大的话复制带来的开销会很大，用cache就是想提升性能呢反过来降落了怎么行。
但这么做就会导致Cache.m里的每一个键值对要被扫描，假如键值对很多的话性能会十分感人。
如许看起来是“不得不用指针”的场景。真的是如许吗？考虑到cache自己就是空间换时间的做法，我们不妨再多用点空间：
goos: windows
goarch: amd64
cpu: Intel(R) Core(TM) i5-10200H CPU @ 2.40GHz
│ old.txt │ new.txt │
│ sec/op │ sec/op vs base │
GCScan-8 379.0µ ± 2% 298.0µ ± 2%-21.51% (p=0.000 n=10)
│ old.txt │ new.txt │
│ B/op │ B/op vs base │
GCScan-8 1.563Mi ± 0% 1.563Mi ± 0%~ (p=0.438 n=10)
│old.txt │ new.txt │
│ allocs/op│ allocs/op vs base │
GCScan-8 20.00 ± 0% 20.00 ± 0%~ (p=1.000 n=10) ¹
¹ all samples are equal我们用一个slice来存所有的值，然后再把key映射到值在slice中的索引上。对于slice的元素，我们是可以取地址的，因此可以简单拿到值的指针，对于值的更新也可以基于这个Get拿到的指针，时间复杂度不变，简单又方便。
然后我们再来看，现在buf和m都没有指针了，只要K和V不包含指针，那么不管我们的cache里存了多少东西对gc来说都只要看看外层的Cache对象是否存活就够了。
但是这么做会有代价：
<ul>Get会稍微慢一点，因为不但要做额外的查抄，还必要从两个不同的数据布局里拿数据，对缓存不友好
存数据进Cache的时间不可制止地必要一次复制
Get返回的指针没有稳定性，在底层的buf扩容后就会失效
删除元素会很慢，这怪我们用了slice，你可以再多浪费点内存用墓碑标志来模仿删除大概干脆不提供删除功能（不好做就干脆不做，这是非常golang的做法
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]