قفل المزامنة sync.Mutex

القفل هو بديلة مزامنة مهمة في أنظمة التشغيل، توفر لغة Go في مكتبتها القياسية نوعين من الأقفال: قفل المزامنة وقفل القراءة والكتابة، يقابلهما:

• sync.Mutex، قفل المزامنة، قراءة-قراءة حصري، قراءة-كتابة حصري، كتابة-كتابة حصري
• sync.RWMutex، قفل القراءة والكتابة، قراءة-قراءة مشترك، قراءة-كتابة حصري، كتابة-كتابة حصري

حالات استخدامهما التجارية شائعة جدًا، تُستخدم لحماية منطقة ذاكرة مشتركة للوصول والتعديل المتسلسل في ظل التزامن، كما في المثال التالي:

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)
}

بدون حماية القفل، نتيجة تنفيذ هذه الدالة قد تكون مختلفة في كل مرة، غير قابلة للتنبؤ، ومن الواضح أن معظم السيناريوهات لا نريد حدوث ذلك. هذا المثال بسيط لمعظم الناس، ربما تكون قد أتقنت استخدام الأقفال، لكن قد لا تفهم كيفية تنفيذ القفل داخليًا في لغة Go، الكود نفسه ليس معقدًا، سأشرحه بالتفصيل فيما يلي.

Locker

قبل البدء، لننظر أولاً لنوع sync.Locker، وهي واجهة عرّفتها Go:

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

الطرق التي توفرها بسيطة وسهلة الفهم، وهي القفل والفتح، لكن بسبب ميزة تطبيق الواجهة优于 الاتفاق في Go، لذلك معظم الناس ربما لم يروها أبدًا، هنا أذكرها باختصار، لأنها ليست مهمة جدًا، القفلان اللذان سنتحدث عنهما لاحقًا يطبقان هذه الواجهة.

Mutex

قفل المزامنة Mutex معرف في ملف sync/mutex.go، وهو نوع بنية:

type Mutex struct {
	state int32
	sema  uint32
}

شرح الحقول:

• state، يمثل حالة القفل
• sema، أي السيمافور semaphore، سيُشرح لاحقًا

لنبدأ بشرح state:

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

state عدد صحيح 32-بت، البتات الثلاثة السفلى تمثل ثلاث حالات، هذه الحالات ليست مستقلة، يمكن أن تتواجد معًا.

• mutexLocked=1، مقفل
• mutexWoken=2، مستيقظ
• mutexStarving=4، وضع المجاعة

الباقي 29 بت تُستخدم لتمثيل عدد الكوروتينات التي تنتظر القفل، لذلك نظريًا قفل المزامنة يمكن استخدامه حتى 2^29+1 كوروتين في نفس الوقت، لكن في الواقع من غير المرجح وجود هذا العدد من الكوروتينات، حتى لو كان كل واحد يشغل 2KB فقط (حجم المكدس الأولي)، المساحة اللازمة لإنشاء هذا العدد من الكوروتينات حوالي 1TB.

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

لقفل المزامنة وضعان للعمل: الوضع الطبيعي، ووضع المجاعة. الوضع الطبيعي يعني أن الكوروتينات تحصل على القفل حسب ترتيب وصولها لطابور الانتظار، أي FIFO، وهذا هو الحال العام، وأفضل أداء، لأن الجميع يحصل على القفل حسب ترتيب الوصول. وضع المجاعة هو الحالة غير الطبيعية، والمجاعة تعني أن الكوروتين المنتظر لا يتمكن من الحصول على القفل لفترة طويلة ويبقى في حالة حجب، وليس أن القفل نفسه في حالة مجاعة. متى يكون الكوروتين في حالة مجاعة؟ تعطي Go مثالًا: وصل كوروتين أولاً، وبسبب عدم تمكنه من الحصول على القفل حُجب، ثم أُيقظ بعد تحرير القفل، وفي هذه اللحظة وصل كوروتين آخر بدأ لتوه تنفيذ الكود في هذه المنطقة (يحب التقدم)، لأن الأخير في حالة تنفيذ (يستخدم شريحة وقت CPU)، احتمال نجاحه في الحصول على القفل عالي جدًا، وفي الحالات القصوى قد يكون هناك العديد من الكوروتينات مثل هذا، فالكوروتين المُوقظ لن يتمكن من الحصول على القفل (تقدم مستمر)، رغم أنه وصل أولاً!

