Golang并发编程：万字长文说明白GPM协程调度模型原理（源码解析）

爱被打了一巴掌 2022-09-04 15:57 421阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3BicmxvdmVqYXZh_size_16_color_FFFFFF_t_70]

### 文章目录 ###

*   *  一、系统初始化
     *   *  核心流程
         *  osinit
         *  schedinit
         *  mcommoninit
         *  procresize
     *  二、协程创建
     *   *  核心流程
         *  newproc
         *  newproc1
         *  gfget
         *  runqput
         *  wakeup
         *  startm
         *  newm
     *  三、调度过程
     *   *  mstart
         *  schedule
         *  execute
         *  Goexit
     *  四、核心流程总结
     *   *  核心流程图

> 本文基于go1.15.8源码对GPM的系统初始化、协程创建及调度进行了分析，GPM调度器的相关结构体存在于runtime/runtime2.go中，而调度过程的源码主要集中在runtime/proc.go中。

> 关于GPM调度器的结构及容器参考：  
> https://arong.blog.csdn.net/article/details/117604004

## 一、系统初始化 ##

### 核心流程 ###

Go程序的引导程序启动进行系统初始化，核心步骤：

1.  调用osinit方法，根据操作系统CPU核数为ncpu字段赋值
2.  调用schedinit方法，调度器初始化
3.  调用newproc方法，创建G并加入队列中
4.  调用mstart方法，开始初始化M，关联P对G进行调度

在系统初始化阶段，osinit及schedinit都是为运行时系统及调度器系统进行初始化，而newproc及mstart则是进行第一轮调度，主要是用于执行g0及main函数的主goroutine，后续创建用户G时仍会调用newproc及mstart对其进行调度。

### osinit ###

Go程序在启动时会根据当前所处的操作系统环境，去调用特定操作系统包的osinit方法：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3BicmxvdmVqYXZh_size_16_color_FFFFFF_t_70 1]

osinit方法不仅对CPU核数进行判断，还会获取到当前操作系统下的页存大小：

func osinit() { 
       // 获取CPU核数
       ncpu = getncpu()
       if physPageSize == 0 { 
          physPageSize = getPageSize()
       }
    }

### schedinit ###

schedinit方法主要对M和P进行初始化，M最大数量被限制为10000，M实例会在需要时被初始化，并在不需要时进行销毁，而P数量默认为cpu核心数和参数GOMAXPROCS即为ncpu数目，并且会被直接初始化。

func schedinit() { 
            // 从TLS或者专用寄存器获取当前g的指针类型
            _g_ := getg()
            // 设置M最大的数量
            sched.maxmcount = 10000
            // 初始化栈的复用空间
            stackinit()
            // 初始化当前M
            mcommoninit(_g_.m)
            // P数目默认为cpu核心数和参数GOMAXPROCS即为ncpu数目
            procs := ncpu
            if n, ok := atoi32(gogetenv("GOMAXPROCS")); ok && n > 0 { 
                  procs = n
            }
            // 生成设定数量的P
            if procresize(procs) != nil { 
                    throw("unknown runnable goroutine during bootstrap")
            }
    }

### mcommoninit ###

mcommoninit对当前M进行初始化，实际上是把M与allm列表进行关联，将M设置为allm列表表头。

func mcommoninit(mp *m) { 
            // 获取当前的G
            _g_ := getg()
            // 判断M数量是否比maxmcount设定的要多
            checkmcount()
            // 将当前M设置为allm列表表头
            mp.alllink = allm
    }

### procresize ###

procresize可以初始化或更改指定数目的P，并将其添加到allp列表中，在调用procresize过程中，Go程序处于Stop the world状态，调度器也无法执行。

