Go Mutex 源码详解

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// 互斥锁
type Mutex struct {
	state int32		// 状态
	sema  uint32  // 信号量
}


const (
	mutexLocked = 1 << iota // 1
	mutexWoken // 2
	mutexStarving // 4
	mutexWaiterShift = iota // 3

	starvationThresholdNs = 1e6 // 判断是否要进入饥饿状态的阈值
)

1
2
3
4
5
6
7
8
9

// 上锁
func (m *Mutex) Lock() {
	// fastpath：期望当前锁没有被占用，可以快速获取到锁， CAS 修改 state 最后一位的值为1（标记锁是否被占用）
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
		return
	}
	// Slow path : 单独抽出来放到一个函数里，方便 fast path 被内联
	m.lockSlow()
}

  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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166

func (m *Mutex) lockSlow() {
	var waitStartTime int64 // // 记录等待时间
	starving := false // 当前的 goroutine 是否已经饥饿了（如果已经饥饿，就会将 state 的饥饿状态置为 1）
	awoke := false  // 当前的 goroutine 是否被唤醒的
	iter := 0	// 自旋次数
	old := m.state // 保存当前的 state 状态
	for {
    
    /*
    自旋：如果满足如下条件，就会进入 if 语句，然后 continue，不断自旋:
    1. 锁被占用，且不处于饥饿模式（饥饿状态直接去排队，不允许尝试获取锁）
    2. 基于当前自旋的次数，再次自旋有意义 runtime_canSpin(iter)
    
    那么退出自旋的条件也就是：
    1. 锁被释放了，当前处于没被占用状态（说明等到了，该goroutine就会立即去获取锁）
		2. mutex进入了饥饿模式，不自旋了，没意义（饥饿状态会直接把锁交给等待队列队首的goroutine）
		3. 不符合自旋状态（自旋次数太多了，自旋失去了意义）
    
    如下代码是位操作：
    mutexLocked|mutexStarving = 00000...101
    mutexLocked = 00000...001
    如果要满足 old & 00000...101 = 00000...001，需要 old = ...0*1,即状态为：锁被占用，且不处于饥饿状态 
    
    runtime_canSpin(iter) 会根据自旋次数，判断是否可以继续自旋
    */
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
			
      /*
      如果 
      	1. 当前goroutine不是被唤醒的 (awoke=false) 
      	2. 锁状态唤醒标志位为0（old&mutexWoken == 0） 
      	3. 等待者数量不为0 （old>>mutexWaiterShift != 0  右移三位得到的就是等待者数量）
      
      	那么利用CAS，将 state 的唤醒标记置为1，标记自己是被唤醒的 （将state的唤醒标记置为1，说明外面有唤醒着的goroutine，那么在释放锁的时候，就不去等待队列叫号了，毕竟已经有唤醒的了）
      	
      	如果有其他 goroutine 已经设置了 state 的唤醒标记位，那么本次就会失败
      */

			if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
				atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
				awoke = true
			}
			runtime_doSpin()
      
      // 迭代次数加一
			iter++
      
      // 获取最新的状态
			old = m.state
      
      // 想再次自旋，看看锁释放了没
			continue
		}
    
    // 到这里，说明退出了自旋，当前锁没被占用 或者  系统处于饥饿模式 或者 自旋次数太多导致不符合自旋条件
  
    // new 代表当前goroutine 基于当前状态要设置的新状态
		new := old
    
    // 只要不是饥饿状态，就需要获取锁（饥饿状态直接去排队，不能抢锁）
		if old&mutexStarving == 0 {
			new |= mutexLocked
		}
    
    // 锁被占用 或者 处于饥饿模式下，新增一个等待者
		if old&(mutexLocked|mutexStarving) != 0 {
			new += 1 << mutexWaiterShift
		}
    
    // 当前 goroutine 已经进入饥饿了，且锁还没有释放，需要把 Mutex 的状态改为饥饿状态
		if starving && old&mutexLocked != 0 {
			new |= mutexStarving
		}
    
    // 如果是被唤醒的，把唤醒标志位置0，表示外面没有被唤醒的goroutine了（抢到就获得锁、抢不到就睡眠，把唤醒标志置0）
		if awoke {
			
      // 由于是被唤醒的，new 里面的 唤醒标记位一定是 1
			if new&mutexWoken == 0 {
				throw("sync: inconsistent mutex state")
			}
      
      // a &^ b 的意思就是 清零a中，ab都为1的位，即清除唤醒标记
			new &^= mutexWoken
		}
    
    /*
      利用CAS,将状态设置为新的
      1. 如果是饥饿状态，只增加一个等待者数量
      2. 正常状态，加锁标记置为 1，如果锁已被占用增加一个等待者数量
      3. 如果当前 goroutine 已经饥饿了，将 饥饿标记 置为 1
      4. 如果是被唤醒的，清除唤醒标记
    */
    
		if atomic.CompareAndSwapInt32(&m.state, old, new) {
      
      // 如果改状态之前，锁未被占用 且 处于正常模式，那么就相当于获取到锁了
			if old&(mutexLocked|mutexStarving) == 0 {
				break 
			}
      
      // 到这里说明：1. 之前锁被占用  或者 2.之前是处于饥饿状态 
      
      // 判断之前是否等待过（是否从队列里唤醒的），之前等待过，再次排队放在队首
			queueLifo := waitStartTime != 0
      
      // 如果之前没等过（新来的），设置等待起始时间当前时间
			if waitStartTime == 0 {
				waitStartTime = runtime_nanotime()
			}
      
     // 之前排过队的老人，放到等待队列队首；新人放到队尾，然后等待获取信号量
			runtime_SemacquireMutex(&m.sema, queueLifo, 1)
      
      // 锁被释放，goroutine 被唤醒
      
      // 设置当前 goroutine 饥饿状态，如果之前已经饥饿，或者距离等待开始时间超过了 1ms，也变饥饿
			starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
      
      // 获取最新的状态
			old = m.state
      
      // 如果 state 饥饿标记为1，说明当前在饥饿模式，饥饿模式下被唤醒，已经获取到锁了；
      // 饥饿状态下，释放锁没有更新等待者数量和饥饿标记，需要获得锁的goroutine去更新状态
			if old&mutexStarving != 0 {

        // 正确性校验：
        // 1. 锁还是锁住的状态（锁已经释放给当前goroutine了，不应该被锁住）
				// 2. 或者有被唤醒的goroutine（饥饿模式下不应该有醒着的goroutine，都应该去乖乖等着）
				// 3. 或者当前goroutine 的等待者数量为0（当前goroutine就是等待者）
				// 这三种情况不应该出现，与预期状态不符
        
				if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
					throw("sync: inconsistent mutex state")
				}
        
        // 加锁，减去一个等待者
				delta := int32(mutexLocked - 1<<mutexWaiterShift)
        
        // 如果当前的 goroutine 非饥饿，或者等待者只有一个（也就是只有当前goroutine，等待队列空了），可以取消饥饿状态，进入正常状态
				if !starving || old>>mutexWaiterShift == 1 {
					delta -= mutexStarving
				}
        
        // 修改状态：
        // 加锁，减去一个等待者： m.state + mutexLocked - 1<<mutexWaiterShift : 
        // 满足非饥饿条件，加锁，减去一个等待者，取消饥饿状态：
        // m.state + mutexLocked - 1<<mutexWaiterShift - mutexStarving: 
				atomic.AddInt32(&m.state, delta)
        
        // 饥饿模式下被唤醒，相当于获得锁了，可以结束
				break
			}
      
      // 之前是处于锁被占用且非饥饿状态，被唤醒，去继续抢锁
			awoke = true
      
      // 新唤醒的，自旋数量置0
			iter = 0
		} else {
      
      // 修改新状态失败，状态有更新，需要重试
			old = m.state
		}
	}
}

