sync.Mutex

Kilitler işletim sistemlerinde önemli bir senkronizasyon ilkelidir. Go standart kütüphanesinde iki uygulama sağlar: mutex ve read-write lock, karşılık gelen:

• sync.Mutex, mutex kilidi: read-read özel, read-write özel, write-write özel
• sync.RWMutex, read-write kilidi: read-read paylaşımlı, read-write özel, write-write özel

Kullanım senaryoları çok yaygındır, eşzamanlı koşullar altında paylaşılan belleği sıralı erişim ve değiştirme için korumak için kullanılır, aşağıdaki örnekte gösterildiği gibi:

import (
	"fmt"
	"sync"
)

func main() {
	var i int
	var wg sync.WaitGroup
	var mu sync.Mutex

	for range 10 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			mu.Lock()
			viewI := i
			mu.Unlock()

			viewI++

			mu.Lock()
			i = viewI
			mu.Unlock()
		}()
	}

	wg.Wait()
	fmt.Println(i)
}

Kilit koruması olmadan, bu fonksiyonun çıktısı her yürütüldüğünde farklı olabilir ve öngörülemez. Açıkçası, çoğu senaryoda böyle durumların oluşmasını istemeyiz. Bu örnek çoğu insan için çok basittir ve zaten kilitleri kullanmada yetkin olabilirsiniz, ancak Go'nun kilitlerinin dahili olarak nasıl uygulandığını anlamıyor olabilirsiniz. Kodun kendisi karmaşık değildir ve bu makale detaylı bir açıklama sağlayacaktır.

Locker

Başlamadan önce, sync.Locker tipine bakalım, bu Go tarafından tanımlanan bir arayüzdür:

// A Locker represents an object that can be locked and unlocked.
type Locker interface {
	Lock()
	Unlock()
}

Sağladığı yöntemler çok basit ve kolay anlaşılır: kilitle ve kilidi aç. Ancak, Go'nun arayüz uygulaması sözleşmeden daha iyi olduğu için, çoğu insan bunu hiç görmemiş olabilir. Burada sadece kısaca bahsediliyor çünkü o kadar önemli değil. Daha sonra tartışılan her iki kilit de bu arayüzü uygular.

Mutex

Mutex Mutex tip tanımı sync/mutex.go dosyasında bulunur. Bir struct tipidir:

type Mutex struct {
	state int32
	sema  uint32}

Alan tanımları:

• state, kilidin durumunu temsil eden alan
• sema, yani semaphore, daha sonra açıklanacaktır

İlk olarak state hakkında konuşalım:

const (
    mutexLocked = 1 << iota // mutex is locked
    mutexWoken
    mutexStarving
)

state 32-bit tamsayı tipidir. Düşük 3 bit yukarıdaki üç durumu temsil etmek için kullanılır. Bu üç durum bağımsız değildir; bir arada bulunabilirler.

• mutexLocked=1, kilitli
• mutexWoken=2, uyanık
• mutexStarving=4, açlık modu

Yüksek 29 bit kaç goroutine'in kilit için beklediğini temsil etmek için kullanılır. Yani teorik olarak, bir mutex en fazla 2^29+1 goroutine tarafından aynı anda kullanılabilir. Ancak gerçekte, bu kadar çok goroutine olması pek olası değildir. Her biri sadece 2KB (başlangıç yığın alanı) kaplasa bile, bu sayıda goroutine oluşturmak için gereken bellek alanı yaklaşık 1TB olurdu.

+-----------------------------------+---------------+------------+-------------+
|              waiter               | mutexStarving | mutexWoken | mutexLocked |
+-----------------------------------+---------------+------------+-------------+
|              29 bits              | 1 bit         | 1 bit      | 1 bit       |
+-----------------------------------+---------------+------------+-------------+

Bir mutex'in iki çalışma modu vardır: normal mod ve açlık modu. Normal mod goroutine'lerin engelleme bekleme kuyruğuna varış sırasını takip eder, yani FIFO. Bu genel durumdur ve ayrıca performansın en iyi olduğu zamandır, çünkü herkes erişim sırasını takip ederek kilidi alır. Açlık modu olağandışı durumdur. Bu açlık bekleyen goroutine'lerin uzun süre kilit alamaması ve engellenmiş kalması anlamına gelir. Mutex'in kendisinin açlık durumunda olduğu anlamına gelmez. Peki goroutine'ler ne zaman açlık durumuna girer? Yetkililer bir örnek verir: daha önce gelen bir goroutine mutex'i alamadığı için engellenir. Daha sonra, kilit serbest bırakıldığında ve uyanıldığında, bu noktaya yeni çalışan başka bir goroutine kilidi almaya çalışır (sıraya girmeyi sever). İkincisi çalışma durumunda olduğundan (CPU zaman dilimini kaplar), ikincinin kilidi başarıyla alma olasılığı yüksektir. Aşırı durumlarda, bu tür birçok goroutine olabilir, bu yüzden yeni uyanan goroutine asla kilidi alamaz (sonsuz sıraya girme). İlk geldi ama asla kilidi alamaz.