func procresize(nprocs int32) *p { 
       // 获取旧的P数目
       old := gomaxprocs
       if old < 0 || nprocs <= 0 { 
          throw("procresize: invalid arg")
       }
       now := nanotime()
       if sched.procresizetime != 0 { 
           // 累加所有P初始化花费的时间
          sched.totaltime += int64(old) * (now - sched.procresizetime)
       }
       // 为调度器标记P初始化的时间
       sched.procresizetime = now
       // 如果需要初始化的P数目大于旧P列表数目，则重新初始化allp列表
       // 首先进行旧P列表的复制形成nallp，再复制回去allp = nallp，这是扩容操作
       if nprocs > int32(len(allp)) { 
          // Synchronize with retake, which could be running
          // concurrently since it doesn't run on a P.
          lock(&allpLock)
          if nprocs <= int32(cap(allp)) { 
             allp = allp[:nprocs]
          } else { 
             nallp := make([]*p, nprocs)
             // Copy everything up to allp's cap so we
             // never lose old allocated Ps.
             copy(nallp, allp[:cap(allp)])
             allp = nallp
          }
          unlock(&allpLock)
       }
        // 对扩容过的allp中，未初始化的P进行初始化
       for i := old; i < nprocs; i++ { 
          pp := allp[i]
          if pp == nil { 
             pp = new(p)
          }
          pp.init(i)
          atomicstorep(unsafe.Pointer(&allp[i]), unsafe.Pointer(pp))
       }
       // 对当前G及其关联的P进行校验，若已经存在关联的P则继续运行，不存在则从allp列表中取表头
       // 关联到该G
       _g_ := getg()
       if _g_.m.p != 0 && _g_.m.p.ptr().id < nprocs { 
          _g_.m.p.ptr().status = _Prunning
          _g_.m.p.ptr().mcache.prepareForSweep()
       } else { 
          // release the current P and acquire allp[0].
          //
          // We must do this before destroying our current P
          // because p.destroy itself has write barriers, so we
          // need to do that from a valid P.
          if _g_.m.p != 0 { 
             if trace.enabled { 
                // Pretend that we were descheduled
                // and then scheduled again to keep
                // the trace sane.
                traceGoSched()
                traceProcStop(_g_.m.p.ptr())
             }
             _g_.m.p.ptr().m = 0
          }
          _g_.m.p = 0
          _g_.m.mcache = nil
          p := allp[0]
          p.m = 0
          p.status = _Pidle
          acquirep(p)
          if trace.enabled { 
             traceGoStart()
          }
       }
       // 如果是缩小了P列表，则需要销毁掉用不到的P
       for i := nprocs; i < old; i++ { 
          p := allp[i]
          p.destroy()
       }
       // 构建空闲P列表返回
       var runnablePs *p
       for i := nprocs - 1; i >= 0; i-- { 
          p := allp[i]
          if _g_.m.p.ptr() == p { 
             continue
          }
          p.status = _Pidle
          if runqempty(p) { 
             pidleput(p)
          } else { 
             p.m.set(mget())
             p.link.set(runnablePs)
             runnablePs = p
          }
       }
       stealOrder.reset(uint32(nprocs))
       var int32p *int32 = &gomaxprocs // make compiler check that gomaxprocs is an int32
       atomic.Store((*uint32)(unsafe.Pointer(int32p)), uint32(nprocs))
       return runnablePs
    }

## 二、协程创建 ##

### 核心流程 ###

使用go func()关键字开启协程时，实质上运行时系统会调用runtime/proc.go下的newproc函数获取到一个G并封装需要执行的func()，并且调用startm方法开启M并绑定P，其源码中的核心步骤如下：

1.  调用gfget方法，从本地P的gfree列表获取空闲G，不存在则从sched调度器的全局gfreeStack及gfreeNoStack列表获取空闲G，在此处将全局空闲G偷取都本地空闲P中
2.  若当前不存在空闲G，则调用malg方法申请一个G的栈内存（默认为2kb），创建新的G，并将其由Gidle状态置换为GDead状态
3.  将新生成的G放入全局G列表中（runtime.allgs）
4.  为该G绑定当前传入的func()函数
5.  将该G状态由GDead状态设置为GRunnable状态，并执行runqput方法，绑定当前P.runnext为该G，使其能够被运行，并将该G放入sched调度器的全局可运行列表runq中
6.  调用wakep方法，本质是调用startm方法唤醒或创建M，并关联P
7.  startm方法会对M进行一系列内存分配、内核线程绑定操作（newm），最终执行mstart方法开启GPM调度

### newproc ###

