Golang协程调度探究

type m struct {
    g0      *g     // 带有调度栈的goroutine
    gsignal       *g         // 处理信号的goroutine
    tls           [6]uintptr // thread-local storage
    mstartfn      func()
    curg          *g       // 当前运行的goroutine
    caughtsig     guintptr
    p             puintptr // 关联p和执行的go代码
    nextp         puintptr
    id            int32
    mallocing     int32 // 状态
    spinning      bool // m是否out of work
    blocked       bool // m是否被阻塞
    inwb          bool // m是否在执行写屏蔽
    .........
}

type gobuf struct {
    // 保存CPU的rsp的寄存器值
	sp   uintptr
    // 保存CPU的rip寄存器值
	pc   uintptr
    // 记录gobuf 属于哪个 协程goroutine
	g    guintptr
	ctxt unsafe.Pointer
    // 保存调用的返回值
	ret  uintptr
	lr   uintptr
    // 保存rbp寄存器的值
	bp   uintptr // for framepointer-enabled architectures
}

type m struct {
    ......
    tls           [tlsSlots]uintptr // thread-local storage (for x86 extern register)
    ......
}

func schedule() {
    // 获取到当前的goroutine
	_g_ := getg()
        ...


	var gp *g
	var inheritTime bool

	// 检查goroutine是否需要唤醒
	tryWakeP := false
	if trace.enabled || trace.shutdown {
		gp = traceReader()
		if gp != nil {
			casgstatus(gp, _Gwaiting, _Grunnable)
			traceGoUnpark(gp, 0)
			tryWakeP = true
		}
	}
	if gp == nil && gcBlackenEnabled != 0 {
		gp = gcController.findRunnableGCWorker(_g_.m.p.ptr())
		if gp != nil {
			tryWakeP = true
		}
	}
	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.
		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 {
		gp, inheritTime = runqget(_g_.m.p.ptr())
		// We can see gp != nil here even if the M is spinning,
		// if checkTimers added a local goroutine via goready.
	}
	if gp == nil {
		gp, inheritTime = findrunnable() // blocks until work is available
	}
}

type p struct {
	id          int32
	status      uint32 // one of pidle/prunning/...
        .....

	// Queue of runnable goroutines. Accessed without lock.
        // 可运行的协程队列
	runqhead uint32
	runqtail uint32
        // 256个局部运行队列
	runq     [256]guintptr
        // 指向下一个要执行的协程
	runnext guintptr
        .....
}

// 有一个头，看起来类似next的链表
type gQueue struct {
	head guintptr
	tail guintptr
}

// 这里是全局列表的代码
type schedt struct {
	// accessed atomically. keep at top to ensure alignment on 32-bit systems.
	goidgen   uint64
	// Global runnable queue.
        // 全局队列源码
	runq     gQueue
	runqsize int32
        ....
}

当P中执行了61次调度之后，就必须从全局队列获取一个G，到当前P中执行

	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.
		// 如果可以被61整除并且全局队列的数量大于0
		if _g_.m.p.ptr().schedtick%61 == 0 && sched.runqsize > 0 {
			lock(&sched.lock)
			// 获取一个g
			gp = globrunqget(_g_.m.p.ptr(), 1)
			unlock(&sched.lock)
		}
	}

func runqget(_p_ *p) (gp *g, inheritTime bool) {
	// 获取下一个可用的G
	next := _p_.runnext
	// 如果下一个next地址不等于0并且cas成功，直接返回一个G
	// 如果下一个next地址不等于0并且cas失败，那么它只能被另一个P抢夺
	// 其他P可以设置为0，当前只有P可以设置非0
	// 只有CAS失败，则不需要重试
	if next != 0 && _p_.runnext.cas(next, 0) {
		return next.ptr(), true
	}

	for {
		// 获取头部
		h := atomic.LoadAcq(&_p_.runqhead) // 加载，获取，和其他P同步
		// 获取尾部
		t := _p_.runqtail
		// 如果头尾相等，那么没有协程可以运行
		if t == h {
			return nil, false
		}
		// 访问加锁
		gp := _p_.runq[h%uint32(len(_p_.runq))].ptr()
		if atomic.CasRel(&_p_.runqhead, h, h+1) { // 推送执行了多少G
			return gp, false
		}
	}
}

// Try get a batch of G's from the global runnable queue.
// sched.lock must be held.
func globrunqget(_p_ *p, max int32) *g {
	assertLockHeld(&sched.lock)

	if sched.runqsize == 0 {
		return nil
	}

	// 平均分配
	// n就是每个p获取的g的数量
	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
	}

	// 全局队列 - n
	sched.runqsize -= n

	// gp = 队伍的最后一个
	gp := sched.runq.pop()
	// 递减
	n--
	// 如果 n > 0
	for ; n > 0; n-- {
		gp1 := sched.runq.pop()
		// 上传
		runqput(_p_, gp1, false)
	}
	return gp
}