const (
	starvationThresholdNs = 1e6
)

Bu durumu önlemek için, Go bir bekleme eşiği starvationThresholdNs ayarlar. Eğer bir goroutine 1ms'den fazla süre hala kilidi alamazsa, mutex açlık moduna girer. Açlık modunda, mutex sahipliği doğrudan bekleme kuyruğunun başındaki goroutine'e aktarılır. Yeni goroutine'ler kilidi almaya çalışmaz ve bunun yerine kuyruğun arkasına gider. Bu şekilde, açlık modunda, mutex sahipliği bekleme kuyruğundaki goroutine'ler tarafından sırayla tutulur (kuyruktaki insanların önce kilidi almasına izin ver, sıraya girenler arkaya gider). Bir goroutine başarıyla kilidi aldığında, eğer son bekleyen goroutine ise veya bekleme süresi 1ms'den azsa, mutex'i normal moda geri değiştirir. Bu açlık modu tasarımı bazı goroutine'lerin uzun süre kilit alamamasını ve "açlıktan ölmesini" önler.

TryLock

Mutex kilitlemek için iki yöntem sağlar:

• Lock(), engelleme modunda kilit al
• TryLock(), engelleme olmayan modda kilit al

İlk olarak TryLock koduna bakalım, çünkü uygulaması daha basittir:

func (m *Mutex) TryLock() bool {
	old := m.state
	if old&(mutexLocked|mutexStarving) != 0 {
		return false
	}

	if !atomic.CompareAndSwapInt32(&m.state, old, old|mutexLocked) {
		return false
	}

	return true
}

Bir kontrolle başlar. Eğer kilit zaten tutuluyorsa veya açlık modundaysa (yani birçok goroutine kilit için bekliyorsa), mevcut goroutine kilidi alamaz. Aksi takdirde, CAS operasyonu kullanarak durumu mutexLocked olarak güncellemeye çalışır. Eğer CAS operasyonu false döndürürse, bu dönemde başka bir goroutine'in kilidi başarıyla aldığı anlamına gelir, bu yüzden mevcut goroutine kilidi alamaz. Aksi takdirde, başarıyla kilidi alır. Bu koddan görülebilir ki, TryLock() çağırıcısı sıraya girmeye çalışan kişidir, çünkü bekleyen goroutine olup olmadığına bakılmaksızın doğrudan kilidi kapar (old 0 olmayabilir).

Lock

Aşağıda Lock kodu bulunmaktadır. Ayrıca doğrudan kilidi tutmak için CAS operasyonu kullanır, ancak daha "dürüst"tür. Sadece engellenmiş bekleyen goroutine olmadığında (old=0) doğrudan kilidi tutar.

func (m *Mutex) Lock() {
    // Fast path: grab unlocked mutex.
    if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
       return
    }
    // Slow path (outlined so that the fast path can be inlined)
    m.lockSlow()
}

Eğer engellenmiş bekleyen goroutine'ler bulursa, "dürüstçe" arkadan kuyruğa girer ve kilidi beklemek için lockSlow döngü akışına girer (mutex'in çekirdeği). İlk olarak, bazı değişkenler hazırlanır:

func (m *Mutex) lockSlow() {
    var waitStartTime int64
    starving := false
    awoke := false
    iter := 0
    old := m.state

• waitStartTime, açlık moduna girip girmeyeceğini kontrol etmek için bekleme başlangıç zamanını kaydetmek için kullanılır.
• starving, mevcut goroutine'in 1ms'den fazla süredir kilit alamadığını gösterir.
• awoke, mevcut goroutine'in uyanıp uyanmadığını işaretler.
• iter, sayaç, döngü yineleme sayısını kaydeder.
• old, mevcut mutex durumunu alır.

Sonra bir for döngüsüne girer ve mevcut goroutine'in döngü durumuna girip giremeyeceğini belirler.

Döngü çoklu iş parçacıklı bir senkronizasyon mekanizmasıdır, meşgul bekleme olarak da bilinir. Bir iş parçacığı kilidi tutmadığında, doğrudan askıya alıp iş parçacığı bağlamını değiştirmez ancak boş döngüye girer. Bu süreçte, sürekli CPU zaman dilimlerini kaplar. Eğer kilit çekişmesi düşükse veya kilit tutma süresi çok kısaysa, bunu yapmak sık iş parçacığı bağlamı değiştirmelerinden kaçınabilir ve performansı etkili bir şekilde artırabilir. Ancak, bir panzehir değildir. Go'da, döngüyü kötüye kullanmak aşağıdaki tehlikeli sonuçlara yol açabilir:

• Yüksek CPU kullanımı: Çok fazla döngü goroutine'i大量 CPU kaynağı tüketir, özellikle kilit uzun süre tutulduğunda.
• Goroutine zamanlamasını etkiler: Toplam işlemci P sayısı sınırlıdır. Eğer birçok döngü goroutine'i P'yi kaplarsa, diğer goroutine'ler zamanında zamanlanamaz.
• Önbellek tutarlılık sorunları: Spin kilitlerinin meşgul bekleme özelliği iş parçacıklarının kilit durumunu önbellekte tekrar tekrar okumasına neden olur. Eğer döngü goroutine'leri farklı çekirdeklerde çalışıyorsa ve kilit durumu zamanında global belleğe güncellenmezse, goroutine'ler yanlış kilit durumu okur. Sık önbellek tutarlılık senkronizasyonu da performansı önemli ölçüde azaltır.

Bu yüzden tüm goroutine'ler döngü durumuna giremez. Aşağıdaki sıkı yargıdan geçmelidir:

for {
    if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
        if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
        atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
            awoke = true
        }
        runtime_doSpin()
        iter++
        old = m.state
        continue
    }
    ...
}

