التزامن

البرمجة المتزامنة موضوع واسع، ولا متسع هنا إلا لعدد من النقاط الخاصة بـ Go.

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

ولتشجيع هذه الطريقة في التفكير، اخترنا لها شعارًا:

لا تتواصل لمشاركة الذاكرة؛ بل شارك الذاكرة بالتواصل.

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

وإليك طريقة للتفكير في هذا النموذج: تأمّل برنامجًا أحاديّ الخيط نموذجيًا يعمل على معالج واحد (CPU). فهو لا يحتاج إلى أي أدوات تزامن (synchronization primitives). الآن شغّل نسخة أخرى من هذا البرنامج؛ فهي أيضًا لا تحتاج إلى تزامن. والآن دع النسختين تتواصلان؛ فإذا كان التواصل هو أداة التزامن، فلا حاجة إلى تزامن آخر. فأنابيب Unix، مثلًا، تناسب هذا النموذج تمامًا. ورغم أن منهج Go في التزامن ينشأ عن «العمليات التسلسلية المتواصل» (CSP) لـ Tony Hoare، فيمكن أيضًا رؤيته تعميمًا آمن النوع لأنابيب Unix.

الخيوط المتوازية (goroutines)

تُسمّى بهذه الاسم لأن المصطلحات القائمة — الخيوط (threads) والـ coroutines والعمليات (processes) وما شابه — تحمل دلالات مضلِّلة. فـ goroutine له نموذج بسيط: هي دالة تُنفَّذ بالتزامن مع goroutines أخرى في نطاق العناوين نفسه. وهي خفيفة الوزن، فلا تكلّف أكثر قليلًا من تخصيص مساحة مكدّس. والمكدّسات تبدأ صغيرة، فهي رخيصة، وتنمو بتخصيص (وتحرير) تخزين الـ heap عند الحاجة.

تُوزَّع الـ goroutines على عدّة خيوط نظام تشغيل، فإن اعترت إحداها حجب — مثلًا أثناء انتظار إدخال/إخراج (I/O) — واصلت غيرها العمل. ويخفي تصميمها كثيرًا من تعقيدات إنشاء الخيوط وإدارتها.

بادأ نداء دالة أو طريقة بالكلمة المفتاحية go لتشغيل النداء في goroutine جديدة. وحين يكتمل النداء، تنتهي الـ goroutine في صمت. (الأثر مشابه لِرمز & في صدفة Unix الذي يشغّل أمرًا في الخلفية.)

go list.Sort()  // run list.Sort concurrently; don't wait for it.

ويمكن أن يكون الدالة الحرفية (function literal) مفيدة في نداء goroutine:

func Announce(message string, delay time.Duration) {
    go func() {
        time.Sleep(delay)
        fmt.Println(message)
    }()  // Note the parentheses - must call the function.
}

وفي Go، الدوال الحرفية هي إغلاقيات (closures): فالتنفيذ يضمن بقاء المتغيّرات التي تشير إليها الدالة حيّة ما دامت نشطة.

وهذه الأمثلة ليست عملية جدًّا لأن الدوال لا تملك وسيلة للإشارة إلى اكتمالها. ونتاجًا لذلك نحتاج إلى القنوات.

القنوات

مثل الخرائط تمامًا، تُخصَّص القنوات بـ make، وتتصرّف القيمة الناتجة كمرجع إلى بنية بيانات تقف خلفها. وإن قُدّم وسيط صحيح اختياري، فإنها تحدّد حجم مخزن القناة. والقيمة الافتراضية صفر، أي قناة غير مخزَّنة أو متزامنة:

ci := make(chan int)            // unbuffered channel of integers
cj := make(chan int, 0)         // unbuffered channel of integers
cs := make(chan *os.File, 100)  // buffered channel of pointers to Files

وتدمج القنوات غير المخزَّنة بين التواصل — أي تبادل قيمة — والتزامن، فيضمن أن حسابين (goroutines) في حالة معروفة.

وتوجد أعراف جميلة كثيرة تستعمل القنوات. وإليك واحدًا لنبدأ به. في القسم السابق أطلقنا عملية ترتيب في الخلفية. وتستطيع القناة أن تجعل الـ goroutine المُطلِقة تنتظر اكتمال الترتيب:

c := make(chan int)  // Allocate a channel.
// Start the sort in a goroutine; when it completes, signal on the channel.
go func() {
    list.Sort()
    c <- 1  // Send a signal; value does not matter.
}()
doSomethingForAWhile()
<-c   // Wait for sort to finish; discard sent value.

يحظر المستقبِل دائمًا حتى تتوفّر بيانات للاستلام. فإن كانت القناة غير مخزَّنة، فإن المرسِل يحظر حتى يستلم المستقبِل القيمة. أمّا إذا كان للقناة مخزن، فإن المرسِل لا يحظر إلا حتى تُنسخ القيمة إلى المخزن؛ وإذا كان المخزن ممتلئًا، فهذا يعني الانتظار حتى يستخرج أحد المستقبِلين قيمة.

