关于垃圾回收的基本知识已经介绍的差不多了，只是要知道垃圾回收过程是需要耗费CPU时间的，那就有可能会影响到用户协程的调度，所以在某些场景需要垃圾回收相关调优。本篇文章主要介绍垃圾回收的触发时机，以及垃圾回收器的几种调度模式，只有了解这些才能知道如何调优；最后结合常用的缓存框架bigcache，分析如何减少垃圾回收的压力。

  什么时候触发垃圾回收呢？首先内存使用增长一定比例时有可能会触发（总不能任由内存增长吧），还有其他方式吗？我们也可以通过runtime.GC函数手动触发（会阻塞用户协程，直到垃圾回收流程结束），另外，Go辅助线程也会检测，如果超过2分钟没有执行垃圾回收，则强制启动垃圾回收。三种触发方式定义如下：

// gcTriggerHeap indicates that a cycle should be started when
// the heap size reaches the trigger heap size computed by the
// controller.
gcTriggerHeap gcTriggerKind = iota   //内存增长到触发门限

// gcTriggerTime indicates that a cycle should be started when
// it's been more than forcegcperiod nanoseconds since the
// previous GC cycle.
gcTriggerTime                        //定时触发

// gcTriggerCycle indicates that a cycle should be started if
// we have not yet started cycle number gcTrigger.n (relative
// to work.cycles).
gcTriggerCycle                      //可用于强制触发


//包含触发类型，当前时间，周期数
type gcTrigger struct {
    kind gcTriggerKind
    now  int64  // gcTriggerTime: current time
    n    uint32 // gcTriggerCycle: cycle number to start
}

  还记得上一篇文章介绍垃圾回收入口函数是gcstart，该函数的输入参数就是gcTrigger，每一次开启垃圾回收之前，都会检测是否应该触发垃圾回收，检测方式如下：

func (t gcTrigger) test() bool {
    switch t.kind {
    case gcTriggerHeap:
        //内存达到门限
        return gcController.heapLive >= gcController.trigger
    case gcTriggerTime:
        //可关闭GC：Initialized from GOGC. GOGC=off means no GC.
        if gcController.gcPercent.Load() < 0 {
            return false
        }

        //forcegcperiod定义为2分钟
        lastgc := int64(atomic.Load64(&memstats.last_gc_nanotime))
        return lastgc != 0 && t.now-lastgc > forcegcperiod
    case gcTriggerCycle:
        // 可用来强制触发
        return int32(t.n-work.cycles) > 0
    }
    return true
}

  总给下来，垃圾回收总共有三种触发方式：申请内存，定时触发，主动触发。主动触发与定时触发的逻辑比较简单，这里就不做过多介绍了，我们重点了解申请内存触发的垃圾回收。

  申请内存如何能触发垃圾回收呢？想想申请内存的入口函数是不是mallocgc，所以只需要在这里判断就可以了：

func mallocgc(size uintptr, typ *_type, needzero bool) unsafe.Pointer {
    //只有当一次申请内存过大（超过32768）才会检测是否开启GC
    if size <= maxSmallSize {

    } else {
        shouldhelpgc = true
    }

    if shouldhelpgc {
        if t := (gcTrigger{kind: gcTriggerHeap}); t.test() {
            gcStart(t)
        }
    }
}

  不过，你有没有想过，如何统计当前分配的内存字节数呢？以及如何计算下一次垃圾回收触发的内存门限呢？分配内存字节数理论上也应该在mallocgc函数更新吧，不过Go语言并没有这么做，而是从mcentral获取mspan缓存到mcache时候更新的（也就是说缓存到mcache就算"申请"了），所以这个数据其实并不是真正用户代码已分配的内存数（略大）。

func (c *mcache) refill(spc spanClass) {
    //该mspan已使用的内存
    usedBytes := uintptr(s.allocCount) * s.elemsize
    //更新全局统计变量
    gcController.update(int64(s.npages*pageSize)-int64(usedBytes), int64(c.scanAlloc))
}

func (c *gcControllerState) update(dHeapLive, dHeapScan int64) {
    if dHeapLive != 0 {
        atomic.Xadd64(&gcController.heapLive, dHeapLive)
    }
}

  另一个问题呢，如何计算下一次垃圾回收触发的内存门限呢？当然在每一次垃圾回收结束之后，需要更新下一次垃圾回收触发的内存门限，该门限与上一次垃圾回收的标记内存数以及触发比例有关，触发比例是多少呢？与环境变量GOGC有关。触发门限计算过程如下：