newproc方法获取了方法参数地址和当前的G，并通过getcallerpc方法获取开启该G的父G程序片段地址，通过g0调用newproc1执行创建或获取可用的G

func newproc(siz int32, fn *funcval) { 
            // 获取参数地址
            argp := add(unsafe.Pointer(&fn), sys.PtrSize)
            // 获取当前执行的G
            gp := getg()
            // 获取开启该G的父G程序片段地址
            pc := getcallerpc()
            systemstack(func() { 
                    // g0执行newproc1
                    newproc1(fn, (*uint8)(argp), siz, gp, pc)
            })
    }

### newproc1 ###

newproc1的作用是初始化G，分配内存空间，保存调用现场信息并关联func，最后生成一个可运行的G并放入队列中，唤醒M执行。

func newproc1(fn *funcval, argp *uint8, narg int32, callergp *g, callerpc uintptr) { 
       // 获取当前G
       _g_ := getg()
       // 要携带的func不能为nil
       if fn == nil { 
          _g_.m.throwing = -1 // do not dump full stacks
          throw("go of nil func value")
       }
       // 获取当前G的M，加锁禁止其他G占有该M
       acquirem()
       // 判断func参数是否过多导致栈内存溢出
       siz := narg
       siz = (siz + 7) &^ 7
       if siz >= _StackMin-4*sys.RegSize-sys.RegSize { 
          throw("newproc: function arguments too large for new goroutine")
       }
       // 查找是否有可用的G
       _p_ := _g_.m.p.ptr()
       newg := gfget(_p_)
       if newg == nil { 
          // 不存在可用的G则创建一个新的G，栈内存为2kb
          newg = malg(_StackMin)
          // CAS更新该G的状态为Gdead
          casgstatus(newg, _Gidle, _Gdead)
          // 将新的G添加到runtime.allg列表末尾
          allgadd(newg)
       }
       if newg.stack.hi == 0 { 
          throw("newproc1: newg missing stack")
       }
       if readgstatus(newg) != _Gdead { 
          throw("newproc1: new g is not Gdead")
       }
       totalSize := 4*sys.RegSize + uintptr(siz) + sys.MinFrameSize // extra space in case of reads slightly beyond frame
       totalSize += -totalSize & (sys.SpAlign - 1)                  // align to spAlign
       sp := newg.stack.hi - totalSize
       spArg := sp
       if usesLR { 
          // caller's LR
          *(*uintptr)(unsafe.Pointer(sp)) = 0
          prepGoExitFrame(sp)
          spArg += sys.MinFrameSize
       }
       // 如果func需要携带参数则封装进入栈中
       if narg > 0 { 
          memmove(unsafe.Pointer(spArg), unsafe.Pointer(argp), uintptr(narg))
          // This is a stack-to-stack copy. If write barriers
          // are enabled and the source stack is grey (the
          // destination is always black), then perform a
          // barrier copy. We do this *after* the memmove
          // because the destination stack may have garbage on
          // it.
          if writeBarrier.needed && !_g_.m.curg.gcscandone { 
             f := findfunc(fn.fn)
             stkmap := (*stackmap)(funcdata(f, _FUNCDATA_ArgsPointerMaps))
             if stkmap.nbit > 0 { 
                // We're in the prologue, so it's always stack map index 0.
                bv := stackmapdata(stkmap, 0)
                bulkBarrierBitmap(spArg, spArg, uintptr(bv.n)*sys.PtrSize, 0, bv.bytedata)
             }
          }
       }
       // 保留调用现场信息、相关栈顶指针信息
       memclrNoHeapPointers(unsafe.Pointer(&newg.sched), unsafe.Sizeof(newg.sched))
       newg.sched.sp = sp
       newg.stktopsp = sp
       // 保存goexit指令指针和关联G，使G在执行完成func后退出
       newg.sched.pc = funcPC(goexit) + sys.PCQuantum // +PCQuantum so that previous instruction is in same function
       newg.sched.g = guintptr(unsafe.Pointer(newg))
       // 保存func指针信息，使其能够执行func
       gostartcallfn(&newg.sched, fn)
       newg.gopc = callerpc
       newg.ancestors = saveAncestors(callergp)
       newg.startpc = fn.fn
       if _g_.m.curg != nil { 
          newg.labels = _g_.m.curg.labels
       }
       if isSystemGoroutine(newg, false) { 
          atomic.Xadd(&sched.ngsys, +1)
       }
       newg.gcscanvalid = false
       // G初始化完成，将其转化为G runnable状态使其能够被执行
       casgstatus(newg, _Gdead, _Grunnable)
    
       if _p_.goidcache == _p_.goidcacheend { 
          // Sched.goidgen is the last allocated id,
          // this batch must be [sched.goidgen+1, sched.goidgen+GoidCacheBatch].
          // At startup sched.goidgen=0, so main goroutine receives goid=1.
          _p_.goidcache = atomic.Xadd64(&sched.goidgen, _GoidCacheBatch)
          _p_.goidcache -= _GoidCacheBatch - 1
          _p_.goidcacheend = _p_.goidcache + _GoidCacheBatch
       }
       newg.goid = int64(_p_.goidcache)
       _p_.goidcache++
       if raceenabled { 
          newg.racectx = racegostart(callerpc)
       }
       if trace.enabled { 
          traceGoCreate(newg, newg.startpc)
       }
       // 将G放入到P的本地队列或全局队列
       runqput(_p_, newg, true)
       // 判断是否有空闲P，且是否需要唤醒一个M来执行G
       if atomic.Load(&sched.npidle) != 0 && atomic.Load(&sched.nmspinning) == 0 && mainStarted { 
          wakep()
       }
       // 执行完成需要释放M
       releasem(_g_.m)
    }