Koşullar şöyledir:

1. Mevcut kilit zaten tutuluyor ve açlık modunda olamaz. Aksi takdirde, bazı goroutine'lerin uzun süredir kilit alamadığı anlamına gelir, bu yüzden doğrudan engelleme akışına girer.

2. runtime.sync_runtime_canSpin yargı akışına girer:

    const (
   	active_spin = 4
   )
   func sync_runtime_canSpin(i int) bool {
   	if i >= active_spin || ncpu <= 1 || gomaxprocs <= sched.npidle.Load()+sched.nmspinning.Load()+1 {
   		return false
   	}
   	if p := getg().m.p.ptr(); !runqempty(p) {
   		return false
   	}
   	return true
   }

3. Döngü sayısı runtime.active_spin'den azdır, varsayılan 4 seferdir. Daha fazlası kaynak israfıdır.

4. CPU çekirdek sayısı 1'den büyüktür. Tek çekirdekli sistemlerde döngünün anlamı yoktur.

5. Mevcut gomaxprocs boş P ve döngü P toplamından 1 fazla olmalıdır, yani döngü için yeterli işlemci yoktur.

6. Mevcut P'nin yerel kuyruğu boş olmalıdır. Aksi takdirde, yürütülecek başka kullanıcı görevleri olduğu anlamına gelir ve döngü yapılamaz.

Eğer döngüye izin verilirse, runtime.sync_runtime_doSpin çağırılır ve döngüye girer. Aslında PAUSE talimatını 30 kez yürütür.

const (
	active_spin_cnt = 30
)

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

Eğer döngü yapılamazsa, sadece iki sonuç vardır: başarıyla kilidi alır veya bekleme kuyruğuna girer ve engellenir. Ancak, ondan önce, birçok şey vardır:

1. Eğer açlık modunda değilse, kilidi almaya çalış:

new := old
if old&mutexStarving == 0 {
	new |= mutexLocked
}

2. Eğer kilit zaten doluysa veya şimdi açlık modundaysa, bekleyen goroutine sayısını artır:

if old&(mutexLocked|mutexStarving) != 0 {
	new += 1 << mutexWaiterShift
}

3. Eğer mevcut goroutine zaten açlık durumundaysa ve kilit hala doluysa, açlık moduna gir:

if starving && old&mutexLocked != 0 {
	new |= mutexStarving
}

4. Eğer mevcut goroutine döngü sırasında uyanmışsa, mutexWoken bayrağını ekle:

if awoke {
	new &^= mutexWoken
}

Sonra CAS aracılığıyla kilit durumunu güncellemeye başla. Eğer güncelleme başarısız olursa, doğrudan bir sonraki yinelemeye başla:

if atomic.CompareAndSwapInt32(&m.state, old, new) {
    ...
}else {
    ...
}

Eğer güncelleme başarılı olursa, aşağıdaki yargıya başla:

1. Orijinal durum açlık modu değil ve hiçbir goroutine kilidi kaplamıyor. Sonra mevcut goroutine doğrudan kilidi tutabilir, akıştan çıkabilir ve kullanıcı kodunu yürütmeye devam edebilir.

   if old&(mutexLocked|mutexStarving) == 0 {
   		break
   }

2. Kilidi almaya çalış başarısız. Bekleme süresini kaydet, burada LIFO true ise kuyruk son-giren-ilk-çıkar, aksi takdirde FIFO ilk-giren-ilk-çıkar.

   queueLifo := waitStartTime != 0
   if waitStartTime == 0 {
   	waitStartTime = runtime_nanotime()
   }

3. Semaphore almaya çalış, runtime.semacquire1 fonksiyonuna gir. Eğer semaphore alınabilirse, doğrudan döner, engellemez. Aksi takdirde, runtime.gopark çağırır ve mevcut goroutine'i askıya alır ve semaphore serbest bırakılmasını bekler.

   runtime_SemacquireMutex(&m.sema, queueLifo, 1)

