sync.Mutex源码解读

未加锁

package main

import (
	"fmt"
	"sync"
	"time"
)

func main() {
    
	x := 0
	now := time.Now()
	var group sync.WaitGroup
	group.Add(1000)
	for i := 0; i < 1000; i++ {
    
		go func() {
    
			x++
			defer group.Done()
		}()
	}
	// 等待所有的goroutine执行完毕再结束
	group.Wait()
	fmt.Println(x)
	fmt.Println(time.Now().UnixMilli() - now.UnixMilli())
}

正常的期望值应该是1000;但是由于我们没有进行加锁,goroutineA在写的同时,goroutineB也在写,同时的去读,会有两个goroutine读到一样的值,然后去执行x++,导致最后数据不正确只有983

加锁

package main

import (
	"fmt"
	"sync"
	"time"
)

func main() {
    
	x := 0
	now := time.Now()
	var group sync.WaitGroup
	var mutex sync.Mutex
	group.Add(1000)
	for i := 0; i < 1000; i++ {
    
		go func() {
    
			mutex.Lock()
			x++
			defer mutex.Unlock()
			defer group.Done()
		}()
	}
	// 等待所有的goroutine执行完毕再结束
	group.Wait()
	fmt.Println(x)
	fmt.Println(time.Now().UnixMilli() - now.UnixMilli())
}

源码

Mutex

// Mutex是一种互斥锁.
// 互斥锁的0值是未锁定的互斥锁.
// 互斥锁在第一次使用后不能被复制
type Mutex struct {
    
	state int32 // 表示当前锁的状态
	sema  uint32 // 信号量变量,用来控制goroutine阻塞休眠和唤醒
}

// state状态
const (
	mutexLocked = 1 << iota // 互斥锁的锁定状态
	mutexWoken	// 从正常模式被唤醒
	mutexStarving	// 当前互斥锁进入饥饿状态
	mutexWaiterShift = iota // 等待的goroutine数量` 
	starvationThresholdNs = 1e6
)

• Mutex两种操作模式
  • 正常模式(非公平锁)
    • 在正常模式下,等待者按照FIFO(先进先出)顺序排队,但是唤醒的goroutine并不拥有锁,而是与新到达的goroutines争夺锁的所有权,新加入的goroutines有一个优势:**它们已经在CPU上运行,可能是一个也可能是多个,**但是只会唤醒一个goroutine,唤醒的这个goroutine很有可能会输,在这种情况下,它被排在等待队列前面.如果一个等待着获得互斥锁的时间超过1ms,它将从正常模式切换到饥饿模式
  • 饥饿模式(公平锁)
    • 在饥饿模式下,互斥锁的所有权直接从正在解锁的goroutine传递给队列前面的goroutine,新到达的goroutines不会试图获取锁,也不会试图自旋,他们会把自己排在等待队列的尾部
    • 如果获取到的锁goroutine是队列中最后一个goroutine,它将从饥饿模式切换到正常模式
    • 如果获取到锁的goroutine等待时间小于1ms,它将从饥饿模式切换到正常模式
  • 普通模式的性能要好很多,因为即使有阻塞的goroutines,goroutine也可以连续多次获取锁,饥饿模式解决了取锁公平的问题,但是性能会下降,正常模式和饥饿模式来回的切换这其实是性能和公平的一个平衡模式

Mutex.Lock

// 加锁方法,如果进行重复加锁会报错
func (m *Mutex) Lock() {
    
	// 通过CAS判断当前锁的状态;
    // 如果锁是空闲的,即m.state==0
    // 则对其加锁,即m.state=1
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
    
		if race.Enabled {
    
			race.Acquire(unsafe.Pointer(m))
		}
		return
	}
	// 当前锁已被其他goroutine获取到
	m.lockSlow()
}