const (
	starvationThresholdNs = 1e6
)

لتجنب这种情况، حددت Go عتبة انتظار starvationThresholdNs، إذا انتظر كوروتين أكثر من 1ms دون الحصول على القفل، يدخل القفل في وضع المجاعة. في وضع المجاعة، تُنقل ملكية القفل مباشرة للكوروتين الأول في طابور الانتظار، الكوروتينات الجديدة لن تحاول الحصول على القفل، بل تدخل في نهاية الطابور. هكذا، في وضع المجاعة تُنقل ملكية القفل بالتتابع للكوروتينات في طابور الانتظار (من ينتظر أولاً يحصل على القفل أولاً)، وعندما يحصل الكوروتين على القفل بنجاح، إذا كان آخر كوروتين منتظر أو كان وقت انتظاره أقل من 1ms، يُعاد القفل للوضع الطبيعي. هذا التصميم لتجنب "مجاعة" بعض الكوروتينات لعدم تمكنها من الحصول على القفل لفترة طويلة.

TryLock

يوفر قفل المزامنة طريقتين للقفل:

• Lock()، الحصول على القفل بطريقة حاجبة
• TryLock()، الحصول على القفل بطريقة غير حاجبة

لننظر أولاً لكود TryLock لأن تنفيذه أبسط:

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
}

في البداية يتحقق، إذا كان القفل مملوكًا بالفعل، أو في وضع المجاعة (أي العديد من الكوروتينات تنتظر القفل)، فالكوروتين الحالي لا يمكنه الحصول على القفل. وإلا يحاول عبر عملية CAS تحديث الحالة لـ mutexLocked، إذا أعادت CAS false، فهذا يعني أن كوروتينًا آخر حصل على القفل خلال هذه الفترة، فالكوروتين الحالي لا يمكنه الحصول على القفل، وإلا نجح في الحصول على القفل. من هذا الكود نرى أن مستدعي TryLock() هو الشخص الذي يحاول التقدم، لأنه لا يهتم بوجود كوروتينات تنتظر، بل ي snatch القفل مباشرة.

Lock

فيما يلي كود Lock، يستخدم أيضًا عملية CAS لمحاولة الحصول على القفل مباشرة، لكنه "أكثر صدقًا"، يحاول الحصول على القفل مباشرة فقط عندما لا يكون هناك كوروتينات محجوبة تنتظر (old=0).

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()
}

إذا اكتشف وجود كوروتينات محجوبة تنتظر، فإنه "بصدق" يذهب للانتظار في الطابور، ويدخل عملية الدوران lockslow (جوهر قفل المزامنة). أولًا يُعد بعض المتغيرات:

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

• waitStartTime: لتسجيل وقت بدء الانتظار، للتحقق من الدخول في وضع المجاعة.
• starving: يشير هل الكوروتين الحالي تجاوز 1ms دون الحصول على القفل.
• awoke: يحدد هل الكوروتين الحالي قد أُيقظ.
• iter: عداد، يسجل عدد مرات الدوران.
• old: يحصل على حالة قفل المزامنة الحالي.

ثم يدخل حلقة for، ليحكم هل الكوروتين الحالي يمكنه الدخول في حالة الدوران.

