美文网首页GO
Go Mutex 源码详解

Go Mutex 源码详解

作者: 漫漫Coding路 | 来源:发表于2022-05-04 23:37 被阅读0次

    你必须非常努力,才能看起来毫不费力!

    微信搜索公众号[ 漫漫Coding路 ],一起From Zero To Hero !

    前言

    在上一篇文章中,我们一起学习了如何使用 Go 中的互斥锁 Mutex,那么本篇文章,我们就一起来探究下 Mutex 底层是如何实现的,知其然,更要知其所以然!

    说明:本文中的示例,均是基于Go1.17 64位机器

    提示:公众号设置星标,获取最新更新;文末博客内阅读,获得更佳的阅读体验

    Mutex 特性

    Mutex 就是一把互斥锁,可以想象成一个令牌,有且只有这一个令牌,只有持有令牌的 goroutine 才能进入房间(临界区),在房间内执行完任务后,走出房间并把令牌交出来,如果还有其余的 goroutine 等着获取这个令牌,让他们再去抢这个令牌,抢到的重复上述过程,没抢到的继续等。

    上述是从宏观角度来看待互斥锁的,但是在 Mutex 内部,有着非常复杂的抢锁逻辑,Mutex 的发展也经历了几个版本,我们可以用拿令牌进餐厅吃饭来形象比喻下几个主要版本的变化。

    前提:餐厅一次只能进入一个人,餐厅有一个令牌,只有持有这个令牌的人才能进去;从餐厅出来后,需要把这个令牌归还

    版本一

    餐厅在门外设置了一个队伍,如果令牌空闲,拿着令牌去餐厅用餐;如果令牌不是空闲的,新来的人就要去队伍后面排队等待叫号。(不是空闲包含两种情况:持有令牌的人在餐厅里面,队伍是空的;队伍有人排队。)

    此版本的问题就是:只要令牌不是空闲的,新来的人必须直接去排队,没有商量的余地。这样看起来很公平,遵循先来后到的原则,但是对于餐厅来说,营业效率就会有所降低,即单位时间内接待顾客的数量(IO)会减少。为什么这样说呢,举个例子,有个顾客从餐厅出来归还令牌后,需要去等待队列去叫号,被叫到号的这个人需要花费时间走到餐厅(获取到CPU),这中间就浪费了不少时间。

    版本二

    为了提高营业效率,允许刚到门口的顾客和被叫到号的顾客一起去抢令牌,而不是直接去排队,这样就给了新人机会。举个例子:当持有令牌的人从餐厅出来归还令牌后,去等待队列叫个号,如果此时有顾客刚到门口,被叫到号的和新到的顾客一起抢令牌,抢到的就可以直接进入餐厅,抢不到的接着去排队,由于刚到的顾客离门口近(正在占据CPU),被叫到号的顾客离得远(需要等CPU),而且刚到的顾客可能不只一个,所以被叫到号的顾客很大概率抢不到令牌,可能还没走到门口(还没获取到CPU)就被新来的顾客抢走了。不管怎么样,这样提高了餐厅的效率,可以在单位时间内接待更多的客户。

    版本三

    餐厅发现有些人用餐很快,如果让抢不到令牌的先别直接去排队,而是在门口转悠会(当然不能一直转悠,有条件限制,到了限制还是要去排队),这种方式类似乐观锁,那么有顾客从餐厅出来后,就不用去叫号了,直接让门口的这些顾客继续抢就行了,这样就进一步提高了餐厅的运行效率,毕竟叫号真的太浪费时间了。

    版本四

    经过了多个版本的优化,餐厅的运营效率是越来越高了,但是有些人可要准备要骂娘了,这些人是谁呢,当然是已经在队伍里等待的那些人。由于给了新人机会,如果持续有新顾客来,那么已经在队伍里的那些人永远也拿不到令牌,可真的要饿死了。

    Mutex 在这个版本只为三件事:公平、公平、还是tm的公平!坚持让每一个人都不饿肚子的原则,餐厅搞出了一个新的模式:饥饿模式。如果有顾客等的时间超过了阈值(1ms),餐厅变为饥饿模式,在该模式下,所有新来的顾客直接去排队,然后按照先来先到的顺序,依次将令牌给等待队列队首的顾客。

    那么什么时候由饥饿模式变为正常模式呢?当拿到令牌的顾客发现自己从等待到拿到令牌的时间小于阈值(1ms)了,或者等待队伍没人等了,此时餐厅就变为正常模式,毕竟上述两个条件都说明当前餐厅竞争不是很激烈了。

    同时这个版本修复了以前的一个问题:之前从等待队列唤醒的顾客如果没有抢到令牌,再回到队列后是插到队尾,这样对已经排到第一位的顾客太不友好了。在这个版本中修复了该问题,唤醒的顾客如果没有抢到令牌,直接插入到队首,下次叫号还是他。

    特性总结

    经过了多次迭代,目前的版本有了如下特性:

    给新人机会:让刚来的顾客和从队列唤醒的顾客一起去抢令牌,唤醒也是按照先来先到的原则唤醒;

    保持乐观态度:没抢到不是直接去排队,而是可以在门口转悠会,说不定里面的人马上就出来了;

    正常模式和饥饿模式的切换:为了公平起见,正常模式下给了新人机会,一起去抢令牌;饥饿模式下照顾老人,所有人老老实实排队,按照先来先到的顺序拿令牌。整个餐厅既保持了公平,又提高了运行效率,一切井然有序起来了。

    回归正题

    让我们从餐厅回到 Go 中来,Mutex 有两种模式:正常模式和饥饿模式:

    正常模式下,如果当前锁正在被持有,抢不到锁的就会进入一个先进先出的等待队列。当持有锁的 goroutine 释放锁之后,按照从前到后的顺序唤醒等待队列的第一个等待者,但是不会直接给被唤醒者锁,还是需要他去抢,即在唤醒等待队列等待者这个时间,同时也会有正在运行且还未进入等待队列的 goroutine 正在抢锁 (数量可能还很多),这些都会和刚唤醒的等待者一起去抢,刚唤醒的可能还没有分到 CPU,而正在运行的正在占据了CPU,所以正在运行的更有可能获取到锁,被唤醒的等待者可能抢锁失败。如果等待者抢锁失败,他会被放到等待队列的队首,如果超过 1ms 都没抢到锁,就会从 正常模式 切换到 饥饿模式

    饥饿模式下,要释放锁的 goroutine 会将锁直接交给等待队列的第一个等待者,不需要去抢了,而且新来的 goroutine 也不会尝试去抢锁,直接加入到等待队列的尾部。那么什么时候会从饥饿模式切换到正常模式呢:

    (1)如果当前被唤醒的等待者获得到锁后,发现自己是队列中的最后一个,队列中没有其他等待者了,此时会切换到正常模式

    (2)如果当前被唤醒的等待者获得到锁后,发现自己总共的等待时间不超过 1ms,就获得到锁了,此时也会切换到正常模式

    正常模式会带来更高的吞吐量:一个 goroutine 要释放锁,更大可能会被正在运行的 goroutine 抢到,这就避免了协程的上下文切换,运行更多的 goroutine,但是有可能造成一个问题,就是锁始终被新来的 goroutine 抢走,在等待队列中的等待者始终抢不到锁,这就会导致饥饿问题。饥饿模式就是为了解决这个问题出现的,保证了每个 goroutine 都有运行的机会,防止等待时间过长。

    数据结构

    // 互斥锁
    type Mutex struct {
        state int32     // 状态
        sema  uint32  // 信号量
    }
    
    
    const (
        mutexLocked = 1 << iota // 1
        mutexWoken // 2
        mutexStarving // 4
        mutexWaiterShift = iota // 3
    
        starvationThresholdNs = 1e6 // 判断是否要进入饥饿状态的阈值
    )
    

    信号量 sema 就相当于我们说的令牌,state 是 int32 类型,一共 32位,通过每个位记录了当前的状态:

    state字段

    mutexLocked:当前是否已经上锁,state & mutexLocked = 1 表示已经上锁;

    mutexWoken:标记当前是否有唤醒的 goroutine,state & mutexWoken = 1 表示有唤醒的goroutine;

    mutexStarving:当前是否为饥饿状态,state & mutexWoken = 1 表示处于饥饿状态;

    mutexWaiterShift:29位,state >> mutexWaiterShift 得到等待者的数量;

    Lock()

    Lock()加锁方法分为两部分,第一部分是 fast path,可以理解为快捷通道,如果当前锁没被占用,直接获得锁返回;否则需要进入 slow path,判断各种条件去竞争锁,主要逻辑都在此处。

    了解过原子操作的同学,对 CompareAndSwap(CAS) 应该不陌生,CompareAndSwapInt32(addr *int32, old, new int32) 有三个参数,如果地址 addr 指向的值与 old 相等,则将 addr 的值改为 new,否则不变,也就是说在我们修改前,如果有人修改了 addr 指向的值,本次修改就会失败。

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

    加锁的这部分代码,新来的 goroutine 或者从队列里面唤醒的 goroutine 都会进入如下逻辑,相当于给新人机会

    1. 乐观态度的自旋:判断是否可以自旋,如果可以自旋,就自旋等待;如果有可能,把唤醒标记位置为1,标记外面有唤醒的 goroutine,释放锁的时候就不会去队列里面唤醒了,毕竟已经有人在等待了;
    2. 修改系统状态:跳出自旋后,每个 goroutine 根据当前系统状态修改系统状态:
      1. 非饥饿状态,想要加锁(如果本来就是加锁状态,将加锁位 设置为 1 相当于不变)
      2. 锁被占用 或者 处于饥饿模式下,新增一个等待者
      3. 当前 goroutine 已经进入饥饿了,且锁还没有释放,需要把 Mutex 的状态改为饥饿状态
      4. 如果当前 goroutine 是被唤醒的,清除系统唤醒标记
    3. 利用 CAS 修改系统状态,同一时刻只有一个 goroutine 能够设置成功,但是设置成功并不代表获取到锁了:
      1. 之前是非上锁的正常状态,设置成功说明本次抢锁成功,可以返回去操作临界区了;
      2. 之前是上锁状态或者饥饿状态,本次只是新增了一个等待者,然后根据是否是新来的,去队列队尾或者队首排队,等待叫号;
    4. 从队列中被叫号唤醒,不一定是获取到锁了:
      1. 当前是饥饿状态,那么一定是获取到锁了,因为饥饿状态只把锁给队列的第一个 goroutine
      2. 非饥饿状态,将自己状态置为唤醒,再去抢锁,重复上述过程

    问:系统会不会同时存在 唤醒标志和饥饿标志都为1 的情况呢?

    答:不会。只有等待时间大于 1ms 的才会去设置饥饿标记,也就是只有从队列唤醒的才会去设置,那么从队列中唤醒的 goroutine ,自身的 awoke=true,每当去设置饥饿标记的时候会把唤醒标记清除。

    Unlock()

    Unlock()解锁方法也分为两部分,第一部分是 fast path,可以理解为快捷通道,直接把锁状态位清除,如果此时系统状态恢复到初始状态,说明没有 goroutine 在抢锁等锁,直接返回,否则进入 slow path

    slow path 会根据是否为饥饿状态,做出不一样的反应:

    • 正常状态:唤醒一个 goroutine 去抢锁,等待者数量减一,并将唤醒状态置为 1
    • 饥饿状态:直接唤醒等待队列队首的 goroutine,锁的所有权直接移交(修改等待者数量、是否取消饥饿标记,由唤醒的 goroutine 去处理)。
    func (m *Mutex) Unlock() {
    
        // Fast path: 把锁标记清除
        new := atomic.AddInt32(&m.state, -mutexLocked)
        if new != 0 {
            // 清除完锁标记,发现还有其他状态,比如等待队列不为空,需要唤醒其他 goroutine
            m.unlockSlow(new)
        }
    }
    
    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)
        }
    }
    

    总结

    本篇文章首先通过餐厅的示例,形象的介绍了 Mutex 的运行特性,然后对源码进行逐行分析,学习了 Mutex 加锁 Lock() 以及 解锁 Unlock()是如何实现的。如果本篇文章对有所帮助,点个关注 + 转发哦 ^_^

    更多

    个人博客: https://lifelmy.github.io/

    微信公众号:漫漫Coding路

    相关文章

      网友评论

        本文标题:Go Mutex 源码详解

        本文链接:https://www.haomeiwen.com/subject/ekxnyrtx.html