func (c *gcControllerState) commit(triggerRatio float64) {
    //gcPercent数据来源于环境变量GOGC

    // goal下一次垃圾回收触发目标
    goal := ^uint64(0)
    if gcPercent := c.gcPercent.Load(); gcPercent >= 0 {
        goal = c.heapMarked + (c.heapMarked+atomic.Load64(&c.stackScan)+atomic.Load64(&c.globalsScan))*uint64(gcPercent)/100
    }

    //trigger由goal计算得来
    //由于垃圾回收过程中，用户协程还在并发申请内存，所以最终触发门限trigger并不等于goal

    c.trigger = trigger
}

  注意由于垃圾回收过程中，用户协程还在并发申请内存，所以最终触发门限trigger并不等于goal，但是不可否认trigger是由goal计算得来的，还涉及到调步算法，这里就不展开了。

  垃圾回收一方面可以回收无用内存，避免Go程序占用过多内存，但是垃圾回收过程也需要占用CPU时间，就可能会对用户协程的调度有一定影响，所以在某些场景可能需要垃圾回收相关调优，一般也就是调整环境变量GOGC，平衡Go程序内存使用与垃圾回收对CPU的占用。

垃圾回收调度模式

  圾回收过程也需要占用CPU时间，上一篇文章我们看到垃圾回收工作协程gcBgMarkWorker数目与逻辑处理器P的数目保持一致，这些协程同时被调度吗？调度之后是一直运行等到垃圾回收过程结束吗？Go语言是如何保障垃圾回收过程占用一定比例的CPU呢，不至于太影响用户协程，也不至于太慢呢？Go语言定义了三种垃圾回收工作协程调度模式：

// gcMarkWorkerDedicatedMode indicates that the P of a mark
// worker is dedicated to running that mark worker. The mark
// worker should run without preemption.
gcMarkWorkerDedicatedMode      //当前P只能运行垃圾回收工作协程，且不能被抢占

// gcMarkWorkerFractionalMode indicates that a P is currently
// running the "fractional" mark worker. The fractional worker
// is necessary when GOMAXPROCS*gcBackgroundUtilization is not
// an integer and using only dedicated workers would result in
// utilization too far from the target of gcBackgroundUtilization.
// The fractional worker should run until it is preempted and
// will be scheduled to pick up the fractional part of
// GOMAXPROCS*gcBackgroundUtilization.
gcMarkWorkerFractionalMode     //当前P运行垃圾回收工作协程，需要保障总得CPU占用时间

// gcMarkWorkerIdleMode indicates that a P is running the mark
// worker because it has nothing else to do. The idle worker
// should run until it is preempted and account its time
// against gcController.idleMarkTime.
gcMarkWorkerIdleMode          //如果当前P没有用户协程可调度，才调度垃圾回收工作协程

  首先要明确，Go语言保证垃圾回收工作协程CPU利用率为25%左右，如何保障呢？假设Go程序创建了8个逻辑处理器P，25%就相当于在两个逻辑处理器P调度垃圾回收主协程（当前P只运行垃圾回收工作协程），那如果逻辑处理器P的数目不能被4整除怎么办？这时候肯定会有部分P调度模式采用gcMarkWorkerFractionalMode，计算方式如下：

func (c *gcControllerState) startCycle(markStartTime int64, procs int) {

    // gcBackgroundUtilization是常量0.25，procs即逻辑处理器P的数目
    totalUtilizationGoal := float64(procs) * gcBackgroundUtilization
    //向上取整，这些P只能运行垃圾回收工作协程
    c.dedicatedMarkWorkersNeeded = int64(totalUtilizationGoal + 0.5)
    //非整数时，计算误差
    utilError := float64(c.dedicatedMarkWorkersNeeded)/totalUtilizationGoal - 1
    //误差比较大，需要修正
    if utilError < -maxUtilError || utilError > maxUtilError {

        //太多P处于gcMarkWorkerDedicatedMode模式，减1
        if float64(c.dedicatedMarkWorkersNeeded) > totalUtilizationGoal {
            c.dedicatedMarkWorkersNeeded--
        }

        //需要有部分P处于gcMarkWorkerFractionalMode模式，这些P占用CPU时间比例为fractionalUtilizationGoal
        c.fractionalUtilizationGoal = (totalUtilizationGoal - float64(c.dedicatedMarkWorkersNeeded)) / float64(procs)
    } else {
        //误差较小，不需要修正
        c.fractionalUtilizationGoal = 0
    }
}

  垃圾回收启动的时候，会计算好各调度模式下逻辑处理器P的数目，Go语言调度器在调度垃圾回收工作协程时，设置各个工作协程的调度模式，参考函数findRunnableGCWorker的实现：

