详尽干货！从源码角度看 Golang 的调度（上）

// runtime/asm_amd64.s
TEXT runtime·rt0_go(SB),NOSPLIT,$0......此处省略N多代码......ok:        // set the per-goroutine and per-mach "registers"        get_tls(BX)  // 将 g0 放到 tls(thread local storage)里        LEAQ    runtime·g0(SB), CX        MOVQ    CX, g(BX)        LEAQ    runtime·m0(SB), AX
        // save m->g0 = g0  // 将全局M0与全局G0绑定        MOVQ    CX, m_g0(AX)        // save m0 to g0->m        MOVQ    AX, g_m(CX)
        CLD                             // convention is D is always left cleared        CALL    runtime·check(SB)
        MOVL    16(SP), AX              // copy argc        MOVL    AX, 0(SP)        MOVQ    24(SP), AX              // copy argv        MOVQ    AX, 8(SP)        CALL    runtime·args(SB) // 解析命令行参数        CALL    runtime·osinit(SB) // 只初始化了CPU核数        CALL    runtime·schedinit(SB) // 内存分配器、栈、P、GC回收器等初始化
        // create a new goroutine to start program        MOVQ    $runtime·mainPC(SB), AX         //         PUSHQ   AX        PUSHQ   $0                      // arg size        CALL    runtime·newproc(SB) // 创建一个新的G来启动runtime.main        POPQ    AX        POPQ    AX
        // start this M        CALL    runtime·mstart(SB) // 启动M0,开始等待空闲G,正式进入调度循环
        MOVL    $0xf1, 0xf1  // crash        RET

// runtime/os_linux.go
func osinit() {  ncpu = getproccount()}

// runtime/proc.go
// 设置最大M数量sched.maxmcount = 10000
// 初始化当前M,即全局M0mcommoninit(_g_.m)
// 查看应该启动的P数量，默认为cpu core数.// 如果设置了环境变量GOMAXPROCS则以环境变量为准,最大不得超过_MaxGomaxprocs(1024)个procs := ncpuif n, ok := atoi32(gogetenv("GOMAXPROCS")); ok && n > 0 {  procs = n}if procs > _MaxGomaxprocs {  procs = _MaxGomaxprocs}// 调整P数量，此时由于是初始化阶段，所以P都是新建的if procresize(procs) != nil {  throw("unknown runnable goroutine during bootstrap")}

// runtime/proc.go
// The main goroutine.func main() {  ......    // 启动后台监控  systemstack(func() {    newm(sysmon, nil)  })
  ......}

