فصل ۴ — الگوهای همزمانی در Go

اصول اولیهٔ primitiveهای همزمانی Go را بررسی کردیم و دربارهٔ نحوهٔ استفادهٔ صحیح از آن‌ها صحبت کردیم. در این فصل، عمیق‌تر به نحوهٔ ترکیب این primitiveها در الگوهایی می‌پردازیم که به مقیاس‌پذیر و قابل‌نگهداری ماندن سیستم شما کمک می‌کنند.

اما پیش از شروع، باید به قالب برخی از الگوهای این فصل اشاره کنیم. در بسیاری از مثال‌ها، از channelهایی استفاده می‌کنیم که empty interface (interface{}) را منتقل می‌کنند. استفاده از empty interface در Go بحث‌برانگیز است؛ با این حال، من این کار را به دو دلیل انجام داده‌ام. اول اینکه نوشتن مثال‌های مختصر در بقیهٔ کتاب را آسان‌تر می‌کند. دوم اینکه در برخی موارد معتقدم این نماینده‌تر از آن چیزی است که الگو می‌خواهد به دست آورد. این نکته را در بخش «Pipelines» مستقیم‌تر بررسی می‌کنیم.

اگر این موضوع برای شما خیلی بحث‌برانگیز است، به یاد داشته باشید که همیشه می‌توانید Go generator برای این کد بسازید و الگوها را طوری تولید کنید که از نوعی که به آن علاقه دارید استفاده کنند.

با این مقدمه، بیایید وارد یادگیری الگوهای همزمانی در Go شویم!

Confinement

هنگام کار با کد همزمان، چند گزینه برای عملیات امن وجود دارد. دو مورد از آن‌ها را مرور کردیم:

primitiveهای همگام‌سازی برای اشتراک‌گذاری حافظه (مثلاً sync.Mutex)
همگام‌سازی از طریق ارتباط (مثلاً channelها)

با این حال، چند گزینهٔ دیگر هم وجود دارد که به‌طور ضمنی در چندین فرایند همزمان امن هستند:

دادهٔ immutable
دادهٔ محافظت‌شده با confinement

به‌نوعی، دادهٔ immutable ایده‌آل است چون به‌طور ضمنی concurrent-safe است. هر فرایند همزمان می‌تواند روی همان داده کار کند، اما نمی‌تواند آن را تغییر دهد. اگر بخواهد دادهٔ جدید بسازد، باید کپی جدیدی از داده با تغییرات موردنظر ایجاد کند. این نه‌تنها بار شناختی کمتری روی توسعه‌دهنده می‌گذارد، بلکه اگر به critical sectionهای کوچک‌تر (یا حذف کامل آن‌ها) منجر شود، می‌تواند به برنامه‌های سریع‌تر هم منجر شود. در Go می‌توانید با نوشتن کدی که به‌جای pointer به مقادیر در حافظه از کپی مقادیر استفاده می‌کند، به این هدف برسید. برخی زبان‌ها از pointer همراه با مقادیر صریحاً immutable پشتیبانی می‌کنند؛ Go جزو آن‌ها نیست.

Confinement هم می‌تواند بار شناختی کمتری روی توسعه‌دهنده و critical sectionهای کوچک‌تر ایجاد کند. تکنیک‌های confine کردن مقادیر همزمان کمی پیچیده‌تر از صرفاً پاس دادن کپی مقادیر است، بنابراین در این فصل این تکنیک‌های confinement را عمیق بررسی می‌کنیم.

Confinement ایده‌ای ساده اما قدرتمند است: اطمینان از اینکه اطلاعات فقط از یک فرایند همزمان در دسترس باشد. وقتی این محقق شود، برنامهٔ همزمان به‌طور ضمنی امن است و نیازی به همگام‌سازی نیست. دو نوع confinement ممکن است: ad hoc و lexical.

Confinement از نوع ad hoc زمانی است که confinement را از طریق یک قرارداد به دست می‌آورید — چه توسط جامعهٔ زبان، گروه کاری شما، یا codebaseای که روی آن کار می‌کنید. به‌نظر من، پایبندی به قرارداد در پروژه‌های با هر اندازه‌ای دشوار است مگر اینکه ابزاری برای static analysis روی کد هر بار که کسی commit می‌کند داشته باشید. اینجا مثالی از confinement از نوع ad hoc است که دلیل را نشان می‌دهد:

data := make([]int, 4)

loopData := func(handleData chan<- int) {
    defer close(handleData)
    for i := range data {
        handleData <- data[i]
    }
}

handleData := make(chan int)
go loopData(handleData)

for num := range handleData {
    fmt.Println(num)
}

می‌بینیم که slice دادهٔ data از هم تابع loopData و هم حلقه روی channel handleData در دسترس است؛ اما طبق قرارداد، فقط از تابع loopData به آن دسترسی داریم. اما با لمس کد توسط افراد زیاد و نزدیک شدن deadlineها، ممکن است اشتباه رخ دهد و confinement از بین برود و مشکل ایجاد کند. همان‌طور که گفتم، ابزار static analysis ممکن است چنین مسائلی را بگیرد، اما static analysis روی codebase Go سطح بلوغی را پیشنهاد می‌کند که خیلی تیم‌ها به آن نمی‌رسند. به همین دلیل lexical confinement را ترجیح می‌دهم: کامپایلر را به‌کار می‌گیرد تا confinement را اعمال کند.

Lexical confinement شامل استفاده از lexical scope برای در معرض قرار دادن فقط داده و primitiveهای همزمانی صحیح برای استفادهٔ چند فرایند همزمان است. انجام کار اشتباه را غیرممکن می‌کند. در واقع در فصل ۳ به این موضوع اشاره کردیم. بخش channelها را به یاد آورید که دربارهٔ در معرض قرار دادن فقط جنبهٔ read یا write یک channel برای فرایندهای همزمانی که به آن نیاز دارند صحبت می‌کرد. بیایید دوباره به آن مثال نگاه کنیم:

chanOwner := func() <-chan int {
    results := make(chan int, 5)
    go func() {
        defer close(results)
        for i := 0; i <= 5; i++ {
            results <- i
        }
    }()
    return results
}

consumer := func(results <-chan int) {
    for result := range results {
        fmt.Printf("Received: %d\n", result)
    }
    fmt.Println("Done receiving!")
}

results := chanOwner()
consumer(results)

اینجا channel را در lexical scope تابع chanOwner instantiate می‌کنیم. این دامنهٔ جنبهٔ write از channel results را به closure تعریف‌شده در زیر آن محدود می‌کند. به عبارت دیگر، جنبهٔ write این channel را confine می‌کند تا goroutineهای دیگر نتوانند روی آن بنویسند.

اینجا جنبهٔ read از channel را دریافت می‌کنیم و می‌توانیم آن را به consumer بدهیم که جز خواندن از آن کاری نمی‌تواند بکند. باز هم main goroutine را به نمای read-only از channel محدود می‌کنیم.

اینجا کپی read-only از یک channel int دریافت می‌کنیم. با اعلام اینکه تنها استفاده‌ای که نیاز داریم دسترسی read است، استفاده از channel در تابع consumer را فقط به read محدود می‌کنیم.

با این تنظیم، استفاده از channelها در این مثال کوچک غیرممکن است. این مقدمهٔ خوبی برای confinement است، اما شاید مثال خیلی جالب نباشد چون channelها concurrent-safe هستند. بیایید مثالی از confinement ببینیم که از ساختار داده‌ای غیر concurrent-safe استفاده می‌کند، نمونه‌ای از bytes.Buffer:

printData := func(wg *sync.WaitGroup, data []byte) {
    defer wg.Done()

    var buff bytes.Buffer
    for _, b := range data {
        fmt.Fprintf(&buff, "%c", b)
    }
    fmt.Println(buff.String())
}

var wg sync.WaitGroup
wg.Add(2)
data := []byte("golang")
go printData(&wg, data[:3])
go printData(&wg, data[3:])

wg.Wait()

اینجا slice حاوی سه بایت اول ساختار داده را پاس می‌دهیم.

اینجا slice حاوی سه بایت آخر ساختار داده را پاس می‌دهیم.

در این مثال می‌بینید که چون printData روی slice داده close نمی‌کند، نمی‌تواند به آن دسترسی داشته باشد و باید slice از byte برای کار بگیرد. زیرمجموعه‌های مختلف slice را پاس می‌دهیم و goroutineهایی که start می‌کنیم را فقط به بخشی از slice که پاس می‌دهیم محدود می‌کنیم. به‌خاطر lexical scope، انجام کار اشتباه را غیرممکن کرده‌ایم¹ و نیازی به همگام‌سازی دسترسی حافظه یا اشتراک داده از طریق ارتباط نداریم.

پس نکته چیست؟ چرا confinement را دنبال کنیم اگر همگام‌سازی در دسترس داریم؟ پاسخ بهبود عملکرد و کاهش بار شناختی روی توسعه‌دهندگان است. همگام‌سازی هزینه دارد، و اگر بتوانید از آن اجتناب کنید critical section نخواهید داشت و هزینهٔ همگام‌سازی آن‌ها را نمی‌پردازید. همچنین از کل دسته‌ای از مسائل ممکن با همگام‌سازی دور می‌زنید؛ توسعه‌دهندگان ساده‌تر نیازی به نگرانی دربارهٔ این مسائل ندارند. کد همزمانی که از lexical confinement استفاده می‌کند معمولاً ساده‌تر از کد همزمان بدون متغیرهای lexically confined است، چون در context lexical scope می‌توانید کد synchronous بنویسید.

با این حال، برقراری confinement می‌تواند دشوار باشد و گاهی باید به primitiveهای همزمانی عالی Go تکیه کنیم.

¹ احتمال دستکاری دستی حافظه از طریق پکیج unsafe را نادیده می‌گیرم. به همین دلیل unsafe نامیده شده!

The for-select Loop

چیزی که بارها و بارها در برنامه‌های Go می‌بینید، حلقهٔ for-select است. چیزی شبیه این:

for { // Either loop infinitely or range over something
    select {
    // Do some work with channels
    }
}

چند سناریوی مختلف وجود دارد که این الگو ظاهر می‌شود.

ارسال متغیرهای iteration روی یک channel

اغلب می‌خواهید چیزی که قابل iteration است را به مقادیر روی یک channel تبدیل کنید. چیز خاصی نیست و معمولاً شبیه این است:

for _, s := range []string{"a", "b", "c"} {
    select {
    case <-done:
        return
    case stringStream <- s:
    }
}

حلقهٔ بی‌نهایت در انتظار توقف

ساخت goroutineهایی که تا توقف بی‌نهایت loop می‌کنند بسیار رایج است. چند تغییر از این الگو وجود دارد. انتخاب شما صرفاً ترجیح سبکی است.

تغییر اول select را تا حد ممکن کوتاه نگه می‌دارد:

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

    // Do non-preemptable work
}

اگر channel done بسته نشده باشد، از select خارج می‌شویم و به ادامهٔ بدنهٔ حلقهٔ for می‌رویم.

تغییر دوم کار را در clause default از select قرار می‌دهد:

for {
    select {
    case <-done:
        return
    default:
        // Do non-preemptable work
    }
}

وقتی وارد select می‌شویم، اگر channel done بسته نشده باشد، به‌جای آن clause default اجرا می‌شود.

چیز بیشتری در این الگو نیست، اما همه‌جا دیده می‌شود و ارزش اشاره دارد.

Preventing Goroutine Leaks

همان‌طور که در بخش «Goroutines» پوشش دادیم، goroutineها ارزان و ساختنشان آسان است؛ یکی از چیزهایی که Go را زبان پرکاربردی می‌کند. runtime goroutineها را روی هر تعداد thread سیستم‌عامل multiplex می‌کند تا اغلب لازم نباشد نگران آن سطح abstraction باشیم. اما منابع مصرف می‌کنند و runtime آن‌ها را garbage-collect نمی‌کند، پس هرچقدر footprint حافظه‌شان کوچک باشد، نمی‌خواهیم در process رها شوند. پس چطور اطمینان حاصل کنیم که پاک می‌شوند؟

