كيفية إغلاق القنوات بأناقة

قبل عدة أيام، كتبت مقالاً يشرح قواعد القنوات في Go. حصل ذلك المقال على أصوات كثيرة على reddit وHN، لكن وُجّهت أيضًا بعض الانتقادات إلى تفاصيل تصميم قنوات Go.

جمعت بعض الانتقادات الموجّهة إلى التصاميم والقواعد التالية لقنوات Go:

1. لا توجد طرق سهلة وعالمية للتحقق مما إذا كانت القناة مغلقة أم لا دون تعديل حالة القناة.
2. إغلاق قناة مغلقة سيؤدي إلى panic، لذا فمن الخطر إغلاق قناة إذا كان المُغلِقون لا يعرفون ما إذا كانت القناة مغلقة أم لا.
3. إرسال قيم إلى قناة مغلقة سيؤدي إلى panic، لذا فمن الخطر إرسال قيم إلى قناة إذا كان المُرسِلون لا يعرفون ما إذا كانت القناة مغلقة أم لا.

تبدو هذه الانتقادات معقولة، لكنها في الحقيقة ليست كذلك. نعم، لا توجد فعلاً دالة مدمجة للتحقق مما إذا كانت القناة قد أُغلقت أم لا.

توجد بالفعل طريقة بسيطة للتحقق مما إذا كانت القناة مغلقة أم لا إذا استطعت التأكد من أنه لم تُرسَل (ولن تُرسَل) أي قيم إلى القناة على الإطلاق. وقد عُرضت هذه الطريقة في المقال السابق. وهنا، ومن أجل تماسك أفضل، تُعرض الطريقة مرة أخرى في المثال التالي.

package main

import "fmt"

type T int

func IsClosed(ch <-chan T) bool {
	select {
	case <-ch:
		return true
	default:
	}

	return false
}

func main() {
	c := make(chan T)
	fmt.Println(IsClosed(c)) // false
	close(c)
	fmt.Println(IsClosed(c)) // true
}

كما ذُكر أعلاه، هذه ليست طريقة عالمية للتحقق مما إذا كانت القناة مغلقة أم لا.

في الحقيقة، حتى لو وُجدت دالة مدمجة بسيطة باسم closed للتحقق مما إذا كانت القناة قد أُغلقت أم لا، فسيكون نفعها محدودًا جدًا، تمامًا مثل الدالة المدمجة len المستخدمة لمعرفة العدد الحالي للقيم المخزّنة في مخزن القيم المؤقت (value buffer) الخاص بقناة ما. والسبب هو أن حالة القناة التي جرى فحصها قد تتغيّر مباشرة بعد إرجاع استدعاء مثل هذه الدوال، بحيث لم تعد القيمة المُرجَعة قادرة على عكس الحالة الأحدث للقناة التي فُحصت للتو. ورغم أنه لا بأس في التوقف عن إرسال قيم إلى قناة ch إذا أرجع الاستدعاء closed(ch) القيمة true، فإنه ليس آمنًا إغلاق القناة أو الاستمرار في إرسال قيم إليها إذا أرجع الاستدعاء closed(ch) القيمة false.

مبدأ إغلاق القنوات

أحد المبادئ العامة لاستخدام قنوات Go هو: لا تُغلق قناة من جهة المستقبِل، ولا تُغلق قناة إذا كان لها عدة مُرسِلين متزامنين. بعبارة أخرى، ينبغي أن نغلق قناة في goroutine الخاصة بالمُرسِل فقط إذا كان هذا المُرسِل هو المُرسِل الوحيد للقناة.

(في ما يلي، سنسمّي المبدأ أعلاه مبدأ إغلاق القنوات.)