الدوران هو آلية مزامنة بين الخيوط، تُسمى أيضًا الانتظار المشغول (busy-waiting)، الخيط عند عدم ملكيته للقفل لا يُعلق مباشرة ويبدل سياق الخيط بل يدور فارغًا، خلال هذه العملية يستمر في استخدام شريحة وقت CPU، إذا كان التنافس على القفل ضعيفًا أو وقت امتلاك القفل قصيرًا، هذا يمكن أن يتجنب تبديل سياق الخيط المتكرر ويحسن الأداء بفعالية، لكنه ليس حلاً لكل شيء، إساءة استخدام الدوران في Go قد يؤدي لعواقب خطيرة:

• استخدام CPU عالٍ: الكوروتينات الدوارة الكثيرة تستهلك موارد CPU كبيرة، خاصة عندما يكون القفل مشغولًا لفترة طويلة.
• التأثير على جدولة الكوروتين: عدد المعالجات P محدود، إذا كان هناك العديد من الكوروتينات الدوارة تشغل P، فالكوروتينات الأخرى التي تنفذ كود المستخدم لن تُجدول في الوقت المناسب.
• مشاكل تناسق الذاكرة المخبئية: خاصية الانتظار المشغول للقفل الدوار تجعل الخيط يقرأ حالة القفل بشكل متكرر من الذاكرة المخبئية (cache)، إذا كانت الكوروتينات الدوارة تعمل على أنوية مختلفة، ولم تُحدّث حالة القفل للذاكرة العامة في الوقت المناسب، سيقرأ الكوروتين حالة قفل غير دقيقة، والتناسق المتكرر للذاكرة المخبئية سيقلل الأداء بشكل ملحوظ.

لذلك ليس كل الكوروتينات يمكنها الدخول في حالة الدوران، تحتاج لحكم صارم:

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
    }
    ...
}

الشروط كالتالي:

1. القفل مملوك حاليًا ولا يمكن أن يكون في وضع المجاعة، وإلا يعني ذلك أن كوروتينًا لم يتمكن من الحصول على القفل لفترة طويلة، فيدخل مباشرة في عملية الحجب.

2. يدخل في عملية الحكم runtime.sync_runtime_canSpin:

    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
   }

   1. عدد مرات الدوران أقل من runtime.active_spin، الافتراضي 4 مرات، كثرة المرات تضيع الموارد.
   2. عدد أنوية CPU أكبر من 1، الدوران في أنظمة أحادية النواة بلا معنى.
   3. gomaxprocs الحالي أكبر من مجموع عدد P الخاملة و P الدوارة + 1، أي لا توجد معالجات كافية للدوران.
   4. الطابور المحلي لـ P الحالي يجب أن يكون فارغًا، وإلا يعني وجود مهام مستخدم أخرى للتنفيذ، لا يمكن الدوران.

إذا تم الحكم على إمكانية الدوران، يُستدعى runtime.sync_runtime_doSpin للدخول في الدوران، في الواقع ينفذ 30 مرة تعليمة PAUSE.

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

إذا لم يتمكن من الدوران، فلن يكون أمامه إلا مصيران: النجاح في الحصول على القفل أو الدخول في طابور الانتظار والحجب، لكن قبل ذلك هناك أمور كثيرة للمعالجة:

1. إذا لم يكن في وضع المجاعة، يحاول الحصول على القفل:

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

2. إذا كان القفل مشغولًا أو في وضع المجاعة، عدد الكوروتينات المنتظرة +1:

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

3. إذا كان الكوروتين الحالي في حالة مجاعة، والقفل لا يزال مشغولًا، يدخل وضع المجاعة:

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

4. إذا كان الكوروتين الحالي قد أُوقظ من الدوران، يُضاف علم mutexWoken:

if awoke {
	new &^= mutexWoken
}

ثم يبدأ بمحاولة تحديث حالة القفل عبر CAS، إذا فشل التحديث يبدأ جولة جديدة من الحلقة مباشرة:

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

إذا نجح التحديث يبدأ الحكم التالي:

1. الحالة الأصلية ليست وضع المجاعة، ولا يوجد كوروتين يشغل القفل، فالكوروتين الحالي يمكنه امتلاك القفل مباشرة، يخرج من العملية، ويستمر تنفيذ كود المستخدم.

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

2. فشلت محاولة امتلاك القفل، يُسجل وقت الانتظار، LIFO إذا كان true يعني أن الطابور يعمل بطريقة آخر داخل أول خارج، وإلا FIFO أول داخل أول خارج.

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

3. يحاول الحصول على السيمافور، يدخل دالة runtime.semacquire1، إذا تمكن من الحصول على السيمافور يعود مباشرة دون حجب، وإلا يستدعي runtime.gopark لتعليق الكوروتين الحالي وانتظار تحرير السيمافور.

   runtime_SemacquireMutex(&m.sema, queueLifo, 1)

4. الوصول لهذه المرحلة يعني احتمالين: إما النجاح في الحصول على السيمافور مباشرة، أو كان محجوبًا وأُيقظ للتو ونجح في الحصول على السيمافور، في كلتا الحالتين نجح في الحصول على السيمافور، إذا كان الآن في وضع المجاعة، يمكنه الحصول على القفل مباشرة.

   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. إذا لم يكن وضع المجاعة، يُعاد ضبط iter، ويُعاد بدء عملية الدوران.

   awoke = true
   iter = 0

بهذا تنتهي عملية القفل، العملية كلها معقدة نسبيًا، استخدمت طريقتي الانتظار بالدوران والانتظار بالسيمافور، ووازنت بين الأداء والعدالة، مناسبة لمعظم حالات التنافس على الأقفال.

Unlock

عملية فتح القفل أبسط بكثير، أولًا تحاول الفتح السريع، إذا كان new يساوي 0 يعني أنه لا توجد كوروتينات منتظرة، وليس في وضع المجاعة، أي نجح الفتح، يمكن العودة مباشرة.

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

وإلا يجب الدخول في عملية unlockslow:

1. أولًا يتحقق هل تم الفتح بالفعل:

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

2. إذا كان في وضع المجاعة، يحرر السيمافور مباشرة، يكتمل الفتح. في وضع المجاعة، الكوروتين الذي يفتح القفل حاليًا ينقل ملكية القفل مباشرة للكوروتين التالي في الانتظار.

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

3. ليس وضع المجاعة، يدخل عملية الفتح الطبيعية:

   1. إذا لم يكن هناك كوروتينات تنتظر، أو كوروتين آخر أُيقظ حصل على القفل بالفعل، أو دخل القفل في وضع المجاعة:

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

   2. وإلا، يحرر السيمافور لإيقاظ الكوروتين المنتظر التالي، ويضبط حالة القفل الحالي على mutexWoken:

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

بهذا تنتهي عملية فتح القفل.

RWMutex

قفل القراءة والكتابة RWMutex معرف في ملف sync/rwmutex.go، وتطبيقه مبني أيضًا على قفل المزامنة.

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
}

شرح الحقول:

• w، قفل مزامنة، عندما يمتلكه كوروتين الكتابة، تُحجب باقي كوروتينات الكتابة والقراءة.
• writerSem، سيمافور الكتابة، يستخدم لحجب كوروتينات الكتابة لانتظار كوروتينات القراءة، كوروتينات الكتابة تحصل على السيمافور، وكوروتينات القراءة تحرره.
• readerSem، سيمافور القراءة، يستخدم لحجب كوروتينات القراءة لانتظار كوروتينات الكتابة، كوروتينات القراءة تحصل على السيمافور، وكوروتينات الكتابة تحرره.
• readerCount، الحقل الجوهري، كل قفل القراءة والكتابة يعتمد عليه للحفاظ على الحالة.
• readerWait، يمثل عدد كوروتينات القراءة التي يجب انتظارها عندما يكون كوروتين الكتابة محجوبًا.

