Golang中Mutex的实现

一.Golang中的锁

       Golang的提供的同步机制有sync模块下的Mutex、WaitGroup以及语言自身提供的chan等。 这些同步的方法都是以runtime中实现的底层同步机制（cas、atomic、spinlock、sem）为基础的。

1 cas(Compare And Swap)和原子运算是其他同步机制的基础

• 原子操作：指那些不能够被打断的操作被称为原子操作，当有一个CPU在访问这块内容addr时，其他CPU就不能访问。
• CAS：比较及交换，其实也属于原子操作，但它是非阻塞的，所以在被操作值被频繁变更的情况下,CAS操作并不那么容易成功，不得不利用for循环以进行多次尝试。

2 自旋锁(spinlock)

       自旋锁是指当一个线程在获取锁的时候，如果锁已经被其他线程获取，那么该线程将循环等待，然后不断地判断是否能够被成功获取，知直到获取到锁才会退出循环。获取锁的线程一直处于活跃状态 Golang中的自旋锁用来实现其他类型的锁,与互斥锁类似，不同点在于，它不是通过休眠来使进程阻塞，而是在获得锁之前一直处于活跃状态(自旋)。

3 信号量

       实现休眠和唤醒协程的一种方式。

信号量有两个操作P和V
P(S):分配一个资源
1. 资源数减1：S=S-1
2. 进行以下判断
    如果S<0，进入阻塞队列等待被释放
    如果S>=0,直接返回

V(S):释放一个资源
1. 资源数加1：S=S+1
2. 进行如下判断
    如果S>0，直接返回
    如果S<=0,表示还有进程在请求资源，释放阻塞队列中的第一个等待进程
    
golang中信号量操作：runtime/sema.go
P操作：runtime_Semacquire
V操作：runtime_Semrelease

二 Mutex的实现

type Mutex struct {
	state int32     //0(可用) 1(被锁) 2~31等待队列计数
	sema  uint32    //信号量，向处于Gwaitting的G发送信号
}

const (
	mutexLocked = 1 << iota // 1 互斥锁是锁定的
	mutexWoken              // 2 唤醒锁
	mutexWaiterShift = iota // 2 统计阻塞在这个互斥锁上的goroutine数目需要移位的数值
)

2.1 互斥锁

Lock方法申请对mutex加锁的时候分为两种情况：

• 有冲突：通过SAS把当前状态设置为加锁状态。
• 无冲突：通过调用semacquire函数进入休眠状态，等待其他协程释放锁的时候唤醒。

//如果已经加锁，那么当前协程进入休眠阻塞，等待唤醒
func (m *Mutex) Lock() {

    // 快速加锁：CAS更新state为locked
    if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
        return
    }

    awoke := false //当前goroutine是否被唤醒
    for {
        old := m.state // 保存当前state的状态
        new := old | mutexLocked // 新值locked位设置为1
        // 如果当前处于加锁状态，新到来的goroutine进入等待队列
        if old&mutexLocked != 0 {
            new = old + 1<<mutexWaiterShift
        }
        if awoke {
            //如果被唤醒，新值需要重置woken位为 0
            new &^= mutexWoken
        }
        
        // 两种情况会走到这里：1.休眠中被唤醒 2.加锁失败进入等待队列
        // CAS 更新，如果更新失败，说明有别的协程抢先一步，那么重新发起竞争。
        if atomic.CompareAndSwapInt32(&m.state, old, new) {
            // 如果更新成功，有两种情况
            // 1.如果为 1，说明当前 CAS 是为了更新 waiter 计数
            // 2.如果为 0，说明是抢锁成功，那么直接 break 退出。
            if old&mutexLocked == 0 { 
                break
            }
            runtime_Semacquire(&m.sema) // 此时如果 sema <= 0 那么阻塞在这里等待唤醒，也就是 park 住。走到这里都是要休眠了。
            awoke = true  // 有人释放了锁，然后当前 goroutine 被 runtime 唤醒了，设置 awoke true
        }
    }

    if raceenabled {
        raceAcquire(unsafe.Pointer(m))
    }
}

UnLock 解锁分两步：

• 解锁，通过CAS操作把当前状态设置为解锁状态。
• 唤醒休眠协程，CAS操作把当前状态的waiter数减1,然后唤醒休眠goroutine。