4. Bu adıma ulaşmanın iki olasılığı vardır: ya doğrudan başarıyla semaphore aldı veya engellemeden uyanıp başarıyla semaphore aldı. Her iki şekilde de, semaphore başarıyla alındı. Eğer şimdi açlık modundaysa, doğrudan kilidi alabilir.

   starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
   old = m.state
   if old&mutexStarving != 0 {
   	delta := int32(mutexLocked - 1<<mutexWaiterShift)
   	if !starving || old>>mutexWaiterShift == 1 {
   		delta -= mutexStarving
   	}
   	atomic.AddInt32(&m.state, delta)
   	break
   }

5. Eğer açlık modunda değilse, iter'i sıfırla ve döngü akışını yeniden başlat.

   awoke = true
   iter = 0

Bu noktada, kilitleme akışı sona erer. Tüm süreç oldukça karmaşıktır, hem döngü bekleme hem de semaphore engelleme bekleme yöntemlerini kullanır, performans ve dengeyi dengeler, çoğu kilit çekişme senaryosu için uygundur.

Unlock

Kilit açma akışı nispeten çok daha basittir. İlk olarak hızlı kilit açmayı dener. Eğer new 0 ise, bu bekleyen goroutine olmadığı ve açlık modunda olmadığı anlamına gelir, yani kilit açma başarılı, doğrudan dönebilir.

func (m *Mutex) Unlock() {
	new := atomic.AddInt32(&m.state, -mutexLocked)
	if new != 0 {
		m.unlockSlow(new)
	}
}

Aksi takdirde, unlockSlow akışına girmesi gerekir:

1. İlk olarak zaten kilidi açılıp açılmadığını kontrol et:

   if (new+mutexLocked)&mutexLocked == 0 {
   	fatal("sync: unlock of unlocked mutex")
   }

2. Eğer açlık modundaysa, doğrudan semaphore serbest bırak ve kilidi aç. Açlık modunda, mevcut kilit açan goroutine kilit sahipliğini doğrudan bir sonraki bekleyen goroutine'e aktarır.

   if new&mutexStarving == 0 {
   	...
   } else {
   	runtime_Semrelease(&m.sema, true, 1)
   }

3. Eğer açlık modunda değilse, normal kilit açma akışına gir:

   1. Eğer bekleyen goroutine yoksa veya diğer uyanık goroutine'ler zaten kilidi aldıysa veya kilit açlık moduna girdiyse:

      if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
          return
      }

   2. Aksi takdirde, semaphore serbest bırak ve bir sonraki bekleyen goroutine'i uyandır, mevcut kilit durumunu mutexWoken olarak ayarla:

      new = (old - 1<<mutexWaiterShift) | mutexWoken
      if atomic.CompareAndSwapInt32(&m.state, old, new) {
          runtime_Semrelease(&m.sema, false, 1)
          return
      }
      old = m.state

Son olarak, kilit açma akışı sona erer.

RWMutex

Read-write mutex RWMutex tip tanımı sync/rwmutex.go dosyasında bulunur. Uygulaması da mutex'e dayanır.

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 atomic.Int32 // number of pending readers
	readerWait  atomic.Int32 // number of departing readers
}

Alan tanımları:

• w, bir mutex. Yazar goroutine bu kilidi tuttuğunda, diğer yazar goroutine'ler ve okuyucu goroutine'ler engellenir.
• writerSem, yazma semaphore, okuyucu goroutine'lerin tamamlanmasını bekleyen yazar goroutine'leri engellemek için kullanılır. Yazar goroutine semaphore alır, okuyucu goroutine semaphore serbest bırakır.
• readerSem, okuma semaphore, yazar goroutine'lerin tamamlanmasını bekleyen okuyucu goroutine'leri engellemek için kullanılır. Okuyucu goroutine semaphore alır, yazar goroutine semaphore serbest bırakır.
• readerCount, çekirdek alan, tüm read-write lock buna bağlı olarak durumu korur.
• readerWait, yazar goroutine engellendiğinde beklemesi gereken okuyucu goroutine sayısını gösterir.

Read-write lock'un genel prensibi şudur: mutex aracılığıyla yazar goroutine'leri karşılıklı olarak özel kılar, iki semaphore writerSem ve readerSem aracılığıyla read-write karşılıklı olarak özel kılar, read-read paylaşımlı.

readerCount

Bu readerCount oldukça değişir ve çok önemlidir, bu yüzden ayrı olarak tartışılır. Kabaca aşağıdaki durumlara özetlenir:

1. 0: Şu anda ne aktif okuyucu goroutine'ler ne de aktif yazar goroutine'ler var, boş durumda.
2. -rwmutexMaxReaders: Bir yazar goroutine zaten mutex'i tuttu, şu anda aktif okuyucu goroutine'ler yok.
3. -rwmutexMaxReaders+N: Bir yazar goroutine zaten yazma kilidini tuttu, mevcut okuyucu goroutine'ler yazar goroutine yazma kilidini serbest bırakana kadar engellenip beklemeli.
4. N-rwmutexMaxReaders: Bir yazar goroutine zaten mutex'i tuttu, kalan okuyucu goroutine'ler okuma kilidini serbest bırakana kadar engellenip beklemeli.
5. N: Şu anda N aktif okuyucu goroutine var, yani N okuma kilidi eklendi, aktif yazar goroutine'ler yok.

