golang之sync.Mutex互斥锁源码分析

 1Mutex can be in 2 modes of operations: normal and starvation.
 2
 3 In normal mode waiters are queued in FIFO order, but a woken up waiter does not own the mutex and competes with new arriving goroutines over the ownership. New arriving goroutines have an advantage -- they are already running on CPU and there can be lots of them, so a woken up waiter has good chances of losing. In such case it is queued at front of the wait queue. If a waiter fails to acquire the mutex for more than 1ms, it switches mutex to the starvation mode.
 4
 5
 6
 7In starvation mode ownership of the mutex is directly handed off from the unlocking goroutine to the waiter at the front of the queue. New arriving goroutines don't try to acquire the mutex even if it appears to be unlocked, and don't try to spin. Instead they queue themselves at the tail of the wait queue.
 8
 9
10
11If a waiter receives ownership of the mutex and sees that either (1) it is the last waiter in the queue, or (2) it waited for less than 1 ms, it switches mutex back to normal operation mode.
12
13
14
15 Normal mode has considerably better performance as a goroutine can acquire a mutex several times in a row even if there are blocked waiters.
16
17Starvation mode is important to prevent pathological cases of tail latency.

 1互斥量可分为两种操作模式:正常和饥饿。
 2
 3在正常模式下，等待的goroutines按照FIFO（先进先出）顺序排队，但是goroutine被唤醒之后并不能立即得到mutex锁，它需要与新到达的goroutine争夺mutex锁。
 4
 5因为新到达的goroutine已经在CPU上运行了，所以被唤醒的goroutine很大概率是争夺mutex锁是失败的。出现这样的情况时候，被唤醒的goroutine需要排队在队列的前面。
 6
 7如果被唤醒的goroutine有超过1ms没有获取到mutex锁，那么它就会变为饥饿模式。
 8
 9在饥饿模式中，mutex锁直接从解锁的goroutine交给队列前面的goroutine。新达到的goroutine也不会去争夺mutex锁（即使没有锁，也不能去自旋），而是到等待队列尾部排队。
10
11在饥饿模式下，有一个goroutine获取到mutex锁了，如果它满足下条件中的任意一个，mutex将会切换回去正常模式：
12
131. 是等待队列中的最后一个goroutine
14
152. 它的等待时间不超过1ms。
16
17正常模式有更好的性能，因为goroutine可以连续多次获得mutex锁；
18
19饥饿模式对于预防队列尾部goroutine一致无法获取mutex锁的问题。

 1type Mutex struct {
 2
 3    state int32 // 将一个32位整数拆分为 当前阻塞的goroutine数(29位)|饥饿状态(1位)|唤醒状态(1位)|锁状态(1位) 的形式，来简化字段设计
 4
 5    sema uint32 // 信号量
 6
 7}
 8
 9
10
11const (
12
13    mutexLocked = 1 << iota // 1 0001 含义：用最后一位表示当前对象锁的状态，0-未锁住 1-已锁住
14
15    mutexWoken // 2 0010 含义：用倒数第二位表示当前对象是否被唤醒 0-唤醒 1-未唤醒
16
17    mutexStarving // 4 0100 含义：用倒数第三位表示当前对象是否为饥饿模式，0为正常模式，1为饥饿模式。
18
19    mutexWaiterShift = iota // 3，从倒数第四位往前的bit位表示在排队等待的goroutine数
20
21    starvationThresholdNs = 1e6 // 1ms
22
23)

 1state: |32|31|...| |3|2|1|
 2
 3         \__________/ | | |
 4
 5              | | | |
 6
 7              | | | mutex的占用状态（1被占用，0可用）
 8
 9              | | |