//锁没有和某个特定的协程关联，可以由一个协程lock，另一个协程unlock
func (m *Mutex) Unlock() {
    if raceenabled {
        _ = m.state
        raceRelease(unsafe.Pointer(m))
    }

    // CAS更新state的状态为locked ======注意：解锁的瞬间可能会有新的协程到来并抢到锁
    new := atomic.AddInt32(&m.state, -mutexLocked)
    // 释放了一个没上锁的锁会panic：原先的lock位为0
    if (new+mutexLocked)&mutexLocked == 0 { 
        panic("sync: unlock of unlocked mutex")
    }
    
    //判断是否需要释放资源
    old := new
    for {
        /**
         * 不需要唤醒的情况
         * 1.等待队列为0
         * 2.已经有协程抢到锁（上面的瞬间抢锁）
         * 3.已经有协程被唤醒
         */
        if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken) != 0 {
            return
        }
        //将waiter计数位减一，并设置state为woken(唤醒)
        //问：会同时有多个被唤醒的协程存在吗
        new = (old - 1<<mutexWaiterShift) | mutexWoken
        if atomic.CompareAndSwapInt32(&m.state, old, new) {
            runtime_Semrelease(&m.sema) // cas成功后，再做sema release操作，唤醒休眠的 goroutine
            return
        }
        old = m.state
    }
}

一代互斥锁的问题：

• 处于休眠中的goroutine优先级低于当前活跃的，unlock解锁的瞬间最新的goroutine会抢到锁。
• 大多数果锁的时间很短，所有的goroutine都要休眠，增加runtime调度开销。

2.2 自旋锁

Lock 方法申请对 mutex 加锁的时候分三种情况：

1. 无冲突 通过 CAS 操作把当前状态设置为加锁状态。
2. 有冲突 开始自旋，并等待锁释放，如果其他 goroutine 在这段时间内释放了该锁，直接获得该锁；如果没有释放，进入3。
3. 有冲突 通过调用 semacquire 函数来让当前 goroutine 进入等待状态，等待其他协程释放锁的时候唤醒。

func (m *Mutex) Lock() {
    //快速加锁，逻辑不变
    if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
        if race.Enabled {
            race.Acquire(unsafe.Pointer(m))
        }
        return
    }

    awoke := false
    iter := 0
    for {
        old := m.state
        new := old | mutexLocked
        if old&mutexLocked != 0 { // 如果当前己经上锁，那么判断是否可以自旋
            //短暂的自旋过后如果无果，就只能通过信号量让当前goroutine进入休眠等待了
            if runtime_canSpin(iter) {
                // Active spinning makes sense.
                /**
                 * 自旋的操作：设置state为woken,这样在unlock的时候就不会唤醒其他协程.
                 * 自旋的条件：
                 * 1.当前协程未被唤醒 !awoke
                 * 2.其他协程未被唤醒 old&mutexWoken == 0
                 * 3.等待队列大于0
                 */
                if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
                    atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
                    awoke = true
                }
                //进行自旋操作
                runtime_doSpin()
                iter++
                continue
            }
            new = old + 1<<mutexWaiterShit
        }
        if awoke {
            //todo 为什么加这个判断
            if new&mutexWoken == 0 {
                panic("sync: inconsistent mutex state")
            }
            new &^= mutexWoken
        }
        if atomic.CompareAndSwapInt32(&m.state, old, new) {
            if old&mutexLocked == 0 {
                break
            }
            runtime_Semacquire(&m.sema)
            awoke = true
            iter = 0
        }
    }

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

二代自旋锁的问题：

• 还是没有解决休眠进程优先级低的问题

2.3 公平锁

基本逻辑：

1. Mutex 两种工作模式，normal 正常模式，starvation 饥饿模式。normal 情况下锁的逻辑与老版相似，休眠的 goroutine 以 FIFO 链表形式保存在 sudog 中，被唤醒的 goroutine 与新到来活跃的 goroutine 竞解，但是很可能会失败。如果一个 goroutine 等待超过 1ms，那么 Mutex 进入饥饿模式。
2. 饥饿模式下，解锁后，锁直接交给 waiter FIFO 链表的第一个，新来的活跃 goroutine 不参与竞争，并放到 FIFO 队尾。
3. 如果当前获得锁的 goroutine 是 FIFO 队尾，或是等待时长小于 1ms，那么退出饥饿模式。
4. normal 模式下性能是比较好的，但是 starvation 模式能减小长尾 latency。

LOCK流程：

1. 无冲突： 通过 CAS 操作把当前状态设置为加锁状态。
2. 有冲突，开始自旋： 如果是饥饿模式禁止自旋，开始自旋，并等待锁释放，如果其他 goroutine 在这段时间内释放了该锁，直接获得该锁；如果没有释放，进入3。
3. 有冲突，且已经过了自旋阶段 ：通过调用 semacquire 函数来让当前 goroutine 进入等待状态，等待其他协程释放锁的时候唤醒，休眠前：如果是饥饿模式，把当前协程放到队列最前面；唤醒后：如果是饥饿模式唤醒的，直接获得锁。