از ابتدا و گام‌به‌گام فکر کنیم: چرا یک goroutine وجود دارد؟ در فصل ۲ گفتیم goroutineها واحدهای کاری را نمایش می‌دهند که ممکن است با هم موازی اجرا شوند یا نه. چند مسیر برای پایان goroutine وجود دارد:

وقتی کارش تمام شود.
وقتی به‌خاطر خطای غیرقابل‌بازیابی نتواند کارش را ادامه دهد.
وقتی به او گفته شود کار را متوقف کند.

دو مسیر اول را رایگان داریم — این مسیرها الگوریتم شماست — اما لغو کار چطور؟ این مهم‌ترین بخش است به‌خاطر اثر شبکه‌ای: اگر goroutine را start کرده‌اید، احتمالاً با چند goroutine دیگر به‌صورت سازمان‌یافته همکاری می‌کند. حتی می‌توانیم این وابستگی را به‌صورت گراف نمایش دهیم: ادامهٔ اجرای goroutine فرزند ممکن است به دانستن وضعیت چند goroutine دیگر وابسته باشد. goroutine والد (اغلب main goroutine) با این دانش کامل context باید بتواند به goroutineهای فرزند بگوید خاتمه یابند. در فصل بعد به وابستگی بزرگ‌مقیاس goroutineها ادامه می‌دهیم، اما فعلاً ببینیم چطور تضمین کنیم یک goroutine فرزند تکی پاک شود. با مثال سادهٔ leak goroutine شروع کنیم:

doWork := func(strings <-chan string) <-chan interface{} {
    completed := make(chan interface{})
    go func() {
        defer fmt.Println("doWork exited.")
        defer close(completed)
        for s := range strings {
            // Do something interesting
            fmt.Println(s)
        }
    }()
    return completed
}

doWork(nil)
// Perhaps more work is done here
fmt.Println("Done.")

می‌بینیم main goroutine یک channel nil به doWork پاس می‌دهد. بنابراین channel strings هرگز رشته‌ای روی آن نوشته نمی‌شود و goroutine حاوی doWork برای تمام عمر این process در حافظه می‌ماند (اگر goroutine داخل doWork و main goroutine را join کنیم حتی deadlock می‌گیریم).

در این مثال عمر process خیلی کوتاه است، اما در برنامهٔ واقعی goroutineها به‌راحتی در ابتدای برنامهٔ بلندمدت start می‌شوند. در بدترین حالت، main goroutine می‌تواند در طول عمرش مدام goroutine بسازد و باعث creep در مصرف حافظه شود.

راه موفق برای کاهش این، برقراری سیگنال بین goroutine والد و فرزندانش است تا والد بتواند لغو را به فرزندانش اعلام کند. طبق قرارداد، این سیگنال معمولاً channel فقط-خواندنی به نام done است. goroutine والد این channel را به goroutine فرزند پاس می‌دهد و وقتی می‌خواهد goroutine فرزند را لغو کند، channel را می‌بندد. مثال:

doWork := func(
  done <-chan interface{},
  strings <-chan string,
) <-chan interface{} {
    terminated := make(chan interface{})
    go func() {
        defer fmt.Println("doWork exited.")
        defer close(terminated)
        for {
            select {
            case s := <-strings:
                // Do something interesting
                fmt.Println(s)
            case <-done:
                return
            }
        }
    }()
    return terminated
}

done := make(chan interface{})
terminated := doWork(done, nil)

go func() {
    // Cancel the operation after 1 second.
    time.Sleep(1 * time.Second)
    fmt.Println("Canceling doWork goroutine...")
    close(done)
}()

<-terminated
fmt.Println("Done.")

اینجا channel done را به تابع doWork پاس می‌دهیم. طبق قرارداد، این channel اولین پارامتر است.

در این خط الگوی رایج for-select را می‌بینیم. یکی از caseها بررسی می‌کند آیا channel done سیگنال داده است. اگر داده باشد، از goroutine return می‌کنیم.

اینجا goroutine دیگری می‌سازیم که اگر بیش از یک ثانیه بگذرد، goroutine spawn‌شده در doWork را لغو می‌کند.

اینجا goroutine spawn‌شده از doWork را با main goroutine join می‌کنیم.

خروجی:

Canceling doWork goroutine...
doWork exited.
Done.

می‌بینید با وجود پاس دادن nil برای channel strings، goroutine ما همچنان با موفقیت خارج می‌شود. برخلاف مثال قبل، اینجا دو goroutine را join می‌کنیم ولی deadlock نمی‌گیریم. چون پیش از join، goroutine سومی می‌سازیم که پس از یک ثانیه goroutine داخل doWork را لغو کند. leak goroutine را با موفقیت حذف کردیم!

مثال قبلی حالت goroutineهایی که روی channel دریافت می‌کنند را خوب پوشش می‌دهد، اما اگر وضعیت معکوس باشد چطور: goroutineای که روی تلاش برای نوشتن مقدار روی channel block شده؟ مثال سریع:

newRandStream := func() <-chan int {
    randStream := make(chan int)
    go func() {
        defer fmt.Println("newRandStream closure exited.")
        defer close(randStream)
        for {
            randStream <- rand.Int()
        }
    }()

    return randStream
}

randStream := newRandStream()
fmt.Println("3 random ints:")
for i := 1; i <= 3; i++ {
    fmt.Printf("%d: %d\n", i, <-randStream)
}

اینجا وقتی goroutine با موفقیت خاتمه می‌یابد پیام چاپ می‌کنیم.

خروجی:

3 random ints:
1: 5577006791947779410
2: 8674665223082153551
3: 6129484611666145821

از خروجی می‌بینید fmt.Println با defer هرگز اجرا نمی‌شود. پس از iteration سوم، goroutine ما روی ارسال عدد تصادفی بعدی به channelای که دیگر خوانده نمی‌شود block می‌ماند. راهی نداریم به producer بگوییم متوقف شود. راه‌حل، مانند حالت دریافت، دادن channel به goroutine producer است تا بداند خارج شود:

newRandStream := func(done <-chan interface{}) <-chan int {
    randStream := make(chan int)
    go func() {
        defer fmt.Println("newRandStream closure exited.")
        defer close(randStream)
        for {
            select {
            case randStream <- rand.Int():
            case <-done:
                return
            }
        }
    }()

    return randStream
}

done := make(chan interface{})
randStream := newRandStream(done)
fmt.Println("3 random ints:")
for i := 1; i <= 3; i++ {
    fmt.Printf("%d: %d\n", i, <-randStream)
}
close(done)

// Simulate ongoing work
time.Sleep(1 * time.Second)

خروجی:

3 random ints:
1: 5577006791947779410
2: 8674665223082153551
3: 6129484611666145821
newRandStream closure exited.

می‌بینیم goroutine به‌درستی پاک می‌شود.

حالا که می‌دانیم چطور از leak goroutine جلوگیری کنیم، می‌توانیم قراردادی تعیین کنیم: اگر یک goroutine مسئول ساخت goroutine دیگری است، مسئول اطمینان از توقف آن goroutine هم هست.

این قرارداد به composable بودن و مقیاس‌پذیری برنامه کمک می‌کند. این تکنیک و قاعده را در بخش‌های «Pipelines» و «The context Package» دوباره می‌بینیم. نحوهٔ اطمینان از توقف goroutineها بسته به نوع و هدف goroutine متفاوت است، اما همه بر پایهٔ پاس دادن channel done بنا شده‌اند.

The or-channel

گاهی می‌خواهید یک یا چند channel done را در یک channel done واحد ترکیب کنید که اگر هر یک از channelهای جزئی بسته شود، بسته شود. نوشتن select که این coupling را انجام دهد قابل‌قبول اما پرحرف است؛ گاهی تعداد channelهای done را در runtime نمی‌دانید. در این حالت، یا اگر one-liner را ترجیح می‌دهید، می‌توانید با الگوی or-channel آن‌ها را ترکیب کنید.

این الگو یک channel done مرکب از طریق بازگشت (recursion) و goroutine می‌سازد. نگاهی بیندازیم:

var or func(channels ...<-chan interface{}) <-chan interface{}
or = func(channels ...<-chan interface{}) <-chan interface{} {
    switch len(channels) {
    case 0:
        return nil
    case 1:
        return channels[0]
    }

    orDone := make(chan interface{})
    go func() {
        defer close(orDone)

        switch len(channels) {
        case 2:
            select {
            case <-channels[0]:
            case <-channels[1]:
            }
        default:
            select {
            case <-channels[0]:
            case <-channels[1]:
            case <-channels[2]:
            case <-or(append(channels[3:], orDone)...):
            }
        }
    }()
    return orDone
}

اینجا تابع or را داریم که slice متغیر از channelها می‌گیرد و یک channel برمی‌گرداند.

چون تابع بازگشتی است، باید معیارهای خاتمه را تنظیم کنیم. اول: اگر slice متغیر خالی است، nil channel برمی‌گردانیم. با پاس دادن هیچ channelای، انتظار نداریم channel مرکب کاری کند.

معیار خاتمه دوم: اگر slice متغیر فقط یک عنصر دارد، همان عنصر را برمی‌گردانیم.

بدنهٔ اصلی تابع و محل بازگشت. goroutine می‌سازیم تا بدون block روی channelها منتظر بمانیم.

به‌خاطر نحوهٔ بازگشت، هر فراخوانی بازگشتی or حداقل دو channel دارد. برای محدود نگه داشتن تعداد goroutineها، حالت خاص برای فراخوانی or با فقط دو channel داریم.

اینجا به‌صورت بازگشتی or-channel از همهٔ channelهای slice بعد از اندیس سوم می‌سازیم و از آن select می‌کنیم. این رابطهٔ بازگشتی بقیهٔ slice را به or-channelها destructure می‌کند تا درختی بسازد که اولین سیگنال برگردد. orDone را هم پاس می‌دهیم تا وقتی goroutineهای بالای درخت خارج می‌شوند، goroutineهای پایین درخت هم خارج شوند.

تابع نسبتاً مختصری است که هر تعداد channel را در یک channel واحد ترکیب می‌کند که به‌محض بسته شدن یا نوشته شدن روی هر channel جزئی، بسته می‌شود. مثال کوتاه: channelهایی که پس از مدت مشخص بسته می‌شوند و تابع or برای ترکیب در یک channel واحد:

sig := func(after time.Duration) <-chan interface{}{
    c := make(chan interface{})
    go func() {
        defer close(c)
        time.Sleep(after)
    }()
    return c
}

start := time.Now()
<-or(
    sig(2*time.Hour),
    sig(5*time.Minute),
    sig(1*time.Second),
    sig(1*time.Hour),
    sig(1*time.Minute),
)
fmt.Printf("done after %v", time.Since(start))

این تابع channelای می‌سازد که وقتی زمان after بگذرد بسته می‌شود.

اینجا تقریباً زمان شروع block شدن channel از تابع or را ردیابی می‌کنیم.

و اینجا زمان read را چاپ می‌کنیم.

خروجی:

done after 1.000216772s

با وجود قرار دادن چند channel با زمان‌های مختلف بسته شدن در فراخوانی or، channelی که پس از یک ثانیه بسته می‌شود باعث بسته شدن کل channel ساخته‌شده توسط or می‌شود. چون — صرف‌نظر از جایش در درختی که or می‌سازد — همیشه اول بسته می‌شود و channelهای وابسته به بسته شدنش هم بسته می‌شوند.

این اختصار را به قیمت goroutineهای اضافی به دست می‌آوریم — f(x)=⌊x/2⌋ که x تعداد goroutineهاست — اما به یاد داشته باشید یکی از نقاط قوت Go توانایی سریع ساخت، زمان‌بندی و اجرای goroutineهاست و زبان فعالانه استفاده از goroutine برای مدل‌سازی صحیح مسائل را تشویق می‌کند. نگرانی دربارهٔ تعداد goroutineهای ساخته‌شده اینجا احتمالاً بهینه‌سازی زودهنگام است. علاوه بر این، اگر در compile time ندانید با چند channel done کار می‌کنید، راه دیگری برای ترکیب آن‌ها نیست.

این الگو در تقاطع ماژول‌های سیستم مفید است. در این تقاطع‌ها معمولاً چند شرط برای لغو درخت goroutineها در call stack دارید. با تابع or می‌توانید آن‌ها را ترکیب و به پایین stack پاس دهید. راه دیگر در «The context Package» را هم می‌بینیم که شاید توصیفی‌تر باشد.