10
11              | | mutex的当前goroutine是否被唤醒
12
13              | |
14
15              | 饥饿位，0正常，1饥饿
16
17              |
18
19               等待唤醒以尝试锁定的goroutine的计数，0表示没有等待者
20

  1func (m *Mutex) Lock() {
  2
  3    // 如果m.state=0，说明当前的对象还没有被锁住，进行原子性赋值操作设置为mutexLocked状态，CompareAnSwapInt32返回true
  4
  5    // 否则说明对象已被其他goroutine锁住，不会进行原子赋值操作设置，CopareAndSwapInt32返回false
  6
  7    if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) 
  8
  9        if race.Enabled {
 10
 11            race.Acquire(unsafe.Pointer(m))
 12
 13        }
 14
 15        return
 16
 17    }
 18
 19
 20
 21    // 开始等待时间戳
 22
 23    var waitStartTime int64
 24
 25    // 饥饿模式标识
 26
 27    starving := false
 28
 29    // 唤醒标识
 30
 31    awoke := false
 32
 33    // 自旋次数
 34
 35    iter := 0
 36
 37    // 保存当前对象锁状态
 38
 39    old := m.state
 40
 41    // 看到这个for {}说明使用了cas算法
 42
 43    for {
 44
 45        // 相当于xxxx...x0xx & 0101 = 01，当前对象锁被使用
 46
 47        if old&(mutexLocked|mutexStarving) == mutexLocked && 
 48
 49            // 判断当前goroutine是否可以进入自旋锁
 50
 51            runtime_canSpin(iter) {
 52
 53
 54
 55            // 主动旋转是有意义的。试着设置mutexwake标志，告知解锁，不要唤醒其他阻塞的goroutines。
 56
 57            if !awoke &&
 58
 59            // 再次确定是否被唤醒： xxxx...xx0x & 0010 = 0
 60
 61            old&mutexWoken == 0 &&
 62
 63            // 查看是否有goroution在排队
 64
 65            old>>mutexWaiterShift != 0 &&
 66
 67                // 将对象锁改为唤醒状态：xxxx...xx0x | 0010 = xxxx...xx1x 
 68
 69                atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
 70
 71                awoke = true
 72
 73            }//END_IF_Lock
 74
 75
 76
 77            // 进入自旋锁后当前goroutine并不挂起，仍然在占用cpu资源，所以重试一定次数后，不会再进入自旋锁逻辑
 78
 79            runtime_doSpin()
 80
 81            // 自加，表示自旋次数
 82
 83            iter++
 84
 85            // 保存mutex对象即将被设置成的状态
 86
 87            old = m.state
 88
 89            continue
 90
 91        }// END_IF_spin
 92
 93
 94
 95        // 以下代码是不使用**自旋**的情况
 96
 97        new := old
 98
 99