1
2
3
4
5
6
7
8
9

func (m *Mutex) Unlock() {

	// Fast path: 把锁标记清除
	new := atomic.AddInt32(&m.state, -mutexLocked)
	if new != 0 {
		// 清除完锁标记，发现还有其他状态，比如等待队列不为空，需要唤醒其他 goroutine
		m.unlockSlow(new)
	}
}

 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

func (m *Mutex) unlockSlow(new int32) {
  
  /* 状态正确性校验：
  	 1. 如果解锁一个上锁状态的锁，最后一位则为1，fast path 中 new 已经减去了1, 此时 new 最后一位应当为0
		 2. 如果解锁一个未上锁状态的锁，最后一位则为0，fast path 中 new 已经减去了1, 此时 new 最后一位应当为1
	 	 如果 (new+mutexLocked)&mutexLocked == 0，说明 new 当前最后一位是1，那么就是解锁了一个没有上锁的锁，状态有误
	*/
	if (new+mutexLocked)&mutexLocked == 0 {
		throw("sync: unlock of unlocked mutex")
	}
  
  // 正常模式，非饥饿，可能需要唤醒队列中的 goroutine，饥饿状态直接移交锁
	if new&mutexStarving == 0 {
		old := new
		for {

      /* 系统运转正常，锁可以正确交接，可以直接返回了：
        1. 没有等待者了 (没有等锁的了，去唤醒谁？)
        2. 有唤醒状态的 goroutine  (自旋状态的 goroutine，将唤醒状态置为1)
      	3. 有 goroutine 已经获取了锁 (Unlock方法已经将锁标记置为了0，可能自旋的此时已经抢到了锁)
      */
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
      
      // 没有唤醒状态的 goroutine，唤醒一个去抢锁
			// 减去一个等待者，并且将 唤醒标记 置为 1
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
        // 第二个参数为false, 唤醒队首的 goroutine 去抢锁(不一定能抢到)
				runtime_Semrelease(&m.sema, false, 1)
				return
			}
      
      // 上面 CAS 失败，可能由于新增了一个等待者，for 循环重试
			old = m.state
		}
	} else {
		
    /*
     1. 第二个参数为 true，直接将锁的所有权，交给等待队列的第一个等待者
		 2. 注意，此时没有设置 mutexLocked =1 ，被唤醒的 goroutine 会设置
		 3. 虽然没有设置 mutexLocked ，但是饥饿模式下, Mutex 始终被认为是锁住的，都会直接排队等待移交锁
    */
		runtime_Semrelease(&m.sema, true, 1)
	}
}