بالطبع، هذا ليس مبدأً عالميًا لإغلاق القنوات. فالمبدأ العالمي هو: لا تُغلق القنوات المغلقة (ولا تُرسل قيمًا إليها). إذا استطعنا ضمان عدم قيام أي goroutines بإغلاق قناة غير مغلقة وغير nil أو إرسال قيم إليها بعد الآن، فيمكن لـ goroutine عندئذٍ إغلاق القناة بأمان. غير أن تقديم مثل هذه الضمانات من قِبَل المستقبِل أو من قِبَل أحد مُرسِلين متعددين للقناة يتطلب عادةً جهدًا كبيرًا، ويجعل الشيفرة معقّدة في كثير من الأحيان. وعلى العكس، من الأسهل بكثير الالتزام بمبدأ إغلاق القنوات المذكور أعلاه.

حلول تُغلق القنوات بوقاحة

إذا كنت ستغلق قناة من جهة المستقبِل أو في أحد مُرسِلي القناة المتعددين على أي حال، فيمكنك استخدام آلية recover لمنع الـ panic المحتمل من إسقاط برنامجك. إليك مثالاً (افترض أن نوع عنصر القناة هو T).

func SafeClose(ch chan T) (justClosed bool) {
	defer func() {
		if recover() != nil {
			// يمكن تعديل نتيجة الإرجاع
			// في استدعاء دالة defer.
			justClosed = false
		}
	}()

	// افترض أن ch != nil هنا.
	close(ch)   // panic إذا كانت ch مغلقة
	return true // <=> justClosed = true; return
}

من الواضح أن هذا الحل يخرق مبدأ إغلاق القنوات.

يمكن استخدام الفكرة نفسها لإرسال قيم إلى قناة قد تكون مغلقة.

func SafeSend(ch chan T, value T) (closed bool) {
	defer func() {
		if recover() != nil {
			closed = true
		}
	}()

	ch <- value  // panic إذا كانت ch مغلقة
	return false // <=> closed = false; return
}

لا يخرق الحل الوقح مبدأ إغلاق القنوات فحسب، بل قد تحدث أيضًا تسابقات على البيانات (data races) في هذه العملية.

حلول تُغلق القنوات بأدب

يفضّل كثير من الناس استخدام sync.Once لإغلاق القنوات:

type MyChannel struct {
	C    chan T
	once sync.Once
}

func NewMyChannel() *MyChannel {
	return &MyChannel{C: make(chan T)}
}

func (mc *MyChannel) SafeClose() {
	mc.once.Do(func() {
		close(mc.C)
	})
}

بالطبع، يمكننا أيضًا استخدام sync.Mutex لتجنّب إغلاق قناة عدة مرات:

type MyChannel struct {
	C      chan T
	closed bool
	mutex  sync.Mutex
}

func NewMyChannel() *MyChannel {
	return &MyChannel{C: make(chan T)}
}

func (mc *MyChannel) SafeClose() {
	mc.mutex.Lock()
	defer mc.mutex.Unlock()
	if !mc.closed {
		close(mc.C)
		mc.closed = true
	}
}

func (mc *MyChannel) IsClosed() bool {
	mc.mutex.Lock()
	defer mc.mutex.Unlock()
	return mc.closed
}

قد تكون هذه الطرق مهذّبة، لكنها قد لا تتجنّب تسابقات البيانات. فحاليًا، لا تضمن مواصفة Go عدم حدوث تسابقات على البيانات عندما تُنفَّذ عملية إغلاق قناة وعملية إرسال إلى قناة بشكل متزامن. وإذا استُدعيت دالة SafeClose بشكل متزامن مع عملية إرسال إلى القناة نفسها، فقد تحدث تسابقات على البيانات (رغم أن مثل هذه التسابقات لا تُلحق أي ضرر عمومًا).

حلول تُغلق القنوات بأناقة

من عيوب دالة SafeSend أعلاه أن استدعاءاتها لا يمكن استخدامها كعمليات إرسال تتبع الكلمة المفتاحية case في كتل select. ومن العيوب الأخرى لدالتي SafeSend وSafeClose أعلاه أن كثيرًا من الناس، بمن فيهم أنا، سيرون أن الحلول السابقة باستخدام panic/recover وحزمة sync ليست أنيقة. وفي ما يلي، ستُقدَّم بعض الحلول الخالصة بالقنوات، دون استخدام panic/recover وحزمة sync، لجميع أنواع الحالات.