Bunlar arasında, rwmutexMaxReaders bir sabit değerdir, değeri mutex üzerinde engellenip bekleyebilecek goroutine sayısının iki katıdır, çünkü yarısı okuyucu goroutine'ler ve yarısı yazar goroutine'lerdir.

const rwmutexMaxReaders = 1 << 30

Tüm read-write lock'un bir kısmı, bu readerCount en karmaşıktır. Değişikliklerini anlamak read-write lock'un iş akışını açıklığa kavuşturur.

TryLock

Her zamanki gibi, ilk olarak en basit TryLock()'a bakalım:

func (rw *RWMutex) TryLock() bool {
	if !rw.w.TryLock() {
		return false
	}
	if !rw.readerCount.CompareAndSwap(0, -rwmutexMaxReaders) {
		rw.w.Unlock()
		return false
	}
	return true
}

İlk olarak mutex'in TryLock()'ını çağırmayı dener. Eğer başarısız olursa, doğrudan döner. Sonra CAS operasyonu kullanarak readerCount değerini 0'dan -rwmutexMaxReaders'a güncellemeye çalışır. 0 okuyucu goroutine'lerin çalışmadığını temsil eder, -rwmutexMaxReaders yazar goroutine'in zaten mutex'i tuttuğunu gösterir. Eğer CAS operasyonu güncellemesi başarısız olursa, kilidi aç. Eğer başarılı olursa, true döndür.

Lock

Sonraki Lock(), uygulaması da çok basittir.

func (rw *RWMutex) Lock() {
	rw.w.Lock()
	r := rw.readerCount.Add(-rwmutexMaxReaders) + rwmutexMaxReaders
	if r != 0 && rw.readerWait.Add(r) != 0 {
		runtime_SemacquireRWMutex(&rw.writerSem, false, 0)
	}
}

İlk olarak, diğer yazar goroutine'lerle yarışır, mutex'i tutana kadar, sonra bu operasyonu gerçekleştirir: ilk olarak atomik olarak -rwmutexMaxReaders çıkarır, sonra atomik olmayan şekilde yeni değere rwmutexMaxReaders ekler:

r = rw.readerCount.Add(-rwmutexMaxReaders) + rwmutexMaxReaders

İki adıma bölelim:

1. Bu diğer okuyucu goroutine'lere bir yazar goroutine'in şimdi kilidi tutmaya çalıştığını bildirmektir, zaten TryLock bölümünde kapsandı.

   rw.readerCount.Add(-rwmutexMaxReaders)

2. Sonra rwmutexMaxReaders ekler ve r'yi alır, bu şu anda çalışan okuyucu goroutine sayısını temsil eder.

   r = rw.readerCount.Add(-rwmutexMaxReaders) + rwmutexMaxReaders

Sonra okuyucu goroutine'lerin çalışıp çalışmadığını kontrol eder, sonra r'yi readerWait değerine ekler. Eğer hala 0 değilse, bu okuyucu goroutine'lerin işi bitene kadar beklemesi gerektiği anlamına gelir, sonra runtime_SemacquireRWMutex akışına girer ve writerSem semaphore'ını almaya çalışır. Bu semaphore okuyucu goroutine'ler tarafından serbest bırakılır. Eğer semaphore alınabilirse, bu okuyucu goroutine'lerin işi bittiği anlamına gelir. Aksi takdirde, engelleme kuyruğuna girip beklemesi gerekir (bu semaphore mantığı temelde mutex kısmı ile tutarlıdır).

UnLock

Sonra UnLock(), yazma kilidini serbest bırakır.

func (rw *RWMutex) Unlock() {
    r := rw.readerCount.Add(rwmutexMaxReaders)
    if r >= rwmutexMaxReaders {
       fatal("sync: Unlock of unlocked RWMutex")
    }

    for i := 0; i < int(r); i++ {
       runtime_Semrelease(&rw.readerSem, false, 0)
    }

    rw.w.Unlock()
}

Akışı şöyledir:

1. Daha önce belirtildiği gibi, kilitleme sırasında, readerCount negatif değere güncellenir. Burada rwmutexMaxReaders eklemek artık yazar goroutine'lerin çalışmadığı anlamına gelir ve elde edilen değer engellenip bekleyen okuyucu goroutine sayısını temsil eder.

   r := rw.readerCount.Add(rwmutexMaxReaders)

2. Eğer kendisi 0 veya 0'dan büyükse, bu yazma kilidinin zaten serbest bırakıldığı anlamına gelir:

   if r >= rwmutexMaxReaders {
   	fatal("sync: Unlock of unlocked RWMutex")
   }

