互斥锁 #

type Locker interface {
    Lock() // 请求锁
    Unlock() // 释放锁
}

import "sync"

var (
	mu sync.Mutex // guards balance 
	balance int
)

func Deposit(amount int) {
	mu.Lock()
	defer mu.Unlock()
	balance = balance + amount
}

func Balance() int {
	mu.Lock()
	defer mu.Unlock()
	b := balance
	return b
}

import (
    "fmt"
    "sync"
)
    
func main() {
    var count = 0
    // 使用 WaitGroup 等待 10 个 goroutine 完成
    var wg sync.WaitGroup
    wg.Add(10)
    for i := 0; i < 10; i++ {
        go func() {
            defer wg.Done()
            // 对变量 count 执行 10 次加 1
            for j := 0; j < 1000; j++ {
                count++
            }
        }()
    }
    // 等待 10 个 goroutine 完成
    wg.Wait()
    fmt.Println(count)
}

// src/sync/mutex.go#L34
type Mutex struct {
    state int32
	sema  uint32
}

const (
    mutexLocked = 1 << iota // 1
    mutexWoken // 2
    mutexStarving // 4
    mutexWaiterShift = iota // 3
	starvationThresholdNs = 1e6 // 1000000
)

func (m *Mutex) Lock() {
    // Fast path: grab unlocked mutex.
	// 没有 goroutine 持有锁，也没有等待的 goroutine，当前 goroutine 可以直接获得锁
    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()
}

func (m *Mutex) lockSlow() {
	var waitStartTime int64
	starving := false // 当前 goroutine 的饥饿标记
	awoke := false // 唤醒标记
	iter := 0 // 自旋次数
	old := m.state // 当前锁的状态
	for {
		// 锁是非饥饿模式并且还没被释放，尝试自旋
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
			// 尝试设置 mutexWoken 标志来通知解锁，以避免唤醒其他阻塞的 goroutine
			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
        if old&mutexStarving == 0 {
			new |= mutexLocked // 非饥饿状态，加锁
		}
		if old&(mutexLocked|mutexStarving) != 0 {
			new += 1 << mutexWaiterShift // waiter 数量加 1
		}
        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 // 新状态清除唤醒标记
		}
        // 设置新状态
		if atomic.CompareAndSwapInt32(&m.state, old, new) {
            // 再次检查，原来锁的状态已释放，并且不是饥饿状态，正常请求到了锁，返回
			if old&(mutexLocked|mutexStarving) == 0 {
				break // locked the mutex with CAS
			}
			// 处理饥饿状态
			// 如果之前就在该队列里面，就加入到队列头
			queueLifo : waitStartTime != 0
			if waitStartTime == 0 {
				waitStartTime = runtime_nanotime()
			}
            // runtime_SemacquireMutex 通过信号量保证资源不会被两个 goroutine 获取
			// runtime_SemacquireMutex 会在方法中不断尝试获取锁并陷入休眠等待信号量的释放
			// 也就是这里会阻塞等待
			// 一旦当前 goroutine 可以获取信号量，它就会立刻返回，剩余代码也会继续执行
			runtime_SemacquireMutex(&m.sema, queueLifo, 1)
			// 在正常模式下，这段代码会设置唤醒和饥饿标记、重置迭代次数并重新执行获取锁的循环
            // 在饥饿模式下，当前 goroutine 会获得锁，如果等待队列中只存在当前 goroutine，锁还会从饥饿模式中退出
			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 := int32(mutexLocked - 1<<mutexWaiterShift)
				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))
	}	
}

//go:linkname sync_runtime_doSpin sync.runtime_doSpin
//go:nosplit
func sync_runtime_doSpin() {
	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) Unlock() {
	if race.Enabled {
		_ = m.state
		race.Release(unsafe.Pointer(m))
	}

	// Fast path: drop lock bit.
	// new == 0 成功释放锁
	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)
	}
}

func (m *Mutex) unlockSlow(new int32) {
	if (new+mutexLocked)&mutexLocked == 0 { // unlock 一个未加锁的锁
		fatal("sync: unlock of unlocked mutex")
	}
	if new&mutexStarving == 0 { // 正常模式
		old := new
		for {
			// 不存在等待者 或者 mutexLocked、mutexStarving、mutexWoken 状态不都为 0
			// 则不需要唤醒其他等待者
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
			// 存在等待者，通过 runtime_Semrelease 唤醒等待者并移交锁的所有权
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
				runtime_Semrelease(&m.sema, false, 1)
				return
			}
			old = m.state
		}
	} else { // 饥饿模式
		// 直接调用 runtime_Semrelease 将当前锁交给下一个正在尝试获取锁的等待者，等待者被唤醒后会得到锁，在这时还不会退出饥饿状态
		runtime_Semrelease(&m.sema, true, 1)
	}
}