100
101        // 不要试图获得饥饿的互斥，新来的goroutines必须排队。
102
103        // 对象锁饥饿位被改变，说明处于饥饿模式
104
105        // xxxx...x0xx & 0100 = 0xxxx...x0xx
106
107        if old&mutexStarving == 0 {
108
109            // xxxx...x0xx | 0001 = xxxx...x0x1，标识对象锁被锁住
110
111            new |= mutexLocked
112
113        }
114
115        // xxxx...x1x1 & (0001 | 0100) => xxxx...x1x1 & 0101 != 0;当前mutex处于饥饿模式并且锁已被占用，新加入进来的goroutine放到队列后面
116
117        if old&(mutexLocked|mutexStarving) != 0 {
118
119            // 更新阻塞goroutine的数量,表示mutex的等待goroutine数目加1
120
121            new += 1 << mutexWaiterShift
122
123        }
124
125
126
127        // 当前的goroutine将互斥锁转换为饥饿模式。但是，如果互斥锁当前没有解锁，就不要打开开关,设置mutex状态为饥饿模式。Unlock预期有饥饿的goroutine
128
129        if starving && 
130
131            // xxxx...xxx1 & 0001 != 0；锁已经被占用
132
133            old&mutexLocked != 0 {
134
135            // xxxx...xxx | 0101 => xxxx...x1x1，标识对象锁被锁住
136
137            new |= mutexStarving
138
139        }
140
141
142
143        // goroutine已经被唤醒，因此需要在两种情况下重设标志
144
145        if awoke {
146
147            // xxxx...xx1x & 0010 = 0,如果唤醒标志为与awoke不相协调就panic
148
149            if new&mutexWoken == 0 {
150
151                panic("sync: inconsistent mutex state")
152
153            }
154
155            // new & (^mutexWoken) => xxxx...xxxx & (^0010) => xxxx...xxxx & 1101 = xxxx...xx0x ：设置唤醒状态位0,被唤醒
156
157            new &^= mutexWoken
158
159        }
160
161        // 获取锁成功
162
163        if atomic.CompareAndSwapInt32(&m.state, old, new) {
164
165            // xxxx...x0x0 & 0101 = 0，已经获取对象锁
166
167            if old&(mutexLocked|mutexStarving) == 0 {
168
169                // 结束cas
170
171                break
172
173            }
174
175            // 以下的操作都是为了判断是否从饥饿模式中恢复为正常模式
176
177            // 判断处于FIFO还是LIFO模式
178
179            queueLifo := waitStartTime != 0
180
181            if waitStartTime == 0 {
182
183                waitStartTime = runtime_nanotime()
184
185            }
186
187            runtime_SemacquireMutex(&m.sema, queueLifo)
188
189            starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
190
191            old = m.state
192
193            // xxxx...x1xx & 0100 != 0
194
195            if old&mutexStarving != 0 {
196
197                // xxxx...xx11 & 0011 != 0
198
199                if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
200
201                    panic("sync: inconsistent mutex state")
202
203                }
204
205                delta := int32(mutexLocked - 1<<mutexWaiterShift)
206
207                if !starving || old>>mutexWaiterShift == 1 {
208
209                    delta -= mutexStarving
210
211                }
212
213                atomic.AddInt32(&m.state, delta)
214
215                break
216
217            }
218
219            awoke = true
220
221            iter = 0
222
223        } else {
224
225            // 保存mutex对象状态
226
227            old = m.state
228
229        }
230
231    }// cas结束
232
233
234
235    if race.Enabled {
236
237        race.Acquire(unsafe.Pointer(m))
238
239    }
240
241}

 1func (m *Mutex) Unlock() {
 2
 3    if race.Enabled {
 4
 5        _ = m.state
 6
 7        race.Release(unsafe.Pointer(m))
 8
 9    }
10
11
12
13    // state-1标识解锁
14
15    new := atomic.AddInt32(&m.state, -mutexLocked)
16
17    // 验证锁状态是否符合
18
19    if (new+mutexLocked)&mutexLocked == 0 {
20
21        panic("sync: unlock of unlocked mutex")
22
23    }
24
25    // xxxx...x0xx & 0100 = 0 ;判断是否处于正常模式
26
27    if new&mutexStarving == 0 {
28
29        old := new
30
31        for {
32
33            // 如果没有等待的goroutine或goroutine已经解锁完成
34
35            if old>>mutexWaiterShift == 0 || 
36
37            // xxxx...x0xx & (0001 | 0010 | 0100) => xxxx...x0xx & 0111 != 0
38
39            old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
40
41                return
42
43            }
44
45            // Grab the right to wake someone.
46
47            new = (old - 1<<mutexWaiterShift) | mutexWoken
48
49            if atomic.CompareAndSwapInt32(&m.state, old, new) {
50
51                runtime_Semrelease(&m.sema, false)
52
53                return
54
55            }
56
57            old = m.state
58
59        }
60
61    } else {
62
63        // 饥饿模式:将mutex所有权移交给下一个等待的goroutine
64
65        // 注意:mutexlock没有设置，goroutine会在唤醒后设置。
66
67        // 但是互斥锁仍然被认为是锁定的，如果互斥对象被设置，所以新来的goroutines不会得到它
68
69        runtime_Semrelease(&m.sema, true)
70
71    }
72
73}