3. Semaphore readerSem'i serbest bırak, bekleyen okuyucu goroutine'leri uyandır:

   for i := 0; i < int(r); i++ {
   	runtime_Semrelease(&rw.readerSem, false, 0)
   }

4. Son olarak mutex'i serbest bırak, bekleyen yazar goroutine'leri uyandır:

   rw.w.Unlock()

Yazma kilidi serbest bırakma tamamlandı.

TryRLock

Sonraki, okuma kilidi kısmına bakalım. Bu TryRLock kodudur:

func (rw *RWMutex) TryRLock() bool {
	for {
		c := rw.readerCount.Load()
		if c < 0 {
			return false
		}
		if rw.readerCount.CompareAndSwap(c, c+1) {
			return true
		}
	}
}

Sadece iki şey yapar:

1. Yazar goroutine'lerin çalışıp çalışmadığını kontrol et. Eğer evet, kilit başarısız.

   c := rw.readerCount.Load()
   if c < 0 {
   	return false
   }

2. readerCount'u 1 artırmayı dener. Eğer güncelleme başarılı olursa, kilit başarılı:

   if rw.readerCount.CompareAndSwap(c, c+1) {
   	return true
   }

3. Aksi takdirde, çıkana kadar döngü yargısını devam ettir.

Buradaki readerCount bağımlılığının tümü yazma kilidi kısmında korunur. Bu da önce yazma kilidinin tartışılmasının nedenidir, çünkü karmaşık çekirdek kısımların tümü yazma kilidi kısmında korunur.

RLock

RLock mantığı daha da basittir:

func (rw *RWMutex) RLock() {
	if rw.readerCount.Add(1) < 0 {
		runtime_SemacquireRWMutexR(&rw.readerSem, false, 0)
	}
}

readerCount değerini 1 artırmayı dener. Eğer yeni değer hala 0'dan küçükse, bu yazar goroutine'lerin çalıştığı anlamına gelir, sonra readerSem semaphore engelleme akışına girer. Mevcut goroutine engelleme kuyruğuna girip bekler.

RUnLock

RUnLock da eşit derecede basit ve kolay anlaşılır:

func (rw *RWMutex) RUnlock() {
    if r := rw.readerCount.Add(-1); r < 0 {
       rw.rUnlockSlow(r)
    }
}

func (rw *RWMutex) rUnlockSlow(r int32) {
	if r+1 == 0 || r+1 == -rwmutexMaxReaders {
		fatal("sync: RUnlock of unlocked RWMutex")
	}
	if rw.readerWait.Add(-1) == 0 {
		runtime_Semrelease(&rw.writerSem, false, 1)
	}
}

İlk olarak readerCount'u bir azaltmayı dener, bu aktif okuyucu goroutine sayısının bir azaldığını gösterir. Eğer elde edilen değer 0'dan büyükse, bu doğrudan serbest bırakılabileceği anlamına gelir, çünkü şimdi hiçbir yazar goroutine mutex'i tutmuyor. Eğer 0'dan küçükse, bu yazar goroutine'in zaten mutex'i tuttuğu anlamına gelir, tüm mevcut okuyucu goroutine'lerin işi bitene kadar bekliyor. Sonra runlockSlow akışına girer:

1. Eğer orijinal readerCount değeri 0 ise (kilit boş) veya -rwmutexMaxReaders (yazar goroutine'in bekleyeceği okuyucu goroutine'leri yok, yani okuma kilitlerinin tümü zaten serbest bırakıldı), bu şu anda aktif okuyucu goroutine'lerin olmadığı anlamına gelir, kilidi açmaya gerek yok:

   if r+1 == 0 || r+1 == -rwmutexMaxReaders {
   	fatal("sync: RUnlock of unlocked RWMutex")
   }

2. Eğer aktif okuyucu goroutine'ler varsa, readerWait'i bir azalt. Eğer mevcut okuyucu goroutine son aktif okuyucu ise, writerSem semaphore'ını serbest bırak, bekleyen yazar goroutine'leri uyandır:

   if rw.readerWait.Add(-1) == 0 {
   	runtime_Semrelease(&rw.writerSem, false, 1)
   }

Okuma kilidi serbest bırakma akışı sona erer.

Semaphore

Mutex içindeki semaphore basit bir uint32 tamsayısıdır, atomik olarak bir azaltma ve bir artırma aracılığıyla semaphore alımını ve serbest bırakmasını temsil eder. Runtime'da semaphore'ları korumaktan sorumlu yapı runtime.semaRoot'tur, tip tanımı runtime/sema.go dosyasında bulunur. semaRoot semaphore'ları organize etmek ve yönetmek için dengeli bir ikili ağaç (treap) kullanır. Ağaçtaki her düğüm bir semaphore'u temsil eder, düğüm tipi *sudog'dur, bu ilgili semaphore'lar için bekleme kuyruğunu koruyan çift yönlü bağlı bir listedir. Düğümler *sudog.elem (semaphore adresi) aracılığıyla benzersizliği korur ve *sudog.ticket alanı aracılığıyla ağaç dengesini korur.