### gfget ###

gfget查看当前P是否有空闲的G，没有则去全局的freeg队列中查找，并检查获取到的G栈内存是否为空，为空则再次为该G分配栈内存。

func gfget(_p_ *p) *g { 
    retry:
       if _p_.gFree.empty() && (!sched.gFree.stack.empty() || !sched.gFree.noStack.empty()) { 
          lock(&sched.gFree.lock)
          // Move a batch of free Gs to the P.
          for _p_.gFree.n < 32 { 
             // Prefer Gs with stacks.
             gp := sched.gFree.stack.pop()
             if gp == nil { 
                gp = sched.gFree.noStack.pop()
                if gp == nil { 
                   break
                }
             }
             sched.gFree.n--
             _p_.gFree.push(gp)
             _p_.gFree.n++
          }
          unlock(&sched.gFree.lock)
          goto retry
       }
       gp := _p_.gFree.pop()
       if gp == nil { 
          return nil
       }
       _p_.gFree.n--
       if gp.stack.lo == 0 { 
          // Stack was deallocated in gfput. Allocate a new one.
          systemstack(func() { 
             gp.stack = stackalloc(_FixedStack)
          })
          gp.stackguard0 = gp.stack.lo + _StackGuard
       } else { 
          if raceenabled { 
             racemalloc(unsafe.Pointer(gp.stack.lo), gp.stack.hi-gp.stack.lo)
          }
          if msanenabled { 
             msanmalloc(unsafe.Pointer(gp.stack.lo), gp.stack.hi-gp.stack.lo)
          }
       }
       return gp
    }

### runqput ###

runqput方法会把新的G（状态被置为runnable）放入P的本地队列或者p.runnext进行预联，如果P的本地队列过长，则把G放到全局队列(runqputslow)：

func runqput(_p_ *p, gp *g, next bool) { 
            if randomizeScheduler && next && fastrand()%2 == 0 { 
                    next = false
            }
            // 如果next为true，则放入到p.runnext里面，并把原先runnext的G交换出来
            if next { 
            retryNext:
                    oldnext := _p_.runnext
                    if !_p_.runnext.cas(oldnext, guintptr(unsafe.Pointer(gp))) { 
                            goto retryNext
                    }
                    if oldnext == 0 { 
                            return
                    }
                    // Kick the old runnext out to the regular run queue.
                    gp = oldnext.ptr()
            }
    
    retry:
            h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with consumers
            t := _p_.runqtail
            // runq是一个长度为256的数组，是P的本地可执行G队列，如果可以容纳则添加到该队列中
            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
            }
            // runq无法容纳G，放到全局队列
            if runqputslow(_p_, gp, h, t) { 
                    return
            }
            // the queue is not full, now the put above must succeed
            goto retry
    }

