互斥锁Mutex

互斥锁Mutex

使用锁保护共享资源

Locker()是一个interface。

type Locker interface {
    Lock()
    Unlock()
}

mutex结构体

type Mutex struct {
    state int32
    sema  uint32
}

state:

state 是一个复合型的字段，一个字段包含多个意义，这样可以通过尽可能少的内存来实现
互斥锁。这个字段的第一位（最小的一位）来表示这个锁是否被持有，第二位代表是否有
唤醒的 goroutine，第三位代表锁释放处于饥饿状态，剩余的位数代表的是等待此锁的 goroutine 数。

sema:

sema是个信号量变量，用来控制等待 goroutine 的阻塞休眠和唤醒。

Lock()

Mutex 可以处于 2 种操作模式：正常模式和饥饿模式。
在正常模式下，waiters按 FIFO 顺序排队，但是被唤醒的waiters不拥有mutex并与新到达的 goroutines 竞争mutex所有权。新到达的 goroutine 有一个优势——它们是已经在 CPU 上运行并且可能有很多，所以被唤醒waiter可能会失败。在这种情况下，唤醒的waiter会在排在等待队列的前面。如果waiter超过 1ms 未能获取mutex，它将mutex切换到饥饿模式。
在饥饿模式下，mutex的所有权直接从解锁 goroutine 移交给队列前面的waiter。
新到达的 goroutine 不会尝试获取互斥锁，即使它看起来已解锁，也不会尝试自旋。相反，他们将自己排在等待队列的尾部。
如果waiter收到mutex的所有权并看到

(1) 它是队列中的最后一个等待者
(2) 它等待的时间不到 1ms，

它将互斥锁切换回正常操作模式。
正常模式具有更好的性能，因为即使有阻塞的waiter，goroutine 可以连续多次获得mutex。
饥饿模式对于防止尾部延迟的病理情况很重要。

func (m *Mutex) Lock() {
    // Fast path: grab unlocked mutex.
    // cas获取到锁
    if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
        if race.Enabled {
            race.Acquire(unsafe.Pointer(m))
        }
        return
    }
    // Slow path (outlined so that the fast path can be inlined)
    // 尝试自旋竞争或饥饿状态下饥饿goroutine竞争
    m.lockSlow()
}

func (m *Mutex) lockSlow() {
    // 记录此 goroutine 请求锁的初始时间
    var waitStartTime int64
    // 此goroutine的饥饿标记
    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.
        // 锁是非饥饿状态，锁未释放，自旋尝试获取锁
        // 如果无法直接获取锁，进行多次自旋尝试;多次尝试失败，进入sema队列休眠
        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.
            // 尝试设置 mutexWoken 标志以通知 Unlock 不要唤醒其他阻塞的 goroutine。
            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 := old
        // Don't try to acquire starving mutex, new arriving goroutines must queue.
        // 不要尝试获取饥饿的互斥锁，新到达的 goroutine 必须排队。
        if old&mutexStarving == 0 {
            // 非饥饿状态，加锁
            new |= mutexLocked
        }
        if old&(mutexLocked|mutexStarving) != 0 {
            // waiter数量加1
            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.
         // 当前的 goroutine 将 mutex 切换到饥饿模式。
        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.
             // goroutine 已经从睡眠中唤醒，
            // 所以我们需要在任何一种情况下重置标志。
            if new&mutexWoken == 0 {
                throw("sync: inconsistent mutex state")
            }
             // 新状态清除唤醒标记
            new &^= mutexWoken
        }
         // cas成功设置新状态
        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.
             // 如果我们之前已经在等待，请在队列的前面排队。
             // 判断是否第一次加入到 waiter 队列
            queueLifo := waitStartTime != 0
            if waitStartTime == 0 {
                waitStartTime = runtime_nanotime()
            }
             // 阻塞等待
             // 将此 waiter 加入到队列，如果是首次，加入到队尾，先进先出。如果不是首次，那么加入到队首，这样等待最久的 goroutine 优先能够获取到锁。此 goroutine 会进行休眠。
            runtime_SemacquireMutex(&m.sema, queueLifo, 1)
             // 唤醒之后检查锁是否应该处于饥饿状态,此时已经被唤醒
             // 检查等待时间是否大于1ms
            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")
                }
                  // 加锁,waiter数减1
                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.
                      // 退出饥饿模式。
                      // 此 waiter 已经是队列中的最后一个 waiter 了，没有其它的等待锁的 goroutine 了；
                    // 此 waiter 的等待时间小于 1 毫秒。
                    delta -= mutexStarving
                }
                atomic.AddInt32(&m.state, delta)
                break
            }
            awoke = true
            iter = 0
        } else {
            old = m.state
        }
    }

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

Unlock()

func (m *Mutex) Unlock() {
    if race.Enabled {
        _ = m.state
        race.Release(unsafe.Pointer(m))
    }

    // Fast path: drop lock bit.
    new := atomic.AddInt32(&m.state, -mutexLocked)
    if new != 0 {
        // Outlined slow path to allow inlining the fast path.
        // To hide unlockSlow during tracing we skip one extra frame when tracing GoUnblock.
        m.unlockSlow(new)
    }
}

func (m *Mutex) unlockSlow(new int32) {
    if (new+mutexLocked)&mutexLocked == 0 {
        fatal("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.
             // 如果 Mutex 处于正常状态，如果没有 waiter，或者已经有在处理的情况了，那么释放就好，不做额外的处理。
            if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
                return
            }
            // Grab the right to wake someone.
             // waiter 数减 1，设置mutexWoken 标志,通过 CAS 更新 state 的值。
            new = (old - 1<<mutexWaiterShift) | mutexWoken
            if atomic.CompareAndSwapInt32(&m.state, old, new) {
                runtime_Semrelease(&m.sema, false, 1)
                return
            }
            old = m.state
        }
    } else {
        // Starving mode: handoff mutex ownership to the next waiter, and yield
        // our time slice so that the next waiter can start to run immediately.
        // Note: mutexLocked is not set, the waiter will set it after wakeup.
        // But mutex is still considered locked if mutexStarving is set,
        // so new coming goroutines won't acquire it.
         // 如果 Mutex 处于饥饿状态，直接唤醒等待队列中的 waiter。
        runtime_Semrelease(&m.sema, true, 1)
    }
}

TryLock()

当一个 goroutine 调用这个TryLock 方法请求锁的时候，如果这把锁没有被其他 goroutine 所持有，那么，这个goroutine 就持有了这把锁，并返回 true；如果这把锁已经被其他 goroutine 所持有，或者是正在准备交给某个被唤醒的 goroutine，那么，这个请求锁的 goroutine 就直接返回
false，不会阻塞在方法调用上。

func (m *Mutex) TryLock() bool {
    old := m.state
    if old&(mutexLocked|mutexStarving) != 0 {
        return false
    }

    // There may be a goroutine waiting for the mutex, but we are
    // running now and can try to grab the mutex before that
    // goroutine wakes up.
    if !atomic.CompareAndSwapInt32(&m.state, old, old|mutexLocked) {
        return false
    }

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