مبدأ قفل القراءة والكتابة تقريبًا: عبر قفل المزامنة يجعل كوروتينات الكتابة متبادلة، عبر السيمافورين writerSem و readerSem يجعل القراءة والكتابة متبادلة، والقراءة-القراءة مشتركة.

readerCount

بما أن readerCount يتغير كثيرًا ومهم، سأشرحه بشكل منفصل، يُلخص في الحالات التالية:

1. 0، لا يوجد كوروتينات قراءة نشطة ولا كوروتينات كتابة نشطة، في حالة خمول.
2. -rwmutexMaxReaders، كوروتين كتابة واحد يمتلك قفل المزامنة، لا يوجد كوروتينات قراءة نشطة حاليًا.
3. -rwmutexMaxReaders+N، كوروتين كتابة واحد يمتلك قفل الكتابة، كوروتينات القراءة الحالية يجب أن تنتظر كوروتين الكتابة ليحرر قفل الكتابة.
4. N-rwmutexMaxReaders، كوروتين كتابة واحد يمتلك قفل المزامنة، يجب انتظار كوروتينات القراءة المتبقية لتحرير قفل القراءة.
5. N، يوجد N كوروتين قراءة نشطة، أي N قفل قراءة، لا يوجد كوروتين كتابة نشط.

حيث rwmutexMaxReaders قيمة ثابتة، قيمتها ضعف عدد الكوروتينات التي يمكن لقفل المزامنة حجبها، لأن النصف لكوروتينات القراءة والنصف الآخر لكوروتينات الكتابة.

const rwmutexMaxReaders = 1 << 30

قفل القراءة والكتابة كله هذا readerCount هو الأكثر تعقيدًا، فهم تغيراته يعني فهم سير عمل قفل القراءة والكتابة.

TryLock

كالعادة، لننظر أولًا لـ TryLock() الأبسط:

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

في البداية، يحاول استدعاء TryLock() لقفل المزامنة، إذا فشل يعود مباشرة. ثم يحاول عبر عملية CAS تحديث قيمة readerCount من 0 لـ -rwmutexMaxReaders. 0 يعني عدم وجود كوروتينات قراءة تعمل، -rwmutexMaxReaders يعني أن كوروتين الكتابة يمتلك قفل المزامنة حاليًا. إذا فشلت عملية CAS يفتح قفل المزامنة، إذا نجحت يعيد true.

Lock

التالي Lock()، تنفيذه بسيط أيضًا:

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)
	}
}

أولًا يتنافس مع كوروتينات الكتابة الأخرى حتى يمتلك قفل المزامنة، ثم يقوم بهذه العملية: يطرح rwmutexMaxReaders ذريًا، ثم يضيف rwmutexMaxReaders غير ذريًا:

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

بخطوتين:

1. هذا لإعلام كوروتينات القراءة الأخرى أن كوروتين كتابة يحاول امتلاك القفل، شرحنا هذا في قسم TryLock.

   rw.readerCount.Add(-rwmutexMaxReaders)

2. ثم يضيف rwmutexMaxReaders للحصول على r، هذا r يمثل عدد كوروتينات القراءة التي تعمل حاليًا.

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

ثم يحكم هل يوجد كوروتينات قراءة تعمل، ثم يضيف r لقيمة readerWait، إذا كانت لا تزال لا تساوي 0 يعني الحاجة لانتظار هذه الكوروتينات القراءة لإنهاء عملها، فيدخل عملية runtime_SemacquireRWMutex لمحاولة الحصول على السيمافور writerSem، هذا السيمافور تحرره كوروتينات القراءة، إذا تمكن من الحصول على السيمافور يعني أن كوروتينات القراءة أنهت عملها، وإلا يدخل في طابور الانتظار المحجوب.

Unlock

ثم Unlock()، تحرير قفل الكتابة:

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()
}