func (c *gcControllerState) findRunnableGCWorker(_p_ *p) *g {
    //函数decIfPositive判断输入参数是否是正数，并减1

    // 调度模式为gcMarkWorkerDedicatedMode
    if decIfPositive(&c.dedicatedMarkWorkersNeeded) {
        _p_.gcMarkWorkerMode = gcMarkWorkerDedicatedMode
    }else if c.fractionalUtilizationGoal == 0 {
        ......
    } else {

        //delta垃圾回收标记开始到现在时间段
        delta := nanotime() - c.markStartTime

        // 计算gcMarkWorkerFractionalMode调度模式下垃圾回收占用的CPU比例，如果超过直接返回
        if delta > 0 && float64(_p_.gcFractionalMarkTime)/float64(delta) > c.fractionalUtilizationGoal {
            ......
        }
        // 运行在gcMarkWorkerFractionalMode调度模式
        _p_.gcMarkWorkerMode = gcMarkWorkerFractionalMode
    }


}

  怎么没有gcMarkWorkerIdleMode呢？想想这种模式的含义是什么：如果当前P没有用户协程可调度，才调度垃圾回收工作协程。所以应该是Go语言调度器在获取可运行用户协程时，发现没有可运行协程而此时正处于垃圾回收过程，则调度垃圾回收工作协程（参考调度器查找协程函数findrunnable）。

  调度模式已经确定了，垃圾回收工作协程gcBgMarkWorker在运行时，会根据调度模式，决定如何执行标记扫描过程：

func gcBgMarkWorker() {
    for {
        gopark(...)

        startTime := nanotime()

        systemstack(func() {
            switch pp.gcMarkWorkerMode {
            default:
                throw("gcBgMarkWorker: unexpected gcMarkWorkerMode")

            //只运行垃圾回收协程，第一次执行标记扫描过程直到被抢占
            case gcMarkWorkerDedicatedMode:
                gcDrain(&pp.gcw, gcDrainUntilPreempt|gcDrainFlushBgCredit)
                //如果被抢占，将当前P队列的协程添加到全局协程队列
                if gp.preempt {
                    if drainQ, n := runqdrain(pp); n > 0 {
                        lock(&sched.lock)
                        globrunqputbatch(&drainQ, int32(n))
                        unlock(&sched.lock)
                    }
                }

                //执行标记扫描，直到任务结束
                gcDrain(&pp.gcw, gcDrainFlushBgCredit)
            //按照一定CPU时间比例执行标记扫描，或者直到被抢占
            case gcMarkWorkerFractionalMode:
                gcDrain(&pp.gcw, gcDrainFractional|gcDrainUntilPreempt|gcDrainFlushBgCredit)
            //只在空闲时执行标记扫描
            case gcMarkWorkerIdleMode:
                gcDrain(&pp.gcw, gcDrainIdle|gcDrainUntilPreempt|gcDrainFlushBgCredit)
            }
        })

        duration := nanotime() - startTime

        //统计gcMarkWorkerFractionalMode模式下，标记扫描过程的运行时间
        if pp.gcMarkWorkerMode == gcMarkWorkerFractionalMode {
            atomic.Xaddint64(&pp.gcFractionalMarkTime, duration)
        }

        ......
    }
}


gcDrainUntilPreempt gcDrainFlags = 1 << iota    // 运行gcDrain直到被抢占
gcDrainFlushBgCredit                            // gcDrain更新全局现金池（回顾辅助标记）
gcDrainIdle                                     // gcDrain只在空闲时运行
gcDrainFractional                                // gcDrain只能占用一定CPU比例

  gcDrain函数主要也是一个循环，循环获取灰色节点并执行标记扫描流程，如何实现这几种调度模式呢？也就是何时结束循环呢？只需要在每次循环开始判断一下就行了，比如在循环条件中判断当然是否被抢占，占用CPU时间是否超过一定比例，当前P是否有用户协程在等待调度等等