(في الأمثلة التالية، تُستخدم sync.WaitGroup لجعل الأمثلة كاملة. وقد لا يكون استخدامها ضروريًا دائمًا في الممارسة الفعلية.)

1. عدد M من المستقبِلين ومُرسِل واحد، حيث يقول المُرسِل «لا مزيد من الإرسال» بإغلاق قناة البيانات

هذه أبسط حالة؛ فما عليك سوى أن تدع المُرسِل يغلق قناة البيانات عندما لا يرغب في إرسال المزيد.

package main

import (
	"log"
	"math/rand"
	"sync"
)

func main() {
	log.SetFlags(0)

	// ...
	const Max = 100000
	const NumReceivers = 100

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(NumReceivers)

	// ...
	dataCh := make(chan int)

	// the sender
	go func() {
		for {
			if value := rand.Intn(Max); value == 0 {
				// المُرسِل الوحيد يمكنه إغلاق
				// القناة بأمان في أي وقت.
				close(dataCh)
				return
			} else {
				dataCh <- value
			}
		}
	}()

	// receivers
	for i := 0; i < NumReceivers; i++ {
		go func() {
			defer wgReceivers.Done()

			// استقبل القيم حتى تُغلق dataCh
			// وتصبح قائمة انتظار مخزن القيم
			// الخاصة بـ dataCh فارغة.
			for value := range dataCh {
				log.Println(value)
			}
		}()
	}

	wgReceivers.Wait()
}

2. مستقبِل واحد وعدد N من المُرسِلين، حيث يقول المستقبِل الوحيد «توقفوا عن الإرسال من فضلكم» بإغلاق قناة إشارة إضافية

هذه حالة أكثر تعقيدًا قليلاً من الحالة السابقة. لا يمكننا أن ندع المستقبِل يغلق قناة البيانات لإيقاف نقل البيانات، لأن فعل ذلك سيخرق مبدأ إغلاق القنوات. لكن يمكننا أن ندع المستقبِل يغلق قناة إشارة إضافية لإخطار المُرسِلين بالتوقف عن إرسال القيم.

package main

import (
	"log"
	"math/rand"
	"sync"
)

func main() {
	log.SetFlags(0)

	// ...
	const Max = 100000
	const NumSenders = 1000

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(1)

	// ...
	dataCh := make(chan int)
	stopCh := make(chan struct{})
		// stopCh هي قناة إشارة إضافية.
		// مُرسِلها هو مستقبِل القناة
		// dataCh، ومستقبِلوها هم
		// مُرسِلو القناة dataCh.

	// senders
	for i := 0; i < NumSenders; i++ {
		go func() {
			for {
				// عملية محاولة الاستقبال (try-receive) هدفها
				// محاولة الخروج من الـ goroutine في أبكر
				// وقت ممكن. في هذا المثال المحدد،
				// ليست ضرورية.
				select {
				case <- stopCh:
					return
				default:
				}

				// حتى لو أُغلقت stopCh، فقد لا يُختار الفرع
				// الأول في كتلة select الثانية في بعض
				// الدورات إذا كان الإرسال إلى dataCh
				// غير محجوب أيضًا. لكن هذا مقبول في
				// هذا المثال، لذا يمكن حذف كتلة select
				// الأولى أعلاه.
				select {
				case <- stopCh:
					return
				case dataCh <- rand.Intn(Max):
				}
			}
		}()
	}

	// the receiver
	go func() {
		defer wgReceivers.Done()

		for value := range dataCh {
			if value == Max-1 {
				// مستقبِل القناة dataCh هو أيضًا
				// مُرسِل stopCh. من الآمن
				// إغلاق قناة stop هنا.
				close(stopCh)
				return
			}

			log.Println(value)
		}
	}()

	// ...
	wgReceivers.Wait()
}