ويمكن، مثلًا، استعمال القناة المخزَّنة كـ semaphore للحدّ من معدّل العمل. ففي هذا المثال تُمرَّر الطلبات الواردة إلى handle، التي ترسل قيمة في القناة، وتُعالج الطلب، ثم تستقبل قيمة من القناة لتُهيّئ «الـ semaphore» للمستهلك التالي. وسعة مخزن القناة تحدّ عدد النداءات المتزامنة لـ process:

var sem = make(chan int, MaxOutstanding)

func handle(r *Request) {
    sem <- 1    // Wait for active queue to drain.
    process(r)  // May take a long time.
    <-sem       // Done; enable next request to run.
}

func Serve(queue chan *Request) {
    for {
        req := <-queue
        go handle(req)  // Don't wait for handle to finish.
    }
}

ومتى أصبح هناك MaxOutstanding معالجات تنفّذ process، فإن أي معالج إضافي سيحظر محاولًا الإرسال في مخزن القناة الممتلئ، حتى ينتهي أحد المعالجات القائمة ويستقبل من المخزن.

ولهذا التصميم مشكلة: فـ Serve تنشئ goroutine جديدة لكل طلب وارد، مع أن MaxOutstanding منها فقط هي التي تستطيع العمل في أي لحظة. ونتيجة لذلك، يستطيع البرنامج استهلاك موارد بلا حدّ إذا وردت الطلبات بسرعة زائدة. يمكننا معالجة هذا النقص بتغيير Serve بحيث تقيّد إنشاء الـ goroutines:

func Serve(queue chan *Request) {
    for req := range queue {
        sem <- 1
        go func() {
            process(req)
            <-sem
        }()
    }
}

(لاحظ أن في إصدارات Go التي تسبق 1.22 خللًا في هذه الشيفرة: فمتغيّر الحلقة مشترك بين كل الـ goroutines. راجع Go wiki للتفاصيل.)

وهناك نهج آخر يدير الموارد جيدًا، وهو تشغيل عدد ثابت من goroutines handle تقرأ جميعها من قناة الطلبات. ويحدّ عدد الـ goroutines عدد النداءات المتزامنة لـ process. كما أن دالة Serve هذه تقبل أيضًا قناة تُبلَّغ فيها للخروج؛ فبعد إطلاق الـ goroutines تحظر مستقبِلةً من تلك القناة:

func handle(queue chan *Request) {
    for r := range queue {
        process(r)
    }
}

func Serve(clientRequests chan *Request, quit chan bool) {
    // Start handlers
    for i := 0; i < MaxOutstanding; i++ {
        go handle(clientRequests)
    }
    <-quit  // Wait to be told to exit.
}

قنوات القنوات

من أهم خصائص Go أن القناة قيمة من الدرجة الأولى، يمكن تخصيصها وتمريرها كأي قيمة أخرى. ومن الاستعمالات الشائعة لهذه الخاصية تنفيذ فصل إشاري (demultiplexing) متوازٍ وآمن.

وفي المثال في القسم السابق، كانت handle معالجًا مثاليًا لطلب، لكننا لم نعرّف النوع الذي يعالجه. فإذا تضمّن ذلك النوع قناة يُردّ عليها، يستطيع كل عميل أن يوفّر له مساره الخاص للإجابة. وإليك تعريفًا مخطّطيًا للنوع Request:

type Request struct {
    args        []int
    f           func([]int) int
    resultChan  chan int
}

يوفّر العميل دالةً ووسائطها، إضافةً إلى قناة داخل كائن الطلب ليستقبل عليها الإجابة:

func sum(a []int) (s int) {
    for _, v := range a {
        s += v
    }
    return
}

request := &Request{[]int{3, 4, 5}, sum, make(chan int)}
// Send request
clientRequests <- request
// Wait for response.
fmt.Printf("answer: %d\n", <-request.resultChan)

وعلى جانب الخادم، لا يتغيّر شيء سوى دالة المعالجة:

func handle(queue chan *Request) {
    for req := range queue {
        req.resultChan <- req.f(req.args)
    }
}

وهناك بطبيعة الحال الكثير ممّا يلزم لجعل هذا واقعيًا، لكن هذه الشيفرة إطار لنظام RPC متوازٍ غير حاجب ومحدود المعدّل، ولا ترى في الأفق قفلًا واحدًا (mutex).

التوازي

تطبيق آخر لهذه الأفكار هو موازنة حساب عبر عدّة أنوية معالج (CPU). فإذا أمكن تفكيك الحساب إلى أجزاء منفصلة تستطيع التنفيذ باستقلال، فيمكن موازنته، مع قناة تشير إلى اكتمال كل جزء.

ولنفترض لدينا عملية مكلفة نريد تنفيذها على متجّهة من العناصر، وأن تكون قيمة العملية على كل عنصر مستقلة عن غيرها، كما في هذا المثال المثالي:

type Vector []float64

// Apply the operation to v[i], v[i+1] ... up to v[n-1].
func (v Vector) DoSome(i, n int, u Vector, c chan int) {
    for ; i < n; i++ {
        v[i] += u.Op(v[i])
    }
    c <- 1    // signal that this piece is done
}

ونطلق الأجزاء باستقلال في حلقة، جزءًا لكل معالج. والجزء قد تنتهي بأي ترتيب ولا يهمّ ذلك؛ فنكتفي بعدّ إشارات الاكتشاف بتفريغ القناة بعد إطلاق كل الـ goroutines:

const numCPU = 4 // number of CPU cores

func (v Vector) DoAll(u Vector) {
    c := make(chan int, numCPU)  // Buffering optional but sensible.
    for i := 0; i < numCPU; i++ {
        go v.DoSome(i*len(v)/numCPU, (i+1)*len(v)/numCPU, u, c)
    }
    // Drain the channel.
    for i := 0; i < numCPU; i++ {
        <-c    // wait for one task to complete
    }
    // All done.
}

وبدلًا من إنشاء قيمة ثابتة لـ numCPU، يمكننا أن نسأل بيئة التشغيل عن القيمة المناسبة. فالدالة runtime.NumCPU تُرجع عدد أنوية المعالج في الآلة، ومن ثمّ يمكننا أن نكتب:

var numCPU = runtime.NumCPU()

وتوجد أيضًا دالة runtime.GOMAXPROCS التي تُبلّغ عن عدد الأنوية الذي حدده المستخدم لبرنامج Go — أو تضبطه — والتي يستطيع البرنامج تشغيلها في آن واحد. وقيمتها الافتراضية هي قيمة runtime.NumCPU، لكن يمكن تجاوزها بضبط متغيّر البيئة بالاسم نفسه في الصدفة، أو بنداء الدالة برقم موجب. أما نداءها بصفر فيكتفي بالاستعلام عن القيمة. فإذا أردنا احترام طلب المستخدم للموارد، فينبغي أن نكتب:

var numCPU = runtime.GOMAXPROCS(0)

واحذر أن تخلط بين فكرتَي التزامن — وهو تنظيم البرنامج كمكوّنات تُنفَّذ باستقلال — والتوازي — وهو تنفيذ الحسابات بالتوازي لتحقيق الكفاءة على عدّة معالجات. فرغم أن مزايا التزامن في Go تجعل بعض المسائل سهلة التنظيم كحسابات متوازية، فإن Go لغة متزامنة لا متوازية، وليس كل مشكلات التوازي يناسبها نموذج Go. ولمزيد من النقاش حول هذا التمييز، راجع المحادثة المُحالة في مشاركة المدونة هذه.

مخزن متسرّب

حتى أدوات البرمجة المتزامنة تجعل التعبير عن أفكار غير متزامنة أسهل. وإليك مثالًا مُجرَّدًا من حزمة RPC. فـ goroutine الخاص بالعميل يدور في حلقة يستقبل فيها البيانات من مصدر ما، ربما شبكة. ولتفادي تخصيص المخازن وتحريرها، يحتفظ بقائمة مخازن متاحة، ويمثّلها بقناة مخزَّنة. فإذا كانت القناة فارغة، خُصِّص مخزن جديد. وحين يجهز مخزن الرسالة، يُرسَل إلى الخادم على serverChan:

var freeList = make(chan *Buffer, 100)
var serverChan = make(chan *Buffer)

func client() {
    for {
        var b *Buffer
        // Grab a buffer if available; allocate if not.
        select {
        case b = <-freeList:
            // Got one; nothing more to do.
        default:
            // None free, so allocate a new one.
            b = new(Buffer)
        }
        load(b)              // Read next message from the net.
        serverChan <- b      // Send to server.
    }
}

حلقة الخادم تستقبل كل رسالة من العميل، وتُعالجها، ثم تُعيد المخزن إلى القائمة المتاحة:

func server() {
    for {
        b := <-serverChan    // Wait for work.
        process(b)
        // Reuse buffer if there's room.
        select {
        case freeList <- b:
            // Buffer on free list; nothing more to do.
        default:
            // Free list full, just carry on.
        }
    }
}

ويحاول العميل استرجاع مخزن من freeList؛ فإن لم يتوفّر أحد خُصِّص مخزن جديد. أما إرسال الخادم إلى freeList فيُعيد b إلى القائمة المتاحة ما لم تكن القائمة ممتلئة، وعندئذٍ يُترك المخزن على الأرض ليستعيده جامع القمامة. (وتُنفَّذ بنود default في جمل select حين لا تكون أي حالة أخرى جاهزة، أي أن selects لا تحجب أبدًا.) وهذا التنفيذ يبني قائمة مخازن بدلوّة متسرّبة (leaky bucket) في بضعة أسطر فقط، بالاعتماد على القناة المخزَّنة وجامع القمامة في مسك الدفاتر.