日本免费高清视频-国产福利视频导航-黄色在线播放国产-天天操天天操天天操天天操|www.shdianci.com

學無先后，達者為師

網站首頁 編程語言 正文

go?RWMutex的實現示例_Golang

作者：機智的小小帥 ? 更新時間： 2022-05-15 編程語言

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() ，因此肯定是讀協程先勝利！

認為寫的比較巧妙的兩個點

  • readerCount 與 rwmutexMaxReaders 的糾纏

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

  • readerCount 與 readerWait 的糾纏

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

原文鏈接：https://www.cnblogs.com/XiaoXiaoShuai-/p/15998061.html

欄目分類
最近更新