عمليته كالتالي:

1. ذكرنا سابقًا أن القفل يحدث عندما يُحدّث readerCount لقيمة سالبة، هنا يضيف rwmutexMaxReaders مرة أخرى، فيعني أن لا يوجد كوروتين كتابة يعمل حاليًا، والقيمة الناتجة هي عدد كوروتينات القراءة المحجوبة المنتظرة.

   r := rw.readerCount.Add(rwmutexMaxReaders)

2. إذا كان 0 أو أكبر من 0، يعني أن قفل الكتابة حُرر بالفعل:

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

3. يحرر السيمافور readerSem، يوقظ كوروتينات القراءة المنتظرة:

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

4. أخيرًا يحرر قفل المزامنة، يوقظ كوروتينات الكتابة المنتظرة.

   rw.w.Unlock()

تحرير قفل الكتابة اكتمل.

TryRLock

الآن لننظر لقفل القراءة، هذا كود TryRLock:

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

يفعل شيئين فقط:

1. يحكم هل يوجد كوروتين كتابة يعمل، إذا وُجد يفشل القفل:

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

2. يحاول زيادة readerCount بواحد، إذا نجح التحديث نجح القفل:

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

3. وإلا يستمر في الحلقة حتى يخرج.

نرى أن readerCount المعتمد هنا يُحافظ عليه في قسم قفل الكتابة، لهذا شرحنا قفل الكتابة أولًا.

RLock

منطق RLock أبسط:

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

يحاول زيادة قيمة readerCount بواحد، إذا كانت القيمة الجديدة لا تزال أقل من 0، يعني أن كوروتين كتابة يعمل، فيدخل عملية حجب السيمافور readerSem، الكوروتين الحالي يدخل طابور الانتظار المحجوب.

RUnlock

RUnlock بسيط وسهل الفهم أيضًا:

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)
	}
}

أولًا يحاول إنقاص readerCount بواحد، يعني أن عدد كوروتينات القراءة النشطة نقص بواحد، إذا كانت القيمة أكبر من 0 يعني يمكن تحريره مباشرة، لأنه لا يوجد كوروتين كتابة يمتلك قفل المزامنة حاليًا، أقل من 0 يعني أن كوروتين كتابة يمتلك قفل المزامنة بالفعل، وهو ينتظر جميع كوروتينات القراءة الحالية لإنهاء عملها. بعدها يدخل عملية runlockSlow:

1. إذا كانت قيمة readerCount الأصلية 0 (القفل خامل) أو -rwmutexMaxReaders (كوروتين الكتابة لا ينتظر كوروتينات قراءة، أي أقفال القراءة حُررت كلها)، يعني أنه لا يوجد كوروتينات قراءة نشطة، لا حاجة للفتح:

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

2. إذا كان هناك كوروتينات قراءة نشطة، يُنقص readerWait بواحد، إذا كان الكوروتين الحالي هو آخر كوروتين قراءة نشط، يحرر السيمافور writerSem، يوقظ كوروتين الكتابة المنتظر:

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

تحرير قفل القراءة انتهى.

Semaphore

السيمافور في قفل المزامنة هو مجرد عدد صحيح uint32، عبر الطرح والإضافة الذرية يمثل الحصول على السيمافور وتحريره، البنية المسؤولة عن الحفاظ على السيمافور في وقت التشغيل هي runtime.semaRoot، معرفة في ملف runtime/sema.go. semaRoot تستخدم شجرة ثنائية متوازنة (treap) لتنظيم وإدارة السيمافورات، كل عقدة في الشجرة تمثل سيمافورًا، نوع العقدة *sudog، وهي قائمة ثنائية الاتجاه، تحافظ على طابور انتظار السيمافور المقابل، العقد تظل فريدة عبر *sudog.elem (عنوان السيمافور)، وتضمن توازن الشجرة عبر الحقل *sudog.ticket.

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.
}