### wakeup ###

当将G放入到可执行队列之后，尝试执行wakeup方法唤醒一个M来执行G，注意，并不一定会执行到该新建的G，而是看具体的情况。

func wakep() { 
       // 一次只需要唤醒1个M
       if !atomic.Cas(&sched.nmspinning, 0, 1) { 
          return
       }
       // 调用startm
       startm(nil, true)
    }

### startm ###

startm的作用是获取空闲的或者新建一个M，并且获取空闲的P，将该P关联为m.nextp，此时形成M-P关联后，就可以通过P获取到G，开始执行相应的func：

func startm(_p_ *p, spinning bool) { 
            lock(&sched.lock)
            if _p_ == nil { 
                    // 如果没有指定P, 则从sched.pidle获取空闲P
                    _p_ = pidleget()
                    if _p_ == nil { 
                            unlock(&sched.lock)
                            // 如果没有获取到P，nmspinning计数器-1
                            if spinning { 
                                    // The caller incremented nmspinning, but there are no idle Ps,
                                    // so it's okay to just undo the increment and give up.
                                    if int32(atomic.Xadd(&sched.nmspinning, -1)) < 0 { 
                                            throw("startm: negative nmspinning")
                                    }
                            }
                            return
                    }
            }
            // 尝试从 sched.midle获取一个空闲M
            mp := mget()
            unlock(&sched.lock)
            if mp == nil { 
                    // 获取不到空闲的M，则创建一个 mspining = true的M，并将P绑定到M上，直接返回,下次唤醒时则有可用的M了
                    var fn func()
                    if spinning { 
                            // The caller incremented nmspinning, so set m.spinning in the new M.
                            fn = mspinning
                    }
                    // 创建内核线程与M关联，并且使M绑定该P及fn
                    newm(fn, _p_)
                    return
            }
            // 当前M必须为自旋状态
            if mp.spinning { 
                    throw("startm: m is spinning")
            }
            // 当前M必须关联P
            if mp.nextp != 0 { 
                    throw("startm: m has p")
            }
            // 当前关联的P必须处于空闲状态，runq队列为空
            if spinning && !runqempty(_p_) { 
                    throw("startm: p has runnable gs")
            }
            // 设定该M为自旋状态并绑定P
            mp.spinning = spinning
            mp.nextp.set(_p_)
            // 唤醒M
            notewakeup(&mp.park)
    }

### newm ###

newm方法真正地创建内核线程与M的关联，并且将P和func绑定在M上,并且执行mstart开启真正的调度过程：

func newm(fn func(), _p_ *p) { 
            // 新建M绑定P及fn
            mp := allocm(_p_, fn)
            // M绑定指定P
            mp.nextp.set(_p_)
            // 创建内核线程与M绑定
            newm1(mp)
    }
    func newm1(mp *m) { 
            if iscgo { 
                    var ts cgothreadstart
                    if _cgo_thread_start == nil { 
                            throw("_cgo_thread_start missing")
                    }
                    ts.g.set(mp.g0)
                    ts.tls = (*uint64)(unsafe.Pointer(&mp.tls[0]))
                    // 绑定该内核线程启动调用mstart方法
                    ts.fn = unsafe.Pointer(funcPC(mstart))
                    if msanenabled { 
                            msanwrite(unsafe.Pointer(&ts), unsafe.Sizeof(ts))
                    }
                    execLock.rlock() // Prevent process clone.
                    asmcgocall(_cgo_thread_start, unsafe.Pointer(&ts))
                    execLock.runlock()
                    return
            }
            execLock.rlock() // Prevent process clone.
            newosproc(mp)
            execLock.runlock()
    }
    
    func newosproc(mp *m) { 
            stk := unsafe.Pointer(mp.g0.stack.hi)
            // Initialize an attribute object.
            var attr pthreadattr
            var err int32
            err = pthread_attr_init(&attr)
    
            // Finally, create the thread. It starts at mstart_stub, which does some low-level
            // setup and then calls mstart.
            var oset sigset
            sigprocmask(_SIG_SETMASK, &sigset_all, &oset)
            // 创建内核线程，并传入启动启动函数 mstart_stub， mstart_stub 之后调用mstart
            err = pthread_create(&attr, funcPC(mstart_stub), unsafe.Pointer(mp))
            sigprocmask(_SIG_SETMASK, &oset, nil)
            if err != 0 { 
                    write(2, unsafe.Pointer(&failthreadcreate[0]), int32(len(failthreadcreate)))
                    exit(1)
            }
    }

