互斥锁Mutex实现

package main

import (
 "fmt"
 "sync"
)

type counter struct {
 count int
 sync.Mutex
}

func (c *counter) add() {
 c.Lock()
 defer c.Unlock()

 c.count++
}

func (c *counter) value() int {
 c.Lock()
 defer c.Unlock()

 return c.count
}

func main() {
 var wg sync.WaitGroup
 var c counter

 wg.Add(2)

 // goroutine 1
 go func() {
  defer wg.Done()

  for i := 0; i < 5000; i++ {
   c.add()
  }
 }()

 // goroutine 2
 go func() {
  defer wg.Done()

  for i := 0; i < 5000; i++ {
   c.add()
  }
 }()

 wg.Wait()

 fmt.Println(c.value())
}

type Mutex struct {
 state int32
 sema  uint32
}

const (
 // state的第一个bit位，表示是否加锁
 mutexLocked = 1 << iota // mutex is locked
 // state的第二个bit位，表示是否被唤醒
 mutexWoken
 // state的第三个bit位，表示是否处于饥饿模式
 mutexStarving
 // state的[4,32]bit位，表示等待锁的goroutine的数量
 mutexWaiterShift = iota
 // 饥饿时间1毫秒
 starvationThresholdNs = 1e6
)

func (m *Mutex) Lock() {
 // happy path, 还没有人获取锁，即只有当前程序在获取锁，可以直接获取到
 // 通过CAS操作设置m.state值，从0修改为1
 if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
  if race.Enabled {
   race.Acquire(unsafe.Pointer(m))
  }
  return
 }
 m.lockSlow()
}

func (m *Mutex) lockSlow() {
 // 记录goroutine等待的时间
 var waitStartTime int64
 // 标记当前的goroutine是否处于饥饿状态
 starving := false
 // goroutine是否被唤醒
 awoke := false
 // 自旋的次数
 iter := 0
 old := m.state
 for {
  // 满足这里的3个条件进行自旋：
  // 1. 互斥锁已经被锁定，即有goroutine正在占用锁
  // 2. 互斥锁当前不处于饥饿模式
  // 3. 满足尝试自旋的条件runtime_canSpin
  if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
   // 走到这里，说明当前的互斥锁处于正常模式，如果当前互斥锁还没有被唤醒，则标记为唤醒状态
   // 唤醒的goroutine就是当前的goroutine. 通过CAS操作，将互斥锁更新为唤醒状态
   if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
    atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
    awoke = true
   }
   // 进行自旋
   runtime_doSpin()
   // 自旋次数+1
   iter++
   // 再次获取锁的状态
   old = m.state
   continue
  }
   ...
 }

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

func sync_runtime_canSpin(i int) bool {
  // sync_runtime_canSpin函数中在以下四种情况返回false

  // 1. 已经执行了很多次
  // 2. 是单核CPU
  // 3. 没有其他正在运行的P
  // 4. 当前P的G队列为空
 // active_spin为4
 if i >= active_spin || ncpu <= 1 || gomaxprocs <= int32(sched.npidle+sched.nmspinning)+1 {
  return false
 }
 if p := getg().m.p.ptr(); !runqempty(p) {
  return false
 }
 return true
}

func sync_runtime_doSpin() {
 // active_spin_cnt为30
 procyield(active_spin_cnt)
}

TEXT runtime·procyield(SB),NOSPLIT,$0-0
    MOVL    cycles+0(FP), AX
again:
    PAUSE
    SUBL    $1, AX
    JNZ again
    RET