type semaRoot struct {
	lock  mutex
	treap *sudog        // root of balanced tree of unique waiters.
	nwait atomic.Uint32 // Number of waiters. Read w/o the lock.
}

Semaphore ağacı semaRoot eşzamanlılık güvenliğini sağlamak için daha düşük seviyeli bir mutex runtime.mutex'e bağlıdır.

var semtable semTable

// Prime to not correlate with any user patterns.
const semTabSize = 251

type semTable [semTabSize]struct {
	root semaRoot
    // Used for memory alignment, improving performance
	pad  [cpu.CacheLinePadSize - unsafe.Sizeof(semaRoot{})]byte
}

semaRoot runtime'da global bir semaTable'de saklanır. Sabit uzunluklu bir dizi gibi görünür, birden fazla semaphore ağacı kök düğüm koleksiyonunu saklamak için kullanılır, ancak aslında operasyonel bakış açısından, aslında bir hash tablosudur. Tablodaki her eleman bir semaRoot ve bazı doldurma baytları (pad) içerir, bellek hizalaması için ve önbellek satırı çekişmesini önlemek için kullanılır. semTabSize semaphore tablosu boyutu sabitidir, tablo uzunluğunu 251 olarak belirtir, genellikle hash çarpışmalarını azaltmak ve hash verimliliğini artırmak için bir asal sayı seçilir.

func (t *semTable) rootFor(addr *uint32) *semaRoot {
	return &t[(uintptr(unsafe.Pointer(addr))>>3)%semTabSize].root
}

rootFor metodu bir hash fonksiyonuna eşdeğerdir. Bir uint32 tipi işaretçi addr (yani semaphore adresi) kabul eder ve o adrese karşılık gelen semaRoot yapısının işaretçisini döndürür. Bu satır kodu önce addr'i uintptr'ye dönüştürür, sonra 3 bit sağa kaydırır, 8'e bölmeye eşdeğerdir (çünkü bir bayt 8 bit kaplar, işaretçi adresini 8'e bölmek dizi indeksine eşleyebilir). semTabSize ile mod alarak indeksin semaphore tablosu boyutu aralığında olmasını sağlar. İndeks aracılığıyla semaRoot elde ettikten sonra, dengeli ağaca gidip semaphore'a karşılık gelen *sudog bekleme kuyruğunu bulur.

Acquire

Semaphore alma, karşılık gelen uygulama runtime.semacquire1 fonksiyonudur:

func semacquire1(addr *uint32, lifo bool, profile semaProfileFlags, skipframes int, reason waitReason)

Aşağıdaki parametreleri alır:

• addr, semaphore adresi
• lifo, dengeli ağaç kuyruk çıkarma sırasını etkiler, varsayılan FIFO'dur, LIFO son-giren-ilk-çıkar, goroutine bekleme kilit süresi 0 olmadığında (en az bir kez engellendi), true'dur
• profile, kilit performans analizi için bayraklar
• skipframes, atlanan yığın kare sayısı
• reason, engelleme nedeni

Aşağıda tüm semaphore alma akışı kısaca açıklanmıştır:

1. Goroutine durumunu kontrol et. Eğer mevcut goroutine zamanlanan goroutine değilse, doğrudan istisna fırlat:

   gp := getg()
   if gp != gp.m.curg {
   	throw("semacquire not on the G stack")
   }

2. Semaphore alınıp alınamayacağını kontrol et ve semaphore'ı engelleme olmayan yöntemle almaya çalış. Eğer alınabilirse, doğrudan dönebilir:

   for {
   	v := atomic.Load(addr)
   	if v == 0 {
   		return false
   	}
   	if atomic.Cas(addr, v, v-1) {
   		return true
   	}
   }

3. Eğer engelleme olmadan alınamazsa, normal yollarla semaphore almak için döngüye gir. İlk olarak, acquireSudog() aracılığıyla önbellekten bir *sudog al, bu yapı engellenmiş bekleme goroutine'ini temsil eder:

   s := acquireSudog()

4. Sonra global tablodan semaphore ağacını al:

   root := semtable.rootFor(addr)

5. Döngüye gir, semaphore ağacını kilitle, semaphore alınıp alınamayacağını tekrar kontrol et. Eğer değilse, onu semaphore ağacına ekle, sonra gopark çağır ve uyanana kadar askıya al ve beklemeye devam et, semaphore alana kadar bu süreci tekrarlayarak döngü:

   for {
   	lockWithRank(&root.lock, lockRankRoot)
   	root.nwait.Add(1)
   	if cansemacquire(addr) {
   		root.nwait.Add(-1)
   		unlock(&root.lock)
   		break
   	}
   	root.queue(addr, s, lifo)
   	goparkunlock(&root.lock, reason, traceBlockSync, 4+skipframes)
   	if s.ticket != 0 || cansemacquire(addr) {
   		break
   	}
   }