كما ذُكر في التعليقات، بالنسبة لقناة الإشارة الإضافية، فإن مُرسِلها هو مستقبِل قناة البيانات. وتُغلق قناة الإشارة الإضافية بواسطة مُرسِلها الوحيد، وهذا يحافظ على مبدأ إغلاق القنوات.

في هذا المثال، لا تُغلق القناة dataCh أبدًا. نعم، لا يلزم إغلاق القنوات. فالقناة ستُجمع في النهاية بواسطة جامع النفايات إذا لم تعد أي goroutines تشير إليها، سواء أُغلقت أم لا. لذا فإن أناقة إغلاق قناة هنا تكمن في عدم إغلاقها.

3. عدد M من المستقبِلين وعدد N من المُرسِلين، حيث يقول أيٌّ منهم «لننهِ اللعبة» بإخطار وسيط (moderator) لإغلاق قناة إشارة إضافية

هذه هي الحالة الأكثر تعقيدًا. لا يمكننا أن ندع أيًّا من المستقبِلين والمُرسِلين يغلق قناة البيانات. ولا يمكننا أن ندع أيًّا من المستقبِلين يغلق قناة إشارة إضافية لإخطار جميع المُرسِلين والمستقبِلين بالخروج من اللعبة. ففعل أيٍّ من ذلك سيخرق مبدأ إغلاق القنوات. ومع ذلك، يمكننا إدخال دور وسيط لإغلاق قناة الإشارة الإضافية. ومن الحيل في المثال التالي كيفية استخدام عملية محاولة الإرسال (try-send) لإخطار الوسيط بإغلاق قناة الإشارة الإضافية.

package main

import (
	"log"
	"math/rand"
	"strconv"
	"sync"
)

func main() {
	log.SetFlags(0)

	// ...
	const Max = 100000
	const NumReceivers = 10
	const NumSenders = 1000

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(NumReceivers)

	// ...
	dataCh := make(chan int)
	stopCh := make(chan struct{})
		// stopCh هي قناة إشارة إضافية.
		// مُرسِلها هو goroutine الوسيط الموضّح
		// أدناه، ومستقبِلوها هم جميع مُرسِلي
		// ومستقبِلي dataCh.
	toStop := make(chan string, 1)
		// تُستخدم القناة toStop لإخطار
		// الوسيط بإغلاق قناة الإشارة
		// الإضافية (stopCh). مُرسِلوها هم أيٌّ من
		// مُرسِلي ومستقبِلي dataCh، ومستقبِلها هو
		// goroutine الوسيط الموضّح أدناه.
		// يجب أن تكون قناة ذات مخزن مؤقت.

	var stoppedBy string

	// moderator
	go func() {
		stoppedBy = <-toStop
		close(stopCh)
	}()

	// senders
	for i := 0; i < NumSenders; i++ {
		go func(id string) {
			for {
				value := rand.Intn(Max)
				if value == 0 {
					// هنا، عملية محاولة الإرسال
					// هدفها إخطار الوسيط بإغلاق
					// قناة الإشارة الإضافية.
					select {
					case toStop <- "sender#" + id:
					default:
					}
					return
				}

				// عملية محاولة الاستقبال هنا هدفها
				// محاولة الخروج من الـ goroutine الخاصة
				// بالمُرسِل في أبكر وقت ممكن. كتل select
				// لمحاولة الاستقبال ومحاولة الإرسال محسّنة
				// خصيصًا من قِبَل المصرِّف القياسي لـ Go،
				// لذا فهي فعّالة جدًا.
				select {
				case <- stopCh:
					return
				default:
				}

				// حتى لو أُغلقت stopCh، فقد لا يُختار الفرع
				// الأول في كتلة select هذه في بعض الدورات
				// (وإلى الأبد نظريًا) إذا كان الإرسال
				// إلى dataCh غير محجوب أيضًا. إذا كان
				// هذا غير مقبول، فعملية محاولة الاستقبال
				// أعلاه ضرورية.
				select {
				case <- stopCh:
					return
				case dataCh <- value:
				}
			}
		}(strconv.Itoa(i))
	}

	// receivers
	for i := 0; i < NumReceivers; i++ {
		go func(id string) {
			defer wgReceivers.Done()

			for {
				// كما في goroutine الخاصة بالمُرسِل،
				// عملية محاولة الاستقبال هنا هدفها
				// محاولة الخروج من goroutine الخاصة
				// بالمستقبِل في أبكر وقت ممكن.
				select {
				case <- stopCh:
					return
				default:
				}

				// حتى لو أُغلقت stopCh، فقد لا يُختار الفرع
				// الأول في كتلة select هذه في بعض الدورات
				// (وإلى الأبد نظريًا) إذا كان الاستقبال
				// من dataCh غير محجوب أيضًا. إذا كان
				// هذا غير مقبول، فعملية محاولة الاستقبال
				// أعلاه ضرورية.
				select {
				case <- stopCh:
					return
				case value := <-dataCh:
					if value == Max-1 {
						// هنا، تُستخدم الحيلة نفسها
						// لإخطار الوسيط بإغلاق قناة
						// الإشارة الإضافية.
						select {
						case toStop <- "receiver#" + id:
						default:
						}
						return
					}

					log.Println(value)
				}
			}
		}(strconv.Itoa(i))
	}

	// ...
	wgReceivers.Wait()
	log.Println("stopped by", stoppedBy)
}

