go sync.Mutex

并发编程中，不免涉及到共享资源的操作，go也提供简易的互斥锁作为访问控制的手段sync.Mutex，通过简单的Lock()进行加锁,Unlock()进行释放。

接下来我们从源码入手，看看加锁和解锁到底做了什么。

1.sync.Mutex结构

// A Mutex is a mutual exclusion lock.
// The zero value for a Mutex is an unlocked mutex.
//
// A Mutex must not be copied after first use.
type Mutex struct {state int32sema  uint32
}const (mutexLocked = 1 << iota // mutex is lockedmutexWokenmutexStarvingmutexWaiterShift = iotastarvationThresholdNs = 1e6
)

其中state字段是，锁的状态，占32位。末3位startving表示，是否是饥饿状态，woken是不是需要唤醒，locked目前该锁是否被持有

sema表示信号量，协程阻塞等待该信号量，解锁的协程释放信号量从而唤醒等待信号量的协程。

2.lock 过程

// Lock locks m.
// If the lock is already in use, the calling goroutine
// blocks until the mutex is available.
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()
}

1.把state从0改到1，如果成功表示获取到锁。state=0表示当前锁，没有被持有

2.如果不成功执行lockSlow

func (m *Mutex) lockSlow() {var waitStartTime int64starving := falseawoke := falseiter := 0old := m.state //老的状态statefor {// Don't spin in starvation mode, ownership is handed off to waiters// so we won't be able to acquire the mutex anyway.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) {//设置目前有自旋状态，需要awokeawoke = true}runtime_doSpin() //自旋时间，是30个时钟周期iter++ //计数old = m.statecontinue}new := old// Don't try to acquire starving mutex, new arriving goroutines must queue.if old&mutexStarving == 0 { //没有在饥饿状态,只是加锁new |= mutexLocked }if old&(mutexLocked|mutexStarving) != 0 {//饥饿状态或者锁，waiter+1new += 1 << mutexWaiterShift }// The current goroutine switches mutex to starvation mode.// But if the mutex is currently unlocked, don't do the switch.// Unlock expects that starving mutex has waiters, which will not// be true in this case.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 //将new中唤醒标志位清零}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 > starvationThresholdNs //时间大于1e6 纳秒old = 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}}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) //30个时钟
}//go:linkname sync_runtime_SemacquireMutex sync.runtime_SemacquireMutex
func sync_runtime_SemacquireMutex(addr *uint32, lifo bool, skipframes int) {semacquire1(addr, lifo, semaBlockProfile|semaMutexProfile, skipframes)
}func semacquire1(addr *uint32, lifo bool, profile semaProfileFlags, skipframes int) {gp := getg()if gp != gp.m.curg {throw("semacquire not on the G stack")}// Easy case.if cansemacquire(addr) {return}// Harder case://  increment waiter count//    try cansemacquire one more time, return if succeeded//  enqueue itself as a waiter//    sleep// (waiter descriptor is dequeued by signaler)s := acquireSudog()root := semroot(addr)t0 := int64(0)s.releasetime = 0s.acquiretime = 0s.ticket = 0if profile&semaBlockProfile != 0 && blockprofilerate > 0 {t0 = cputicks()s.releasetime = -1}if profile&semaMutexProfile != 0 && mutexprofilerate > 0 {if t0 == 0 {t0 = cputicks()}s.acquiretime = t0}for {lock(&root.lock)// Add ourselves to nwait to disable "easy case" in semrelease.atomic.Xadd(&root.nwait, 1)// Check cansemacquire to avoid missed wakeup.if cansemacquire(addr) {atomic.Xadd(&root.nwait, -1)unlock(&root.lock)break}// Any semrelease after the cansemacquire knows we're waiting// (we set nwait above), so go to sleep.root.queue(addr, s, lifo)//加入等待队列goparkunlock(&root.lock, waitReasonSemacquire, traceEvGoBlockSync, 4+skipframes)//go park主goroutingif s.ticket != 0 || cansemacquire(addr) {break}}if s.releasetime > 0 {blockevent(s.releasetime-t0, 3+skipframes)}releaseSudog(s)
}

func sync_runtime_canSpin(i int) bool {// sync.Mutex is cooperative, so we are conservative with spinning.// Spin only few times and only if running on a multicore machine and// GOMAXPROCS>1 and there is at least one other running P and local runq is empty.// As opposed to runtime mutex we don't do passive spinning here,// because there can be work on global runq or on other Ps.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
}const (locked uintptr = 1active_spin     = 4active_spin_cnt = 30passive_spin    = 1
)

自旋条件：小于四次，cpu >1,有runningP的本地本地队列为空

整体流程：

1.如果被锁或者饥饿状态并且可以自旋

2.满足自旋条件下，自旋等待（设置awak）

3.自旋获取不到锁，g入等待队列,gopark当前gorouting，如果超时，设置饥饿模式