همچنین نگاهی به استفاده از تغییر این الگو برای ساخت الگوی پیچیده‌تر در «Replicated Requests» خواهیم داشت.

Error Handling

در برنامه‌های همزمان، error handling را درست پیاده کردن دشوار است. گاهی آن‌قدر روی نحوهٔ اشتراک اطلاعات و هماهنگی فرایندها فکر می‌کنیم که فراموش می‌کنیم چطون gracefully با حالت‌های خطا برخورد کنند. وقتی Go مدل رایج exception برای خطاها را رد کرد، گفت error handling مهم است و همان‌قدر که به الگوریتم‌ها توجه می‌کنیم باید به مسیرهای خطا هم توجه کنیم. در همین روح، ببینیم با چند فرایند همزمان چطور این کار را انجام می‌دهیم.

سؤال بنیادی: «چه کسی مسئول handle کردن خطا باشد؟» در نقطه‌ای برنامه باید از انتقال خطا به بالای stack دست بردارد و کاری با آن بکند. چه چیزی مسئول این است؟

با فرایندهای همزمان این سؤال پیچیده‌تر می‌شود. چون فرایند همزمان مستقل از والد یا خواهر و برادرش کار می‌کند، استدلال دربارهٔ کار درست با خطا دشوار است. کد زیر را ببینید:

checkStatus := func(
  done <-chan interface{},
  urls ...string,
) <-chan *http.Response {
    responses := make(chan *http.Response)
    go func() {
        defer close(responses)
        for _, url := range urls {
            resp, err := http.Get(url)
            if err != nil {
                fmt.Println(err)
                continue
            }
            select {
            case <-done:
                return
            case responses <- resp:
            }
        }
    }()
    return responses
}

done := make(chan interface{})
defer close(done)

urls := []string{"https://www.google.com", "https://badhost"}
for response := range checkStatus(done, urls...) {
    fmt.Printf("Response: %v\n", response.Status)
}

اینجا goroutine بهترین تلاشش را برای سیگنال خطا می‌کند. چه کار دیگری می‌تواند بکند؟ نمی‌تواند برگرداند! چند خطا زیاد است؟ ادامهٔ درخواست‌ها؟

خروجی:

Response: 200 OK
Get https://badhost: dial tcp: lookup badhost on 127.0.1.1:53: no such host

goroutine انتخابی نداشته. نمی‌تواند خطا را ببلعد و تنها کار منطقی چاپ خطا و امید به توجه کسی است. goroutineهایتان را در این موقعیت ناجور قرار ندهید. پیشنهاد می‌کنم concerns را جدا کنید: به‌طور کلی فرایندهای همزمان باید خطاها را به بخش دیگری از برنامه بفرستند که اطلاعات کامل دربارهٔ وضعیت برنامه دارد و تصمیم آگاهانه‌تری بگیرد. مثال صحیح:

type Result struct {
    Error error
    Response *http.Response
}
checkStatus := func(done <-chan interface{}, urls ...string) <-chan Result {
    results := make(chan Result)
    go func() {
        defer close(results)

        for _, url := range urls {
            var result Result
            resp, err := http.Get(url)
            result = Result{Error: err, Response: resp}
            select {
            case <-done:
                return
            case results <- result:
            }
        }
    }()
    return results
}
done := make(chan interface{})
defer close(done)

urls := []string{"https://www.google.com", "https://badhost"}
for result := range checkStatus(done, urls...) {
    if result.Error != nil {
        fmt.Printf("error: %v", result.Error)
        continue
    }
    fmt.Printf("Response: %v\n", result.Response.Status)
}

اینجا نوعی می‌سازیم که هم *http.Response و هم خطای ممکن از iteration حلقه در goroutine را در بر می‌گیرد.

این خط channelای برمی‌گرداند که از آن نتایج iteration حلقه خوانده می‌شود.

اینجا نمونهٔ Result با فیلدهای Error و Response تنظیم‌شده می‌سازیم.

اینجا Result را روی channel می‌نویسیم.

اینجا در main goroutine می‌توانیم با context کامل برنامه با خطاهای خروجی از goroutine checkStatus هوشمندانه برخورد کنیم.

خروجی:

Response: 200 OK
error: Get https://badhost: dial tcp: lookup badhost on 127.0.1.1:53:
no such host

نکتهٔ کلیدی: نتیجهٔ بالقوه را با خطای بالقوه couple کرده‌ایم. این مجموعهٔ کامل نتایج ممکن از goroutine checkStatus را نمایش می‌دهد و به main goroutine اجازه می‌دهد دربارهٔ خطاها تصمیم بگیرد. به‌طور گسترده‌تر، concerns مربوط به error handling را از producer goroutine جدا کردیم. این مطلوب است چون goroutineی که producer را spawn کرده — اینجا main goroutine — context بیشتری دارد و تصمیم هوشمندانه‌تری می‌گیرد.

در مثال قبلی خطاها را به stdio نوشتیم، اما می‌توانستیم کار دیگری بکنیم. برنامه را کمی تغییر دهیم تا اگر سه خطا یا بیشتر رخ داد، دیگر وضعیت را بررسی نکند:

done := make(chan interface{})
defer close(done)

errCount := 0
urls := []string{"a", "https://www.google.com", "b", "c", "d"}
for result := range checkStatus(done, urls...) {
    if result.Error != nil {
        fmt.Printf("error: %v\n", result.Error)
        errCount++
        if errCount >= 3 {
            fmt.Println("Too many errors, breaking!")
            break
        }
        continue
    }
    fmt.Printf("Response: %v\n", result.Response.Status)
}

خروجی:

error: Get a: unsupported protocol scheme ""
Response: 200 OK
error: Get b: unsupported protocol scheme ""
error: Get c: unsupported protocol scheme ""
Too many errors, breaking!

می‌بینید چون خطاها از checkStatus برمی‌گردند و داخل goroutine handle نمی‌شوند، error handling الگوی آشنای Go را دنبال می‌کند. مثال ساده است، اما تصور کنید main goroutine نتایج چند goroutine را هماهنگ می‌کند و قواعد پیچیده‌تری برای ادامه یا لغو goroutineهای فرزند می‌سازد. نکتهٔ اصلی: خطاها باید هنگام ساخت مقادیری که از goroutineها برمی‌گردند، شهروند درجهٔ یک باشند. اگر goroutine می‌تواند خطا تولید کند، آن خطاها باید با نوع نتیجه tightly coupled باشند و از همان خطوط ارتباطی — مانند توابع synchronous معمولی — عبور کنند.

Pipelines

وقتی برنامه می‌نویسید، احتمالاً یک تابع بلند نمی‌نویسید — حداقل امیدوارم نه! abstractionها را به‌صورت تابع، struct، method و غیره می‌سازید. چرا؟ جز برای abstract کردن جزئیاتی که به جریان کلی مهم نیستند، تا بتوانید روی یک بخش کد بدون تأثیر روی بخش‌های دیگر کار کنید. آیا مجبور شده‌اید برای یک تغییر منطقی به چند بخش سیستم دست بزنید؟ شاید به‌خاطر abstraction ضعیف است.

Pipeline فقط ابزار دیگری برای abstraction در سیستم است. به‌ویژه وقتی برنامه باید streamها یا batchهای داده را پردازش کند بسیار قدرتمند است. واژهٔ pipeline گمان می‌رود اولین بار در ۱۸۵۶ استفاده شده و احتمالاً به خط لوله‌ای اشاره داشت که مایع را از جایی به جای دیگر می‌برد. در علوم کامپیوتر هم چیزی را از جایی به جای دیگر منتقل می‌کنیم: داده. pipeline چیزی بیش از مجموعه‌ای از مراحلی نیست که داده می‌گیرند، عملیاتی روی آن انجام می‌دهند و داده را برمی‌گردانند. هر عملیات را stage از pipeline می‌نامیم.

با pipeline، concerns هر stage جدا می‌شود و مزایای زیادی دارد: stageها را مستقل تغییر دهید، نحوهٔ ترکیب stageها را بدون تغییر خود stageها mix and match کنید، هر stage را concurrent با stageهای upstream یا downstream پردازش کنید، و بخش‌هایی از pipeline را fan-out یا rate-limit کنید. fan-out را در «Fan-Out, Fan-In» و rate-limiting را در فصل ۵ پوشش می‌دهیم. فعلاً ساده شروع کنیم و یک stage pipeline بسازیم.

همان‌طور که گفتیم، stage چیزی است که داده می‌گیرد، تبدیل می‌کند و برمی‌گرداند. تابعی که می‌تواند stage pipeline باشد:

multiply := func(values []int, multiplier int) []int {
    multipliedValues := make([]int, len(values))
    for i, v := range values {
        multipliedValues[i] = v * multiplier
    }
    return multipliedValues
}

این تابع slice از int با multiplier می‌گیرد، ضرب می‌کند و slice جدید برمی‌گرداند. stage دیگر:

add := func(values []int, additive int) []int {
    addedValues := make([]int, len(values))
    for i, v := range values {
        addedValues[i] = v + additive
    }
    return addedValues
}

شاید بپرسید چه چیزی این دو تابع را stage pipeline می‌کند نه فقط تابع. ترکیبشان:

ints := []int{1, 2, 3, 4}
for _, v := range add(multiply(ints, 2), 1) {
    fmt.Println(v)
}

خروجی:

add و multiply را در clause range ترکیب کرده‌ایم. چون آن‌ها را با ویژگی‌های stage pipeline ساخته‌ایم، می‌توانیم pipeline بسازیم. ویژگی‌های stage pipeline چیست؟

یک stage همان نوع را consume و return می‌کند.
یک stage باید توسط زبان reified² شود تا قابل پاس دادن باشد. توابع در Go reified هستند و این نقش را خوب بازی می‌کنند.

آشنایان با برنامه‌نویسی تابعی شاید به higher order function و monad فکر کنند. stageهای pipeline به FP نزدیک‌اند و زیرمجموعه‌ای از monadها محسوب می‌شوند. اینجا صریحاً به monad یا FP نمی‌پردازیم، اما دانش هر دو مفید است هرچند ضروری نیست.

add و multiply همهٔ ویژگی‌ها را دارند: هر دو slice از int consume و return می‌کنند و چون Go توابع reified دارد، می‌توانیم آن‌ها را پاس دهیم. این ویژگی‌ها امکان ترکیب stageها در سطح بالاتر بدون تغییر خود stageها را می‌دهد. مثلاً برای stage اضافی ضرب در دو:

ints := []int{1, 2, 3, 4}
for _, v := range multiply(add(multiply(ints, 2), 1), 2) {
    fmt.Println(v)
}

خروجی:

بدون تابع جدید، بدون تغییر توابع موجود، بدون تغییر استفاده از نتیجه pipeline این کار را کردیم. شاید مزایای الگوی pipeline را ببینید. البته می‌توانستیم procedural هم بنویسیم:

ints := []int{1, 2, 3, 4}
for _, v := range ints {
    fmt.Println(2*(v*2+1))
}

در ابتدا ساده‌تر به نظر می‌رسد، اما همان‌طور که می‌بینیم کد procedural همان مزایای pipeline را با stream داده ندارد.

هر stage slice داده می‌گیرد و slice برمی‌گرداند؟ این‌ها batch processing انجام می‌دهند: روی chunk داده یکجا کار می‌کنند نه یک مقدار discrete. نوع دیگر stream processing است: یک عنصر در هر بار دریافت و emit می‌کند.

مزایا و معایب batch در برابر stream را بعداً بحث می‌کنیم. فعلاً برای دست‌نخورده ماندن دادهٔ اصلی، هر stage باید slice جدیدی به همان طول بسازد. یعنی footprint حافظه در هر لحظه دو برابر slice ورودی pipeline است. stageها را stream-oriented کنیم:

multiply := func(value, multiplier int) int {
    return value * multiplier
}

add := func(value, additive int) int {
    return value + additive
}

ints := []int{1, 2, 3, 4}
for _, v := range ints {
    fmt.Println(multiply(add(multiply(v, 2), 1), 2))
}

خروجی:

هر stage یک مقدار discrete دریافت و emit می‌کند و footprint حافظه به اندازهٔ ورودی pipeline برگشته. اما pipeline را به بدنهٔ حلقهٔ for کشیدیم و range کار سنگین feeding را انجام می‌دهد. این reuse نحوهٔ feeding را محدود می‌کند و بعداً مقیاس‌پذیری را هم محدود می‌کند. مشکلات دیگر هم داریم: عملاً pipeline را برای هر iteration می‌سازیم. فراخوانی تابع ارزان است، اما سه فراخوانی برای هر iteration داریم. concurrency چطور؟ گفتم یکی از مزایای pipeline پردازش concurrent هر stage است و fan-out را ذکر کردم. کجا؟

می‌توانستیم multiply و add را بیشتر گسترش دهیم، اما مفهوم pipeline را معرفی کردند. وقت یادگیری best practiceهای ساخت pipeline در Go است و با primitive channel شروع می‌کنیم.

Best Practices for Constructing Pipelines

Channelها برای pipeline در Go مناسب‌اند چون همهٔ نیازهای پایه را برآورده می‌کنند: دریافت و emit مقدار، استفادهٔ امن concurrent، range روی آن‌ها، و reified بودن توسط زبان. مثال قبلی را با channel تبدیل کنیم:

generator := func(done <-chan interface{}, integers ...int) <-chan int {
    intStream := make(chan int)
    go func() {
        defer close(intStream)
        for _, i := range integers {
            select {
            case <-done:
                return
            case intStream <- i:
            }
        }
    }()
    return intStream
}

multiply := func(
  done <-chan interface{},
  intStream <-chan int,
  multiplier int,
) <-chan int {
    multipliedStream := make(chan int)
    go func() {
        defer close(multipliedStream)
        for i := range intStream {
            select {
            case <-done:
                return
            case multipliedStream <- i*multiplier:
            }
        }
    }()
    return multipliedStream
}

add := func(
  done <-chan interface{},
  intStream <-chan int,
  additive int,
) <-chan int {
    addedStream := make(chan int)
    go func() {
        defer close(addedStream)
        for i := range intStream {
            select {
            case <-done:
                return
            case addedStream <- i+additive:
            }
        }
    }()
    return addedStream
}

done := make(chan interface{})
defer close(done)

intStream := generator(done, 1, 2, 3, 4)
pipeline := multiply(done, add(done, multiply(done, intStream, 2), 1), 2)

for v := range pipeline {
    fmt.Println(v)
}

خروجی:

به نظر خروجی مطلوب را تکرار کرده‌ایم، اما با کد بیشتر. چه چیزی به دست آوردیم؟ سه تابع به‌جای دو. همه داخلشان goroutine start می‌کنند و الگوی «Preventing Goroutine Leaks» را با channel برای خروج goroutine به‌کار می‌برند. همه channel برمی‌گردانند و بعضی channel اضافی هم می‌گیرند. بیایید بیشتر بشکنیم:

done := make(chan interface{})
defer close(done)

اولین کار ساخت channel done و close در defer است. خروج تمیز و بدون leak goroutine. بعد تابع generator:

generator := func(done <-chan interface{}, integers ...int) <-chan int {
    intStream := make(chan int)
    go func() {
        defer close(intStream)
        for _, i := range integers {
            select {
            case <-done:
                return
            case intStream <- i:
            }
        }
    }()
    return intStream
}

// ...

intStream := generator(done, 1, 2, 3, 4)

generator slice متغیر از int می‌گیرد، channel بافرشدهٔ int با طول برابر slice می‌سازد، goroutine start می‌کند و channel را برمی‌گرداند. در goroutine ساخته‌شده روی slice متغیر range می‌زند و مقادیر را روی channel می‌فرستد.

send روی channel در select با انتخاب روی channel done اشتراک دارد — الگوی «Preventing Goroutine Leaks». به‌اختصار، generator مجموعهٔ discrete از مقادیر را به stream داده روی channel تبدیل می‌کند. این نوع تابع generator نامیده می‌شود. در ابتدای pipeline همیشه batch داده دارید که باید به channel تبدیل شود. ساخت pipeline:

pipeline := multiply(done, add(done, multiply(done, intStream, 2), 1), 2)

همان pipeline قبلی: ضرب در دو، جمع یک، ضرب دوباره در دو. شبیه pipeline با توابع است اما تفاوت‌های مهم دارد.

اول: از channel استفاده می‌کنیم. در انتها range برای استخراج مقادیر و در هر stage اجرای امن concurrent چون ورودی و خروجی در context همزمان امن‌اند.

دوم: هر stage به‌صورت concurrent اجرا می‌شود. هر stage فقط منتظر ورودی و توانایی send خروجی است. پیامدهای بزرگی در «Fan-Out, Fan-In» دارد؛ فعلاً stageها برای مدتی مستقل اجرا می‌شوند.

سوم: روی pipeline range می‌زنیم و مقادیر از سیستم کشیده می‌شوند:

for v := range pipeline {
    fmt.Println(v)
}

جدول نحوهٔ ورود هر مقدار به هر channel و زمان بسته شدن channelها (Iteration شمارش صفر-based حلقهٔ for است):

Iteration	Generator	Multiply	Add	Multiply	Value
0	1
0		1
0	2		2
0		2		3
0	3		4		6
1		3		5
1	4		6		10
2	(closed)	4		7
2		(closed)	8		14
3			(closed)	9
3				(closed)	18

الگوی سیگنال خروج goroutine را با goroutineهای وابسته بررسی کنیم. اگر پیش از پایان برنامه روی channel done، close کنیم چه می‌شود؟

pipeline := multiply(done, add(done, multiply(done, intStream, 2), 1), 2)

stageها از دو راه به هم وصل‌اند: channel مشترک done و channelهایی که به stageهای بعدی پاس می‌شوند. جدول را دوباره ببینیم و قبل از تکمیل، close(done) کنیم:

Iteration	Generator	Multiply	Add	Multiply	Value
0	1
0		1
0	2		2
0		2		3
1	3		4		6
close(done)	(closed)	3		5
		(closed)	6
			(closed)	7
				(closed)	(exit range)

بستن channel done در pipeline cascade می‌شود. در هر stage دو چیز ممکن است:

range روی channel ورودی. وقتی channel ورودی بسته شود، range خارج می‌شود.
send در select مشترک با channel done.

صرف‌نظر از وضعیت stage — منتظر channel ورودی یا send — بستن channel done stage را خاتمه می‌دهد.

رابطهٔ بازگشتی وجود دارد. در ابتدای pipeline باید مقادیر discrete را به channel تبدیل کنیم. دو نقطه باید preemptable باشند:

ساخت مقدار discrete که تقریباً آنی نیست.
ارسال مقدار discrete روی channel آن.

اول به عهدهٔ شماست. در generator، مقادیر از range روی slice متغیر ساخته می‌شوند که به‌قدر کافی آنی است. دوم با select و channel done handle می‌شود.

در انتهای pipeline، preemptability با استقراء تضمین می‌شود. channelای که range می‌زنیم وقتی preempt شود بسته می‌شود و range می‌شکند. stream وابسته preemptable است.

بین ابتدا و انتها، کد همیشه روی یک channel range می‌زند و روی channel دیگر در select حاوی channel done می‌نویسد.

اگر stage روی دریافت از channel ورودی block باشد، با بسته شدن channel ورودی unblock می‌شود. با استقراء می‌دانیم channel بسته می‌شود چون یا stage مشابه است یا ابتدای preemptable pipeline است. اگر روی send block باشد، به‌خاطر select preemptable است.

پس کل pipeline با بستن channel done همیشه preemptable است.

² در context زبان‌ها، reification یعنی زبان مفهومی را به توسعه‌دهندگان expose می‌کند تا مستقیم با آن کار کنند. توابع در Go reified گفته می‌شوند چون می‌توانید متغیر با نوع امضای تابع تعریف کنید و توابع را در برنامه پاس دهید.

Some Handy Generators

قبلاً گفتم دربارهٔ generatorهای مفید صحبت می‌کنم. یادآوری: generator برای pipeline هر تابعی است که مجموعهٔ discrete از مقادیر را به stream روی channel تبدیل می‌کند. generator به نام repeat:

repeat := func(
    done <-chan interface{},
    values ...interface{},
) <-chan interface{} {
    valueStream := make(chan interface{})
    go func() {
        defer close(valueStream)
        for {
            for _, v := range values {
                select {
                case <-done:
                    return
                case valueStream <- v:
                }
            }
        }
    }()
    return valueStream
}

این تابع مقادیری که پاس می‌دهید را تا گفتن توقف بی‌نهایت تکرار می‌کند.

stage عمومی دیگر مفید با repeat، take:

take := func(
    done <-chan interface{},
    valueStream <-chan interface{},
    num int,
) <-chan interface{} {
    takeStream := make(chan interface{})
    go func() {
        defer close(takeStream)
        for i := 0; i < num; i++ {
            select {
            case <-done:
                return
            case takeStream <- <- valueStream:
            }
        }
    }()
    return takeStream
}

این stage فقط num آیتم اول از valueStream ورودی را می‌گیرد و خارج می‌شود. با هم قدرتمندند:

done := make(chan interface{})
defer close(done)

for num := range take(done, repeat(done, 1), 10) {
    fmt.Printf("%v ", num)
}

خروجی:

1 1 1 1 1 1 1 1 1 1

repeat برای تولید بی‌نهایت یک‌ها ساخته شد، اما فقط ۱۰ تا اول را take می‌گیریم. چون send در repeat روی receive در take block می‌شود، repeat بسیار کارآمد است. با وجود توان تولید stream بی‌نهایت یک‌ها، فقط N+1 نمونه تولید می‌شود که N عددی است که به take پاس می‌دهیم.

می‌توانیم گسترش دهیم. generator تکرارکنندهٔ دیگری که مکرراً تابعی را صدا می‌زند — repeatFn:

repeatFn := func(
    done <-chan interface{},
    fn func() interface{},
) <-chan interface{} {
    valueStream := make(chan interface{})
    go func() {
        defer close(valueStream)
        for {
            select {
            case <-done:
                return
            case valueStream <- fn():
            }
        }
    }()
    return valueStream
}

۱۰ عدد تصادفی:

done := make(chan interface{})
defer close(done)

rand := func() interface{} { return rand.Int()}

for num := range take(done, repeatFn(done, rand), 10) {
    fmt.Println(num)
}

خروجی:

5577006791947779410
8674665223082153551
6129484611666145821
4037200794235010051
3916589616287113937
6334824724549167320
605394647632969758
1443635317331776148
894385949183117216
2775422040480279449

channel بی‌نهایت اعداد تصادفی بر اساس نیاز!

شاید بپرسید چرا همهٔ generatorها و stageها روی channelهای interface{} دریافت و ارسال می‌کنند. می‌توانستیم type-specific بنویسیم یا Go generator بسازیم.

Empty interface در Go کمی taboo است، اما برای stageهای pipeline به‌نظر من OK است با channelهای interface{} کار کنیم تا از کتابخانهٔ استاندارد الگوهای pipeline استفاده کنید. utility زیاد pipeline از stageهای قابل‌استفادهٔ مجدد می‌آید. این وقتی بهترین است که stageها در سطح specificity مناسب خودشان کار کنند. در repeat و repeatFn نگرانی تولید stream با loop روی لیست یا operator است. در take نگرانی محدود کردن pipeline است. هیچ‌کدام به اطلاعات نوع نیاز ندارند، فقط به arity پارامترها.

وقتی به نوع خاص نیاز دارید، stageی برای type assertion قرار دهید. overhead عملکرد stage اضافی (و goroutine) و type assertion ناچیز است. مثال toString:

toString := func(
    done <-chan interface{},
    valueStream <-chan interface{},
) <-chan string {
    stringStream := make(chan string)
    go func() {
        defer close(stringStream)
        for v := range valueStream {
            select {
            case <-done:
                return
            case stringStream <- v.(string):
            }
        }
    }()
    return stringStream
}

استفاده:

done := make(chan interface{})
defer close(done)

var message string
for token := range toString(done, take(done, repeat(done, "I", "am."), 5)) {
    message += token
}

fmt.Printf("message: %s...", message)

خروجی:

message: Iam.Iam.I...