type Mutex struct {
	state int32 
	sema  **uint32**
}

// A Locker represents an object that can be locked and unlocked.
type Locker interface {
	Lock()
	Unlock()
}

//为什么使用位掩码表达式
//第3位到第32位表示等待在mutex上协程数量
const (
	mutexLocked = 1 << iota // mutex is locked 
	mutexWoken                                  
	mutexStarving           //新增饥饿状态
	mutexWaiterShift = iota                     
	starvationThresholdNs = 1e6 //饥饿状态的阈值：等待时间超过1ms就会进入饥饿状态
)


func (m *Mutex) Lock() {
	//快速加锁：逻辑不变
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
		if race.Enabled {
			race.Acquire(unsafe.Pointer(m))
		}
		return
	}

	var waitStartTime int64 //等待时间
	starving := false   //饥饿标记
	awoke := false      //唤醒标记
	iter := 0           //循环计数器
	old := m.state      //保存当前锁状态
	for {
	    // 自旋的时候增加了一个判断：如果处于饥饿状态就不进入自旋，因为饥饿模式下，释放的锁会直接给等待队列的第一个，当前协程直接进入等待队列
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
			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
		// 当mutex不处于饥饿状态的时候，将new值设置为locked，也就是说如果是饥饿状态，新到来的goroutine直接排队
		if old&mutexStarving == 0 {
			new |= mutexLocked
		}
		// 当mutex处于加锁锁或者饥饿状态时，新到来的goroutine进入等待队列
		if old&(mutexLocked|mutexStarving) != 0 {
			new += 1 << mutexWaiterShift
		}
		// 当等待时间超过阈值，当前goroutine切换mutex为饥饿模式，如果未加锁，就不需要切换
		if starving && old&mutexLocked != 0 {
			new |= mutexStarving
		}
		if awoke {
			if new&mutexWoken == 0 {
				throw("sync: inconsistent mutex state")
			}
			new &^= mutexWoken
		}
		if atomic.CompareAndSwapInt32(&m.state, old, new) {
		    // mutex 处于未加锁，正常模式下，当前 goroutine 获得锁
			if old&(mutexLocked|mutexStarving) == 0 {
				break // locked the mutex with CAS
			}
			// 如果已经在排队了，就排到队伍的最前面
			queueLifo := waitStartTime != 0
			if waitStartTime == 0 {
				waitStartTime = runtime_nanotime()
			}
			// queueLifo 为真的时候，当前goroutine会被放到队头，
			// 也就是说被唤醒却没抢到锁的goroutine放到最前面
			runtime_SemacquireMutex(&m.sema, queueLifo)
			// 当前goroutine等待时间超过阈值，切换为饥饿模式，starving设置为true
			starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
			old = m.state
			//如果当前是饥饿模式
			if old&mutexStarving != 0 {
				if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
					throw("sync: inconsistent mutex state")
				}
				// 如果切换为饥饿模式，等待队列计数减1
				delta := int32(mutexLocked - 1<<mutexWaiterShift)
				// 如果等待时间小于1ms或者自己是最后一个被唤醒的，退出饥饿模式
				if !starving || old>>mutexWaiterShift == 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 解锁分两步：

1. 解锁，通过CAS操作把当前状态设置为解锁状态。
2. 唤醒休眠协程，CAS操作把当前状态的waiter数减1,然后唤醒休眠goroutine，如果是饥饿模式的话，唤醒等待队列的第一个。

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

	new := atomic.AddInt32(&m.state, -mutexLocked)
	if (new+mutexLocked)&mutexLocked == 0 {
		throw("sync: unlock of unlocked mutex")
	}
	
	if new&mutexStarving == 0 {
	// 正常模式
		old := new
		for {
		    /**
             * 不需要唤醒的情况
             * 1.等待队列为0
             * 2.已经有协程抢到锁（上面的瞬间抢锁）
             * 3.已经有协程被唤醒
             * 4.处于饥饿模式 在饥饿模式获取到锁的协程仍然处于饥饿状态，新的goroutine无法获取到锁
             */
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
			// Grab the right to wake someone.
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
				runtime_Semrelease(&m.sema, false)
				return
			}
			old = m.state
		}
	} else {
	    // 饥饿模式
		runtime_Semrelease(&m.sema, true)
	}
}