Golang sync.Mutex 源码分析

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

// 忙等待
wait(s) {
    while(s <= 0); // do nothing

    s--;
}

// 阻塞
wait(semaphore *s) {
    if (s.value <= 0) {
        // add this process to  s.list;
        // block()
    }

    s--;
}

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

func count() {
	var mutex sync.Mutex
	count := 0

	for r := 0; r < 50; r++ {
		go func() {
			mutex.Lock()
			count += 1
			mutex.Unlock()
		}()
	}

	time.Sleep(time.Second)
	fmt.Println("the count is : ", count) // res:  50
}

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

// A Mutex is a mutual exclusion lock. 
// The zero value for a Mutex is an unlocked mutex.
//
// A Mutex must not be copied after first use.
// Mutex 是一个互斥锁， 0 代表解除互斥，第一次使用后不允许复制。
type Mutex struct {
	state int32  // 状态 
	sema  uint32 // 非负数， 信号量
}

const (
	mutexLocked = 1 << iota // mutex is locked
	mutexWoken
	mutexStarving
	mutexWaiterShift = iota

    // Mutex can be in 2 modes of operations: normal and starvation. 互斥有两种模式：  正常和饥饿
}

 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

// Lock locks m.
// If the lock is already in use, the calling goroutine
// blocks until the mutex is available.
func (m *Mutex) Lock() {        // 如果已经被某个 goroutine 使用，将阻塞直到（ mutex.sema = 0 ）mutex 可用
	// Fast path: grab unlocked mutex.
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {    // CAS 比较是否为 0 如果为 0 则置 1 表示 lock 成功
		if race.Enabled {  // 竞态检测，并标记 [详细了解 race](https://www.cnblogs.com/yjf512/p/5144211.html)
			race.Acquire(unsafe.Pointer(m))  
		}
		return
	}

	var waitStartTime int64
	starving := false
	awoke := false
    // 自旋数
	iter := 0
    // 保存当前协程状态
	old := m.state
	for {
		// Don't spin in starvation mode, ownership is handed off to waiters
		// so we won't be able to acquire the mutex anyway.
        // 第一个条件是 state 已被锁状态，且不是饥饿状态。如果时饥饿状态，自旋是没有用的，锁的拥有权直接交给了等待队列的第一个。
        // 第二个条件是还可以自旋，多核、压力不大并且在一定次数内可以自旋
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) { // 自旋 设状态为 mutexWoken
            // remove detail code
			continue
		}
		new := old
		// Don't try to acquire starving mutex, new arriving goroutines must queue.
		if old&mutexStarving == 0 {
			new |= mutexLocked
		}
		if old&(mutexLocked|mutexStarving) != 0 { 
			new += 1 << mutexWaiterShift
		}
		// The current goroutine switches mutex to starvation mode.
		// But if the mutex is currently unlocked, don't do the switch.
		// Unlock expects that starving mutex has waiters, which will not
		// be true in this case.
		if starving && old&mutexLocked != 0 { // 任然为 mutexLocked|mutexStarving 状态，标志位偏移，加入阻塞队列
			new |= mutexStarving
		}
		if awoke {
			// The goroutine has been woken from sleep,
			// so we need to reset the flag in either case.
			if new&mutexWoken == 0 {
				throw("sync: inconsistent mutex state")
			}
			new &^= mutexWoken
		}
		if atomic.CompareAndSwapInt32(&m.state, old, new) {
			if old&(mutexLocked|mutexStarving) == 0 {
				break // locked the mutex with CAS
			}
			// If we were already waiting before, queue at the front of the queue.
			queueLifo := waitStartTime != 0
			if waitStartTime == 0 {
				waitStartTime = runtime_nanotime()
			}
			runtime_SemacquireMutex(&m.sema, queueLifo)
			starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs // 等待时间超过 1 ns，进入饥饿状态
			old = m.state
			if old&mutexStarving != 0 {
				// If this goroutine was woken and mutex is in starvation mode,
				// ownership was handed off to us but mutex is in somewhat
				// inconsistent state: mutexLocked is not set and we are still
				// accounted as waiter. Fix that.
				if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
					throw("sync: inconsistent mutex state")
				}
				delta := int32(mutexLocked - 1<<mutexWaiterShift)
				if !starving || old>>mutexWaiterShift == 1 {
					// Exit starvation mode.
					// Critical to do it here and consider wait time.
					// Starvation mode is so inefficient, that two goroutines
					// can go lock-step infinitely once they switch mutex
					// to starvation mode.
					delta -= mutexStarving
				}
				atomic.AddIntt32(&m.state, delta)
				break
			}
			awoke = true
			iter = 0
		} else {
			old = m.state
		}
	}

	if race.Enabled {
		race.Acquire(unsafe.Pointer(m))
	}
}

 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

// 当没有锁定的对象进行解锁将会是一个运行时错误，允许一个 goroutine 锁定一个互斥锁安排另一个 goroutine 解锁
func (m *Mutex) Unlock() {

	// Fast path: drop lock bit.
    // 如果state不是处于锁的状态, 那么就是Unlock根本没有加锁的mutex, panic
	new := atomic.AddInt32(&m.state, -mutexLocked) // state--;
	if (new+mutexLocked)&mutexLocked == 0 { // -> (new + 1) & 1 == 0 ？-> new == 0 ？
		throw("sync: unlock of unlocked mutex")
	}

    // 释放了锁，还得需要通知其它等待者
    // 锁如果处于饥饿状态，直接交给等待队列的第一个, 唤醒它，让它去获取锁
    // 锁如果处于正常状态，

    // 正常状态
	if new&mutexStarving == 0 {
		old := new
		for {
			// If there are no waiters or a goroutine has already
			// been woken or grabbed the lock, no need to wake anyone.
			// In starvation mode ownership is directly handed off from unlocking
			// goroutine to the next waiter. We are not part of this chain,
			// since we did not observe mutexStarving when we unlocked the mutex above.
			// So get off the way.
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
			// 将等待的goroutine数减一，并设置woken标识，并通过信号量唤醒某个阻塞的 goroutine 去获取锁.
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
				runtime_Semrelease(&m.sema, false)
				return
			}
			old = m.state
		}
	} else {
        // 饥饿模式下， 直接将锁的拥有权传给等待队列中的第一个.
        // 注意此时state的mutexLocked还没有加锁，唤醒的goroutine会设置它。
        // 在此期间，如果有新的goroutine来请求锁， 因为mutex处于饥饿状态， mutex还是被认为处于锁状态，
        // 新来的goroutine不会把锁抢过去.
		runtime_Semrelease(&m.sema, true)
	}
}