شجرة السيمافور semaRoot تعتمد على قفل مزامنة من مستوى أدنى runtime.mutex لضمان أمان التزامن.

var semtable semTable

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

type semTable [semTabSize]struct {
	root semaRoot
    // لمحاذاة الذاكرة، تحسين الأداء
	pad  [cpu.CacheLinePadSize - unsafe.Sizeof(semaRoot{})]byte
}

semaRoot يُخزن في وقت التشغيل في semaTable عام، يبدو كمصفوفة ثابتة الطول، تُستخدم لتخزين مجموعة عقد جذر أشجار السيمافور المتعددة، لكن من حيث طريقة العمل، هي في الواقع جدول هاش. كل عنصر في الجدول يحتوي على semaRoot وبعض البايتات للحشو (pad)، لمحاذاة الذاكرة وتجنب تنافس سطر الذاكرة المخبئية. semTabSize هو ثابت حجم جدول السيمافور، يحدد طول الجدول بـ 251، عادةً يُختار عدد أولي لتقليل تصادم الهاش وتحسين كفاءة التشتت.

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

طريقة rootFor تكافئ دالة الهاش، تقبل مؤشر addr من نوع uint32 (أي عنوان السيمافور)، وتعيد مؤشر للبنية semaRoot المقابلة لهذا العنوان. هذا السطر يحول addr أولًا لـ uintptr، ثم يزيحه لليمين 3 بتات، أي يقسم على 8 (لأن البايت يشغل 8 بتات، قسمة عنوان المؤشر على 8 تحوله لفهرس المصفوفة)، عبر أخذ الباقي من القسمة على semTabSize، يضمن أن الفهرس ضمن نطاق حجم جدول السيمافور، بعد الحصول على semaRoot عبر الفهرس، يذهب للبحث في الشجرة المتوازنة عن طابور انتظار *sudog المقابل للسيمافور.

Acquire

الحصول على السيمافور، التنفيذ المقابل هو دالة runtime.semacquire1:

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

تستقبل المعاملات التالية:

• addr، عنوان السيمافور
• lifo، يؤثر على ترتيب الخروج من الشجرة المتوازنة، الافتراضي هو FIFO، LIFO أي آخر داخل أول خارج، عندما يكون وقت انتظار الكوروتين للقفل ليس 0 (حُجب مرة واحدة على الأقل)، يكون true
• profile، علم لتحليل أداء القفل
• skipframes، عدد إطارات المكدس المتخطاة
• reason، سبب الحجب

فيما يلي ملخص لعملية الحصول على السيمافور:

1. يحكم على حالة الكوروتين، إذا لم يكن الكوروتين الحالي هو الكوروتين المجدول حاليًا، يرمي استثناءً مباشرة:

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

2. يحكم هل يمكن الحصول على السيمافور، ويحاول الحصول عليه بطريقة غير حاجبة، إذا تمكن يعود مباشرة:

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

3. إذا لم يتمكن من الحصول عليه بطريقة غير حاجبة، يدخل حلقة للحصول عليه بالطريقة العادية، أولًا يحصل على *sudog من المخبأ عبر acquireSudog()، هذه البنية تمثل كوروتينًا محجوبًا ينتظر:

   s := acquireSudog()

4. ثم يحصل على شجرة السيمافور من الجدول العام:

   root := semtable.rootFor(addr)

5. يدخل الحلقة، يقفل شجرة السيمافور، يحكم مرة أخرى هل يمكن الحصول على السيمافور، إذا لم يمكن يضيفها لشجرة السيمافور، ثم يستدعي gopark لتعليقها وانتظار الإيقاظ لمتابعة تكرار هذه العملية، يستمر في الحلقة حتى يحصل على السيمافور:

   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. أخيرًا عند الإيقاظ يُحرر *sudog، يُعاده للمخبأ:

   releaseSudog(s)

Release