في هذا المثال، لا يزال مبدأ إغلاق القنوات محفوظًا.

يرجى ملاحظة أن حجم المخزن المؤقت (السعة) للقناة toStop هو واحد. والهدف من ذلك تجنّب فقدان الإخطار الأول عندما يُرسَل قبل أن يصبح goroutine الوسيط جاهزًا لاستقبال الإخطار من toStop.

يمكننا أيضًا ضبط سعة القناة toStop على مجموع عدد المُرسِلين والمستقبِلين، وعندئذٍ لن نحتاج إلى كتلة select لمحاولة الإرسال لإخطار الوسيط.

...
toStop := make(chan string, NumReceivers + NumSenders)
...
			value := rand.Intn(Max)
			if value == 0 {
				toStop <- "sender#" + id
				return
			}
...
				if value == Max-1 {
					toStop <- "receiver#" + id
					return
				}
...

4. صيغة مختلفة من حالة «عدد M من المستقبِلين ومُرسِل واحد»: طلب الإغلاق يأتي من goroutine طرف ثالث

أحيانًا، يُطلب أن تصدر إشارة الإغلاق من goroutine طرف ثالث. في مثل هذه الحالات، يمكننا استخدام قناة إشارة إضافية لإخطار المُرسِل بإغلاق قناة البيانات. مثلاً،

package main

import (
	"log"
	"math/rand"
	"sync"
	"time"
)

func main() {
	log.SetFlags(0)

	// ...
	const Max = 100000
	const NumReceivers = 100
	const NumThirdParties = 15

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(NumReceivers)

	// ...
	dataCh := make(chan int)
	closing := make(chan struct{}) // قناة إشارة
	closed := make(chan struct{})

	// يمكن استدعاء الدالة stop
	// عدة مرات بأمان.
	stop := func() {
		select {
		case closing<-struct{}{}:
			<-closed
		case <-closed:
		}
	}

	// بعض goroutines الأطراف الثالثة
	for i := 0; i < NumThirdParties; i++ {
		go func() {
			r := 1 + rand.Intn(3)
			time.Sleep(time.Duration(r) * time.Second)
			stop()
		}()
	}

	// the sender
	go func() {
		defer func() {
			close(closed)
			close(dataCh)
		}()

		for {
			select{
			case <-closing: return
			default:
			}

			select{
			case <-closing: return
			case dataCh <- rand.Intn(Max):
			}
		}
	}()

	// receivers
	for i := 0; i < NumReceivers; i++ {
		go func() {
			defer wgReceivers.Done()

			for value := range dataCh {
				log.Println(value)
			}
		}()
	}

	wgReceivers.Wait()
}