## 三、调度过程 ##

经过系统初始化以及go func创建协程后，G会封装好需要执行的func位于p.runnext、p.runq或sched.runq这三个队列之一，接下来就是调用mstart开启真正的调度过程（startm是对M进行初始化操作，而不是进行调度）。需要注意的是，调度程序会在多个M实例中进行，并且会在以下场景中进行触发schedule函数进行调度：

*  Go程序初始化
 *  使用go关键字开启协程
 *  执行runtime.Gosched，当前G暂停运行让给其他G运行机会
 *  执行runtime.Goexit，当前G结束运行程序
 *  G状态转化时

### mstart ###

mstart在Go程序初始化或使用go关键字开启协程时被调用，本质上也是调用schedule函数进行调度：

func mstart() { 
       _g_ := getg()
    
       osStack := _g_.stack.lo == 0
       if osStack { 
          // Initialize stack bounds from system stack.
          // Cgo may have left stack size in stack.hi.
          // minit may update the stack bounds.
          size := _g_.stack.hi
          if size == 0 { 
             size = 8192 * sys.StackGuardMultiplier
          }
          _g_.stack.hi = uintptr(noescape(unsafe.Pointer(&size)))
          _g_.stack.lo = _g_.stack.hi - size + 1024
       }
       // Initialize stack guard so that we can start calling regular
       // Go code.
       _g_.stackguard0 = _g_.stack.lo + _StackGuard
       // This is the g0, so we can also call go:systemstack
       // functions, which check stackguard1.
       _g_.stackguard1 = _g_.stackguard0
       mstart1()
    
       // Exit this thread.
       if GOOS == "windows" || GOOS == "solaris" || GOOS == "illumos" || GOOS == "plan9" || GOOS == "darwin" || GOOS == "aix" { 
          // Windows, Solaris, illumos, Darwin, AIX and Plan 9 always system-allocate
          // the stack, but put it in _g_.stack before mstart,
          // so the logic above hasn't set osStack yet.
          osStack = true
       }
       mexit(osStack)
    }
    
    func mstart1() { 
       _g_ := getg()
       // 保存调用信息
       save(getcallerpc(), getcallersp())
       asminit()
       // 对M进行初始化
       minit()
    
       // 执行m0的任务
       if _g_.m == &m0 { 
          mstartm0()
       }
       // 如果该M存在startfn，则执行startfn
       if fn := _g_.m.mstartfn; fn != nil { 
          fn()
       }
       // 开始进入调度循环
       schedule()
    }

### schedule ###

schedule函数又称为一轮调度函数，它的作用是全力查找可执行的G并最后执行execute函数运行，在恶执行完用户代码后，会执行go exit命令， 会跳转到runtime.Goexit0函数，该函数最后仍调用了schedule函数，形成了一个调度闭环操作。

一轮调度的执行核心流程如下，需要注意的是当当前M已经被特定的G锁定（lockedg）时，则需要停止已锁定的当前M，待该锁定的G达到可运行状态时对该M进行唤醒，这是因为M与G锁定后，该G会占用M相关栈空间，所以不能够使用无关的M来运行已经被其他M锁定的G。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3BicmxvdmVqYXZh_size_16_color_FFFFFF_t_70 2]