func (m *Mutex) lockSlow() {
 ...
 for {
   ...
  // 走到这里有四种情况：
  // 1. 当前互斥锁处于正常模式，并且锁还没有被释放
  // 2. 当前互斥锁处于饥饿模式，并且锁还没有被释放
  // 3. 当前互斥锁处于正常模式，并且锁已经被释放
  // 4. 当前互斥锁处于饥饿模式，并且锁已经被释放
  new := old
 
  // 之前处于非饥饿状态，新状态进行加锁
  if old&mutexStarving == 0 {
   new |= mutexLocked
  }
  // 之前处于加锁状态或是饥饿状态，当前的goroutine需要等待
  // 之前处于加锁状态，则当前的goroutine还是不会获取到锁的，所以waiter+1
  // 之前是饥饿状态，则当前的goroutine不会设置新状态锁，因为锁被转移给队列中的第一个goroutine.
  if old&(mutexLocked|mutexStarving) != 0 {
   new += 1 << mutexWaiterShift
  }
  
  // 如果当前的goroutine正处于饥饿模式并且旧状态中锁还是处于锁定状态
  // 则将新状态标记为饥饿模式
  if starving && old&mutexLocked != 0 {
   new |= mutexStarving
  }
  // 如果当前的goroutine已经设置为唤醒状态，需要清除新状态中的唤醒标记，因为走到这里，
  // 当前的goroutine要么拿到锁，要么会进入休眠状态，反正不在是唤醒状态
  if awoke {
   if new&mutexWoken == 0 {
    throw("sync: inconsistent mutex state")
   }
   // 当前的goroutine已经唤醒了，重置清理掉new中的唤醒标志
   new &^= mutexWoken
  }
   ...
 }

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

func (m *Mutex) lockSlow() {
    ...
 for {
  ...
  // 成功设置为新状态
  if atomic.CompareAndSwapInt32(&m.state, old, new) {
   // 原来锁的状态已经释放并且不是处于饥饿状态，当前的goroutine请求到了锁的所有权直接返回
   if old&(mutexLocked|mutexStarving) == 0 {
    break // locked the mutex with CAS
   }
   
   // queueLifo表示当前的goroutine以前是否在队列里面，queueLifo为true表示以前就在队列里面
   queueLifo := waitStartTime != 0
   if waitStartTime == 0 {
    // 当前的goroutine以前不在队列里面，将当前的时间保存起来
    waitStartTime = runtime_nanotime()
   }
   // 休眠当前goroutine等待，如果是新的goroutine，它的queueLifo为false，会加入到
   // 等待队列的尾部 ；如果是唤醒的goroutine，它的queueLifo为true，会加入到等待队列的队头
   runtime_SemacquireMutex(&m.sema, queueLifo, 1)
   // 休眠之后被唤醒，计算它是否处处于饥饿状态
   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")
    }
    // delta为7，二级制为0111,此时old二进制最低3位为100,所以old-delta，
    // 相当于waiter减1并加锁
    delta := int32(mutexLocked - 1<<mutexWaiterShift)
    // 如果没有处于饥饿模式或者是最后一个waiter，需要将锁设置为正常模式
    if !starving || old>>mutexWaiterShift == 1 {
     delta -= mutexStarving
    }
    // 设置state为新状态，因为已经获得了锁，直接break返回
    atomic.AddInt32(&m.state, delta)
    break
   }
   awoke = true
   iter = 0
  } else {
   old = m.state
  }
 }

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

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

 // 解锁
 new := atomic.AddInt32(&m.state, -mutexLocked)
 // 如果new为0，说明只有当前的执行体持有锁，不用做其他的处理
 // 如果new非0，需要执行其他操作，尝试唤醒等待者
 if new != 0 {
  m.unlockSlow(new)
 }
}

func (m *Mutex) unlockSlow(new int32) {
 if (new+mutexLocked)&mutexLocked == 0 {
  throw("sync: unlock of unlocked mutex")
 }
 // 没有处于饥饿模式，即处于正常模式
 if new&mutexStarving == 0 {
  old := new
  for {
   // 下面四个条件满足任一一个就直接返回，不做唤醒其他goroutine处理：
   // 1. 没有waiter了，所以也就没有唤醒的对象了
   // 2. 锁处于锁定状态，表明被其他goroutine抢到了
   // 3. 锁处于唤醒状态，表明有其他等待的goroutine被唤醒
   // 4. 锁处于饥饿模式下，将会解锁队列头的goroutine
   if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
    return
   }
   // 将等待的waiter数减一，并设置为唤醒状态
   new = (old - 1<<mutexWaiterShift) | mutexWoken
   // 原子操作设置m.state为新状态new
   if atomic.CompareAndSwapInt32(&m.state, old, new) {
    // 唤醒一个waiter然后返回
    runtime_Semrelease(&m.sema, false, 1)
    return
   }
   // 执行CAS操作失败会走到这里，操作失败的原因是锁的状态已经被别的goroutine改变了
   // 这里更新状态后执行下一个循环
   old = m.state
  }
 } else {
  // 处于饥饿状态，直接唤醒队列头部的goroutine
  runtime_Semrelease(&m.sema, true, 1)
 }
}