الفكرة المستخدمة في الدالة stop مأخوذة من تعليق كتبه Roger Peppe.

5. صيغة مختلفة من حالة «عدد N من المُرسِلين»: يجب إغلاق قناة البيانات لإخبار المستقبِلين بانتهاء إرسال البيانات

في الحلول الخاصة بحالات N من المُرسِلين أعلاه، وللحفاظ على مبدأ إغلاق القنوات، تجنّبنا إغلاق قنوات البيانات. غير أنه يُشترط أحيانًا أن تُغلق قنوات البيانات في النهاية لتُعلم المستقبِلين بانتهاء إرسال البيانات. في مثل هذه الحالات، يمكننا تحويل حالة N من المُرسِلين إلى حالة مُرسِل واحد باستخدام قناة وسيطة. فالقناة الوسيطة لها مُرسِل واحد فقط، لذا يمكننا إغلاقها بدلاً من إغلاق قناة البيانات الأصلية.

package main

import (
	"log"
	"math/rand"
	"strconv"
	"sync"
	"time"
)

func main() {
	log.SetFlags(0)

	// ...
	const Max = 1000000
	const NumReceivers = 10
	const NumSenders = 1000
	const NumThirdParties = 15

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(NumReceivers)

	// ...
	dataCh := make(chan int)     // ستُغلق
	middleCh := make(chan int)   // لن تُغلق أبدًا
	closing := make(chan string) // قناة إشارة
	closed := make(chan struct{})

	var stoppedBy string

	// يمكن استدعاء الدالة stop
	// عدة مرات بأمان.
	stop := func(by string) {
		select {
		case closing <- by:
			<-closed
		case <-closed:
		}
	}

	// the middle layer
	go func() {
		exit := func(v int, needSend bool) {
			close(closed)
			if needSend {
				dataCh <- v
			}
			close(dataCh)
		}

		for {
			select {
			case stoppedBy = <-closing:
				exit(0, false)
				return
			case v := <- middleCh:
				select {
				case stoppedBy = <-closing:
					exit(v, true)
					return
				case dataCh <- v:
				}
			}
		}
	}()

	// بعض goroutines الأطراف الثالثة
	for i := 0; i < NumThirdParties; i++ {
		go func(id string) {
			r := 1 + rand.Intn(3)
			time.Sleep(time.Duration(r) * time.Second)
			stop("3rd-party#" + id)
		}(strconv.Itoa(i))
	}

	// senders
	for i := 0; i < NumSenders; i++ {
		go func(id string) {
			for {
				value := rand.Intn(Max)
				if value == 0 {
					stop("sender#" + id)
					return
				}

				select {
				case <- closed:
					return
				default:
				}

				select {
				case <- closed:
					return
				case middleCh <- value:
				}
			}
		}(strconv.Itoa(i))
	}

	// receivers
	for range [NumReceivers]struct{}{} {
		go func() {
			defer wgReceivers.Done()

			for value := range dataCh {
				log.Println(value)
			}
		}()
	}

	// ...
	wgReceivers.Wait()
	log.Println("stopped by", stoppedBy)
}

المزيد من الحالات؟

ينبغي أن توجد المزيد من صيغ الحالات، لكن الحالات المعروضة أعلاه هي الأكثر شيوعًا وأساسية. وباستخدام القنوات (وتقنيات البرمجة المتزامنة الأخرى) بذكاء، أعتقد أنه يمكن دائمًا إيجاد حل يحافظ على مبدأ إغلاق القنوات لكل صيغة حالة.

الخاتمة

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

البرمجة بقنوات Go تشبه صنع الفن.