اثبات کنیم هزینهٔ عملکرد generic کردن بخش‌های pipeline ناچیز است. دو benchmark: generic و type-specific:

func BenchmarkGeneric(b *testing.B) {
    done := make(chan interface{})
    defer close(done)

    b.ResetTimer()
    for range toString(done, take(done, repeat(done, "a"), b.N)) {
    }
}

func BenchmarkTyped(b *testing.B) {
    repeat := func(done <-chan interface{}, values ...string) <-chan string {
        valueStream := make(chan string)
        go func() {
            defer close(valueStream)
            for {
                for _, v := range values {
                    select {
                    case <-done:
                        return
                    case valueStream <- v:
                    }
                }
            }
        }()
        return valueStream
    }

    take := func(
        done <-chan interface{},
        valueStream <-chan string,
        num int,
    ) <-chan string {
        takeStream := make(chan string)
        go func() {
            defer close(takeStream)
            for i := num; i > 0 || i == -1; {
                if i != -1 {
                    i--
                }
                select {
                case <-done:
                    return
                case takeStream <- <-valueStream:
                }
            }
        }()
        return takeStream
    }

    done := make(chan interface{})
    defer close(done)

    b.ResetTimer()
    for range take(done, repeat(done, "a"), b.N) {
    }
}

نتایج:

BenchmarkGeneric-4          1000000                   2266     ns/op
BenchmarkTyped-4            1000000                   1181     ns/op
PASS
ok                          command-line-arguments    3.486s

stageهای type-specific دو برابر سریع‌ترند، اما در magnitude فقط کمی سریع‌تر. معمولاً محدودکنندهٔ pipeline یا generator یا stage محاسباتی سنگین است. اگر generator از حافظه نسازد (مثل repeat و repeatFn)، احتمالاً I/O bound هستید. خواندن از دیسک یا شبکه این overhead را می‌پوشاند.

اگر stage محاسباتی گران باشد، آن هم این overhead را می‌پوشاند. اگر هنوز خوشتان نمی‌آید، Go generator برای stageهای generator بنویسید. اگر یک stage محاسباتی گران باشد، چطور کمک کنیم؟ کل pipeline را rate-limit نمی‌کند؟

برای کاهش این، fan-out و fan-in را بحث می‌کنیم.

Fan-Out, Fan-In

pipeline آماده است. داده زیبا جریان دارد و در stageهای زنجیره‌شده transform می‌شود. stream زیبا، stream کند — چرا این‌قدر طول می‌کشد؟

گاهی stageهای pipeline محاسباتی گران‌اند. stageهای upstream در انتظار stageهای گران block می‌شوند. کل pipeline هم طولانی اجرا می‌شود. چه کار کنیم؟

یکی از ویژگی‌های جالب pipeline توانایی کار روی stream با ترکیب stageهای جدا و اغلب قابل بازچینش است. حتی می‌توانید stage را چندبار reuse کنید. جالب نیست یک stage را روی چند goroutine برای parallelize کردن pull از upstream reuse کنیم؟ شاید عملکرد pipeline بهتر شود.

در واقع می‌شود و نامش fan-out, fan-in است.

Fan-out: start کردن چند goroutine برای handle کردن ورودی pipeline.

Fan-in: ترکیب چند نتیجه در یک channel.

چه stageای برای این الگو مناسب است؟ اگر هر دو برقرار باشند:

به مقادیری که stage قبلاً محاسبه کرده وابسته نیست.
اجرایش طولانی است.

order-independence مهم است چون تضمینی نیست کپی‌های concurrent stage به چه ترتیبی اجرا یا برگردند.

مثال: روش ناکارآمد یافتن اعداد اول. از stageهای «Pipelines» استفاده می‌کنیم:

rand := func() interface{} { return rand.Intn(50000000) }

done := make(chan interface{})
defer close(done)

start := time.Now()

randIntStream := toInt(done, repeatFn(done, rand))
fmt.Println("Primes:")
for prime := range take(done, primeFinder(done, randIntStream), 10) {
    fmt.Printf("\t%d\n", prime)
}

fmt.Printf("Search took: %v", time.Since(start))

خروجی:

Primes:
    24941317
    36122539
    6410693
    10128161
    25511527
    2107939
    14004383
    7190363
    45931967
    2393161
Search took: 23.437511647s

stream اعداد تصادفی تا ۵۰٬۰۰۰٬۰۰۰، تبدیل به stream int، سپس primeFinder. primeFinder سعی می‌کند عدد را بر همهٔ اعداد کوچک‌ترش تقسیم کند. روش وحشتناکی برای پیدا کردن اول است، اما طولانی اجرا می‌شود — همان چیزی که می‌خواهیم.

در حلقهٔ for روی اعداد اول، چاپ و با take بستن pipeline پس از ۱۰ اول. زمان جستجو چاپ می‌شود و done با defer بسته و pipeline تخریب می‌شود.

برای جلوگیری از تکرار در نتایج می‌توانستیم stage cache با set اضافه کنیم؛ برای سادگی نادیده می‌گیریم.

حدود ۲۳ ثانیه برای ۱۰ اول. معمولاً الگوریتم را بهبود می‌دهیم، اما هدف اینجا کند بودن stage است، پس fan-out می‌کنیم.

مثال ساده: دو stage — تولید عدد تصادفی و غربال اول. در برنامهٔ بزرگ‌تر stageهای بیشتر؛ کدام را fan-out کنیم؟ معیارها: order-independence و duration. تولید عدد تصادفی order-independent است اما سریع. primeFinder هم order-independent و با الگوریتم naive کند است. کاندیدای خوب fan-out.

fan-out کردن stage در pipeline فوق‌العاده آسان است: چند نسخه از همان stage start کنید. به‌جای:

primeStream := primeFinder(done, randIntStream)

این:

numFinders := runtime.NumCPU()
finders := make([]<-chan int, numFinders)
for i := 0; i < numFinders; i++ {
    finders[i] = primeFinder(done, randIntStream)
}

به‌اندازهٔ تعداد CPU کپی start می‌کنیم. روی کامپیوتر من runtime.NumCPU() هشت برمی‌گرداند. در production احتمالاً تست empirical برای تعداد بهینه؛ اینجا ساده فرض می‌کنیم یک CPU با یک کپی findPrimes مشغول می‌ماند.

هشت goroutine از generator اعداد تصادفی pull می‌کنند و اول بودن را بررسی می‌کنند. تولید تصادفی سریع است و هر goroutine findPrimes بلافاصله عدد بعدی دارد.

اما مشکل: چهار goroutine یعنی چهار channel، اما range روی primes فقط یک channel انتظار دارد. بخش fan-in.

Fan-in یعنی multiplex یا join کردن چند stream در یک stream. الگوریتم:

fanIn := func(
    done <-chan interface{},
    channels ...<-chan interface{},
) <-chan interface{} {
    var wg sync.WaitGroup
    multiplexedStream := make(chan interface{})

    multiplex := func(c <-chan interface{}) {
        defer wg.Done()
        for i := range c {
            select {
            case <-done:
                return
            case multiplexedStream <- i:
            }
        }
    }

    // Select from all the channels
    wg.Add(len(channels))
    for _, c := range channels {
        go multiplex(c)
    }

    // Wait for all the reads to complete
    go func() {
        wg.Wait()
        close(multiplexedStream)
    }()

    return multiplexedStream
}

done استاندارد برای tear down goroutineها و slice متغیر channelهای interface{} برای fan-in.

sync.WaitGroup تا drain شدن همهٔ channelها.

تابع multiplex: از channel می‌خواند و روی multiplexedStream می‌نویسد.

WaitGroup به تعداد channelهای multiplex افزایش می‌یابد.

goroutine منتظر drain شدن همهٔ channelها برای close کردن multiplexedStream.

به‌اختصار fan-in: ساخت channel multiplexed برای consumer، یک goroutine برای هر channel ورودی، و یک goroutine برای بستن multiplexed وقتی ورودی‌ها بسته شدند. چون goroutineی منتظر N goroutine دیگر است، sync.WaitGroup منطقی است. multiplex به WaitGroup اعلام پایان می‌کند.

یادآوری اضافی
پیاده‌سازی naive الگوریتم fan-in و fan-out فقط وقتی درست کار می‌کند که ترتیب رسیدن نتایج مهم نباشد. کاری نکرده‌ایم ترتیب خواندن از randIntStream در غربال حفظ شود. بعداً مثالی برای حفظ ترتیب می‌بینیم.

همه را کنار هم:

done := make(chan interface{})
defer close(done)

start := time.Now()

rand := func() interface{} { return rand.Intn(50000000) }

randIntStream := toInt(done, repeatFn(done, rand))

numFinders := runtime.NumCPU()
fmt.Printf("Spinning up %d prime finders.\n", numFinders)
finders := make([]<-chan interface{}, numFinders)
fmt.Println("Primes:")
for i := 0; i < numFinders; i++ {
    finders[i] = primeFinder(done, randIntStream)
}

for prime := range take(done, fanIn(done, finders...), 10) {
    fmt.Printf("\t%d\n", prime)
}

fmt.Printf("Search took: %v", time.Since(start))

خروجی:

Spinning up 8 prime finders.
Primes:
    6410693
    24941317
    10128161
    36122539
    25511527
    2107939
    14004383
    7190363
    2393161
    45931967
Search took: 5.438491216s

از ~۲۳ ثانیه به ~۵ ثانیه — کاهش ~۷۸٪ بدون تغییر شدید ساختار برنامه. منفعت fan-out و fan-in و utility pipeline.

The or-done-channel

گاهی با channelهایی از بخش‌های مختلف سیستم کار می‌کنید. برخلاف pipelineها، نمی‌توانید دربارهٔ رفتار channel وقتی کد با channel done لغو می‌شود فرضیه بزنید. یعنی نمی‌دانید لغو goroutine شما یعنی channelای که از آن می‌خوانید هم لغو شده. پس طبق «Preventing Goroutine Leaks» باید read از channel را با select که از channel done هم select می‌کند wrap کنیم. این قابل‌قبول است اما کدی که به‌راحتی خوانده می‌شود:

for val := range myChan {
    // Do something with val
}

به این تبدیل می‌شود:

loop:
for {
    select {
    case <-done:
        break loop
    case maybeVal, ok := <-myChan:
        if ok == false {
            return // or maybe break from for
        }
        // Do something with val
    }
}

با حلقه‌های تو در تو شلوغ می‌شود. با تم تمام استفاده از goroutine برای کد همزمان خواناتر و بدون بهینه‌سازی زودهنگام، با یک goroutine درست می‌کنیم:

orDone := func(done, c <-chan interface{}) <-chan interface{} {
    valStream := make(chan interface{})
    go func() {
        defer close(valStream)
        for {
            select {
            case <-done:
                return
            case v, ok := <-c:
                if ok == false {
                    return
                }
                select {
                case valStream <- v:
                case <-done:
                }
            }
        }
    }()
    return valStream
}

باز به حلقهٔ for ساده:

for val := range orDone(done, myChan) {
    // Do something with val
}

edge caseهایی با حلقهٔ تنگ select ممکن است؛ ابتدا خوانایی را امتحان کنید و از بهینه‌سازی زودهنگام اجتناب کنید.

The tee-channel

گاهی می‌خواهید مقادیر ورودی از یک channel را split کنید تا به دو بخش جدا از codebase بفرستید. channel دستورات کاربر: stream دستورات را بگیرید، به اجراکننده بفرستید و به logger برای audit بعدی.

به‌نام دستور tee در سیستم‌های شبیه Unix، tee-channel همین کار را می‌کند. channel برای خواندن می‌گیرد و دو channel جدا که همان مقدار را می‌گیرند برمی‌گرداند:

tee := func(
    done <-chan interface{},
    in <-chan interface{},
) (_, _ <-chan interface{}) { <-chan interface{}) {
    out1 := make(chan interface{})
    out2 := make(chan interface{})
    go func() {
        defer close(out1)
        defer close(out2)
        for val := range orDone(done, in) {
            var out1, out2 = out1, out2
            for i := 0; i < 2; i++ {
                select {
                case <-done:
                case out1<-val:
                    out1 = nil
                case out2<-val:
                    out2 = nil
                }
            }
        }
    }()
    return out1, out2
}

