• <menu id="sssag"></menu>
  • <menu id="sssag"></menu>
  • go RWMutex 的實現

    Overview

    go 里面的 rwlock 是 write preferred 的,可以避免寫鎖饑餓。

    讀鎖和寫鎖按照先來后到的規則持有鎖,一旦有協程持有了寫鎖,后面的協程只能在寫鎖被釋放后才能得到讀鎖。

    同樣,一旦有 >= 1 個協程寫到了讀鎖,只有等這些讀鎖全部釋放后,后面的協程才能拿到寫鎖。

    下面了解一下 Go 的 RWMutex 是如何實現的吧,下面的代碼取自 go1.17.2/src/sync/rwmutex.go,并刪減了 race 相關的代碼。

    PS: rwmutex 的代碼挺短的,其實讀源碼也沒那么可怕...

    RWMutex 的結構

    RWMutex 總體上是通過: 普通鎖和條件變量來實現的

    type RWMutex struct {
    	w           Mutex  // held if there are pending writers
    	writerSem   uint32 // semaphore for writers to wait for completing readers
    	readerSem   uint32 // semaphore for readers to wait for completing writers
    	readerCount int32  // number of pending readers
    	readerWait  int32  // number of departing readers
    }
    
    Lock
    func (rw *RWMutex) Lock() {
    	// First, resolve competition with other writers.
    	rw.w.Lock()
    	// Announce to readers there is a pending writer.
    	r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    	// Wait for active readers.
    	if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
    		runtime_SemacquireMutex(&rw.writerSem, false, 0)
    	}
    }
    
    Unlock
    const rwmutexMaxReaders = 1 << 30
    
    func (rw *RWMutex) Unlock() {
    	// Announce to readers there is no active writer.
    	r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    	// Unblock blocked readers, if any.
    	for i := 0; i < int(r); i++ {
    		runtime_Semrelease(&rw.readerSem, false, 0)
    	}
    	// Allow other writers to proceed.
    	rw.w.Unlock()
    }
    
    RLock
    func (rw *RWMutex) RLock() {
    	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
    		// A writer is pending, wait for it.
    		runtime_SemacquireMutex(&rw.readerSem, false, 0)
    	}
    }
    
    RUnlock
    func (rw *RWMutex) RUnlock() {
    	if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
    		// Outlined slow-path to allow the fast-path to be inlined
    		rw.rUnlockSlow(r)
    	}
    }
    func (rw *RWMutex) rUnlockSlow(r int32) {
    	// A writer is pending.
    	if atomic.AddInt32(&rw.readerWait, -1) == 0 {
    		// The last reader unblocks the writer.
    		runtime_Semrelease(&rw.writerSem, false, 1)
    	}
    }
    
    Q1: 多個協程并發拿讀鎖,如何保證這些讀鎖協程都不會被阻塞?
    func (rw *RWMutex) RLock() {
    	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
    		// A writer is pending, wait for it.
    		runtime_SemacquireMutex(&rw.readerSem, false, 0)
    	}
    }
    

    拿讀鎖時,僅僅會增加 readerCount,因此讀鎖之間是可以正常并發的

    Q2: 多個協程并發拿寫鎖,如何保證只會有一個協程拿到寫鎖?
    func (rw *RWMutex) Lock() {
    	// First, resolve competition with other writers.
    	rw.w.Lock()
    	// Announce to readers there is a pending writer.
    	r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    	// Wait for active readers.
    	if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
    		runtime_SemacquireMutex(&rw.writerSem, false, 0)
    	}
    }
    

    拿寫鎖時,會獲取 w.Lock,自然能保證同一時間只會有一把寫鎖

    Q3: 在讀鎖被拿到的情況下,新協程拿寫鎖,如果保證寫鎖現成會被阻塞?
    func (rw *RWMutex) Lock() {
    	// First, resolve competition with other writers.
    	rw.w.Lock()
    	// Announce to readers there is a pending writer.
    	r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    	// Wait for active readers.
    	if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
    		runtime_SemacquireMutex(&rw.writerSem, false, 0)
    	}
    }
    

    假設此時有 5 個協程拿到讀鎖,則 readerCount = 5,假設 rwmutexMaxReaders = 100。

    此時有一個新的協程 w1 想要拿寫鎖。

    在執行

    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    

    后, rw.readerCount = -95,r = 5。

    在執行

    atomic.AddInt32(&rw.readerWait, r)
    

    后,rw.readerWait = 5。

    readerWait 記錄了在獲取寫鎖的這一瞬間有多少個協程持有讀鎖。這一瞬間之后,就算有新的協程嘗試獲取讀鎖,也只會增加 readerCount ,而不會動到 readerWait。

    之后執行 runtime_SemacquireMutex() 睡在了 writerSem 這個信號量上面。

    Q4: 在讀鎖被拿到的情況下,新協程拿寫鎖被阻塞,當舊有的讀鎖協程全部釋放,如何喚醒等待的寫鎖協程
    func (rw *RWMutex) RUnlock() {
    	if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
    		// Outlined slow-path to allow the fast-path to be inlined
    		rw.rUnlockSlow(r)
    	}
    }
    func (rw *RWMutex) rUnlockSlow(r int32) {
    	// A writer is pending.
    	if atomic.AddInt32(&rw.readerWait, -1) == 0 {
    		// The last reader unblocks the writer.
    		runtime_Semrelease(&rw.writerSem, false, 1)
    	}
    }
    

    繼續上一步的場景,每當執行 RUnlock 時,readerCount 都會減去1。當 readerCount 為負數時,意味著有協程正在持有或者正在等待持有寫鎖。

    之前的五個讀協程中的四個,每次 RUnlock() 之后,readerCount = -95 - 4 = -99,readerWait = 5 - 4 = 1。

    當最后一個讀協程調用 RUnlock() 之后,readerCount 變成了 -100,readerWait 變成 0,此時會喚醒在 writerSem 上沉睡的協程 w1。

    Q5: 在寫鎖被拿到的情況下,新協程拿讀鎖,如何讓新協程被阻塞?
    func (rw *RWMutex) RLock() {
    	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
    		// A writer is pending, wait for it.
    		runtime_SemacquireMutex(&rw.readerSem, false, 0)
    	}
    }
    

    繼續上面的場景,readerCount = -100 + 1 = -99 < 0。

    新的讀協程 r1 被沉睡在 readerSem 下面。

    假設此時再來一個讀協程 r2,則 readerCount = -98,依舊沉睡。

    Q6: 在寫鎖被拿到的情況下,新協程拿讀鎖,寫鎖協程釋放,如何喚醒等待的讀鎖協程?

    繼續上面的場景,此時協程 w1 釋放寫鎖

    func (rw *RWMutex) Unlock() {
    	// Announce to readers there is no active writer.
    	r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    	// Unblock blocked readers, if any.
    	for i := 0; i < int(r); i++ {
    		runtime_Semrelease(&rw.readerSem, false, 0)
    	}
    	// Allow other writers to proceed.
    	rw.w.Unlock()
    }
    

    在執行

    atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    

    后,r = readerCount = -98 + 100 = 2,代表此時有兩個讀協程 r1 和 r2 在等待

    ps: 如果此時有一些新的協程想要拿讀鎖,他會因為 readerCount = 2 + 1 = 3 > 0 而順利執行下去,不會被阻塞

    之后 for 循環執行兩次,將協程 r1 和 協程 r2 都喚醒了。

    Q7: 在寫鎖被拿到的情況下,有兩個協程分別去搶讀鎖和寫鎖,當寫鎖被釋放時,這兩個協程誰會勝利?
    func (rw *RWMutex) Unlock() {
    	// Announce to readers there is no active writer.
    	r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    	// Unblock blocked readers, if any.
    	for i := 0; i < int(r); i++ {
    		runtime_Semrelease(&rw.readerSem, false, 0)
    	}
    	// Allow other writers to proceed.
    	rw.w.Unlock()
    }
    

    由于是先喚醒讀鎖,再調用 w.Unlock() ,因此肯定是讀協程先勝利!

    認為寫的比較巧妙的兩個點
    1. readerCount 與 rwmutexMaxReaders 的糾纏

      通過 readerCount + rwmutexMaxReaders 以及 readerCount - rwmutexMaxReaders 這兩個操作可以得知當前是否有協程等待/持有寫鎖以及當前等待/持有讀鎖的協程數量

    2. readerCount 與 readerWait 的糾纏

      在 Lock() 時直接將 readerCount 的值賦給 readerWait,在 readerWait = 0 而非 readerCount = 0 是喚醒寫協程,可以避免在 Lock() 后來達到的讀協程先于寫協程被執行。

    posted @ 2022-03-12 18:49  機智的小小帥  閱讀(35)  評論(0編輯  收藏  舉報
    国产在线码观看超清无码视频,人妻精品动漫H无码,十大看黄台高清视频,国产在线无码视频一区二区三区,国产男女乱婬真视频免费,免费看女人的隐私超爽,狠狠色狠狠色综合久久蜜芽