【重造轮子】golang实现可重入锁

介绍

开新坑啦！！
从这篇文章开始，尝试造轮子，包括一些可能有用、也可能没用的轮子。
温故而知新，我相信时常回顾基础的东西能让我们受益良多，这点我深有体会，每过一段时间我都会把《程序员的自我修养》拿出来翻翻，常翻常新，每次读都能有新的收获，开始吧。

“转向毕竟是一个很长的过程，先做起来吧，给我和别的生命一个活下去的机会。”—《三体》

目标

用go 实现可重入锁；

正文

Golang的sync.Mutex是并发场景下的“灵丹妙药”，但是我们真的了解吗？
本文通过对源码的剖析，让我们重新、更全面的认识sync.Mutex，尤其是其优缺点。
在对sync.Mutex有了深入了解后，我们尝试对其进行魔改，实现可重入锁。

sync.Mutex

点击上方的sync.Mutex进入golang.tour我们可以看到，sync.Mutex的简单介绍；
总结如下：

sync.Mutex介绍

当多个goroutine之间需要通信，尤其需要访问（同一份）数据时，需要互斥（锁）来保证一次只有一个goroutine访问数据。
当然，sync.Mutex作为一个同步原语实现了Locker接口（后面会提到），所以只有Lock()、Unlock()两个接口。

多协程计数器demo

比如大家在许多地方看到的例子，多协程累加计数：

var (counter = 1
)func incrCounter() {var wg sync.WaitGroupwg.Add(11)i := 0for {go func() {defer wg.Done()iter:=0for {counter ++if iter > 4 {break}iter ++}}()if i > 9 {break}i ++}wg.Wait()fmt.Println("incrCounter:",counter)os.Exit(1)
}

本地运行结果是incrCounter: 56；大家可以试试本地运行的结果。

多协程计数器加锁

接下来就说到今天的主角了；在上面的多协程计数器代码上，集成sync.Mutex，加两行代码，分别是Lock、Unlock；
代码如下

var (mtx sync.Mutexcounter = 1
)
func incrCounter() {var wg sync.WaitGroupwg.Add(11)i := 0for {go func() {defer wg.Done()iter:=0for {mtx.Lock() counter ++mtx.Unlock()if iter > 4 {break}iter ++}}()if i > 9 {break}i ++}wg.Wait()fmt.Println("incrCounter:",counter)os.Exit(1)
}

本地运行结果是incrCounter: 67；大家可以试试本地运行的结果。(相信很多人看到67会觉得奇怪，没错我是故意的，就是给粗心的同学卖了一个坑，想想为什么是67而不是66？！)
奇怪哎，为啥加了sync.Mutex不一样了呢？！
这里一定要回顾下开头sync.Mutex的介绍！

源码剖析

点击sync.Mutex,我们可以看到它的数据结构；

Mutex数据结构

简单解释下分别是：
state状态位（如果不是远古版本，分为了4段），以及sema信号量变量；不急，后面会细说，这里先了解基本构成；

type Mutex struct {state int32sema  uint32
}

回顾多协程计数器中sync.Mutex的使用例子，核心方法只有两个，为什么只有两个呢？看源码发现原来是实现了Locker interface，因为实现了Locker所以有Lock()、UnLock()；
这里需要重点说一下，golang中的同步原语都会实现Locker ，比如RWMutex；所以以后提到Lock、UnLock那么就可以思考是不是实现了Locker interface；

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

接下来，看下Lock()的实现；看看golang是如何加锁的。

Lock()

照例，点进去看下源码；
如果没加锁，运气很好，加锁就行然后返回；如果已经加过锁了，那么就进入lockSlow，也是加锁逻辑最复杂的地方；
race是做死锁检查的，先不管，捋主体逻辑先；
这里多提一下，fast path一般用来表示捷径或者幸运case，意思是直接成功，不用再执行复杂的逻辑，如果大家看多了开源项目看到fast path就可以跳过这段代码，因为不用看你也能猜到这段代码的意思；

func (m *Mutex) Lock() {// Fast path: grab unlocked mutex.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)m.lockSlow()
}

加锁核心逻辑

先看几个变量；
表示饥饿模式的starving，
唤醒状态的标记awoke，
迭代次数统计的iter，
当前的加锁状态old；

var waitStartTime int64starving := falseawoke := falseiter := 0old := m.state

接下来是饥饿模式的自旋逻辑；

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.statecontinue}

如果是饥饿模式，那么直接拿到锁，新到的goroutine会放入等待队列（等待队列数+1）；

new := old
if old&mutexStarving == 0 {new |= mutexLocked}
if old&(mutexLocked|mutexStarving) != 0 {new += 1 << mutexWaiterShift}

如果当前协程是饥饿模式，并且Mutex并没有标记为饥饿模式，那么就把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.if new&mutexWoken == 0 {throw("sync: inconsistent mutex state")}new &^= mutexWoken}

紧接着，将改变的状态同步到Mutex.state字段；
如果到目前为止当前协程没有获取到锁也没有进入饥饿模式，就可以提前结束当前流程（等待下一次唤醒）；
判断运行时间，并调用runtime_SemacquireMutex休眠，并尝试获取信号量；
运行时间超过1ms就自动进入饥饿模式（starving = 1）；
一旦当前Mutex被标记为饥饿模式，将状态保存到Mutex.state中；
这里需要注意state（int32）中各段的：
第一段（最左边29位）为等待协程的数量；
第二段（1位）饥饿模式标记；
第三段（1位）唤醒标记；
第四段（1位）是否加锁；
如果没有进入饥饿模式，那么将唤醒标记为true，并且重新开始（继续尝试获取锁）；

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 != 0if waitStartTime == 0 {waitStartTime = runtime_nanotime()}runtime_SemacquireMutex(&m.sema, queueLifo, 1)starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNsold = m.stateif 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 = trueiter = 0} else {old = m.state}

UnLock()

记得前面的fast path这个case吗；如果state为1就直接释放锁然后就结束了；
这里需要结合加锁逻辑去看；
回顾下组成state的四个部分：
第一段（最左边29位）为等待协程的数量；
第二段（1位）饥饿模式标记；
第三段（1位）唤醒标记；
第四段（1位）是否加锁；
那么state为，说明：没有等待的协程，没有饥饿模式和唤醒标记，仅仅Mutex被加锁了；

func (m *Mutex) Unlock() {if race.Enabled {_ = m.staterace.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)}
}

否则，进入unlockSlow逻辑；
入口时异常判断，如果释放一个没有加锁的锁则抛出异常；
如果是饥饿模式，将锁直接给饥饿模式的协程，注意是饥饿模式的协程不是等待队列中的等待协程；
不是饥饿模式（比如正常的等待协程）是正常模式，判断锁是否已被锁定或者是否存在唤醒或者是否是饥饿模式，则直接放回，并不释放锁；否则唤醒等待队列中的协程，直接移交给等待者；

func (m *Mutex) unlockSlow(new int32) {if (new+mutexLocked)&mutexLocked == 0 {fatal("sync: unlock of unlocked mutex")}if new&mutexStarving == 0 {old := newfor {// 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}// Grab the right to wake someone.new = (old - 1<<mutexWaiterShift) | mutexWokenif 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.runtime_Semrelease(&m.sema, true, 1)}
}

重入锁与可重入锁

未完待续

魔改 sync.Mutex

未完待续

参考文档

认识可重入锁
Mutex
饥饿模式