深入理解Go调度器：如何初始化P（四）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

// src/runtime/runtime2.go#L609
type p struct {
	id          int32
	status      uint32 // P 的状态
	link        puintptr
	schedtick   uint32     // 被调度的次数，每调度一次递增
	syscalltick uint32     // 执行系统调用的次数，每执行一次递增
	sysmontick  sysmontick // sysmon 最近一次运行的时间
	m           muintptr   // P 所关联的 M，如果 P 为空闲状态，则为 nil
	mcache      *mcache    // 小对象缓存，可以无锁访问
	pcache      pageCache  // 页缓存，也可以无锁访问
	raceprocctx uintptr

	deferpool    []*_defer // pool of available defer structs (see panic.go)
	deferpoolbuf [32]*_defer

	// Cache of goroutine ids, amortizes accesses to runtime·sched.goidgen.
  // goroutine ids 缓存
	goidcache    uint64
	goidcacheend uint64

	// Queue of runnable goroutines. Accessed without lock.
  // P 的本地可运行队列，LRQ，可无锁方位
  // 由 runqhead, runqtail, runq 三者组成的无锁循环队列
  // 最多存放 256 个可运行的g
	runqhead uint32
	runqtail uint32
	runq     [256]guintptr
	// runnext, if non-nil, is a runnable G that was ready'd by
	// the current G and should be run next instead of what's in
	// runq if there's time remaining in the running G's time
	// slice. It will inherit the time left in the current time
	// slice. If a set of goroutines is locked in a
	// communicate-and-wait pattern, this schedules that set as a
	// unit and eliminates the (potentially large) scheduling
	// latency that otherwise arises from adding the ready'd
	// goroutines to the end of the run queue.
	//
	// Note that while other P's may atomically CAS this to zero,
	// only the owner P can CAS it to a valid G.
  // 待运行的 g
  // 如果不为空，且当前时间片还有时间的话，优先运行runnext
  // 其他 P 可以将其设置为 0，但只有自身的P 可以将其设置为非零
	runnext guintptr

	// Available G's (status == Gdead)
  // 已结束运行的 G
	gFree struct {
		gList
		n int32
	}

	sudogcache []*sudog
	sudogbuf   [128]*sudog

	// Cache of mspan objects from the heap.
  // 来自堆上的 mspan 缓存
	mspancache struct {
		// We need an explicit length here because this field is used
		// in allocation codepaths where write barriers are not allowed,
		// and eliminating the write barrier/keeping it eliminated from
		// slice updates is tricky, moreso than just managing the length
		// ourselves.
		len int
		buf [128]*mspan
	}
	
  // gc 相关字段
	gcMarkWorkerMode gcMarkWorkerMode
	gcMarkWorkerStartTime int64
	gcw gcWork

  ......

	// preempt is set to indicate that this P should be enter the
	// scheduler ASAP (regardless of what G is running on it).
  // 抢占标记
	preempt bool
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

// src/runtime/runtime2.go#L106
const (
	// P status

	// _Pidle means a P is not being used to run user code or the
	// scheduler. Typically, it's on the idle P list and available
	// to the scheduler, but it may just be transitioning between
	// other states.
	//
	// The P is owned by the idle list or by whatever is
	// transitioning its state. Its run queue is empty.
	_Pidle = iota

	// _Prunning means a P is owned by an M and is being used to
	// run user code or the scheduler. Only the M that owns this P
	// is allowed to change the P's status from _Prunning. The M
	// may transition the P to _Pidle (if it has no more work to
	// do), _Psyscall (when entering a syscall), or _Pgcstop (to
	// halt for the GC). The M may also hand ownership of the P
	// off directly to another M (e.g., to schedule a locked G).
	_Prunning

	// _Psyscall means a P is not running user code. It has
	// affinity to an M in a syscall but is not owned by it and
	// may be stolen by another M. This is similar to _Pidle but
	// uses lightweight transitions and maintains M affinity.
	//
	// Leaving _Psyscall must be done with a CAS, either to steal
	// or retake the P. Note that there's an ABA hazard: even if
	// an M successfully CASes its original P back to _Prunning
	// after a syscall, it must understand the P may have been
	// used by another M in the interim.
	_Psyscall

	// _Pgcstop means a P is halted for STW and owned by the M
	// that stopped the world. The M that stopped the world
	// continues to use its P, even in _Pgcstop. Transitioning
	// from _Prunning to _Pgcstop causes an M to release its P and
	// park.
	//
	// The P retains its run queue and startTheWorld will restart
	// the scheduler on Ps with non-empty run queues.
	_Pgcstop

	// _Pdead means a P is no longer used (GOMAXPROCS shrank). We
	// reuse Ps if GOMAXPROCS increases. A dead P is mostly
	// stripped of its resources, though a few things remain
	// (e.g., trace buffers).
	_Pdead
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// src/runtime/proc.go#L740 这段代码在 schedinit() 函数内部
	lock(&sched.lock)
	sched.lastpoll.Store(nanotime())
	procs := ncpu
	if n, ok := atoi32(gogetenv("GOMAXPROCS")); ok && n > 0 {
		procs = n
	}
	if procresize(procs) != nil {
		throw("unknown runnable goroutine during bootstrap")
	}
	unlock(&sched.lock)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

// src/runtime/proc.go#L4937
// 入参：需要创建的 P 的个数
// 出参：P 的列表
func procresize(nprocs int32) *p {
  // 环境检查，确保已经上锁，确保worldstop
	assertLockHeld(&sched.lock)
	assertWorldStopped()

  // 参数检查，程序启动时 gomaxprocs为0
	old := gomaxprocs
	if old < 0 || nprocs <= 0 {
		throw("procresize: invalid arg")
	}
	if trace.enabled {
		traceGomaxprocs(nprocs)
	}

	// update statistics
	now := nanotime()
	if sched.procresizetime != 0 {
		sched.totaltime += int64(old) * (now - sched.procresizetime)
	}
	sched.procresizetime = now

	maskWords := (nprocs + 31) / 32

	// Grow allp if necessary.
	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)) {
      // 如果需要的 P 个数小于 allp 的容量，缩小 allp 容量
			allp = allp[:nprocs]
		} else {
      // 否则，创建临时 nallp 数组，并 allp 全部拷贝至 nallp
			nallp := make([]*p, nprocs)
			// Copy everything up to allp's cap so we
			// never lose old allocated Ps.
			copy(nallp, allp[:cap(allp)])
      // 更新 allp
			allp = nallp
		}
		// 以上将 allp 调整到需要的大小，即 nproc
		// .....
	}

	// initialize new P's
  // 初始化 P, 初次启动时,old为0
	for i := old; i < nprocs; i++ {
		pp := allp[i]
    // 申请新 P 对象
		if pp == nil {
			pp = new(p)
		}
    // 初始化 P，每个 P 都有一个ID，该ID就是 P 在 allp 的索引
		pp.init(i)
    // 将 P 放入 allp 中
		atomicstorep(unsafe.Pointer(&allp[i]), unsafe.Pointer(pp))
	}
  // 以上，初始化了所有 p 对象，存放在 allp 数组中

  // 获取当前正在运行的 g,初始化时后，这里就是g0
  // 初始化时，g0所在的 p 还是0，所以会进入 else 逻辑
	gp := getg()
	if gp.m.p != 0 && gp.m.p.ptr().id < nprocs {
		// continue to use the current P
    // 继续使用当前的 P
		gp.m.p.ptr().status = _Prunning
		gp.m.p.ptr().mcache.prepareForSweep()
	} else {
    // 初始化会进到这个分支，
    // ......
    // 将 g0 所在 p 清空
		gp.m.p = 0
    // 取第 0 号 p，清空m, 并设置p状态为 _Pidle
		pp := allp[0]
		pp.m = 0
		pp.status = _Pidle
    // acquirep() 函数将 m0/p0 关联起来
		acquirep(pp)
	}

	// g.m.p is now set, so we no longer need mcache0 for bootstrapping.
  // 此时 m0,g0 已经设置，不在需要mcache0
	mcache0 = nil

	// release resources from unused P's
  // 释放不在使用的 P 的资源，但不会收回 P，因为有可能处于系统调用的M还会关联 P
	for i := nprocs; i < old; i++ {
		pp := allp[i]
		pp.destroy()
		// can't free P itself because it can be referenced by an M in syscall
	}

	// Trim allp.
  // 如果 allp 超过了 nprocs 进行缩容，一般运行时不调整 nprocs 所以不进入这个逻辑
	if int32(len(allp)) != nprocs {
		lock(&allpLock)
		allp = allp[:nprocs]
		idlepMask = idlepMask[:maskWords]
		timerpMask = timerpMask[:maskWords]
		unlock(&allpLock)
	}
	// 下面这个 for 循环检查 allp，将 空闲的P 放入全局空闲P列表；
  // 如果 P 的 LRQ 不为空，即有工作可做，将这些可运行的 P 链起来
	var runnablePs *p
	for i := nprocs - 1; i >= 0; i-- {
		pp := allp[i]
    // allp[0] 已经跟 m0 关联了，所以要跳过
		if gp.m.p.ptr() == pp {
			continue
		}
    // 将状态转为 _Pidle
		pp.status = _Pidle
    // 如果 P 的 LRQ 为空，将其加入 全局空闲P列表
		if runqempty(pp) {
			pidleput(pp, now)
		} else {
			pp.m.set(mget())
			pp.link.set(runnablePs)
			runnablePs = pp
		}
	}
	stealOrder.reset(uint32(nprocs))
	var int32p *int32 = &gomaxprocs // make compiler check that gomaxprocs is an int32
	atomic.Store((*uint32)(unsafe.Pointer(int32p)), uint32(nprocs))
	if old != nprocs {
		// Notify the limiter that the amount of procs has changed.
		gcCPULimiter.resetCapacity(now, nprocs)
	}
  // 返回所有可运行的 P 列表
	return runnablePs
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// src/runtime/proc.go#L5088
// Associate p and the current m.
// 关联 P 到当前的 M
func acquirep(pp *p) {
	// wirep 执行真正的关联
	wirep(pp)
  // 整理mcache
	pp.mcache.prepareForSweep()
	// ......
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// src/runtime/proc.go
func wirep(pp *p) {
	gp := getg()

	if gp.m.p != 0 {
		throw("wirep: already in go")
	}
	if pp.m != 0 || pp.status != _Pidle {
		id := int64(0)
		if pp.m != 0 {
			id = pp.m.ptr().id
		}
		print("wirep: p->m=", pp.m, "(", id, ") p->status=", pp.status, "\n")
		throw("wirep: invalid p state")
	}
  // 将 P / M 相互引用，并设置 P 为 _Prunning
	gp.m.p.set(pp)
	pp.m.set(gp.m)
	pp.status = _Prunning
}

1
2
3
4

// src/runtime/proc.go	
	if procresize(procs) != nil {
		throw("unknown runnable goroutine during bootstrap")
	}