نسخهٔ محلی out1 و out2 می‌خواهیم — shadow می‌کنیم.

یک select تا write به out1 و out2 یکدیگر را block نکنند. دو iteration select: یکی برای هر channel خروجی.

پس از write به یک channel، کپی shadowed را nil می‌کنیم تا write بعدی block شود و channel دیگر ادامه دهد.

write به out1 و out2 tightly coupled است. iteration روی in تا write به هر دو ادامه نمی‌یابد. معمولاً مشکل نیست چون throughput reader هر channel concern جای دیگری است، اما worth noting.

مثال:

done := make(chan interface{})
defer close(done)

out1, out2 := tee(done, take(done, repeat(done, 1, 2), 4))

for val1 := range out1 {
    fmt.Printf("out1: %v, out2: %v\n", val1, <-out2)
}

با این الگو channelها همچنان join pointهای سیستم می‌مانند.

The bridge-channel

گاهی می‌خواهید از دنباله‌ای از channelها مقدار consume کنید:

<-chan <-chan interface{}

کمی متفاوت از coalesce کردن slice channelها در یک channel (مثل «The or-channel» یا «Fan-Out, Fan-In»). دنبالهٔ channelها نوشتن مرتب را پیشنهاد می‌دهد، هرچند از منابع مختلف. مثال: stage pipeline با عمر intermittent. با الگوهای «Confinement» و مالکیت channel توسط goroutine نویسنده، هر restart stage در goroutine جدید channel جدید می‌سازد — دنبالهٔ channelها. در «Healing Unhealthy Goroutines» بیشتر بررسی می‌کنیم.

به‌عنوان consumer شاید مهم نباشد مقادیر از دنبالهٔ channelها می‌آیند. در آن صورت channel of channels دست‌وپاگیر است. تابعی که channel of channels را به channel ساده destructure می‌کند — bridging — کار consumer را آسان‌تر می‌کند:

bridge := func(
    done <-chan interface{},
    chanStream <-chan <-chan interface{},
) <-chan interface{} {
    valStream := make(chan interface{})
    go func() {
        defer close(valStream)
        for {
            var stream <-chan interface{}
            select {
            case maybeStream, ok := <-chanStream:
                if ok == false {
                    return
                }
                stream = maybeStream
            case <-done:
                return
            }
            for val := range orDone(done, stream) {
                select {
                case valStream <- val:
                case <-done:
                }
            }
        }
    }()
    return valStream
}

channelی که همهٔ مقادیر از bridge برمی‌گردد.

حلقهٔ بیرونی channelها را از chanStream می‌کشد و به حلقهٔ تو در تو می‌دهد.

حلقهٔ درونی از channel داده می‌خواند و روی valStream تکرار می‌کند. وقتی stream فعلی بسته شود، از حلقهٔ خواندن خارج و iteration بعدی channel جدید — stream بی‌وقفه.

کد straightforward است. bridge برای facade تک‌channel روی channel of channels. مثال: ۱۰ channel با یک عنصر:

genVals := func() <-chan <-chan interface{} {
    chanStream := make(chan (<-chan interface{}))
    go func() {
        defer close(chanStream)
        for i := 0; i < 10; i++ {
            stream := make(chan interface{}, 1)
            stream <- i
            close(stream)
            chanStream <- stream
        }
    }()
    return chanStream
}

for v := range bridge(nil, genVals()) {
    fmt.Printf("%v ", v)
}

خروجی:

0 1 2 3 4 5 6 7 8 9

با bridge، channel of channels را در یک range استفاده می‌کنیم و روی منطق حلقه تمرکز می‌کنیم. destructure به کد مخصوص این concern واگذار می‌شود.

Queuing

گاهی مفید است پذیرش کار برای pipeline را شروع کنید حتی وقتی pipeline هنوز آمادهٔ بیشتر نیست. این queuing نامیده می‌شود.

یعنی پس از تکمیل کار، stage آن را در حافظهٔ موقت ذخیره می‌کند تا stageهای دیگر بعداً بردارند و stage نیازی به نگه‌داشتن reference ندارد. در بخش «Channels» buffered channel به‌عنوان نوعی queue بحث شد، اما زیاد استفاده نکردیم — و دلیل خوبی دارد.

queuing در سیستم مفید است، اما معمولاً یکی از آخرین تکنیک‌های بهینه‌سازی است. اضافه کردن زودهنگام می‌تواند مسائل همگام‌سازی مثل deadlock و livelock را پنهان کند و با نزدیک شدن به correctness شاید به queue بیشتر یا کمتر نیاز داشته باشید.

queuing برای چیست؟ با خطای رایج شروع کنیم: اضافه کردن queue برای بهبود عملکرد. queue تقریباً هرگز کل runtime برنامه را سریع‌تر نمی‌کند؛ فقط رفتار برنامه را متفاوت می‌کند.

برای درک، pipeline ساده:

done := make(chan interface{})
defer close(done)

zeros := take(done, 3, repeat(done, 0))
short := sleep(done, 1*time.Second, zeros)
long := sleep(done, 4*time.Second, short)
pipeline := long

چهار stage زنجیره‌شده:

repeat — stream بی‌نهایت صفر.
لغو پس از سه آیتم.
stage «short» — sleep یک ثانیه.
stage «long» — sleep چهار ثانیه.

فرض کنیم stageهای ۱ و ۲ آنی‌اند و روی stageهای sleep تمرکز کنیم.

جدول زمان t، iteration i، و زمان باقی‌مانده long و short:

Time(t)	i	Long stage	Short stage
0	0		1s
1	0	4s	1s
2	0	3s	(blocked)
3	0	2s	(blocked)
4	0	1s	(blocked)
5	1	4s	1s
6	1	3s	(blocked)
7	1	2s	(blocked)
8	1	1s	(blocked)
9	2	4s	(close)
10	2	3s
11	2	2s
12	2	1s
13	3	(close)

pipeline حدود ۱۳ ثانیه طول می‌کشد. short حدود ۹ ثانیه.

اگر buffer اضافه کنیم؟ buffer اندازه ۲ بین long و short:

done := make(chan interface{})
defer close(done)

zeros := take(done, 3, repeat(done, 0))
short := sleep(done, 1*time.Second, zeros)
buffer := buffer(done, 2, short)    // Buffers sends from short by 2
long := sleep(done, 4*time.Second, short)
pipeline := long

Time(t)	i	Long stage	Buffer	Short stage
0	0		0/2	1s
1	0	4s	0/2	1s
2	0	3s	1/2	1s
3	0	2s	2/2	(close)
4	0	1s	2/2
5	1	4s	1/2
6	1	3s	1/2
7	1	2s	1/2
8	1	1s	1/2
9	2	4s	0/2
10	2	3s	0/2
11	2	2s	0/2
12	2	1s	0/2
13	3	(close)

کل pipeline هنوز ۱۳ ثانیه! اما runtime short فقط ۳ ثانیه به‌جای ۹. چطور کمک می‌کند؟

pipeline دیگر:

p := processRequest(done, acceptConnection(done, httpHandler))

pipeline تا لغو خارج نمی‌شود و acceptConnection تا لغو pipeline اتصال نمی‌پذیرد. نمی‌خواهید اتصال‌ها timeout شوند چون processRequest روی acceptConnection block است. می‌خواهید acceptConnection تا حد ممکن unblock باشد وگرنه درخواست‌ها رد می‌شوند.

پس utility queue کاهش runtime یک stage نیست، بلکه کاهش زمان حالت block است تا stage کارش را ادامه دهد. کاربران lag می‌بینند، نه رد کامل سرویس.

utility واقعی queue: decouple کردن stageها تا runtime یک stage روی دیگری تأثیر نگذارد. این رفتار کل سیستم را تغییر می‌دهد — خوب یا بد.

سؤال tune کردن: queue کجا؟ اندازه buffer چقدر؟ بسته به ماهیت pipeline.

برای افزایش عملکرد کل سیستم، فقط این موارد:

اگر batch کردن درخواست‌ها در یک stage زمان صرفه‌جویی کند.
اگر تأخیر در یک stage feedback loop به سیستم ایجاد کند.

مثال اول: بافر ورودی در چیزی سریع‌تر (حافظه) نسبت به مقصد (دیسک). هدف پکیج bufio Go. مقایسهٔ write بافرشده و غیربافرشده:

func BenchmarkUnbufferedWrite(b *testing.B) {
    performWrite(b, tmpFileOrFatal())
}

func BenchmarkBufferedWrite(b *testing.B) {
    bufferredFile := bufio.NewWriter(tmpFileOrFatal())
    performWrite(b, bufio.NewWriter(bufferredFile))
}

func tmpFileOrFatal() *os.File {
    file, err := ioutil.TempFile("", "tmp")
    if err != nil {
        log.Fatal("error: %v", err)
    }
    return file
}

func performWrite(b *testing.B, writer io.Writer) {
    done := make(chan interface{})
    defer close(done)

    b.ResetTimer()
    for bt := range take(done, repeat(done, byte(0)), b.N) {
        writer.Write([]byte{bt.(byte)})
    }
}

go test -bench=. src/concurrency-patterns-in-go/queuing/buffering_test.go

نتایج:

BenchmarkUnbufferedWrite-8    500000                   3969     ns/op
BenchmarkBufferedWrite-8      1000000                  1356     ns/op
PASS
ok                            command-line-arguments   3.398s

write بافرشده سریع‌تر است. در bufio.Writer writeها داخلاً در buffer تا رسیدن chunk کافی queue می‌شوند — chunking.

chunking سریع‌تر است چون bytes.Buffer برای ذخیره باید حافظه allocate‌شده را grow کند. grow کردن حافظه گران است؛ پس کمتر grow کنیم سیستم کارآمدتر است. queue عملکرد کل را افزایش داد.

مثال ساده in-memory است؛ chunking در میدان رایج است. هر عملیاتی با overhead — باز کردن transaction دیتابیس، checksum پیام، allocate فضای پیوسته — شاید از chunking سود ببرد.

علاوه بر chunking، queue اگر الگوریتم با lookbehind یا ordering بهینه شود هم کمک می‌کند.

سناریوی دوم: تأخیر در stage باعث ورودی بیشتر به pipeline — سخت‌تر تشخیص دادن، مهم‌تر چون به فروپاشی سیستمیک upstream منجر می‌شود.

negative feedback loop، downward-spiral، یا death-spiral. رابطهٔ بازگشتی بین pipeline و upstream؛ نرخ ارسال درخواست جدید به کارایی pipeline وابسته است.

اگر کارایی زیر آستانهٔ بحرانی بیفتد، upstream ورودی را افزایش می‌دهد، pipeline کارایی را از دست می‌دهد و death-spiral شروع می‌شود. بدون fail-safe، سیستم هرگز بازیابی نمی‌شود.

queue در ورود pipeline feedback loop را می‌شکند به قیمت lag برای درخواست‌ها. از دید caller درخواست در حال پردازش است اما خیلی طولانی. تا timeout نشود pipeline پایدار می‌ماند. اگر timeout شود، check readiness هنگام dequeue لازم است وگرنه پردازش درخواست‌های مرده کارایی را کاهش می‌دهد و feedback loop می‌سازد.

آیا death-spiral دیده‌اید؟
اگر هنگام راه‌اندازی سیستم داغ (سرور بازی جدید، سایت launch محصول و غیره) سایت با وجود تلاش توسعه‌دهندگان down می‌ماند، تبریک! احتمالاً negative feedback loop دیده‌اید. تیم توسعه چیزهای مختلف امتحان می‌کند تا کسی بفهمد queue لازم است و عجولانه پیاده می‌شود. بعد مشتریان از زمان queue شکایت می‌کنند!

الگو: queue در:

ورود pipeline.
stageهایی که batching کارایی را بالا می‌برد.

وسوسهٔ queue جای دیگر — مثلاً بعد از stage گران — را اجتناب کنید! فقط چند موقعیت runtime pipeline را کاهش می‌دهد و پخش کردن queue برای دور زدن این می‌تواند فاجعه‌بار باشد.

برای درک چرا، throughput pipeline را بحث کنیم. در نظریهٔ queue، Little's Law با نمونه‌گیری کافی throughput را پیش‌بینی می‌کند: L=λW که:

L = میانگین تعداد واحدها در سیستم.
λ = میانگین نرخ ورود واحدها.
W = میانگین زمان ماندن واحد در سیستم.

فقط برای سیستم‌های stable. در pipeline، stable یعنی ingress برابر egress. اگر ingress از egress بیشتر باشد، سیستم unstable و death-spiral. اگر کمتر، منابع کامل استفاده نمی‌شوند.

فرض stable. برای کاهش W به عامل n، فقط میانگین L را کاهش دهید: L/n = λ * W/n — یعنی افزایش egress. اگر queue به stageها اضافه کنیم L را افزایش می‌دهیم (nL = nλ * W یا nL = λ * nW). با Little's Law ثابت کردیم queue زمان ماندن در سیستم را کاهش نمی‌دهد.

کاهش W در کل pipeline بین همهٔ stageها توزیع می‌شود:

L = λΣiWi

pipeline فقط به‌اندازهٔ کندترین stage سریع است. بی‌تفاوت optimize نکنید!

Little's Law جالب است. با سه stage فرضی:

چند request در ثانیه؟ فرض ۱ request حدود ۱ ثانیه طول بکشد:

3r = λr/s * 1s
3r/s = λr/s
λr/s = 3r/s

L=3 چون هر stage یک request پردازش می‌کند. W=1 ثانیه. سه request در ثانیه.

اندازه queue برای ۱۰۰٬۰۰۰ request در ثانیه؟ فرض ۱ ms پردازش:

Lr-3r = 100,000r/s * 0.0001s
Lr-3r = 10r
Lr = 7r

سه stage، L را ۳ کم می‌کنیم. λ=100٬۰۰۰ r/s. برای این نرخ، ظرفیت queue باید ۷ باشد. با افزایش queue، کار بیشتر طول می‌کشد تا از سیستم عبور کند — utilization را با lag معامله می‌کنید.

Little's Law دربارهٔ failure insight نمی‌دهد. اگر pipeline panic کند، همهٔ requestهای queue از دست می‌روند. اگر بازسازی سخت است، queue size صفر یا persistent queue (queue persisted برای خواندن بعدی).

Queue در سیستم مفید است، اما به‌خاطر پیچیدگی معمولاً یکی از آخرین بهینه‌سازی‌هایی است که پیشنهاد می‌کنم.

The context Package

در برنامه‌های همزمان اغلب باید عملیات را به‌خاطر timeout، cancellation، یا failure بخش دیگری از سیستم preempt کنیم. idiom ساخت channel done که در برنامه جریان می‌یابد و همهٔ عملیات blockکنندهٔ همزمان را لغو می‌کند را دیدیم. خوب کار می‌کند اما محدود است.

مفید است اطلاعات اضافی کنار اعلان لغو منتقل کنیم: چرا لغو رخ داد، یا آیا تابع deadline دارد.

نیاز wrap کردن channel done با این اطلاعات در سیستم‌های با هر اندازه رایج است و نویسندگان Go الگوی استاندارد ساختند. ابتدا آزمایش خارج standard library بود، اما در Go 1.7 پکیج context به standard library آمد — idiom استاندارد Go برای کد همزمان.

نگاهی به پکیج context:

var Canceled = errors.New("context canceled")
var DeadlineExceeded error = deadlineExceededError{}

type CancelFunc
type Context

func Background() Context
func TODO() Context
func WithCancel(parent Context) (ctx Context, cancel CancelFunc)
func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc)
func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc)
func WithValue(parent Context, key, val interface{}) Context

فعلاً روی نوع Context تمرکز کنیم. مانند channel done در سیستم جریان می‌یابد. با پکیج context، هر تابع downstream از فراخوانی همزمان سطح بالا Context را به‌عنوان اولین آرگومان می‌گیرد:

type Context interface {

    // Deadline returns the time when work done on behalf of this
    // context should be canceled. Deadline returns ok==false when no
    // deadline is set. Successive calls to Deadline return the same
    // results.
    Deadline() (deadline time.Time, ok bool)

    // Done returns a channel that's closed when work done on behalf
    // of this context should be canceled. Done may return nil if this
    // context can never be canceled. Successive calls to Done return
    // the same value.
    Done() <-chan struct{}

    // Err returns a non-nil error value after Done is closed. Err
    // returns Canceled if the context was canceled or
    // DeadlineExceeded if the context's deadline passed. No other
    // values for Err are defined. After Done is closed, successive
    // calls to Err return the same value.
    Err() error

    // Value returns the value associated with this context for key,
    // or nil if no value is associated with key. Successive calls to
    // Value with the same key returns the same result.
    Value(key interface{}) interface{}
}

ساده به نظر می‌رسد. متد Done channelای برمی‌گرداند که وقتی تابع باید preempt شود بسته می‌شود. متدهای جدید اما ساده: Deadline برای زمان لغو، Err برای خطای لغو. Value کمی عجیب — برای چیست؟

یکی از کاربردهای اصلی goroutineها برنامه‌هایی است که request سرویس می‌دهند. معمولاً اطلاعات مخصوص request علاوه بر preemption باید عبور کند. هدف Value همین است. پکیج context دو هدف اصلی دارد:

API برای لغو شاخه‌های call-graph.
data-bag برای انتقال دادهٔ request-scoped در call-graph.

روی لغو تمرکز کنیم. طبق «Preventing Goroutine Leaks»، لغو در تابع سه جنبه دارد:

والد goroutine ممکن است بخواهد آن را لغو کند.
goroutine ممکن است بخواهد فرزندانش را لغو کند.
هر عملیات blockکننده در goroutine باید preemptable باشد.

پکیج context هر سه را مدیریت می‌کند.

Context اولین آرگومان تابع است. روی interface Context چیزی برای mutate کردن state ساختار زیرین نیست. چیزی برای لغو Context توسط تابعی که آن را می‌پذیرد نیست. این توابع بالای call stack را از لغو توسط فرزندان محافظت می‌کند. با متد Done و channel done، Context لغو را از اجداد به‌صورت امن مدیریت می‌کند.

سؤال: اگر Context immutable است، چطور رفتار لغو در توابع پایین‌تر call stack را تحت تأثیر قرار دهیم؟

توابع پکیج context مهم‌اند:

func WithCancel(parent Context) (ctx Context, cancel CancelFunc)
func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc)
func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc)

همه Context می‌گیرند و Context برمی‌گردانند. بعضی deadline یا timeout هم می‌گیرند. همه نمونهٔ جدید Context با گزینه‌های مربوطه می‌سازند.

WithCancel: Context جدید که با فراخوانی cancel برگشتی، channel done بسته می‌شود.

WithDeadline: Context جدید که وقتی ساعت از deadline گذشت، done بسته می‌شود.

WithTimeout: Context جدید که پس از مدت timeout، done بسته می‌شود.

اگر تابع باید لغو را به پایین call-graph تغییر دهد، یکی از این توابع را صدا می‌زند، Context دریافتی را پاس می‌دهد و Context برگشتی را به فرزندان می‌دهد. اگر نیازی به تغییر رفتار لغو نیست، همان Context دریافتی را پاس می‌دهد.

لایه‌های متوالی call-graph می‌توانند Context متناسب با نیازشان بسازند بدون تأثیر روی والد — راه‌حل composable و elegant برای شاخه‌های call-graph.

نمونه‌های Context باید در call-graph برنامه جریان یابند. در paradigm شیءگرا مرجع دادهٔ پرکاربرد را member variable نگه می‌دارند، اما با context.Context این کار را نکنید. نمونه‌ها از بیرون یکسان به نظر می‌رسند اما داخلاً در هر stack-frame ممکن است عوض شوند. همیشه Context را به توابع پاس دهید تا Context موردنظر خودشان را بگیرند، نه Context stack-frame N سطح بالاتر.

در بالای call-graph ناهمزمان، احتمالاً Context به شما پاس نشده. برای شروع زنجیره، دو تابع برای نمونهٔ خالی Context:

func Background() Context
func TODO() Context

Background Context خالی برمی‌گرداند. TODO برای production نیست اما Context خالی برمی‌گرداند؛ placeholder وقتی نمی‌دانید کدام Context یا انتظار دارید upstream هنوز نداده.

بیایید استفاده کنیم. برنامه‌ای که همزمان greeting و farewell چاپ می‌کند — با الگوی channel done:

func main() {
    var wg sync.WaitGroup
    done := make(chan interface{})
    defer close(done)

    wg.Add(1)
    go func() {
        defer wg.Done()
        if err := printGreeting(done); err != nil {
            fmt.Printf("%v", err)
            return
        }
    }()

    wg.Add(1)
    go func() {
        defer wg.Done()
        if err := printFarewell(done); err != nil {
            fmt.Printf("%v", err)
            return
        }
    }()

    wg.Wait()
}

func printGreeting(done <-chan interface{}) error {
    greeting, err := genGreeting(done)
    if err != nil {
        return err
    }
    fmt.Printf("%s world!\n", greeting)
    return nil
}

func printFarewell(done <-chan interface{}) error {
    farewell, err := genFarewell(done)
    if err != nil {
        return err
    }
    fmt.Printf("%s world!\n", farewell)
    return nil
}

func genGreeting(done <-chan interface{}) (string, error) {
    switch locale, err := locale(done); {
    case err != nil:
        return "", err
    case locale == "EN/US":
        return "hello", nil
    }
    return "", fmt.Errorf("unsupported locale")
}

func genFarewell(done <-chan interface{}) (string, error) {
    switch locale, err := locale(done); {
    case err != nil:
        return "", err
    case locale == "EN/US":
        return "goodbye", nil
    }
    return "", fmt.Errorf("unsupported locale")
}

func locale(done <-chan interface{}) (string, error) {
    select {
    case <-done:
        return "", fmt.Errorf("canceled")
    case <-time.After(1*time.Minute):
    }
    return "EN/US", nil
}

خروجی:

goodbye world!
hello world!

(race condition را نادیده می‌گیریم — ممکن است farewell قبل از greeting بیاید!) دو شاخه همزمان اجرا می‌شوند. روش preempt استاندارد: channel done در call-graph. بستن done در main هر دو شاخه را لغو می‌کند.

با goroutine در main، کنترل برنامه به روش‌های مختلف ممکن است. شاید genGreeting اگر طولانی شد timeout شود. شاید genFarewell اگر والد به‌زودی لغو می‌شود locale را صدا نزند. در هر stack-frame تابع می‌تواند کل call stack زیرش را تحت تأثیر قرار دهد.

با الگوی channel done می‌توانستیم done ورودی را wrap کنیم و با fire شدن هر کدام return کنیم، اما deadline و خطاهای Context را نداشتیم.

برای مقایسهٔ آسان‌تر، برنامه را به‌صورت درخت نمایش دهیم. هر گره فراخوانی تابع است.

شکل ۴-۱ — call-graph برنامه greeting/farewell با channel done
main
├── goroutine 1 → printGreeting → genGreeting → locale
└── goroutine 2 → printFarewell → genFarewell → locale
1
2
3

برنامه را با پکیج context به‌جای channel done تغییر دهیم. با انعطاف context.Context سناریوی جالب:

genGreeting فقط یک ثانیه برای locale صبر کند — timeout یک ثانیه. در main اگر printGreeting ناموفق بود، printFarewell را هم لغو کنیم. معنی ندارد goodbye بگوییم اگر hello نگفته‌ایم!

با پکیج context پیاده‌سازی trivial است:

func main() {
    var wg sync.WaitGroup
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    wg.Add(1)
    go func() {
        defer wg.Done()

        if err := printGreeting(ctx); err != nil {
            fmt.Printf("cannot print greeting: %v\n", err)
            cancel()
        }
    }()

    wg.Add(1)
    go func() {
        defer wg.Done()
        if err := printFarewell(ctx); err != nil {
            fmt.Printf("cannot print farewell: %v\n", err)
        }
    }()

    wg.Wait()
}

func printGreeting(ctx context.Context) error {
    greeting, err := genGreeting(ctx)
    if err != nil {
        return err
    }
    fmt.Printf("%s world!\n", greeting)
    return nil
}

func printFarewell(ctx context.Context) error {
    farewell, err := genFarewell(ctx)
    if err != nil {
        return err
    }
    fmt.Printf("%s world!\n", farewell)
    return nil
}