6. Son olarak, uyanıldığında, *sudog'u serbest bırak, önbelleğe döndür:

   releaseSudog(s)

Release

Semaphore serbest bırak, engellenmiş bekleme goroutine'lerini uyandır. Bu işlevsellik runtime.semrelease1 fonksiyonu tarafından uygulanır:

func semrelease1(addr *uint32, handoff bool, skipframes int)

Aşağıdaki parametreleri alır:

• addr, semaphore adresi
• handoff, P tarafından şu anda zamanlanan G'yi uyanık G'ye doğrudan değiştirip değiştirmeyeceğini gösterir, sadece açlık modunda true
• skipframes, atlanan yığın kare sayısı

Aşağıda tüm serbest bırakma süreci kısaca açıklanmıştır:

1. Semaphore ağacını al, sonra semaphore'ı bir artır, bu bir semaphore serbest bırakıldığını gösterir:

   root := semtable.rootFor(addr)
   atomic.Xadd(addr, 1)

2. Eğer bekleyen goroutine sayısı 0 ise, doğrudan dön:

   if root.nwait.Load() == 0 {
   	return
   }

3. Semaphore ağacını kilitle, ikincil kontrol bekleyen goroutine olup olmadığı:

   lockWithRank(&root.lock, lockRankRoot)
   if root.nwait.Load() == 0 {
   	unlock(&root.lock)
   	return
   }

4. Semaphore ağacından bir engellenmiş bekleme goroutine'i al, nwait bir azalt, sonra semaphore kilidini serbest bırak:

   s, t0, tailtime := root.dequeue(addr)
   if s != nil {
   	root.nwait.Add(-1)
   }
   unlock(&root.lock)

5. Semaphore alınıp alınamayacağını kontrol et:

   if handoff && cansemacquire(addr) {
   	s.ticket = 1
   }

6. readyWithTime fonksiyonu doğrudan uyanık goroutine G'yi P için çalışacak bir sonraki goroutine olarak kullanır, yani *p.runnext=g değiştirir:

   readyWithTime(s, 5+skipframes)

7. Eğer handoff true ise, o zaman goyield mevcut semaphore serbest bırakan goroutine G'yi mevcut M'den ayırır ve P yerel çalıştırma kuyruğunun sonuna yeniden ekler, sonra yeni zamanlama döngüsünü başlatır, böylece uyanık goroutine G hemen zamanlanabilir:

   if s.ticket == 1 && getg().m.locks == 0 {
   	goyield()
   }

Semaphore alma ve serbest bırakma akışları bunlardır. Go dili semaphore'ları sadece mutex kilitlerinde kullanmaz. Burada yerleştirilmiştir çünkü semaphore'lar mutex kilitleriyle en büyük ilişkiye sahiptir. Yetkililer yorumlarda şöyle yazmıştır:

// Asynchronous semaphore for sync.Mutex.

Semaphore'ları anladıktan sonra, mutex kilitlerine geri bakmak daha net olur.

TIP

semaRoot semaphore ağacı hakkında, kuyruk çıkarma ve ekleme kendi kendine dengeleme operasyonlarını içerir ve uygulaması oldukça karmaşıktır. Bunlara derinlemesine dalmak bu makalenin temasıyla ilgisizdir ve anlamsızdır, bu yüzden korunmuştur. Eğer ilgileniyorsanız, kaynak kodu kendiniz keşfedebilirsiniz.

Özet

Mutex sync.Mutex goroutine beklemesini döngü ve semaphore iki mekanizma aracılığıyla uygular. Döngü engelleme değildir ancak CPU kaynakları tükettiği için sıkı sınırlamalar gerekir; semaphore'lar engellemedir ve gereksiz kaynak tüketimini etkili bir şekilde önleyebilir. Daha adil bir yarış mekanizması uygulamak için, Go goroutine'lerin kilitler için daha dengeli bir şekilde yarışabilmesini sağlamak için normal mod ve açlık modu arasında ayrım yapar. runtime.mutex bu tür düşük seviyeli kilide kıyasla, sync.Mutex kullanıcıya yönelik bir kilit olarak tasarım sırasında daha fazla gerçek kullanım senaryosunu dikkate alır.

Read-write lock sync.RWMutex yazma-yazma karşılıklı özel kılmayı mutex sync.Mutex aracılığıyla uygular ve bu temel üzerine ayrıca iki semaphore ekler, read-write karşılıklı özel kılma ve read-read paylaşımı uygulamak için kullanılır, böylece çoklu eşzamanlı senaryoları destekler.

Kilit uygulaması karmaşık görünse de, Mutex prensiplerini anladıktan sonra, sync standart kütüphanesindeki diğer senkronizasyon araçlarını öğrenmek çok daha kolay olur.