func gcDrain(gcw *gcWork, flags gcDrainFlags) {

    // 是否可被抢占
    preemptible := flags&gcDrainUntilPreempt != 0

    //循环结束检测方法
    if flags&(gcDrainIdle|gcDrainFractional) != 0 {
        if idle {
            check = pollWork     //检测是否有用户协程等待调度
        } else if flags&gcDrainFractional != 0 {
            check = pollFractionalWorkerExit   //检测CPU时间占用比例
        }
    }

    //如果允许抢占且被抢占，结束
    for !(gp.preempt && (preemptible || atomic.Load(&sched.gcwaiting) != 0)) {
        //标记扫描

        //校验是否结束循环
        if check != nil && check() {
            break
        }

    }

    ......
}

bigcache概述

  垃圾回收如何调优呢？一方面可以针对业务类型调整环境变量GOGC，平衡Go程序内存使用与垃圾回收对CPU的占用；另一方面可以尽量减少用户代码分配内存的数量，比如使用对象池（复用）。

  还有其他方案吗？回想一下标记扫描整个过程：1）从灰色对象集合中选择一个对象，标为黑色；2）扫描该对象指向的所有对象，将其加入到灰色对象集合；3）不断重复步骤1/2。本质上就是只要对象包含指针，就需要继续扫描，所以Go语言才会将每种mspan分为两种规格，有指针与无指针，而不包含指针的mspan是不需要继续扫描的。

  通常为了提升服务性能，都会使用本地缓存，那用户代码就必然会大量分配内存，垃圾回收的压力也会非常大，这时候该如何解决呢？bigcache是常用的本地内存缓存组件，就是通过去除指针来减少垃圾回收扫描的压力。

  缓存组件还能去除指针？想想一般缓存数据存储怎么设计呢：通常有一个map，存储缓存key-value对象，一般还会基于LRU实现缓存淘汰算法。bigcache也是是用的map存储缓存对象，那怎么说去除了指针呢？因为缓存的key和value都是整数！key按hash值存储，那字符串key存储在哪呢？value又是怎么按照整数存储呢？其实value存储的也是位置索引，真正的数据entry存储在字节数组，并不像传统map，entry是一个个独立的对象/节点。

type cacheShard struct {
    // map定义，key-value都是整数
    hashmap     map[uint64]uint32
    //真正存储数据entry，是一个字节数组
    entries     queue.BytesQueue
    //读写需要加锁
    lock        sync.RWMutex
}

  看到了吧map的定义是不包含指针的，而且数据key-value是编码为二进制存储在entries字节数组的，整个也不包含指针。下面我们简单看看bigcache的数据查找逻辑：

func (s *cacheShard) get(key string, hashedKey uint64) ([]byte, error) {
    s.lock.RLock()

    // 或者entry的字节编码
    wrappedEntry, err := s.getWrappedEntry(hashedKey)

    // readKeyFromEntry函数将字节数组解码为
    if entryKey := readKeyFromEntry(wrappedEntry); key != entryKey {
        // 哈希冲突，返回错误
    }

    // 解码
    entry := readEntry(wrappedEntry)
    s.lock.RUnlock()

    return entry, nil
}

func (s *cacheShard) getWrappedEntry(hashedKey uint64) ([]byte, error) {
    itemIndex := s.hashmap[hashedKey]

    // itemIndex就是字节数组索引
    wrappedEntry, err := s.entries.Get(int(itemIndex))
    return wrappedEntry, err
}

  bigcache的数据存储不包含指针，通过这种方式来减少垃圾回收扫描的压力，不过由于entries是普通的字节数组，所以也就无法实现灵活的缓存淘汰策略了。

  本篇文章主要介绍垃圾回收的触发时机，以及垃圾回收器的几种调度模式，你需要了解垃圾回收占用CPU资源，可能会影响用户协程的调度执行，所以某些业务场景需要垃圾回收调优。最后结合常用的缓存框架bigcache，学习如何减少垃圾回收的压力（无指针）。