func genGreeting(ctx context.Context) (string, error) {
    ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
    defer cancel()

    switch locale, err := locale(ctx); {
    case err != nil:
        return "", err
    case locale == "EN/US":
        return "hello", nil
    }
    return "", fmt.Errorf("unsupported locale")
}

func genFarewell(ctx context.Context) (string, error) {
    switch locale, err := locale(ctx); {
    case err != nil:
        return "", err
    case locale == "EN/US":
        return "goodbye", nil
    }
    return "", fmt.Errorf("unsupported locale")
}

func locale(ctx context.Context) (string, error) {
    select {
    case <-ctx.Done():
        return "", ctx.Err()
    case <-time.After(1 * time.Minute):
    }
    return "EN/US", nil
}

main با context.Background() و context.WithCancel برای لغو Context جدید می‌سازد.

اگر خطا از printGreeting بیاید، main Context را لغو می‌کند.

genGreeting با context.WithTimeout wrap می‌کند — پس از ۱ ثانیه خودکار لغو و لغو فرزندان مثل locale.

ctx.Err() دلیل لغو را برمی‌گرداند و تا main bubble می‌کند و لغو در آنجا رخ می‌دهد.

خروجی:

cannot print greeting: context deadline exceeded
cannot print farewell: context canceled

با call-graph بفهمیم چه می‌گذرد:

شکل ۴-۲ — call-graph با context و timeout در genGreeting

main (WithCancel)
├── goroutine 1 → printGreeting → genGreeting (WithTimeout 1s) → locale [timeout]
└── goroutine 2 → printFarewell → genFarewell → locale [canceled]

سیستم کامل کار می‌کند. چون locale حداقل یک دقیقه طول می‌کشد، فراخوانی در genGreeting همیشه timeout می‌شود و main call-graph زیر printFarewell را لغو می‌کند.

genGreeting context.Context سفارشی ساخت بدون تأثیر روی Context والد. اگر موفق بود و printGreeting فراخوانی دیگری می‌خواست، بدون leak اطلاعات عملکرد genGreeting انجام می‌داد. composability برای سیستم‌های بزرگ بدون مخلوط کردن concerns در call-graph.

بهبود دیگر: چون locale حدود یک دقیقه طول می‌کشد، در locale بررسی کنیم deadline داریم و آیا به آن می‌رسیم — با Deadline:

func main() {
    var wg sync.WaitGroup
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    wg.Add(1)
    go func() {
        defer wg.Done()

        if err := printGreeting(ctx); err != nil {
            fmt.Printf("cannot print greeting: %v\n", err)
            cancel()
        }
    }()

    wg.Add(1)
    go func() {
        defer wg.Done()
        if err := printFarewell(ctx); err != nil {
            fmt.Printf("cannot print farewell: %v\n", err)
        }
    }()

    wg.Wait()
}

func printGreeting(ctx context.Context) error {
    greeting, err := genGreeting(ctx)
    if err != nil {
        return err
    }
    fmt.Printf("%s world!\n", greeting)
    return nil
}

func printFarewell(ctx context.Context) error {
    farewell, err := genFarewell(ctx)
    if err != nil {
        return err
    }
    fmt.Printf("%s world!\n", farewell)
    return nil
}

func genGreeting(ctx context.Context) (string, error) {
    ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
    defer cancel()

    switch locale, err := locale(ctx); {
    case err != nil:
        return "", err
    case locale == "EN/US":
        return "hello", nil
    }
    return "", fmt.Errorf("unsupported locale")
}

func genFarewell(ctx context.Context) (string, error) {
    switch locale, err := locale(ctx); {
    case err != nil:
        return "", err
    case locale == "EN/US":
        return "goodbye", nil
    }
    return "", fmt.Errorf("unsupported locale")
}

func locale(ctx context.Context) (string, error) {
    if deadline, ok := ctx.Deadline(); ok {
        if deadline.Sub(time.Now().Add(1*time.Minute)) <= 0 {
            return "", context.DeadlineExceeded
        }
    }

    select {
    case <-ctx.Done():
        return "", ctx.Err()
    case <-time.After(1 * time.Minute):
    }
    return "EN/US", nil
}

بررسی deadline در Context. اگر ساعت از deadline گذشته، با context.DeadlineExceeded fail fast.

تفاوت کوچک است اما locale را fail fast می‌کند. در برنامه‌های با هزینهٔ بالای فراخوانی بعدی زمان زیادی صرفه‌جویی می‌کند؛ حداقل بلافاصله fail می‌کند به‌جای انتظار timeout واقعی. شرط: ایده‌ای از مدت call-graph زیردست — تمرین دشوار.

نیمهٔ دیگر پکیج context: data-bag برای ذخیره و بازیابی دادهٔ request-scoped. وقتی تابع goroutine و Context می‌سازد، اغلب فرایندی برای سرویس request است و توابع پایین‌تر به اطلاعات request نیاز دارند. ذخیره و بازیابی:

func main() {
    ProcessRequest("jane", "abc123")
}

func ProcessRequest(userID, authToken string) {
    ctx := context.WithValue(context.Background(), "userID", userID)
    ctx = context.WithValue(ctx, "authToken", authToken)
    HandleResponse(ctx)
}

func HandleResponse(ctx context.Context) {
    fmt.Printf(
        "handling response for %v (%v)",
        ctx.Value("userID"),
        ctx.Value("authToken"),
    )
}

خروجی:

handling response for jane (abc123)

ساده است. شرایط:

کلید باید comparable باشد؛ == و != درست کار کنند.
مقادیر برگشتی باید از چند goroutine امن باشند.

چون key و value هر دو interface{} هستند، type-safety Go را از دست می‌دهیم. نویسندگان Go چند قاعده پیشنهاد می‌کنند.

اول: نوع کلید سفارشی در پکیج تعریف کنید. با همان کار در پکیج‌های دیگر از برخورد در Context جلوگیری می‌شود. یادآوری:

type foo int
type bar int

m := make(map[interface{}]int)
m[foo(1)] = 1
m[bar(1)] = 2

fmt.Printf("%v", m)

خروجی:

map[1:1 1:2]

مقادیر زیرین یکسان اما اطلاعات نوع در map تفکیک می‌کند. چون نوع کلید unexported است، پکیج‌های دیگر با کلیدهای شما conflict نمی‌کنند.

چون کلیدها export نمی‌شوند، توابع export برای بازیابی داده لازم است — type-safe و static.

جمع‌بندی:

func main() {
    ProcessRequest("jane", "abc123")
}

type ctxKey int

const (
    ctxUserID ctxKey = iota
    ctxAuthToken
)

func UserID(c context.Context) string {
    return c.Value(ctxUserID).(string)
}

func AuthToken(c context.Context) string {
    return c.Value(ctxAuthToken).(string)
}

func ProcessRequest(userID, authToken string) {
    ctx := context.WithValue(context.Background(), ctxUserID, userID)
    ctx = context.WithValue(ctx, ctxAuthToken, authToken)
    HandleResponse(ctx)
}

func HandleResponse(ctx context.Context) {
    fmt.Printf(
        "handling response for %v (auth: %v)",
        UserID(ctx),
        AuthToken(ctx),
    )
}

خروجی:

handling response for jane (auth: abc123)

راه type-safe برای بازیابی از Context — consumer در پکیج دیگر کلیدهای ذخیره را نمی‌داند.

مشکل: فرض کنید HandleResponse در پکیج response و ProcessRequest در پکیج process است. process باید response import کند، اما HandleResponse به accessorهای process دسترسی ندارد چون import process circular dependency می‌سازد. چون انواع کلید private به process هستند، response نمی‌تواند داده را بخواند!

معماری را به سمت پکیج‌های حول نوع داده که از چند جا import می‌شوند سوق می‌دهد. بد نیست، اما awareness لازم است.

پکیج context جالب است اما یکدست تحسین نشده. در جامعهٔ Go کمی controversial است. جنبهٔ لغو خوب پذیرفته شده، اما ذخیرهٔ دادهٔ arbitrary در Context و type-unsafe بودن آن divisiveness ایجاد کرده. با accessorها تا حدی type-safety را جبران کردیم، اما هنوز می‌توان با نوع اشتباه bug ساخت.

مسئلهٔ بزرگ‌تر: چه چیزی در نمونه‌های Context ذخیره شود.

راهنمای prevalent، این نظر نسبتاً مبهم در پکیج:

Use context values only for request-scoped data that transits processes and API boundaries, not for passing optional parameters to functions.

optional parameter روشن است (از Context برای optional parameter استفاده نکنید)، اما «request-scoped data» چیست؟ «transits processes and API boundaries» — چیزهای زیادی. بهترین تعریف: heuristics با تیم و ارزیابی در code review. heuristics من:

داده باید از مرز process یا API عبور کند. اگر در حافظهٔ process تولید می‌کنید، شاید candidate خوبی نباشد مگر از مرز API هم عبور کند.
داده باید immutable باشد. وگرنه از request نیامده.
داده به سمت انواع ساده متمایل باشد. عبور از مرز API با import گراف پیچیدهٔ پکیج‌ها برای طرف دیگر سخت‌تر است.
داده باشد، نه نوع با method. عملیات logic است و روی consumer داده است.
داده عملیات را decorate کند، نه drive کند. اگر الگوریتم بر اساس محتوای Context متفاوت رفتار کند، احتمالاً وارد optional parameter شده‌اید.

قواعد سخت نیستند؛ heuristics. اگر دادهٔ Context هر پنج را نقض کند، بازنگری کنید.

بعد دیگر: چند لایه تا استفاده؟ اگر چند framework و ده‌ها تابع بین پذیرش و استفاده، signatureهای verbose و self-documenting با پارامتر اضافه؟ یا در Context با dependency نامرئی؟ هر دو merit دارند — تصمیم تیم.

حتی با heuristics، request-scoped بودن سخت است. جدول نظر من دربارهٔ انواع داده و پنج heuristics:

Data	1	2	3	4	5
Request ID	✓	✓	✓	✓	✓
User ID	✓	✓	✓	✓
URL	✓	✓
API Server Connection
Authorization Token	✓	✓	✓	✓
Request Token	✓	✓	✓

گاهی روشن است چیزی نباید در context باشد — مثل اتصال API server. گاهی مبهم — token احراز هویت؟ immutable و slice بایت، اما receiver برای field کردن request از آن استفاده نمی‌کند؟

آنچه برای یک تیم قابل‌قبول است برای دیگری شاید نباشد.

پاسخ آسان نیست. پکیج در standard library است — باید نظری داشته باشید که بسته به پروژه عوض شود. توصیهٔ نهایی: قابلیت لغو Context بسیار مفید است و احساسات دربارهٔ data-bag نباید شما را از استفاده باز دارد.

Summary

در این فصل زیاد پیش رفتیم. primitiveهای همزمانی Go را ترکیب کردیم تا الگوهایی برای کد همزمان قابل‌نگهداری بسازیم. حالا که با این الگوها آشنا شدید، می‌توانیم ببینیم چطور در الگوهای دیگر برای سیستم‌های بزرگ گنجانده شوند. فصل بعد نمای کلی تکنیک‌ها برای همین کار را می‌دهد.

فصل ۴ — الگوهای همزمانی در Go ​

Confinement ​

The for-select Loop ​

ارسال متغیرهای iteration روی یک channel ​

حلقهٔ بی‌نهایت در انتظار توقف ​

Preventing Goroutine Leaks ​

The or-channel ​

Error Handling ​

Pipelines ​

Best Practices for Constructing Pipelines ​

Some Handy Generators ​

Fan-Out, Fan-In ​

The or-done-channel ​

The tee-channel ​

The bridge-channel ​

Queuing ​

The context Package ​

Summary ​

فصل ۴ — الگوهای همزمانی در Go

Confinement

The for-select Loop

ارسال متغیرهای iteration روی یک channel

حلقهٔ بی‌نهایت در انتظار توقف

Preventing Goroutine Leaks

The or-channel

Error Handling

Pipelines

Best Practices for Constructing Pipelines

Some Handy Generators

Fan-Out, Fan-In

The or-done-channel

The tee-channel

The bridge-channel

Queuing

The context Package

Summary