func (m *Mutex) lockSlow() {
    
    // 用来计算等待时间
	var waitStartTime int64 
    // 饥饿模式标记,如果等待时间超过1ms,starving设置为true
    // 后续操作会把m也标记为饥饿状态
	starving := false 
    // 表示当前goroutine是否唤醒
    // 当goroutine在自旋时,相当于CPU上已经有在等待的goroutine
    // 为避免m解锁时在唤醒其他的goroutine,自旋时要把m置为唤醒状态
    // m处于唤醒状态时,awoke也置为true
	awoke := false
    // 自旋次数
	iter := 0
    // 当前锁的状态
	old := m.state
	for {
    
		// 判断是否可以进入自旋
        // 1.当前处于非饥饿模式
        // 2.runtime_canSpin(iter);自旋次数小于4
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
    
            // !awoke:判断当前goroutine不是在唤醒状态
            // old&mutexWoken == 0: 当前没有其它正在唤醒的goroutine
            // old>>mutexWaiterShift != 0: 等待队列中有正在等待的goroutine
            // atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken): 尝试将当前锁的低2位的Woken状态位设置为1,表示已被唤醒,这是为了通知在解锁Unlock()中不要再唤醒其他的goroutine了
			if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
				atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
    
                 // 将当前goroutine是唤醒
				awoke = true
			}
            // 进行自旋
			runtime_doSpin()
            // 自旋次数++
			iter++
            // 记录当前锁的状态
			old = m.state
			continue
		}
        // 基于old声明一个新状态
		new := old
		// 不要试图获取饥饿的互斥锁,新到达的goroutine必须排队
		if old&mutexStarving == 0 {
    
			new |= mutexLocked
		}
        // 如果old已经加锁或处于集合模式,按照FIFO排队
		if old&(mutexLocked|mutexStarving) != 0 {
    
			new += 1 << mutexWaiterShift
		}
		// 如果m处于饥饿模式,就将starving置为1表示饥饿
		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) {
    
			if old&(mutexLocked|mutexStarving) == 0 {
    
                // 如果原来的锁没有加锁并且不是饥饿模式
                // 表示当前goroutine获取到锁了,直接break
				break 
			}
			// 还没有获取倒锁;waitStartTime就是等待时间
            // 如果不为0就放置到对头
			queueLifo := waitStartTime != 0
            // 为0放置到队尾
			if waitStartTime == 0 {
    
				waitStartTime = runtime_nanotime()
			}
            // 阻塞等待
			runtime_SemacquireMutex(&m.sema, queueLifo, 1)
            // 唤醒后,检查是否为饥饿模式
            // 1.当前为饥饿模式
            // 2. 等待超过1ms
			starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
            // 再次获取m的状态
			old = m.state
			if old&mutexStarving != 0 {
    
				if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
    
                    // 当前锁既不是被获取状态也不是被唤醒状态,或者等待队列为空
                    // 代表出现互斥的状态不一致问题
					throw("sync: inconsistent mutex state")
				}
                // 当前goroutine获取到锁了,等待队列-1
				delta := int32(mutexLocked - 1<<mutexWaiterShift)
				if !starving || old>>mutexWaiterShift == 1 {
    
                    // 当前非饥饿模式或队列仅剩一个goroutine退出饥饿模式
					delta -= mutexStarving
				}
                // 更新状态值并break
				atomic.AddInt32(&m.state, delta)
				break
			}
            // 设置状态为唤醒状态;并重置自旋次数
			awoke = true
			iter = 0
		} else {
    
            // 锁被其它goroutine占用,还原状态继续for循环
			old = m.state
		}
	}

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

Mutex.Unlock

// 解锁方法,如果没有加锁就进行解锁会报错
func (m *Mutex) Unlock() {
    
	if race.Enabled {
    
		_ = m.state
		race.Release(unsafe.Pointer(m))
	}

	// 将m.state置为0,并复制给new
    // 如果new为0;表示解锁成功;直接退出
	new := atomic.AddInt32(&m.state, -mutexLocked)
	if new != 0 {
    
		// new不为0;说明当前锁没有被占用,但是有等待的goroutine未被唤醒
		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直接返回
            // 如果锁处于加锁状态;代表有goroutine获取到锁;直接返回
            // 如果锁处于唤醒状态;代表有goroutine被唤醒,直接返回
            // 如果当前处于饥饿模式,锁之后会按照FIFO规则给对头的goroutine
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
    
				return
			}
			// 把锁唤醒,等待队列-1
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
    
                // 抢占成功唤醒一个goroutine返回
				runtime_Semrelease(&m.sema, false, 1)
				return
			}
            // 抢占不成功;更新状态信息;继续for
			old = m.state
		}
	} else {
    
		// 饥饿模式
        // 直接按照FIFO规则给对头的goroutine
		runtime_Semrelease(&m.sema, true, 1)
	}
}