Pipeline قابل للإلغاء

Pipeline قابل للإلغاء — Cancellable Pipeline

الـ pipeline مفيد عندما تمر البيانات على مراحل: إنتاج، تحويل، ثم استهلاك. في Go، context يجعل الإلغاء جزءاً واضحاً من التصميم.

تخيل مساراً يقرأ أرقاماً من مصدر، يحولها، ثم يرسل النتائج إلى مستهلك. هذا هو معنى pipeline: كل مرحلة تفعل شيئاً واحداً، وتسلّم الناتج للمرحلة التالية عبر channel. المشكلة تظهر عندما يتوقف المستهلك مبكراً. إذا لم تعرف المراحل السابقة أن العمل انتهى، قد تبقى goroutines معلقة تحاول إرسال قيم لا يستقبلها أحد.

هنا يأتي دور context. هو لا يقتل goroutine بالقوة، بل يرسل إشارة منظمة: “توقف عندما تصل لنقطة فحص مناسبة”. لذلك يجب أن تكتب كل مرحلة بحيث تراقب ctx.Done() أثناء الإرسال أو الانتظار. هذا التصميم مهم في HTTP requests، jobs طويلة، وأي معالجة يمكن أن يلغيها المستخدم أو ينتهي وقتها.

الخطوة 1: مولّد أرقام

main.go

package main

import "fmt"

func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

func main() {
    for n := range generate(1, 2, 3) {
        fmt.Println(n)
    }
}

Output:

الدالة generate ترجع قناة استقبال فقط <-chan int. هذا يخبر المستدعي أنه يستطيع القراءة من القناة، لكنه لا يرسل إليها ولا يغلقها. هذا تفصيل تصميم مهم: من ينشئ القناة ويمتلك goroutine هو من يغلقها. كلما كانت الملكية أوضح، قل احتمال إغلاق قناة من المكان الخطأ.

الخطوة 2: مرحلة تحويل

نضيف مرحلة تضرب كل رقم في نفسه.

main.go

package main

import "fmt"

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

func main() {
    input := make(chan int)
    go func() {
        defer close(input)
        for i := 1; i <= 3; i++ {
            input <- i
        }
    }()

    for n := range square(input) {
        fmt.Println(n)
    }
}

Output:

مرحلة square لا تعرف مصدر الأرقام ولا تعرف أين ستذهب النتائج. هي تستقبل قناة، تنشئ قناة جديدة، ثم تحوّل كل قيمة. هذه قابلية التركيب هي قوة pipeline. تستطيع لاحقاً وضع مرحلة filter قبلها أو مرحلة format بعدها دون تغيير منطق التربيع.

الخطوة 3: أضف context للإلغاء

كل مرحلة تفحص ctx.Done() حتى لا تبقى goroutine عالقة.

main.go

package main

import (
    "context"
    "fmt"
)

func generate(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case <-ctx.Done():
                return
            case out <- n:
            }
        }
    }()
    return out
}

func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case <-ctx.Done():
                return
            case out <- n * n:
            }
        }
    }()
    return out
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    values := square(ctx, generate(ctx, 1, 2, 3, 4))
    for n := range values {
        fmt.Println(n)
        if n >= 4 {
            cancel()
        }
    }
}

Output:

لاحظ أن select يحيط بعملية الإرسال. هذا مهم لأن الإرسال على قناة غير buffered قد ينتظر إذا لم يكن هناك مستقبل جاهز. عند الإلغاء، نريد للمرحلة أن تستطيع الخروج بدلاً من البقاء معلقة على out <- value. لذلك نضع احتمالين: إما أن يتم الإرسال، أو تصل إشارة الإلغاء.

قواعد تصميم pipeline

كل مرحلة تبدأ goroutine يجب أن تكون مسؤولة عن إغلاق القناة التي تنتجها. وكل مرحلة تستقبل context.Context يجب أن تفحصه في أماكن الانتظار لا في بداية الدالة فقط. لا تخزن context داخل struct طويل العمر بدون سبب؛ مرره مع العملية التي تحتاج الإلغاء. ولا تستخدم context.Background() داخل مرحلة داخلية إذا كان المستدعي أعطاك سياقاً، لأنك بذلك تقطع سلسلة الإلغاء.

من الأخطاء الشائعة أن يطبع المستهلك قيمتين ثم يخرج من الحلقة دون إلغاء. في هذه الحالة قد تبقى مرحلة الإنتاج تحاول إرسال القيمة الثالثة. الإلغاء ليس للزينة؛ هو رسالة للمراحل السابقة أن التوقف مقصود وآمن.

تحدي موجه

أكمل pipeline بحيث يطبع أول مربعين فقط ثم يلغي العمل.

تحدي — Challenge

package main

import (
    "context"
    "fmt"
)

func generate(ctx context.Context) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 1; i <= 5; i++ {
            select {
            case <-ctx.Done():
                return
            case out <- i:
            }
        }
    }()
    return out
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    count := 0
    for n := range generate(ctx) {
        fmt.Println(n * n)
        count++
        // ألغِ بعد قيمتين
    }
    fmt.Println("تم الإلغاء")
}

خلاصة

الـ pipeline الجيد في Go يجمع بين قناتين من التفكير: تدفق البيانات وتدفق الإلغاء. channels تنقل القيم بين المراحل، وcontext ينقل قرار التوقف. عندما تكتب المراحل بهذه الحدود، تستطيع بناء معالجة متقدمة تبقى قابلة للفهم: كل مرحلة صغيرة، كل قناة لها مالك، وكل goroutine تعرف كيف تخرج عند انتهاء الحاجة إليها.