basic knowledge
Scheme for acquiring locks
Generally speaking, there are two solutions for acquiring locks, one is continuous spin + CAS, and the other is blocking + wakeup. There are pros and cons to both approaches. The Go language combines these two solutions, automatically judges the competition situation of the current lock, first tries to spin several times, and if the lock has not been released, then joins the blocking queue.
fairness
There are two modes of lock in Go language: normal mode and starvation mode.
Normal mode: Blocked coroutines enter the FIFO blocking queue, and all coroutines perform preemptive scheduling on locks. That is to say, when a coroutine is woken up from the blocking queue, it cannot directly acquire the lock, but has to preempt it again by itself (at this time, there may be other coroutines that have just entered the Lock() process and are still in the process of locking themselves). spin attempt, without entering the blocking queue). As mentioned in the comments of the source code, in this mode, coroutines awakened from the blocking queue are at a disadvantage. Because the newly arrived coroutines are already taking up the CPU, and there may be a lot of them.
Hunger mode: As mentioned above, the coroutines awakened from the blocking queue are at a disadvantage in competition. It is possible that a coroutine has been blocking the queue and cannot get the lock. In order to avoid this situation, when a coroutine is blocked in the blocking queue for a long time (1ms), the lock will be set to starvation mode. In starvation mode, all coroutines that want to acquire locks will not spin and wait, but will directly enter the tail of the blocking queue to queue.
Mutex structure
type Mutex struct {
state int32
sema uint32
}
const (
mutexLocked = 1 << iota // mutex is locked
mutexWoken
mutexStarving
mutexWaiterShift = iota
starvationThresholdNs = 1e6 //1ms
)
The state indicates all the information of the lock, including mutexLocked (locked or not), mutexWoken (whether there is a running coroutine trying to seize the lock, which is convenient for judging whether to wake up the coroutine from the blocking queue when Unlock ing), mutexStarving (whether in starvation mode), and the number of coroutines in the blocking queue.
The bit layout of the state is as follows:
|31-----------------3|-------2-----|-----1----|-----0-----| |Number of coroutines in the blocking queue|mutexStarving|mutexWoken|mutexLocked|
Lock()
Let's take a look at the specific implementation of the Lock method.
func (m *Mutex) Lock() {
if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
......
return
}
// Slow path (outlined so that the fast path can be inlined)
m.lockSlow()
}
The first if condition means that if the lock is a brand new lock (no other coroutines have touched it), just lock it directly and return. Otherwise, enter the lockSlow() process.
lockSlow()
func (m *Mutex) lockSlow() {
var waitStartTime int64
starving := false
awoke := false
iter := 0
old := m.state
//...
}
The meaning of local variables:
waitStartTime: The total time spent waiting in the blocking queue, this value will be accumulated.
starving: Mark whether to change the lock to starvation mode.
awoke: Mark whether to modify the mutexWoken bit of the lock.
iter: The number of times to try the spin.
old: The old state of the lock, used for CAS operations, etc.
for {
// Don't spin in starvation mode, ownership is handed off to waiters
// so we won't be able to acquire the mutex anyway.
if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
// Active spinning makes sense.
// Try to set mutexWoken flag to inform Unlock
// to not wake other blocked goroutines.
if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
awoke = true
}
runtime_doSpin()
iter++
old = m.state
continue
}
//...
}
1. First judge the state of the lock, the lock must be occupied, not in starvation mode (starvation mode cannot be preempted, it must be queued in the blocking queue), and can continue to spin before entering the logic of the first if.
2. If the current lock mutexWoken is 0, that is, no other coroutines are running and waiting, and the size of the blocking queue is not 0, you can mark mutexWoken as 1.
for{
//...
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 {
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
}
//...
}
The logic here is that the spin ends, a new state is constructed, and the lock is ready to be acquired.
1. If the mutexStarving of old is 0, it means that the current coroutine can be locked, and the mutexLocked bit is written as 1 in new.
2. If the old mutexLocked or mutexStarving is 1, the current coroutine must not be locked, and the value of the blocking queue is increased by 1.
3. If the current coroutine is in the starving state (waiting in the blocking queue for too long) and the old state is locked, write the mutexStarving bit of new as 1.
4. If the current coroutine is in the awoke state, it is necessary to modify the value of the mutexWoken bit of new and write it as 0 (the subsequent current thread only has two possibilities of acquiring locks and blocking).
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, 1)
//wake up
starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
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.AddInt32(&m.state, delta)
break
}
awoke = true
iter = 0
} else {
old = m.state
}
}
Here is an attempt to update the lock status. If the CAS operation fails, update the value of old and continue the cycle.
If successful:
1. If old is not locked and is not in starvation mode, it means that the current coroutine is successfully locked and the function can return.
2. Determine whether the current coroutine just came or woke up from the blocking queue. If it is just coming, put it at the end of the blocking queue, otherwise put it at the head of the blocking queue.
3. runtime_SemacquireMutex(&m.sema, queueLifo, 1) means put into the blocking queue and block.
4. Update the value of starving after being awakened.
5. If the current lock is in starvation mode, the current coroutine can directly acquire the lock, otherwise it will return to the loop and try to preempt.
6. When acquiring a lock, the mutexLocked bit and the size of the blocking queue must be updated. If the current coroutine is not in the starving state or there is only one coroutine in the blocking queue, change the lock to normal mode.
Unlock()
Relatively simple, temporarily omitted.