// runtime/proc.go
func startm(_p_ *p, spinning bool) {  lock(&sched.lock)  if _p_ == nil {    // 从空闲P里获取一个    _p_ = pidleget()        ......  }  // 获取一个空闲的m  mp := mget()  unlock(&sched.lock)  // 如果没有空闲M，则new一个  if mp == nil {    var fn func()    if spinning {      // The caller incremented nmspinning, so set m.spinning in the new M.      fn = mspinning    }    newm(fn, _p_)    return  }    ......    // 唤醒M  notewakeup(&mp.park)}
func newm(fn func(), _p_ *p) {  // 创建一个M对象,且与P关联  mp := allocm(_p_, fn)  // 暂存P  mp.nextp.set(_p_)  mp.sigmask = initSigmask    ......    execLock.rlock() // Prevent process clone.  // 创建系统内核线程  newosproc(mp, unsafe.Pointer(mp.g0.stack.hi))  execLock.runlock()}
// runtime/os_linux.gofunc newosproc(mp *m, stk unsafe.Pointer) {  // Disable signals during clone, so that the new thread starts  // with signals disabled. It will enable them in minit.  var oset sigset  sigprocmask(_SIG_SETMASK, &sigset_all, &oset)  ret := clone(cloneFlags, stk, unsafe.Pointer(mp), unsafe.Pointer(mp.g0), unsafe.Pointer(funcPC(mstart)))  sigprocmask(_SIG_SETMASK, &oset, nil)}
func allocm(_p_ *p, fn func()) *m {  ......    mp := new(m)  mp.mstartfn = fn // 设置启动函数  mcommoninit(mp)  // 初始化m
  // 创建g0  // In case of cgo or Solaris, pthread_create will make us a stack.  // Windows and Plan 9 will layout sched stack on OS stack.  if iscgo || GOOS == "solaris" || GOOS == "windows" || GOOS == "plan9" {    mp.g0 = malg(-1)  } else {    mp.g0 = malg(8192 * sys.StackGuardMultiplier)  }  // 把新创建的g0与M做关联  mp.g0.m = mp
  ......    return mp}
func mstart() {  ......    mstart1()}
func mstart1() {
  ......    // 进入调度循环(阻塞不返回)  schedule()}

// runtime/proc.go
func schedule() {  _g_ := getg()
  // 进入gc MarkWorker 工作模式  if gp == nil && gcBlackenEnabled != 0 {    gp = gcController.findRunnableGCWorker(_g_.m.p.ptr())  }  if gp == nil {    // Check the global runnable queue once in a while to ensure fairness.    // Otherwise two goroutines can completely occupy the local runqueue    // by constantly respawning each other.    // 每处理n个任务就去全局队列获取G任务,确保公平    if _g_.m.p.ptr().schedtick%61 == 0 && sched.runqsize > 0 {      lock(&sched.lock)      gp = globrunqget(_g_.m.p.ptr(), 1)      unlock(&sched.lock)    }  }  // 从P本地获取  if gp == nil {    gp, inheritTime = runqget(_g_.m.p.ptr())    if gp != nil && _g_.m.spinning {      throw("schedule: spinning with local work")    }  }  // 从其它地方获取G,如果获取不到则沉睡M，并且阻塞在这里，直到M被再次使用  if gp == nil {    gp, inheritTime = findrunnable() // blocks until work is available  }
  ......    // 执行找到的G  execute(gp, inheritTime)}
// 从P本地获取一个可运行的Gfunc runqget(_p_ *p) (gp *g, inheritTime bool) {  // If there's a runnext, it's the next G to run.  // 优先从runnext里获取一个G，如果没有则从runq里获取  for {    next := _p_.runnext    if next == 0 {      break    }    if _p_.runnext.cas(next, 0) {      return next.ptr(), true    }  }
  // 从队头获取  for {    h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with other consumers    t := _p_.runqtail    if t == h {      return nil, false    }    gp := _p_.runq[h%uint32(len(_p_.runq))].ptr()    if atomic.Cas(&_p_.runqhead, h, h+1) { // cas-release, commits consume      return gp, false    }  }}
// 从其它地方获取Gfunc findrunnable() (gp *g, inheritTime bool) {  ......
  // 从本地队列获取  if gp, inheritTime := runqget(_p_); gp != nil {    return gp, inheritTime  }
  // 全局队列获取  if sched.runqsize != 0 {    lock(&sched.lock)    gp := globrunqget(_p_, 0)    unlock(&sched.lock)    if gp != nil {      return gp, false    }  }    // 从epoll里取  if netpollinited() && sched.lastpoll != 0 {    if gp := netpoll(false); gp != nil { // non-blocking      ......            return gp, false    }  }    ......    // 尝试4次从别的P偷  for i := 0; i < 4; i++ {    for enum := stealOrder.start(fastrand()); !enum.done(); enum.next() {      if sched.gcwaiting != 0 {        goto top      }      stealRunNextG := i > 2 // first look for ready queues with more than 1 g      // 在这里开始针对P进行偷取操作      if gp := runqsteal(_p_, allp[enum.position()], stealRunNextG); gp != nil {        return gp, false      }    }  }}
// 尝试从全局runq中获取G// 在"sched.runqsize/gomaxprocs + 1"、"max"、"len(_p_.runq))/2"三个数字中取最小的数字作为获取的G数量func globrunqget(_p_ *p, max int32) *g {  if sched.runqsize == 0 {    return nil  }
  n := sched.runqsize/gomaxprocs + 1  if n > sched.runqsize {    n = sched.runqsize  }  if max > 0 && n > max {    n = max  }  if n > int32(len(_p_.runq))/2 {    n = int32(len(_p_.runq)) / 2  }
  sched.runqsize -= n  if sched.runqsize == 0 {    sched.runqtail = 0  }
  gp := sched.runqhead.ptr()  sched.runqhead = gp.schedlink  n--  for ; n > 0; n-- {    gp1 := sched.runqhead.ptr()    sched.runqhead = gp1.schedlink    runqput(_p_, gp1, false) // 放到本地P里  }  return gp}

// runtime/proc.go
func runqput(_p_ *p, gp *g, next bool) {  if randomizeScheduler && next && fastrand()%2 == 0 {    next = false  }
  // 尝试把G添加到P的runnext节点，这里确保runnext只有一个G，如果之前已经有一个G则踢出来放到runq里  if next {  retryNext:    oldnext := _p_.runnext    if !_p_.runnext.cas(oldnext, guintptr(unsafe.Pointer(gp))) {      goto retryNext    }    if oldnext == 0 {      return    }    // 把老的g踢出来，在下面放到runq里    gp = oldnext.ptr()  }
retry:  // 如果_p_.runq队列不满，则放到队尾就结束了。  // 试想如果不放到队尾而放到队头里会怎样？如果频繁的创建G则可能后面的G总是不被执行，对后面的G不公平  h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with consumers  t := _p_.runqtail  if t-h < uint32(len(_p_.runq)) {    _p_.runq[t%uint32(len(_p_.runq))].set(gp)    atomic.Store(&_p_.runqtail, t+1) // store-release, makes the item available for consumption    return  }  //如果队列满了，尝试把G和当前P里的一部分runq放到全局队列  //因为操作全局需要加锁,所以名字里带个slow  if runqputslow(_p_, gp, h, t) {    return  }  // the queue is not full, now the put above must succeed  goto retry}
func runqputslow(_p_ *p, gp *g, h, t uint32) bool {  var batch [len(_p_.runq)/2 + 1]*g
  // First, grab a batch from local queue.  n := t - h  n = n / 2  if n != uint32(len(_p_.runq)/2) {    throw("runqputslow: queue is not full")  }  // 从runq头部开始取出一半的runq放到临时变量batch里  for i := uint32(0); i < n; i++ {    batch[i] = _p_.runq[(h+i)%uint32(len(_p_.runq))].ptr()  }  if !atomic.Cas(&_p_.runqhead, h, h+n) { // cas-release, commits consume    return false  }  // 把要put的g也放进batch去  batch[n] = gp
  if randomizeScheduler {    for i := uint32(1); i <= n; i++ {      j := fastrandn(i + 1)      batch[i], batch[j] = batch[j], batch[i]    }  }
  // 把取出来的一半runq组成链表  for i := uint32(0); i < n; i++ {    batch[i].schedlink.set(batch[i+1])  }
  // 将一半的runq放到global队列里,一次多转移一些省得转移频繁  lock(&sched.lock)  globrunqputbatch(batch[0], batch[n], int32(n+1))  unlock(&sched.lock)  return true}
func globrunqputbatch(ghead *g, gtail *g, n int32) {  gtail.schedlink = 0  if sched.runqtail != 0 {    sched.runqtail.ptr().schedlink.set(ghead)  } else {    sched.runqhead.set(ghead)  }  sched.runqtail.set(gtail)  sched.runqsize += n}

// runtime/proc.go
// 从其它地方获取Gfunc findrunnable() (gp *g, inheritTime bool) {  ......    // 尝试4次从别的P偷  for i := 0; i < 4; i++ {    for enum := stealOrder.start(fastrand()); !enum.done(); enum.next() {      if sched.gcwaiting != 0 {        goto top      }      stealRunNextG := i > 2 // first look for ready queues with more than 1 g      // 在这里开始针对P进行偷取操作      if gp := runqsteal(_p_, allp[enum.position()], stealRunNextG); gp != nil {        return gp, false      }    }  }}

// runtime/proc.go
// 偷取P2一半到本地运行队列，失败则返回nilfunc runqsteal(_p_, p2 *p, stealRunNextG bool) *g {  t := _p_.runqtail  n := runqgrab(p2, &_p_.runq, t, stealRunNextG)  if n == 0 {    return nil  }  n--  // 返回尾部的一个G  gp := _p_.runq[(t+n)%uint32(len(_p_.runq))].ptr()  if n == 0 {    return gp  }  h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with consumers  if t-h+n >= uint32(len(_p_.runq)) {    throw("runqsteal: runq overflow")  }  atomic.Store(&_p_.runqtail, t+n) // store-release, makes the item available for consumption  return gp}
// 从P里获取一半的G,放到batch里func runqgrab(_p_ *p, batch *[256]guintptr, batchHead uint32, stealRunNextG bool) uint32 {  for {    // 计算一半的数量    h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with other consumers    t := atomic.Load(&_p_.runqtail) // load-acquire, synchronize with the producer    n := t - h    n = n - n/2        ......        // 将偷到的任务转移到本地P队列里    for i := uint32(0); i < n; i++ {      g := _p_.runq[(h+i)%uint32(len(_p_.runq))]      batch[(batchHead+i)%uint32(len(batch))] = g    }    if atomic.Cas(&_p_.runqhead, h, h+n) { // cas-release, commits consume      return n    }  }}