3.unlock


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)}
}func (m *Mutex) unlockSlow(new int32) {if (new+mutexLocked)&mutexLocked == 0 { //unlock已经unlock的panicthrow("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 {//如果没有等待的 goroutine，或者当前有醒着的 goroutine，就不用进行任何操作，直接返回，否则进入下一步，去唤醒某一个 goroutine，并将唤醒标志置为1return}// Grab the right to wake someone.new = (old - 1<<mutexWaiterShift) | mutexWokenif atomic.CompareAndSwapInt32(&m.state, old, new) {runtime_Semrelease(&m.sema, false, 1) //handOff == truereturn}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)// handOff == true}
}func sync_runtime_Semrelease(addr *uint32, handoff bool, skipframes int) {semrelease1(addr, handoff, skipframes)
}func semrelease1(addr *uint32, handoff bool, skipframes int) {root := semroot(addr)atomic.Xadd(addr, 1)// Easy case: no waiters?// This check must happen after the xadd, to avoid a missed wakeup// (see loop in semacquire).if atomic.Load(&root.nwait) == 0 {return}// Harder case: search for a waiter and wake it.lock(&root.lock)if atomic.Load(&root.nwait) == 0 {// The count is already consumed by another goroutine,// so no need to wake up another goroutine.unlock(&root.lock)return}s, t0 := root.dequeue(addr)//出队if s != nil {atomic.Xadd(&root.nwait, -1)}unlock(&root.lock)if s != nil { // May be slow or even yield, so unlock firstacquiretime := s.acquiretimeif acquiretime != 0 {mutexevent(t0-acquiretime, 3+skipframes)}if s.ticket != 0 {throw("corrupted semaphore ticket")}if handoff && cansemacquire(addr) {s.ticket = 1}readyWithTime(s, 5+skipframes)if s.ticket == 1 && getg().m.locks == 0 {// Direct G handoff// readyWithTime has added the waiter G as runnext in the// current P; we now call the scheduler so that we start running// the waiter G immediately.// Note that waiter inherits our time slice: this is desirable// to avoid having a highly contended semaphore hog the P// indefinitely. goyield is like Gosched, but it emits a// "preempted" trace event instead and, more importantly, puts// the current G on the local runq instead of the global one.// We only do this in the starving regime (handoff=true), as in// the non-starving case it is possible for a different waiter// to acquire the semaphore while we are yielding/scheduling,// and this would be wasteful. We wait instead to enter starving// regime, and then we start to do direct handoffs of ticket and// P.// See issue 33747 for discussion.goyield()//调度}}
}func goyield() {checkTimeouts()mcall(goyield_m)
}

go sync.Mutex相关推荐

Go 学习笔记（66）— Go 并发同步原语（sync.Mutex、sync.RWMutex、sync.Once）
1. 竞态条件一旦数据被多个线程共享,那么就很可能会产生争用和冲突的情况.这种情况也被称为竞态条件(race condition),这往往会破坏共享数据的一致性. 举个例子,同时有多个线程连续向同一 ...
Go 学习笔记（23）— 并发（02）[竞争，锁资源，原子函数sync/atomic、互斥锁sync.Mutex]
本文参考 <Go 语言实战> 1. 竞争状态简述如果两个或者多个 goroutine 在没有互相同步的情况下,访问某个共享的资源,并试图同时读和写这个资源,就处于相互竞争的状态,这种情况 ...
对于sync.Mutex使用注意事项
1.sync.Mutex的初始化注意事项 type MemProvider struct { lock *sync.Mutex //用来锁 sessions map[ ...
golang sync.Mutex 互斥锁使用实例
实例: var mutex sync.Mutex //互斥锁 func printer(str string){mutex.Lock() //加锁defer mutex.Unlock() //解锁fo ...
Go的sync.Mutex(七)：互斥锁锁定一个资源只有一个协程操作其他等待
简介多个协程会操作一个特定资源,就会出现意想不到的错误类比脏读幻读等,所以我们使用互斥锁, 一个协程使用特定资源的时候进行锁定,用完解锁, 再让其他协程使用,所以其他协程想使用此资源,必须自己给资源 ...
Golang sync.Mutex 与 sync.RWMutex
文章目录 1.sync.Mutex 2.sync.RWMutex 2.1 Lock()与Unlock() 2.2 RLock() 和 RUnlock() 2.3 错误使用异常参考文献 Golang ...
sync.Mutex 与 sync.WaitGroup 使用示例
使用 sync.Mutex 与 sync.WaitGroup 线程不安全的用法: {var wg sync.WaitGroupcount := 0for i := 0; i < 10; i++ ...
go vet 报错：xx_xx passes lock by value: sync.Map contains sync.Mutex； call of xx_xx copies lock value
问题描述: test.go: package main import "fmt" import "sync"var syncMapTest sync.Mapfu ...
Golang sync.Mutex分析
sync.Mutex是一个不可重入的排他锁.当一个 goroutine 获得了这个锁的拥有权后, 其它请求锁的 goroutine 就会阻塞在 Lock 方法的调用上,直到锁被释放. sync.Mut ...

go sync.Mutex

1.sync.Mutex结构

2.lock 过程

3.unlock

go sync.Mutex相关推荐

最新文章

热门文章