//从任何P中进行窃取的函数
func stealWork(now int64) (gp *g, inheritTime bool, rnow, pollUntil int64, newWork bool) {
	//获取到G
	pp := getg().m.p.ptr()

	ranTimer := false

	//重复4次操作
	const stealTries = 4
	for i := 0; i < stealTries; i++ {
		stealTimersOrRunNextG := i == stealTries-1

		// 二层循环，找到可窃取的P马上返回
		for enum := stealOrder.start(fastrand()); !enum.done(); enum.next() {
			if sched.gcwaiting != 0 {
				// GC work may be available.
				return nil, false, now, pollUntil, true
			}
			p2 := allp[enum.position()]
			if pp == p2 {
				continue
			}

			// 从P2里面进行窃取
			if stealTimersOrRunNextG && timerpMask.read(enum.position()) {
				tnow, w, ran := checkTimers(p2, now)
				now = tnow
				if w != 0 && (pollUntil == 0 || w < pollUntil) {
					pollUntil = w
				}
				if ran {
					// 检查现在是否要运行的本地G
					if gp, inheritTime := runqget(pp); gp != nil {
						return gp, inheritTime, now, pollUntil, ranTimer
					}
					ranTimer = true
				}
			}

			//如果P2空闲，不进行窃取
			if !idlepMask.read(enum.position()) {
				if gp := runqsteal(pp, p2, stealTimersOrRunNextG); gp != nil {
					return gp, false, now, pollUntil, ranTimer
				}
			}
		}
	}

	// 没有发现的协程G，持续等待或者推出
	return nil, false, now, pollUntil, ranTimer
}

	const stealTries = 4
	for i := 0; i < stealTries; i++ {
		stealTimersOrRunNextG := i == stealTries-1

		// 二层循环，找到可窃取的P马上返回
		for enum := stealOrder.start(fastrand()); !enum.done(); enum.next() {
			if sched.gcwaiting != 0 {
				// GC work may be available
	.....

func runqgrab(_p_ *p, batch *[256]guintptr, batchHead uint32, stealRunNextG bool) uint32 {
	for {
		h := atomic.LoadAcq(&_p_.runqhead) // load-acquire, synchronize with other consumers
		t := atomic.LoadAcq(&_p_.runqtail) // load-acquire, synchronize with the producer
		n := t - h
		// 偷取一半的G个数
		n = n - n/2
		.....
		// 放入自己的队列
		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.CasRel(&_p_.runqhead, h, h+n) { // cas-release, commits consume
			return n
		}
	}
}

func goschedImpl(gp *g) {
	// 判断g的状态
	status := readgstatus(gp)
	// 如果g不是_Grunning状态
	if status&^_Gscan != _Grunning {
		// 获取g的详细信息
		dumpgstatus(gp)
		// 输出提示
		throw("bad g status")
	}
	// 取消g和m之间的绑定关系
	casgstatus(gp, _Grunning, _Grunnable)
	dropg()
	// 加锁g
	lock(&sched.lock)
	// g放入全局运行队列
	globrunqput(gp)
	// 关闭锁
	unlock(&sched.lock)
	// 新一轮调度
	schedule()
}

func gopark(unlockf func(*g, unsafe.Pointer) bool, lock unsafe.Pointer, reason waitReason, traceEv byte, traceskip int) {
	if reason != waitReasonSleep {
		checkTimeouts() // timeouts may expire while two goroutines keep the scheduler busy
	}
	mp := acquirem()
	gp := mp.curg
	status := readgstatus(gp)
	if status != _Grunning && status != _Gscanrunning {
		throw("gopark: bad g status")
	}
	mp.waitlock = lock
	mp.waitunlockf = unlockf
	gp.waitreason = reason
	mp.waittraceev = traceEv
	mp.waittraceskip = traceskip
	releasem(mp)
	//执行被动调度
	mcall(park_m)
}

func park_m(gp *g) {
	_g_ := getg()

	if trace.enabled {
		traceGoPark(_g_.m.waittraceev, _g_.m.waittraceskip)
	}

	// 解除G和M的关系
	casgstatus(gp, _Grunning, _Gwaiting)
	dropg()

	// 判定执行被动调度的原因
	if fn := _g_.m.waitunlockf; fn != nil {
		ok := fn(gp, _g_.m.waitlock)
		// 执行waitlock
		_g_.m.waitunlockf = nil
		_g_.m.waitlock = nil
		if !ok {
			if trace.enabled {
				traceGoUnpark(gp, 2)
			}
			casgstatus(gp, _Gwaiting, _Grunnable)
			execute(gp, true) // Schedule it back, never returns.
		}
	}
	// 重新调度
	schedule()
}

func retake(now int64) uint32 {
	n := 0
	lock(&allpLock)
	// 重新获取allp
	for i := 0; i < len(allp); i++ {
		_p_ := allp[i]
		if _p_ == nil {
			// This can happen if procresize has grown
			// allp but not yet created new Ps.
			continue
		}
		pd := &_p_.sysmontick
		s := _p_.status
		sysretake := false
		if s == _Prunning || s == _Psyscall {
			// Preempt G if it's running for too long.
			t := int64(_p_.schedtick)
			if int64(pd.schedtick) != t {
				pd.schedtick = uint32(t)
				pd.schedwhen = now
			} else if pd.schedwhen+forcePreemptNS <= now {
				preemptone(_p_)
				// 系统调用 没有连接到P的M
				sysretake = true
			}
		}
		if s == _Psyscall {
			// 如果存在1个sysmon，从系统调用中获取P
			t := int64(_p_.syscalltick)
			if !sysretake && int64(pd.syscalltick) != t {
				pd.syscalltick = uint32(t)
				pd.syscallwhen = now
				continue
			}
			// 防止进入深度睡眠无法huan'x
			if runqempty(_p_) && atomic.Load(&sched.nmspinning)+atomic.Load(&sched.npidle) > 0 && pd.syscallwhen+10*1000*1000 > now {
				continue
			}
			unlock(&allpLock)
			// 较少空闲锁定M的数量
			// 增加nmidle报告死锁
			incidlelocked(-1)
			if atomic.Cas(&_p_.status, s, _Pidle) {
				if trace.enabled {
					traceGoSysBlock(_p_)
					traceProcStop(_p_)
				}
				n++
				_p_.syscalltick++
				handoffp(_p_)
			}
			incidlelocked(1)
			lock(&allpLock)
		}
	}
	unlock(&allpLock)
	return uint32(n)
}