تحرير السيمافور، إيقاظ الكوروتين المحجوب المنتظر، هذه الوظيفة تُنفذها دالة runtime.semrelease1:

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

تستقبل المعاملات التالية:

• addr، عنوان السيمافور
• handoff، يشير هل يجب تبديل G الذي يجدوله P الحالي مباشرة لـ G المُوقظ، يكون true فقط في وضع المجاعة
• skipframes، عدد إطارات المكدس المتخطاة

فيما يلي ملخص لعملية التحرير:

1. يحصل على شجرة السيمافور، ثم يزيد السيمافور بواحد، يعني تحرير سيمافور واحد:

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

2. إذا كان عدد الكوروتينات المنتظرة 0، يعود مباشرة:

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

3. يقفل شجرة السيمافور، يحكم مرة ثانية هل يوجد كوروتينات منتظرة:

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

4. يحصل على كوروتين محجوب ينتظر من شجرة السيمافور، يُنقص nwait بواحد، ثم يحرر قفل السيمافور:

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

5. يحكم هل يمكن الحصول على السيمافور:

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

6. دالة readyWithTime ستجعل الكوروتين المُوقظ G هو الكوروتين التالي الذي سيعمل على P، أي تعدل *p.runnext=g:

   readyWithTime(s, 5+skipframes)

7. إذا كان handoff يساوي true، فإن goyield ستجعل الكوروتين G الذي حرر السيمافور حاليًا ينفصل عن M الحالي، ويعاد إضافته لذيل طابور التشغيل المحلي لـ P، ثم تبدأ جولة جديدة من حلقة الجدولة، لكي يمكن جدولة الكوروتين G المُوقظ فورًا:

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

هذه هي عملية الحصول على السيمافور وتحريره، لغة Go تستخدم السيمافور في أكثر من قفل المزامنة، وضعته هنا لأن ارتباطه بقفل المزامنة هو الأكبر، حتى أن المسؤولين كتبوا في التعليقات:

// Asynchronous semaphore for sync.Mutex.

بعد فهم السيمافور، العودة لقراءة قفل المزامنة ستكون أوضح.

TIP

بخصوص شجرة السيمافور semaRoot، عمليات الدخول والخروج منها لأنها تتضمن عمليات التوازن الذاتي معقدة التنفيذ، التعمق في هذه التفاصيل ليس له علاقة بموضوع المقال وليس له معنى، لذا تجاوزتها، يمكنك فهم الكود المصدري بنفسك إذا كنت مهتمًا.

ملخص

قفل المزامنة sync.Mutex يحقق انتظار الكوروتينات عبر آليتي الدوران والسيمافور. الدوران غير حاجب، لكن يحتاج لقيود صارمة في الاستخدام، لأنه يستهلك موارد CPU؛ أما السيمافور فهو حاجب، يمكنه تجنب استهلاك الموارد غير الضروري بفعالية. لتحقيق آلية تنافس أكثر عدالة، ميزت Go بين الوضع الطبيعي ووضع المجاعة لضمان توازن أكبر في عملية تنافس الكوروتينات على القفل. مقارنة بـ runtime.mutex القفل منخفض المستوى، sync.Mutex كقفل موجه للمستخدم، صُمم مع مراعاة المزيد من سيناريوهات الاستخدام الفعلي.

قفل القراءة والكتابة sync.RWMutex، يحقق التبادل بين الكتابة والكتابة عبر قفل المزامنة sync.Mutex، وعلى هذا الأساس أضاف سيمافورين إضافيين، لتحقيق التبادل بين القراءة والكتابة والمشاركة بين القراءة والقراءة، لدعم سيناريوهات التزامن المتعددة.

رغم أن تنفيذ القفل يبدو معقدًا نسبيًا، لكن بمجرد فهم مبدأ Mutex، يصبح تعلم أدوات المزامنة الأخرى في المكتبة القياسية sync أسهل بكثير.