func schedule() { 
            _g_ := getg()
    
            if _g_.m.locks != 0 { 
                    throw("schedule: holding locks")
            }
            // 当前M有lockg，停止执行当前M
            if _g_.m.lockedg != 0 { 
                    // 停止当前M，直到其绑定的G处于可运行状态
                    stoplockedm()
                    // 开始执行绑定的G
                    execute(_g_.m.lockedg.ptr(), false) // Never returns.
            }
    
    top:
            // GC进行中，停止M
            if sched.gcwaiting != 0 { 
                    gcstopm()
                    // 跳转到top循环，再次寻找可执行G
                    goto top
            }
            
            // 需要执行的G
            var gp *g
            // 继承时间片
            var inheritTime bool
            
            if gp == nil { 
                    // 每隔61次从全局队列上获取可执行的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)
                    }
            }
            
            if gp == nil { 
                    // 从P的本地队列获取可执行的G
                    gp, inheritTime = runqget(_g_.m.p.ptr())
                    if gp != nil && _g_.m.spinning { 
                            throw("schedule: spinning with local work")
                    }
            }
            
            if gp == nil { 
                    // 如果从全局队列以及P本地队列都无法获取到可执行的P，开始全力查找可执行的G
                    gp, inheritTime = findrunnable()        
            }
    
            if _g_.m.spinning { 
                    // 如果M是自旋状态，取消自旋
                    resetspinning()
            }
            if gp.lockedm != 0 { 
                    // 如果当前G已经被其他M锁定，则唤醒与该G锁定的M
                    startlockedm(gp)
                    // 跳转到top循环，再次寻找可执行G
                    goto top
            }
            // 开始执行这个G
            execute(gp, inheritTime)
    }

### execute ###

开始真正执行G，在执行完G携带的func后，会执行runtime.Goexit函数：

func execute(gp *g, inheritTime bool) { 
            _g_ := getg()
            // 更改G的状态为running
            casgstatus(gp, _Grunnable, _Grunning)
            gp.waitsince = 0
            // 不允许抢占
            gp.preempt = false
            gp.stackguard0 = gp.stack.lo + _StackGuard
            if !inheritTime { 
                // 每调度一次则计数+1
                _g_.m.p.ptr().schedtick++
            }
            // 关联M->G
            _g_.m.curg = gp
            // 关联G->M
            gp.m = _g_.m
            // 开始执行G的代码
            gogo(&gp.sched)
    }

### Goexit ###

最后一个环节，执行完G的代码后进行程序退出，执行链路是gogo->runtime.Goexit->runtime.goexit1->runtime.goexit0,在goexit0函数中会真正地把已经执行完的G状态变更为dead，然后再次执行schedule函数，进入下一个调度循环之中,这就是整个GPM调度器的执行核心流程了。

func Goexit() { 
       // 忽略
       goexit1()
    }
    func goexit1() { 
        // 忽略
       // 调用goexit0
       mcall(goexit0)
    }
    func goexit0(gp *g) { 
       _g_ := getg()
        // 将运行完成G状态转换为dead
       casgstatus(gp, _Grunning, _Gdead)
       // 忽略
       schedule()
    }

## 四、核心流程总结 ##

根据上面的源码可以梳理一个核心的协程调度流程图，总结下整个调度过程，关键的步骤就是Go程序启动时对调度器的初始化，会把M和P存储到GPM容器中，然后在go func启动协程时，会进行G的获取及存储，封装需要执行的func，最后调用schedule函数启动调度器，调度过程是一个闭环操作，该过程存在于各个M实例之中，所以每个M都可以独立进行调度，启动调度程序后会寻找可执行的G，最后执行完G的程序后，会执行runtime.Goexit退出协程程序，此时会跳转到schedule函数继续执行调度操作，形成闭环。

### 核心流程图 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3BicmxvdmVqYXZh_size_16_color_FFFFFF_t_70 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3BicmxvdmVqYXZh_size_16_color_FFFFFF_t_70]: /images/20220829/cd9b59070a47411ea27e7592e5cf8c3e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3BicmxvdmVqYXZh_size_16_color_FFFFFF_t_70 1]: /images/20220829/dc2beed6a74b4d4eaea30dea103cbccb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3BicmxvdmVqYXZh_size_16_color_FFFFFF_t_70 2]: /images/20220829/59024f41376f4fa3b289a93b9a746193.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3BicmxvdmVqYXZh_size_16_color_FFFFFF_t_70 3]: /images/20220829/d5543fff29794687899b56ade064f5fe.png