فصل ۳ — بلوک‌های ساختمانی همزمانی Go

در این فصل، دربارهٔ مجموعهٔ غنی ویژگی‌های Go که از داستان همزمانی آن پشتیبانی می‌کنند بحث می‌کنیم. تا پایان این فصل، باید درک خوبی از syntax، توابع و پکیج‌های در دسترس و عملکردشان داشته باشید.

Goroutines

Goroutineها یکی از پایه‌ای‌ترین واحدهای سازمان‌دهی در یک برنامهٔ Go هستند، بنابراین مهم است بدانیم چه هستند و چگونه کار می‌کنند. در واقع، هر برنامهٔ Go حداقل یک goroutine دارد: goroutine اصلی (main goroutine)، که به‌طور خودکار هنگام شروع process ساخته و اجرا می‌شود. در تقریباً هر برنامه‌ای، زود یا دیر به goroutine نیاز پیدا می‌کنید تا در حل مسائل کمک کند. پس چه هستند؟

به‌طور بسیار ساده، goroutine تابعی است که به‌صورت همزمان (به یاد داشته باشید: لزوماً موازی نیست!) در کنار کد دیگر اجرا می‌شود. می‌توانید با قرار دادن کلیدواژهٔ go قبل از یک تابع آن را شروع کنید:

func main() {
    go sayHello()
    // continue doing other things
}

func sayHello() {
    fmt.Println("hello")
}

توابع anonymous هم کار می‌کنند! اینجا مثالی است که همان کار مثال قبلی را انجام می‌دهد؛ اما به‌جای ساخت goroutine از یک تابع، goroutine را از یک تابع anonymous می‌سازیم:

go func() {
    fmt.Println("hello")
}()
// continue doing other things

توجه کنید که برای استفاده از کلیدواژهٔ go باید تابع anonymous را فوراً فراخوانی کنیم.

به‌عنوان جایگزین، می‌توانید تابع را به یک متغیر اختصاص دهید و تابع anonymous را این‌گونه فراخوانی کنید:

sayHello := func() {
    fmt.Println("hello")
}
go sayHello()
// continue doing other things

چقدر جالب است! می‌توانیم با یک تابع و یک کلیدواژهٔ تنها، بلوک منطقی همزمان بسازیم. باور کنید یا نه، برای شروع کار با goroutineها همین کافی است. دربارهٔ نحوهٔ استفادهٔ صحیح، همگام‌سازی و سازمان‌دهی آن‌ها مطالب زیادی وجود دارد، اما برای شروع واقعاً همین را باید بدانید. بقیهٔ این فصل عمیق‌تر به ماهیت و نحوهٔ کار goroutineها می‌پردازد. اگر فقط به نوشتن کدی علاقه دارید که با goroutineها درست کار کند، می‌توانید به بخش بعدی بروید.

پس ببینیم پشت صحنه چه می‌گذرد: goroutineها واقعاً چگونه کار می‌کنند؟ آیا threadهای OS هستند؟ green thread؟ چند تا می‌توانیم بسازیم؟

Goroutineها منحصربه‌فرد Go هستند (اگرچه برخی زبان‌های دیگر primitive همزمانی مشابهی دارند). آن‌ها threadهای OS نیستند و دقیقاً green thread هم نیستند — threadهایی که توسط runtime زبان مدیریت می‌شوند — بلکه سطح بالاتری از abstraction به نام coroutine هستند. Coroutineها به‌سادگی subroutineهای همزمان (توابع، closureها یا methodها در Go) هستند که nonpreemptive‌اند — یعنی نمی‌توان آن‌ها را قطع کرد. در عوض، coroutineها نقاط متعددی دارند که امکان تعلیق یا ورود مجدد را فراهم می‌کنند.

آنچه goroutineها را منحصربه‌فرد Go می‌کند، یکپارچگی عمیق آن‌ها با runtime Go است. Goroutineها نقاط تعلیق یا ورود مجدد خود را تعریف نمی‌کنند؛ runtime Go رفتار runtime goroutineها را مشاهده می‌کند و وقتی block می‌شوند به‌طور خودکار آن‌ها را تعلیق و وقتی unblock می‌شوند از سر می‌گیرد. به نوعی preemptable هستند، اما فقط در نقاطی که goroutine block شده است. این همکاری ظریفی بین runtime و منطق goroutine است. بنابراین goroutineها را می‌توان دستهٔ ویژه‌ای از coroutine دانست.

Coroutineها و در نتیجه goroutineها، سازه‌های ضمناً همزمان هستند، اما همزمانی ویژگی coroutine نیست: چیزی باید چند coroutine را همزمان میزبانی کند و به هر کدام فرصت اجرا بدهد — وگرنه همزمان نخواهند بود! توجه کنید این به معنای موازی بودن ضمنی coroutineها نیست. قطعاً ممکن است چند coroutine به‌صورت متوالی اجرا شوند تا توهم موازی‌سازی ایجاد کنند و در Go این اتفاق مرتب می‌افتد.

مکانیزم Go برای میزبانی goroutineها پیاده‌سازی چیزی به نام scheduler از نوع M:N است، یعنی M green thread را به N thread OS نگاشت می‌کند. سپس goroutineها روی green threadها زمان‌بندی می‌شوند. وقتی goroutine بیشتری نسبت به green threadهای موجود داریم، scheduler توزیع goroutineها روی threadهای موجود را مدیریت می‌کند و تضمین می‌کند وقتی این goroutineها block می‌شوند، goroutineهای دیگر اجرا شوند. نحوهٔ کار همهٔ این‌ها را در فصل ۶ بحث می‌کنیم، اما اینجا نحوهٔ مدل‌سازی همزمانی توسط Go را پوشش می‌دهیم.

Go از مدل همزمانی به نام مدل fork-join پیروی می‌کند.¹ کلمهٔ fork به این اشاره دارد که در هر نقطه از برنامه، می‌توان شاخهٔ فرزند اجرا را برای اجرای همزمان با والد جدا کرد. کلمهٔ join به این اشاره دارد که در آینده، این شاخه‌های همزمان اجرا دوباره به هم می‌پیوندند. جایی که فرزند به والد می‌پیوندد join point نامیده می‌شود. اینجا نمای گرافیکی برای تصویرسازی:

شکل ۳-۱. مدل fork-join

والد (main)
    |
    |---- fork (go) ----> goroutine فرزند ۱
    |
    |---- fork (go) ----> goroutine فرزند ۲
    |
    ... ادامهٔ اجرای والد ...
    |
    +---- join point (مثلاً wg.Wait) ----+
    |                                      |
    goroutine فرزند ۱ به والد می‌پیوندد   |
    goroutine فرزند ۲ به والد می‌پیوندد   |

دستور go نحوهٔ انجام fork توسط Go است و threadهای اجرای forkشده goroutine هستند. به مثال سادهٔ goroutine برگردیم:

sayHello := func() {
    fmt.Println("hello")
}
go sayHello()
// continue doing other things

اینجا تابع sayHello روی goroutine خودش اجرا می‌شود، در حالی که بقیهٔ برنامه ادامه می‌یابد. در این مثال، join point وجود ندارد. Goroutine در حال اجرای sayHello در زمانی نامشخص در آینده خارج می‌شود و بقیهٔ برنامه از قبل ادامه یافته است.

با این حال، یک مشکل در این مثال وجود دارد: همان‌طور که نوشته شده، نامشخص است آیا تابع sayHello اصلاً اجرا می‌شود یا نه. Goroutine ساخته و توسط runtime Go برای اجرا زمان‌بندی می‌شود، اما ممکن است قبل از خروج goroutine اصلی واقعاً فرصت اجرا پیدا نکند.

در واقع، چون برای سادگی بقیهٔ تابع main را حذف کرده‌ایم، وقتی این مثال کوچک را اجرا می‌کنیم، تقریباً قطعی است که برنامه قبل از شروع goroutine میزبان فراخوانی sayHello تمام می‌شود. در نتیجه کلمهٔ «hello» روی stdout چاپ نمی‌شود. می‌توانید بعد از ساخت goroutine یک time.Sleep بگذارید، اما به یاد داشته باشید که این واقعاً join point نمی‌سازد، فقط race condition ایجاد می‌کند. اگر فصل ۱ را به یاد دارید، احتمال اجرای goroutine قبل از خروج را افزایش می‌دهید، اما تضمین نمی‌کنید. Join pointها هستند که صحت برنامه را تضمین و race condition را حذف می‌کنند.

برای ساخت join point باید goroutine اصلی و goroutine sayHello را همگام‌سازی کنید. این را می‌توان به روش‌های مختلف انجام داد، اما از یکی استفاده می‌کنم که در «پکیج sync» بحث می‌کنیم: sync.WaitGroup. فعلاً مهم نیست بفهمید این مثال چگونه join point می‌سازد، فقط اینکه بین دو goroutine یکی می‌سازد. اینجا نسخهٔ صحیح مثال:

var wg sync.WaitGroup
sayHello := func() {
    defer wg.Done()
    fmt.Println("hello")
}
wg.Add(1)
go sayHello()
wg.Wait() // This is the join point.

خروجی:

hello

این مثال به‌طور قطعی goroutine اصلی را block می‌کند تا goroutine میزبان تابع sayHello خاتمه یابد. نحوهٔ کار sync.WaitGroup را در «پکیج sync» یاد می‌گیرید، اما برای صحیح بودن مثال‌ها از این به بعد از آن برای ساخت join point استفاده می‌کنم.

در مثال‌ها زیاد از توابع anonymous برای ساخت سریع مثال‌های goroutine استفاده کردیم. توجه را به closureها معطوف کنیم. Closureها حول lexical scopeای که در آن ساخته شده‌اند بسته می‌شوند و متغیرها را capture می‌کنند. اگر closure را در goroutine اجرا کنید، آیا closure روی کپی این متغیرها کار می‌کند یا روی referenceهای اصلی؟ امتحان کنیم:

var wg sync.WaitGroup
salutation := "hello"
wg.Add(1)
go func() {
    defer wg.Done()
    salutation = "welcome"
}()
wg.Wait()
fmt.Println(salutation) // Here we see the goroutine modifying the value of the variable salutation.

فکر می‌کنید مقدار salutation چه خواهد بود: «hello» یا «welcome»؟ اجرا کنیم:

welcome

جالب است! Goroutineها در همان address spaceای که در آن ساخته شده‌اند اجرا می‌شوند و برنامه کلمهٔ «welcome» را چاپ می‌کند. مثال دیگر:

var wg sync.WaitGroup
for _, salutation := range []string{"hello", "greetings", "good day"} {
    wg.Add(1)
    go func() {
        defer wg.Done()
        fmt.Println(salutation) // Here we reference the loop variable salutation created by ranging over a string slice.
    }()
}
wg.Wait()

پاسخ پیچیده‌تر از انتظار اکثر مردم است و یکی از چند نکتهٔ غافلگیرکننده در Go. اکثر مردم شهوداً فکر می‌کنند کلمات «hello»، «greetings» و «good day» به ترتیبی غیرقطعی چاپ می‌شوند، اما ببینید چه می‌کند:

good day
good day
good day

کمی غافلگیرکننده است! بفهمیم چه می‌گذرد. در این مثال، goroutine closureای اجرا می‌کند که روی متغیر تکرار salutation با نوع string بسته شده است. با هر تکرار حلقه، salutation به مقدار رشتهٔ بعدی در slice literal اختصاص می‌یابد. چون goroutineهای زمان‌بندی‌شده ممکن است در هر زمانی در آینده اجرا شوند، نامشخص است چه مقادیری از داخل goroutine چاپ می‌شوند. روی ماشین من، احتمال زیاد حلقه قبل از شروع goroutineها تمام می‌شود. یعنی متغیر salutation از scope خارج می‌شود. آن‌گاه چه؟ آیا goroutineها هنوز می‌توانند به چیزی که از scope خارج شده reference دهند؟ آیا به حافظه‌ای دسترسی پیدا نمی‌کنند که احتمالاً garbage collect شده؟

این نکتهٔ جانبی جالبی دربارهٔ مدیریت حافظه توسط Go است. Runtime Go به اندازهٔ کافی هوشمند است که بداند reference به متغیر salutation هنوز نگه داشته می‌شود و بنابراین حافظه را به heap منتقل می‌کند تا goroutineها بتوانند به آن دسترسی داشته باشند.

معمولاً روی ماشین من، حلقه قبل از اجرای هر goroutine تمام می‌شود، پس salutation به heap منتقل می‌شود و reference به آخرین مقدار در string slice، «good day» را نگه می‌دارد. بنابراین معمولاً سه بار «good day» می‌بینم. روش صحیح نوشتن این حلقه، عبور دادن کپی salutation به closure است تا وقتی goroutine اجرا می‌شود، روی دادهٔ تکرار خودش از حلقه کار کند:

var wg sync.WaitGroup
for _, salutation := range []string{"hello", "greetings", "good day"} {
    wg.Add(1)
    go func(salutation string) { // Here we declare a parameter, just like any other function. We shadow the original salutation variable to make what's happening more apparent.
        defer wg.Done()
        fmt.Println(salutation)
    }(salutation) // Here we pass in the current iteration's variable to the closure. A copy of the string struct is made, thereby ensuring that when the goroutine is run, we refer to the proper string.
}
wg.Wait()

خروجی صحیح:

good day
hello
greetings

این مثال همان‌طور که انتظار داریم رفتار می‌کند و فقط کمی پرحرف‌تر است.

چون goroutineها در همان address space یکدیگر کار می‌کنند و صرفاً توابع را میزبانی می‌کنند، استفاده از goroutine گسترش طبیعی نوشتن کد غیرهمزمان است. کامپایلر Go به‌خوبی متغیرها را در حافظه pin می‌کند تا goroutineها تصادفاً به حافظهٔ آزادشده دسترسی نکنند و developerها روی problem space تمرکز کنند نه مدیریت حافظه؛ اما این چک سفید نیست.

چون چند goroutine می‌توانند روی همان address space کار کنند، هنوز باید نگران همگام‌سازی باشیم. همان‌طور که بحث کردیم، می‌توانیم دسترسی به حافظهٔ مشترک را همگام‌سازی کنیم، یا از primitiveهای CSP برای share memory by communication استفاده کنیم. این تکنیک‌ها را بعداً در «Channels» و «پکیج sync» بحث می‌کنیم.

مزیت دیگر goroutineها سبکی فوق‌العادهٔ آن‌هاست. اینجا گزیده‌ای از Go FAQ:

A newly minted goroutine is given a few kilobytes, which is almost always enough. When it isn't, the run-time grows (and shrinks) the memory for storing the stack automatically, allowing many goroutines to live in a modest amount of memory. The CPU overhead averages about three cheap instructions per function call. It is practical to create hundreds of thousands of goroutines in the same address space. If goroutines were just threads, system resources would run out at a much smaller number.

چند کیلوبایت به ازای هر goroutine؛ اصلاً بد نیست! خودمان تأیید کنیم. اما قبلش یک نکتهٔ جالب: garbage collector کاری برای جمع‌آوری goroutineهایی که به‌نوعی رها شده‌اند انجام نمی‌دهد. اگر این را بنویسم:

go func() {
    // <operation that will block forever>
}()
// Do work

این goroutine تا خروج process باقی می‌ماند. نحوهٔ رفع این را در فصل ۴ در بخش «جلوگیری از نشت Goroutine» بحث می‌کنیم. در مثال بعد از این به نفعمان استفاده می‌کنیم تا اندازهٔ goroutine را اندازه بگیریم.

در مثال بعد، اینکه goroutineها garbage collect نمی‌شوند را با توانایی runtime برای introspection و اندازه‌گیری حافظهٔ تخصیص‌یافته قبل و بعد از ساخت goroutine ترکیب می‌کنیم:

memConsumed := func() uint64 {
    runtime.GC()
    var s runtime.MemStats
    runtime.ReadMemStats(&s)
    return s.Sys
}

var c <-chan interface{}
var wg sync.WaitGroup
noop := func() { wg.Done(); <-c } // We require a goroutine that will never exit so that we can keep a number of them in memory for measurement. Don't worry about how we're achieving this at this time; just know that this goroutine won't exit until the process is finished.

const numGoroutines = 1e4 // Here we define the number of goroutines to create. We will use the law of large numbers to asymptotically approach the size of a goroutine.
wg.Add(numGoroutines)
before := memConsumed() // Here we measure the amount of memory consumed before creating our goroutines.
for i := numGoroutines; i > 0; i-- {
    go noop()
}
wg.Wait()
after := memConsumed() // And here we measure the amount of memory consumed after creating our goroutines.
fmt.Printf("%.3fkb", float64(after-before)/numGoroutines/1000)

نتیجه:

2.817kb

به نظر می‌رسد مستندات درست است! این‌ها goroutineهای خالی‌اند که کاری نمی‌کنند، اما ایده‌ای از تعداد goroutineهایی که احتمالاً می‌توانیم بسازیم می‌دهد. جدول ۳-۱ برآوردهای تقریبی تعداد goroutineهای ممکن با CPU ۶۴ بیتی بدون swap space را می‌دهد.

جدول ۳-۱. تحلیل تعداد تقریبی goroutineهای ممکن در حافظهٔ داده‌شده

حافظه (GB)	Goroutineها (#/100,000)	مرتبهٔ بزرگی
2^0	3.718	3
2^1	7.436	3
2^2	14.873	6
2^3	29.746	6
2^4	59.492	6
2^5	118.983	6
2^6	237.967	6
2^7	475.934	6
2^8	951.867	6
2^9	1903.735	9

این اعداد بسیار بزرگ‌اند! روی لپ‌تاپم ۸ گیگابایت RAM دارم، یعنی در تئوری می‌توانم میلیون‌ها goroutine بدون swap راه بیندازم. البته برنامه‌های دیگر و محتوای واقعی goroutineها را نادیده می‌گیرد، اما این محاسبهٔ سریع نشان می‌دهد goroutineها چقدر سبک‌اند!

چیزی که ممکن است روحیه را تیره کند context switching است — وقتی میزبان یک process همزمان باید state را ذخیره کند تا process همزمان دیگری را اجرا کند. اگر processهای همزمان زیادی داشته باشیم، ممکن است همهٔ زمان CPU را صرف context switch کنیم و کار واقعی انجام نشود. در سطح OS با threadها، این می‌تواند بسیار پرهزینه باشد. Thread OS باید مقادیر register، lookup table و memory map را ذخیره کند تا بعداً thread فعلی را از سر بگیرد، سپس همان اطلاعات را برای thread ورودی بارگذاری کند.

Context switching در نرم‌افزار نسبتاً بسیار ارزان‌تر است. زیر scheduler تعریف‌شده در نرم‌افزار، runtime می‌تواند انتخابی‌تر باشد که چه چیزی persist شود، چگونه persist شود و چه زمانی. عملکرد نسبی context switching بین threadهای OS و goroutineها را ببینیم. ابتدا از benchmark داخلی Linux برای زمان ارسال پیام بین دو thread روی یک core استفاده می‌کنیم:

taskset -c 0 perf bench sched pipe -T

خروجی:

# Running 'sched/pipe' benchmark:
# Executed 1000000 pipe operations between two threads

     Total time: 2.935 [sec]

       2.935784 usecs/op
         340624 ops/sec

این benchmark زمان ارسال و دریافت پیام روی thread را اندازه می‌گیرد، پس نتیجه را بر دو تقسیم می‌کنیم: ۱.۴۶۷ میکروثانیه به ازای هر context switch. بد به نظر نمی‌رسد، اما تا benchmark goroutineها قضاوت نکنیم.

benchmark مشابه با Go می‌سازیم. از چیزهایی استفاده کرده‌ام که هنوز بحث نکرده‌ایم؛ اگر گیج‌کننده است، calloutها را دنبال کنید و روی نتیجه تمرکز کنید. این مثال دو goroutine می‌سازد و بین آن‌ها پیام می‌فرستد:

func BenchmarkContextSwitch(b *testing.B) {
    var wg sync.WaitGroup
    begin := make(chan struct{})
    c := make(chan struct{})

    var token struct{}
    sender := func() {
        defer wg.Done()
        <-begin // Here we wait until we're told to begin. We don't want the cost of setting up and starting each goroutine to factor into the measurement of context switching.
        for i := 0; i < b.N; i++ {
            c <- token // Here we send messages to the receiver goroutine. A struct{}{} is called an empty struct and takes up no memory; thus, we are only measuring the time it takes to signal a message.
        }
    }
    receiver := func() {
        defer wg.Done()
        <-begin
        for i := 0; i < b.N; i++ {
            <-c // Here we receive a message but do nothing with it.
        }
    }

    wg.Add(2)
    go sender()
    go receiver()
    b.StartTimer() // Here we begin the performance timer.
    close(begin)   // Here we tell the two goroutines to begin.
    wg.Wait()
}

benchmark را با یک CPU اجرا می‌کنیم:

go test -bench=. -cpu=1 \
src/gos-concurrency-building-blocks/goroutines/fig-ctx-switch_test.go

BenchmarkContextSwitch              5000000                  225      ns/op
PASS
ok                                  command-line-arguments   1.393s

۲۲۵ نانوثانیه به ازای هر context switch! یعنی ۰.۲۲۵ میکروثانیه، یا ۹۲٪ سریع‌تر از context switch OS روی ماشین من که ۱.۴۶۷ میکروثانیه بود. سخت است ادعای قطعی دربارهٔ تعداد goroutineهایی که context switching زیاد ایجاد می‌کنند کرد، اما می‌توانیم بگوییم سقف بالا احتمالاً مانع استفاده از goroutineها نیست.

با خواندن این بخش، باید بدانید چگونه goroutineها را شروع کنید و کمی دربارهٔ نحوهٔ کارشان. همچنین باید مطمئن باشید هر وقت problem space اقتضا کند می‌توانید goroutine بسازید. همان‌طور که در «تفاوت همزمانی و موازی‌سازی» بحث شد، هرچه goroutine بیشتر بسازید و اگر problem space با قانون Amdahl به یک segment همزمان محدود نشود، برنامه با پردازنده‌های بیشتر بهتر scale می‌کند. ساخت goroutine بسیار ارزان است و فقط وقتی باید دربارهٔ هزینهٔ آن صحبت کنید که ثابت کرده‌اید علت اصلی مشکل عملکرد هستند.

پکیج sync

پکیج sync primitiveهای همزمانی را دارد که برای همگام‌سازی دسترسی سطح پایین به حافظه مفیدترند. اگر با زبان‌هایی کار کرده‌اید که همزمانی را عمدتاً از طریق همگام‌سازی دسترسی حافظه مدیریت می‌کنند، این انواع آشنا هستند. تفاوت در Go این است که Go مجموعهٔ جدیدی از primitiveهای همزمانی روی primitiveهای همگام‌سازی دسترسی حافظه ساخته تا مجموعهٔ گسترده‌تری در اختیار شما باشد. همان‌طور که در «فلسفهٔ همزمانی Go» بحث شد، این عملیات کاربرد دارند — عمدتاً در scopeهای کوچک مثل یک struct. تصمیم با شماست چه زمانی همگام‌سازی دسترسی حافظه مناسب است. با این حال، primitiveهای مختلف پکیج sync را ببینیم.

WaitGroup

WaitGroup راه خوبی برای انتظار تا مجموعه‌ای از عملیات همزمان تمام شوند است، وقتی یا به نتیجهٔ عملیات همزمان اهمیت نمی‌دهید یا راه دیگری برای جمع‌آوری نتایج دارید. اگر هیچ‌کدام برقرار نیست، پیشنهاد می‌کنم از channelها و دستور select استفاده کنید. WaitGroup آن‌قدر مفید است که اول معرفی می‌شود تا در بخش‌های بعد استفاده شود. مثال پایه:

var wg sync.WaitGroup

wg.Add(1) // Here we call Add with an argument of 1 to indicate that one goroutine is beginning.
go func() {
    defer wg.Done() // Here we call Done using the defer keyword to ensure that before we exit the goroutine's closure, we indicate to the WaitGroup that we've exited.
    fmt.Println("1st goroutine sleeping...")
    time.Sleep(1)
}()

wg.Add(1)
go func() {
    defer wg.Done()
    fmt.Println("2nd goroutine sleeping...")
    time.Sleep(2)
}()

wg.Wait() // Here we call Wait, which will block the main goroutine until all goroutines have indicated they have exited.
fmt.Println("All goroutines complete.")

خروجی:

2nd goroutine sleeping...
1st goroutine sleeping...
All goroutines complete.

می‌توانید WaitGroup را مثل شمارندهٔ concurrent-safe در نظر بگیرید: Add شمارنده را به اندازهٔ عدد ورودی افزایش می‌دهد و Done آن را یک واحد کاهش می‌دهد. Wait تا صفر شدن شمارنده block می‌کند.

توجه کنید Add خارج از goroutineهایی که ردیابی می‌کند فراخوانی می‌شود. اگر نمی‌کردیم race condition ایجاد می‌کردیم، چون از «Goroutines» می‌دانیم تضمینی دربارهٔ زمان زمان‌بندی goroutineها نداریم؛ ممکن است قبل از شروع goroutineها به Wait برسیم. اگر Add داخل closure goroutineها بود، Wait ممکن بود بدون block برگردد چون Add هنوز اجرا نشده بود.

معمولاً Add را تا حد ممکن نزدیک به goroutineهایی که ردیابی می‌کند قرار می‌دهند، اما گاهی یک‌جا برای گروهی از goroutineها فراخوانی می‌شود. معمولاً قبل از حلقهٔ for این‌گونه:

hello := func(wg *sync.WaitGroup, id int) {
    defer wg.Done()
    fmt.Printf("Hello from %v!\n", id)
}

const numGreeters = 5
var wg sync.WaitGroup
wg.Add(numGreeters)
for i := 0; i < numGreeters; i++ {
    go hello(&wg, i+1)
}
wg.Wait()

خروجی:

Hello from 5!
Hello from 4!
Hello from 3!
Hello from 2!
Hello from 1!

Mutex و RWMutex

اگر با زبان‌هایی که همزمانی را از طریق همگام‌سازی دسترسی حافظه مدیریت می‌کنند آشنا هستید، احتمالاً Mutex را فوراً می‌شناسید. اگر نه، نگران نباشید، Mutex بسیار ساده است.

Mutex مخفف «mutual exclusion» است و راهی برای محافظت از critical sectionهای برنامه. از فصل ۱ به یاد دارید critical section ناحیه‌ای است که دسترسی انحصاری به منبع مشترک می‌خواهد. Mutex راه concurrent-safe برای بیان دسترسی انحصاری به این منابع مشترک فراهم می‌کند. با اصطلاح Go، در حالی که channelها با ارتباط، حافظه را به اشتراک می‌گذارند، Mutex با ایجاد قراردادی که developerها باید برای همگام‌سازی دسترسی به حافظه رعایت کنند، حافظه را به اشتراک می‌گذارد. شما مسئول هماهنگی دسترسی به این حافظه با محافظت از آن با mutex هستید. مثال سادهٔ دو goroutine که سعی دارند مقدار مشترک را افزایش و کاهش دهند:

var count int
var lock sync.Mutex

increment := func() {
    lock.Lock()   // Here we request exclusive use of the critical section—in this case the count variable—guarded by a Mutex, lock.
    defer lock.Unlock() // Here we indicate that we're done with the critical section lock is guarding.
    count++
    fmt.Printf("Incrementing: %d\n", count)
}

decrement := func() {
    lock.Lock()
    defer lock.Unlock()
    count--
    fmt.Printf("Decrementing: %d\n", count)
}

// Increment
var arithmetic sync.WaitGroup
for i := 0; i <= 5; i++ {
    arithmetic.Add(1)
    go func() {
        defer arithmetic.Done()
        increment()
    }()
}

// Decrement
for i := 0; i <= 5; i++ {
    arithmetic.Add(1)
    go func() {
        defer arithmetic.Done()
        decrement()
    }()
}

arithmetic.Wait()
fmt.Println("Arithmetic complete.")

خروجی (نمونه):

Decrementing: -1
Incrementing: 0
...
Arithmetic complete.

همیشه Unlock را در defer فراخوانی می‌کنیم. این idiom رایج با Mutex است تا حتی هنگام panic هم اجرا شود. در غیر این صورت احتمال deadlock زیاد است.

Critical sectionها به این نام‌اند چون گلوگاه برنامه را نشان می‌دهند. ورود و خروج از critical section نسبتاً پرهزینه است و معمولاً زمان در critical section کمینه می‌شود.

یکی از راه‌ها کاهش cross-section critical section است. ممکن است حافظه‌ای بین چند process همزمان به اشتراک گذاشته شود اما همه آن‌ها نخوانند و ننویسند. در این صورت می‌توان از sync.RWMutex استفاده کرد.

sync.RWMutex از نظر مفهومی همان Mutex است: از حافظه محافظت می‌کند؛ اما کنترل بیشتری می‌دهد. می‌توانید lock برای خواندن بخواهید و دسترسی می‌گیرید مگر lock برای نوشتن گرفته شده باشد. یعنی تعداد دلخواه reader می‌توانند reader lock داشته باشند تا وقتی writer lock نگرفته شده. مثال producer کم‌فعال‌تر از consumerهای متعدد:

producer := func(wg *sync.WaitGroup, l sync.Locker) { // The producer function's second parameter is of the type sync.Locker. This interface has two methods, Lock and Unlock, which the Mutex and RWMutex types satisfy.
    defer wg.Done()
    for i := 5; i > 0; i-- {
        l.Lock()
        l.Unlock()
        time.Sleep(1) // Here we make the producer sleep for one second to make it less active than the observer goroutines.
    }
}

observer := func(wg *sync.WaitGroup, l sync.Locker) {
    defer wg.Done()
    l.Lock()
    defer l.Unlock()
}

test := func(count int, mutex, rwMutex sync.Locker) time.Duration {
    var wg sync.WaitGroup
    wg.Add(count+1)
    beginTestTime := time.Now()
    go producer(&wg, mutex)
    for i := count; i > 0; i-- {
        go observer(&wg, rwMutex)
    }

    wg.Wait()
    return time.Since(beginTestTime)
}

tw := tabwriter.NewWriter(os.Stdout, 0, 1, 2, ' ', 0)
defer tw.Flush()

var m sync.RWMutex
fmt.Fprintf(tw, "Readers\tRWMutext\tMutex\n")
for i := 0; i < 20; i++ {
    count := int(math.Pow(2, float64(i)))
    fmt.Fprintf(
        tw,
        "%d\t%v\t%v\n",
        count,
        test(count, &m, m.RLocker()),
        test(count, &m, &m),
    )
}

خروجی (نمونه):

Readers      RWMutext            Mutex
1            38.343µs            15.854µs
...
524288       303.865661ms        231.072729ms

در این مثال خاص، کاهش cross-section critical section تقریباً از حدود ۲¹³ reader سودمند می‌شود. بسته به کار critical section متفاوت است، اما معمولاً وقتی از نظر منطقی مناسب است RWMutex به‌جای Mutex توصیه می‌شود.

Cond

کامنت نوع Cond هدف آن را خوب توصیف می‌کند:

...a rendezvous point for goroutines waiting for or announcing the occurrence of an event.

در این تعریف، «event» هر سیگنال دلخواه بین دو یا چند goroutine است که جز اینکه رخ داده اطلاع دیگری نمی‌برد. اغلب می‌خواهید قبل از ادامهٔ اجرا روی goroutine منتظر یکی از این سیگنال‌ها بمانید. بدون نوع Cond، یک رویکرد ساده حلقهٔ بی‌نهایت است:

for conditionTrue() == false {
}

این همهٔ چرخه‌های یک core را مصرف می‌کند. با time.Sleep بهتر می‌شود:

for conditionTrue() == false {
    time.Sleep(1*time.Millisecond)
}

بهتر است اما هنوز ناکارآمد است و باید مدت sleep را حدس بزنید: زیاد — عملکرد را مصنوعیاً پایین می‌آورد؛ کم — CPU زیاد مصرف می‌شود. بهتر است goroutine تا سیگنال بیداری کارآمد بخوابد. دقیقاً همان کاری است که Cond انجام می‌دهد:

c := sync.NewCond(&sync.Mutex{})
c.L.Lock()
for conditionTrue() == false {
    c.Wait() // Here we wait to be notified that the condition has occurred. This is a blocking call and the goroutine will be suspended.
}
c.L.Unlock()

• اینجا Cond جدید می‌سازیم. NewCond نوعی که sync.Locker را satisfy کند می‌گیرد تا Cond هماهنگی concurrent-safe با goroutineهای دیگر را تسهیل کند.
• اینجا Locker این condition را lock می‌کنیم. لازم است چون Wait هنگام ورود خودکار Unlock روی Locker می‌زند.
• اینجا Locker را unlock می‌کنیم. لازم است چون هنگام خروج از Wait، Lock روی Locker فراخوانی می‌شود.

این رویکرد بسیار کارآمدتر است. Wait فقط block نمی‌کند، goroutine فعلی را تعلیق می‌کند تا goroutineهای دیگر روی thread OS اجرا شوند. هنگام Wait چیزهای دیگری هم رخ می‌دهد: هنگام ورود به Wait، Unlock روی Locker متغیر Cond فراخوانی می‌شود و هنگام خروج Lock. به نظرم کمی عادت می‌خواهد؛ عملاً side effect پنهان متد است. انگار کل زمان انتظار lock را نگه داشته‌ایم، اما واقعاً چنین نیست. هنگام خواندن کد باید این الگو را ببینید.

مثال را گسترش می‌دهیم: goroutine منتظر سیگنال و goroutine ارسال‌کنندهٔ سیگنال. صف با طول ثابت ۲ و ۱۰ آیتم برای enqueue. می‌خواهیم به‌محض جا شدن enqueue کنیم، پس به‌محض جا شدن در صف مطلع شویم. با Cond:

c := sync.NewCond(&sync.Mutex{}) // First, we create our condition using a standard sync.Mutex as the Locker.
queue := make([]interface{}, 0, 10) // Next, we create a slice with a length of zero. Since we know we'll eventually add 10 items, we instantiate it with a capacity of 10.

removeFromQueue := func(delay time.Duration) {
    time.Sleep(delay)
    c.L.Lock() // We once again enter the critical section for the condition so we can modify data pertinent to the condition.
    queue = queue[1:] // Here we simulate dequeuing an item by reassigning the head of the slice to the second item.
    fmt.Println("Removed from queue")
    c.L.Unlock() // Here we exit the condition's critical section since we've successfully dequeued an item.
    c.Signal()   // Here we let a goroutine waiting on the condition know that something has occurred.
}

for i := 0; i < 10; i++{
    c.L.Lock() // We enter the critical section for the condition by calling Lock on the condition's Locker.
    for len(queue) == 2 { // Here we check the length of the queue in a loop. This is important because a signal on the condition doesn't necessarily mean what you've been waiting for has occurred—only that something has occurred.
        c.Wait() // We call Wait, which will suspend the main goroutine until a signal on the condition has been sent.
    }
    fmt.Println("Adding to queue")
    queue = append(queue, struct{}{})
    go removeFromQueue(1*time.Second) // Here we create a new goroutine that will dequeue an element after one second.
    c.L.Unlock() // Here we exit the condition's critical section since we've successfully enqueued an item.
}

خروجی (نمونه):

Adding to queue
Adding to queue
Removed from queue
...

برنامه هر ۱۰ آیتم را با موفقیت به صف اضافه می‌کند (قبل از dequeue دو آیتم آخر خارج می‌شود). همیشه تا حداقل یک dequeue منتظر می‌ماند قبل از enqueue بعدی.

متد جدید Signal داریم. یکی از دو متد Cond برای اطلاع goroutineهای blockشده روی Wait است. دیگری Broadcast است. درون runtime لیست FIFO از goroutineهای منتظر نگه داشته می‌شود؛ Signal طولانی‌ترین انتظار را مطلع می‌کند، Broadcast به همهٔ منتظرها. Broadcast جالب‌تر است چون با چند goroutine همزمان ارتباط می‌گیرد. رفتار Signal را با channelها به‌راحتی می‌توان تقلید کرد (در «Channels» می‌بینیم)، اما تکرار Broadcast سخت‌تر است. Cond از channelها کارآمدتر است.

برای حس استفاده از Broadcast، GUI با دکمه تصور کنید. می‌خواهیم تعداد دلخواه تابع ثبت کنیم که با کلیک اجرا شوند. Cond عالی است چون Broadcast همهٔ handlerهای ثبت‌شده را مطلع می‌کند:

type Button struct {
    Clicked *sync.Cond
}
button := Button{ Clicked: sync.NewCond(&sync.Mutex{}) } // We define a type Button that contains a condition, Clicked.

subscribe := func(c *sync.Cond, fn func()) { // Here we define a convenience function that will allow us to register functions to handle signals from a condition. Each handler is run on its own goroutine, and subscribe will not exit until that goroutine is confirmed to be running.
    var goroutineRunning sync.WaitGroup
    goroutineRunning.Add(1)
    go func() {
        goroutineRunning.Done()
        c.L.Lock()
        defer c.L.Unlock()
        c.Wait()
        fn()
    }()
    goroutineRunning.Wait()
}

var clickRegistered sync.WaitGroup // Here we create a WaitGroup. This is done only to ensure our program doesn't exit before our writes to stdout occur.
clickRegistered.Add(3)
subscribe(button.Clicked, func() { // Here we register a handler that simulates maximizing the button's window when the button is clicked.
    fmt.Println("Maximizing window.")
    clickRegistered.Done()
})
subscribe(button.Clicked, func() { // Here we register a handler that simulates displaying a dialog box when the mouse is clicked.
    fmt.Println("Displaying annoying dialog box!")
    clickRegistered.Done()
})
subscribe(button.Clicked, func() {
    fmt.Println("Mouse clicked.")
    clickRegistered.Done()
})

button.Clicked.Broadcast() // Here we set a handler for when the mouse button is raised. It in turn calls Broadcast on the Clicked Cond to let all handlers know that the mouse button has been clicked (a more robust implementation would first check that it had been depressed). Next, we simulate a user raising the mouse button from having clicked the application's button.

clickRegistered.Wait()

خروجی:

Mouse clicked.
Maximizing window.
Displaying annoying dialog box!

با یک Broadcast روی Clicked Cond، هر سه handler اجرا می‌شوند. بدون clickRegistered WaitGroup می‌توانستیم چند بار Broadcast بزنیم و هر بار هر سه handler فراخوانی شوند. channelها این را به‌راحتی نمی‌توانند و یکی از دلایل اصلی استفاده از Cond است.

مانند بیشتر چیزهای پکیج sync، Cond بهترین کار را وقتی می‌کند که در scope تنگ محدود شود یا از طریق نوعی که آن را encapsulate می‌کند در scope گسترده‌تر expose شود.

Once

فکر می‌کنید این کد چه چاپ می‌کند؟

var count int

increment := func() {
    count++
}

var once sync.Once

var increments sync.WaitGroup
increments.Add(100)
for i := 0; i < 100; i++ {
    go func() {
        defer increments.Done()
        once.Do(increment)
    }()
}

increments.Wait()
fmt.Printf("Count is %d\n", count)

وسوسه‌انگیز است بگوییم Count is 100، اما sync.Once و wrap کردن increment در Do را دیده‌اید. در واقع:

Count is 1

همان‌طور که نام می‌برد، sync.Once با primitiveهای داخلی sync تضمین می‌کند فقط یک بار فراخوانی به Do تابع ورودی را صدا بزند — حتی روی goroutineهای مختلف.

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

grep -ir sync.Once $(go env GOROOT)/src |wc -l

70

چند نکته دربارهٔ sync.Once. مثال دیگر:

var count int
increment := func() { count++ }
decrement := func() { count-- }

var once sync.Once
once.Do(increment)
once.Do(decrement)

fmt.Printf("Count: %d\n", count)

خروجی:

Count: 1

غافلگیرکننده است که ۱ و نه ۰؟ چون sync.Once فقط تعداد فراخوانی‌های Do را می‌شمارد، نه تعداد توابع یکتای ورودی به Do. کپی‌های sync.Once به‌طور نزدیک به توابعی که باید با آن‌ها فراخوانی شوند couple هستند؛ دوباره می‌بینیم انواع پکیج sync در scope تنگ بهترند. این coupling را با wrap کردن sync.Once در بلوک lexical کوچک — تابع کوچک یا نوع — رسمی کنید. مثال دیگر:

var onceA, onceB sync.Once
var initB func()
initA := func() { onceB.Do(initB) }
initB = func() { onceA.Do(initA) }
onceA.Do(initA) // This call can't proceed until the call at returns.

برنامه deadlock می‌کند چون Do در آخرین خط تا خروج Do در خط قبل پیش نمی‌رود — deadlock کلاسیک. برای برخی کمی غیرشهودی است چون انگار sync.Once را درست برای جلوگیری از initialization چندگانه استفاده کرده‌ایم، اما sync.Once فقط تضمین می‌کند توابع فقط یک بار فراخوانی شوند. گاهی با deadlock برنامه را متوقف و نقص منطق را expose می‌کند — اینجا reference دایره‌ای.

Pool

Pool پیاده‌سازی concurrent-safe الگوی object pool است. توضیح کامل object pool به ادبیات design pattern² واگذار می‌شود؛ چون Pool در پکیج sync است، به‌اختصار بحث می‌کنیم چرا مفید است.

در سطح بالا، الگوی pool راهی برای ساخت و در دسترس قرار دادن تعداد ثابت چیزهاست. معمولاً برای محدود کردن ساخت چیزهای پرهزینه (مثل اتصال پایگاه داده) استفاده می‌شود تا فقط تعداد ثابتی ساخته شوند، اما تعداد نامشخصی عملیات همچنان دسترسی بخواهند. در sync.Pool Go، این نوع داده را چند goroutine می‌توانند به‌امن استفاده کنند.

رابط اصلی Pool متد Get است. Get ابتدا بررسی می‌کند instance آزاد در pool هست؛ اگر نه، New را برای ساخت جدید صدا می‌زند. پایان کار، Put instance را به pool برمی‌گرداند:

myPool := &sync.Pool{
    New: func() interface{} {
        fmt.Println("Creating new instance.")
        return struct{}{}
    },
}

myPool.Get() // Here we call Get on the pool. These calls will invoke the New function defined on the pool since instances haven't yet been instantiated.
instance := myPool.Get()
myPool.Put(instance) // Here we put an instance previously retrieved back in the pool. This increases the available number of instances to one.
myPool.Get()         // When this call is executed, we will reuse the instance previously allocated and put back in the pool. The New function will not be invoked.

خروجی:

Creating new instance.
Creating new instance.

چرا pool و نه instantiate در حین کار؟ Go garbage collector دارد و اشیاء پاک می‌شوند. نکته چیست؟

var numCalcsCreated int
calcPool := &sync.Pool {
    New: func() interface{} {
        numCalcsCreated += 1
        mem := make([]byte, 1024)
        return &mem // Notice that we are storing the address of the slice of bytes.
    },
}

// Seed the pool with 4KB
calcPool.Put(calcPool.New())
calcPool.Put(calcPool.New())
calcPool.Put(calcPool.New())
calcPool.Put(calcPool.New())

const numWorkers = 1024*1024
var wg sync.WaitGroup
wg.Add(numWorkers)
for i := numWorkers; i > 0; i-- {
    go func() {
        defer wg.Done()

        mem := calcPool.Get().(*[]byte) // And here we are asserting the type is a pointer to a slice of bytes.
        defer calcPool.Put(mem)

        // Assume something interesting, but quick is being done with
        // this memory.
    }()
}

wg.Wait()
fmt.Printf("%d calculators were created.", numCalcsCreated)

خروجی:

8 calculators were created.

بدون sync.Pool، در بدترین حالت ممکن بود یک گیگابایت حافظه allocate کنم، اما فقط ۴ کیلوبایت allocate شده.

موقعیت رایج دیگر: گرم کردن cache اشیاء از پیش allocateشده برای عملیاتی که باید سریع‌ترین اجرا را داشته باشند. اینجا به‌جای محدود کردن تعداد اشیاء، زمان consumer را با پیش‌بارگذاری زمان گرفتن reference محافظت می‌کنیم. در سرورهای شبکهٔ پرتوان رایج است. سناریویی شبیه‌سازی کنیم.

تابعی که اتصال به سرویس را شبیه‌سازی می‌کند — اتصال طولانی:

func connectToService() interface{} {
    time.Sleep(1*time.Second)
    return struct{}{}
}

عملکرد سرویس شبکه اگر برای هر درخواست اتصال جدید باز شود:

func startNetworkDaemon() *sync.WaitGroup {
    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        server, err := net.Listen("tcp", "localhost:8080")
        if err != nil {
            log.Fatalf("cannot listen: %v", err)
        }
        defer server.Close()

        wg.Done()

        for {
            conn, err := server.Accept()
            if err != nil {
                log.Printf("cannot accept connection: %v", err)
                continue
            }
            connectToService()
            fmt.Fprintln(conn, "")
            conn.Close()
        }
    }()
    return &wg
}

benchmark:

func init() {
    daemonStarted := startNetworkDaemon()
    daemonStarted.Wait()
}

func BenchmarkNetworkRequest(b *testing.B) {
    for i := 0; i < b.N; i++ {
        conn, err := net.Dial("tcp", "localhost:8080")
        if err != nil {
            b.Fatalf("cannot dial host: %v", err)
        }
        if _, err := ioutil.ReadAll(conn); err != nil {
            b.Fatalf("cannot read: %v", err)
        }
        conn.Close()
    }
}

cd src/gos-concurrency-building-blocks/the-sync-package/pool/ && \
go test -benchtime=10s -bench=.

BenchmarkNetworkRequest-8             10                       1000385643   ns/op
PASS
ok                                    command-line-arguments   11.008s

حدود ۱E9 ns/op — منطقی به نظر می‌رسد. با sync.Pool برای میزبانی اتصالات بهبود دهیم:

func warmServiceConnCache() *sync.Pool {
    p := &sync.Pool {
        New: connectToService,
    }
    for i := 0; i < 10; i++ {
        p.Put(p.New())
    }
    return p
}

func startNetworkDaemon() *sync.WaitGroup {
    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        connPool := warmServiceConnCache()

        server, err := net.Listen("tcp", "localhost:8080")
        if err != nil {
            log.Fatalf("cannot listen: %v", err)
        }
        defer server.Close()

        wg.Done()

        for {
            conn, err := server.Accept()
            if err != nil {
                log.Printf("cannot accept connection: %v", err)
                continue
            }
            svcConn := connPool.Get()
            fmt.Fprintln(conn, "")
            connPool.Put(svcConn)
            conn.Close()
        }
    }()
    return &wg
}

benchmark مجدد:

BenchmarkNetworkRequest-8               5000                     2904307   ns/op
PASS
ok                                      command-line-arguments   32.647s

۲.۹E6 ns/op: سه مرتبهٔ بزرگی سریع‌تر! الگوی object pool برای چیزهای پرهزینهٔ ساخت می‌تواند زمان پاسخ را به‌شدت بهبود دهد.

الگوی object pool بهترین است وقتی processهای همزمان به اشیاء نیاز دارند اما خیلی سریع پس از instantiation رها می‌کنند، یا ساخت آن‌ها روی حافظه تأثیر منفی دارد. یک هشدار: اگر کد استفاده‌کننده از Pool به چیزهای ناهمگن نیاز دارد، ممکن است زمان تبدیل آنچه از Pool گرفته‌اید بیش از instantiate مستقیم شود. مثلاً اگر برنامه به sliceهای با طول تصادفی و متغیر نیاز دارد، Pool کمک زیادی نمی‌کند.

هنگام کار با Pool به یاد داشته باشید:

• هنگام instantiate کردن sync.Pool، New thread-safe بدهید.
• پس از Get، دربارهٔ state شیء فرض نکنید.
• پایان کار Put کنید؛ وگرنه Pool بی‌فایده است. معمولاً با defer.
• اشیاء pool باید تقریباً همگن باشند.

Channels

Channelها یکی از primitiveهای همزمان‌سازی Go مشتق از CSP هور هستند. اگرچه می‌توانند دسترسی حافظه را همگام‌سازی کنند، بهترین کاربردشان ارتباط اطلاعات بین goroutineهاست. همان‌طور که در «فلسفهٔ همزمانی Go» بحث شد، channelها در برنامه‌های هر اندازه به‌خاطر قابلیت ترکیب (composition) بسیار مفیدند. پس از معرفی channel در این بخش، ترکیب را در «دستور select» بررسی می‌کنیم.

مانند رودخانه، channel مجرای جریان اطلاعات است؛ مقادیر از channel عبور و downstream خوانده می‌شوند. به همین دلیل معمولاً نام متغیرهای chan را با «Stream» تمام می‌کنم. با channel مقدار را به متغیر chan می‌فرستید و جای دیگر برنامه می‌خوانید. بخش‌های جدا برنامه نیازی به دانستن یکدیگر ندارند، فقط reference به همان نقطهٔ حافظه که channel در آن است. این با عبور دادن referenceهای channel در برنامه انجام می‌شود.

ساخت channel بسیار ساده است. مثالی که اعلان و instantiation را جدا می‌کند تا هر دو را ببینید. مانند مقادیر دیگر Go، می‌توانید با := در یک گام بسازید، اما اغلب باید channel اعلام کنید:

var dataStream chan interface{} // Here we declare a channel. We say it is "of type" interface{} since the type we've declared is the empty interface.
dataStream = make(chan interface{}) // Here we instantiate the channel using the built-in make function.

این مثال channel dataStream را تعریف می‌کند که هر مقداری روی آن نوشته یا خوانده می‌شود (چون از empty interface استفاده کردیم). Channelها می‌توانند فقط جریان یک‌طرفه داده را پشتیبانی کنند — فقط ارسال یا فقط دریافت. دلیل اهمیت را بعداً توضیح می‌دهم.

برای اعلام channel یک‌طرفه، عملگر <- را می‌گذارید. برای اعلام و ساخت channel فقط خواندنی، <- سمت چپ:

var dataStream <-chan interface{}
dataStream := make(<-chan interface{})

برای فقط ارسال، <- سمت راست:

var dataStream chan<- interface{}
dataStream := make(chan<- interface{})

معمولاً channel یک‌طرفه instantiate نمی‌شوند، اما اغلب به‌عنوان پارامتر و نوع بازگشتی تابع دیده می‌شوند. Go به‌طور ضمنی channel دوطرفه را به یک‌طرفه تبدیل می‌کند:

var receiveChan <-chan interface{}
var sendChan chan<- interface{}
dataStream := make(chan interface{})

// Valid statements:
receiveChan = dataStream
sendChan = dataStream

Channelها typed هستند. در این مثال chan interface{} ساختیم؛ می‌توانیم نوع سخت‌گیرانه‌تری بدهیم:

intStream := make(chan int)

برای استفاده از channelها دوباره از <- استفاده می‌کنیم. ارسال: <- سمت راست channel؛ دریافت: سمت چپ. داده در جهت فلش به متغیر جریان می‌یابد:

stringStream := make(chan string)
go func() {
    stringStream <- "Hello channels!" // Here we pass a string literal onto the channel stringStream.
}()
fmt.Println(<-stringStream) // Here we read the string literal off of the channel and print it out to stdout.

خروجی:

Hello channels!

بسیار ساده! فقط متغیر channel لازم است؛ اما خطا است روی channel فقط خواندنی بنویسید یا از channel فقط نوشتنی بخوانید:

writeStream := make(chan<- interface{})
readStream := make(<-chan interface{})

<-writeStream
readStream <- struct{}{}

خطای کامپایلر:

invalid operation: <-writeStream (receive from send-only type chan<- interface {})
invalid operation: readStream <- struct {} literal (send to receive-only type <-chan interface {})

بخشی از type system Go برای type-safety حتی با primitiveهای همزمانی. بعداً می‌بینیم این راه قدرتمندی برای API و برنامه‌های composable است.

یادآوری: فقط زمان‌بندی goroutine تضمین اجرا قبل از خروج process نیست؛ اما مثال قبلی کامل و صحیح بود. شاید پرسیده باشید چرا goroutine anonymous قبل از goroutine اصلی تمام شد؟

Channelهای Go blocking هستند. Goroutine که به channel پر بنویسد تا خالی شدن منتظر می‌ماند؛ goroutine که از channel خالی بخواند تا قرار گرفتن حداقل یک آیتم منتظر می‌ماند. در مثال، fmt.Println از stringStream می‌کشد و تا قرار گرفتن مقدار می‌ماند. Goroutine anonymous سعی می‌کند روی stringStream بنویسد و تا موفقیت write خارج نمی‌شود. بنابراین goroutine اصلی و anonymous به‌طور قطعی block می‌شوند.

اگر برنامه را درست ساختار ندهید deadlock می‌دهد:

stringStream := make(chan string)
go func() {
    if 0 != 1 {
        return
    }
    stringStream <- "Hello channels!"
}()
fmt.Println(<-stringStream) // Here we ensure the stringStream channel never gets a value placed upon it.

panic:

fatal error: all goroutines are asleep - deadlock!

Goroutine اصلی منتظر مقدار روی stringStream است و به‌خاطر شرط هرگز قرار نمی‌گیرد. وقتی goroutine anonymous خارج می‌شود، Go تشخیص می‌دهد همه goroutineها خواب‌اند و deadlock گزارش می‌کند. بعداً ساختار برنامه برای جلوگیری از چنین deadlockها و در فصل بعد جلوگیری کامل‌تر. فعلاً به خواندن از channel برگردیم.

فرم دریافت <- می‌تواند دو مقدار برگرداند:

stringStream := make(chan string)
go func() {
    stringStream <- "Hello channels!"
}()
salutation, ok := <-stringStream // Here we receive both a string, salutation, and a boolean, ok.
fmt.Printf("(%v): %v", ok, salutation)

خروجی:

(true): Hello channels!

boolean چه معنایی دارد؟ مقدار دوم نشان می‌دهد خواندن از channel ناشی از write در process بوده یا مقدار پیش‌فرض از channel بسته‌شده. Channel بسته‌شده چیست؟

در برنامه‌ها مفید است نشان دهیم دیگر مقداری روی channel فرستاده نمی‌شود. این به processهای downstream کمک می‌کند بدانند چه زمانی ادامه دهند، خارج شوند، روی channel جدید بازگشایی کنند و غیره. می‌توانستیم sentinel ویژه برای هر نوع داشته باشیم، اما بستن channel مانند sentinel جهانی است: «دیگر upstream نمی‌نویسد، هر کاری می‌خواهید بکنید.» برای بستن از close استفاده می‌کنیم:

valueStream := make(chan interface{})
close(valueStream)

از channel بسته هم می‌توان خواند:

intStream := make(chan int)
close(intStream)
integer, ok := <- intStream // Here we read from a closed stream.
fmt.Printf("(%v): %v", ok, integer)

خروجی:

(false): 0

هرگز چیزی روی channel نگذاشتیم؛ فوراً بستیم. همچنان read انجام شد و می‌توان بی‌نهایت از channel بسته خواند — برای پشتیبانی از چند reader downstream از یک writer upstream (در فصل ۴ رایج است). ok برابر false است: مقدار zero value برای int است نه مقدار واقعی stream.

الگوی جدید: range روی channel. range با for از channel پشتیبانی می‌کند و وقتی channel بسته شد حلقه را می‌شکند:

intStream := make(chan int)
go func() {
    defer close(intStream) // Here we ensure that the channel is closed before we exit the goroutine. This is a very common pattern.
    for i := 1; i <= 5; i++ {
        intStream <- i
    }
}()

for integer := range intStream { // Here we range over intStream.
    fmt.Printf("%v ", integer)
}

خروجی:

1 2 3 4 5

حلقه به معیار خروج نیاز ندارد و range boolean دوم را برنمی‌گرداند. مدیریت channel بسته برای شما انجام می‌شود.

بستن channel یکی از راه‌های سیگنال همزمان به چند goroutine است. اگر n goroutine روی یک channel منتظرند، به‌جای n بار نوشتن، channel را ببندید. چون از channel بسته بی‌نهایت خوانده می‌شود، مهم نیست چند goroutine منتظرند؛ بستن هم ارزان‌تر و سریع‌تر از n write است:

begin := make(chan interface{})
var wg sync.WaitGroup
for i := 0; i < 5; i++ {
    wg.Add(1)
    go func(i int) {
        defer wg.Done()
        <-begin // Here the goroutine waits until it is told it can continue.
        fmt.Printf("%v has begun\n", i)
    }(i)
}

fmt.Println("Unblocking goroutines...")
close(begin) // Here we close the channel, thus unblocking all the goroutines simultaneously.
wg.Wait()

خروجی:

Unblocking goroutines...
4 has begun
2 has begun
3 has begun
0 has begun
1 has begun

در «پکیج sync» با sync.Cond همان رفتار را بحث کردیم. می‌توانید از آن استفاده کنید، اما channelها composable هستند و این روش مورد علاقهٔ من برای unblock همزمان چند goroutine است.

می‌توانیم buffered channel بسازیم — channelهایی که هنگام instantiation ظرفیت دارند. یعنی حتی بدون read، goroutine می‌تواند n بار بنوید که n ظرفیت buffered channel است:

var dataStream chan interface{}
dataStream = make(chan interface{}, 4) // Here we create a buffered channel with a capacity of four. This means that we can place four things onto the channel regardless of whether it's being read from.

اعلان buffered channel از unbuffered تفاوتی ندارد. Goroutineی که channel را می‌سازد کنترل می‌کند buffered باشد یا نه — یعنی ساخت channel بهتر است به goroutineهایی که write می‌کنند tightly couple شود.

Unbuffered channel از نظر buffered channel با ظرفیت ۰ تعریف می‌شود:

a := make(chan int)
b := make(chan int, 0)

هر دو int channel با ظرفیت صفر. «پر» و «خالی» تابع ظرفیت هستند. Unbuffered قبل از هر write پر است. Buffered بدون receiver و ظرفیت ۴ بعد از چهار write پر و write پنجم block می‌شود. Buffered channelها هم blocking هستند؛ فقط پیش‌شرط‌ها متفاوت‌اند. buffered channel صف FIFO در حافظه برای ارتباط processهای همزمان است.

برای درک، buffered channel با ظرفیت ۴:

c := make(chan rune, 4)

شکل ۳-۲. بافر channel خالی با ظرفیت ۴

[ _ | _ | _ | _ ]

c <- 'A'

شکل ۳-۳. پس از نوشتن 'A' بدون reader

[ A | _ | _ | _ ]

c <- 'B'
c <- 'C'
c <- 'D'

شکل ۳-۴. بافر پر شده تا 'D'

[ A | B | C | D ]

پس از چهار write، channel پر است. write پنجم:

c <- 'E'

شکل ۳-۵. goroutine نویسنده block شده — بافر پر

[ A | B | C | D ]  ← goroutine نویسنده 'E' منتظر جا

Goroutine نویسنده block می‌ماند تا read جا باز کند:

<-c

شکل ۳-۶. پس از read — 'A' خارج شد، 'E' وارد شد

[ B | C | D | E ]

read اولین rune قرارگرفته، A، را می‌گیرد؛ write blockشده unblock و E انتهای بافر قرار می‌گیرد.

اگر buffered channel خالی باشد و receiver داشته باشد، بافر bypass می‌شود و مقدار مستقیم از sender به receiver می‌رود. در عمل شفاف است اما برای performance profile مفید است.

Buffered channel در موقعیت‌های خاص مفید است اما با احتیاط بسازید. در فصل بعد می‌تواند بهینه‌سازی زودهنگام شود و deadlock را با کم‌محتمل‌تر کردن پنهان کند — بهتر است deadlock را هنگام نوشتن کد پیدا کنید نه نیمه‌شب در production.

مثال کامل‌تر buffered channel:

var stdoutBuff bytes.Buffer
defer stdoutBuff.WriteTo(os.Stdout) // Here we ensure that the buffer is written out to stdout before the process exits.

intStream := make(chan int, 4) // Here we create a buffered channel with a capacity of one.
go func() {
    defer close(intStream)
    defer fmt.Fprintln(&stdoutBuff, "Producer Done.")
    for i := 0; i < 5; i++ {
        fmt.Fprintf(&stdoutBuff, "Sending: %d\n", i)
        intStream <- i
    }
}()

for integer := range intStream {
    fmt.Fprintf(&stdoutBuff, "Received %v.\n", integer)
}

• اینجا بافر در حافظه برای کاهش غیرقطعیت خروجی (تضمین نمی‌دهد اما سریع‌تر از stdout مستقیم).
• goroutine anonymous هر پنج نتیجه را روی intStream می‌گذارد و قبل از pull اول توسط main خارج می‌شود.

خروجی نمونه:

Sending: 0
Sending: 1
Sending: 2
Sending: 3
Sending: 4
Producer Done.
Received 0.
Received 1.
Received 2.
Received 3.
Received 4.

بهینه‌سازی مفید در شرایط درست: اگر goroutine نویسنده تعداد writeها را می‌داند، buffered channel با ظرفیت برابر تعداد writeها و write سریع مفید است. محدودیت‌ها در فصل بعد.

unbuffered، buffered، دوطرفه و یک‌طرفه را پوشش دادیم. تنها جنبهٔ باقی‌مانده: مقدار پیش‌فرض channel — nil. خواندن از nil channel:

var dataStream chan interface{}
<-dataStream

fatal error: all goroutines are asleep - deadlock!

خواندن از nil channel block می‌کند (لزوماً deadlock). نوشتن:

var dataStream chan interface{}
dataStream <- struct{}{}

fatal error: all goroutines are asleep - deadlock!

نوشتن هم block می‌کند. close:

var dataStream chan interface{}
close(dataStream)

panic: close of nil channel

بدترین نتیجه: panic. channelها را همیشه ابتدا initialize کنید.

جدول ۳-۲. نتیجهٔ عملیات channel با توجه به state

عملیات	state channel	نتیجه
Read	nil	Block
Read	Open و Not Empty	Value
Read	Open و Empty	Block
Read	Closed	<default value>, false
Read	Write Only	Compilation Error
Write	nil	Block
Write	Open و Full	Block
Write	Open و Not Full	Write Value
Write	Closed	panic
Write	Receive Only	Compilation Error
close	nil	panic
close	Open و Not Empty	Channel بسته می‌شود؛ read تا drain موفق، سپس default value
close	Open و Empty	Channel بسته؛ read مقدار پیش‌فرض
close	Closed	panic
close	Receive Only	Compilation Error

سه عملیات block و سه عملیات panic! اما با انگیزهٔ نتایج و قاب‌بندی استفاده از channel منطقی می‌شود.

اولین کار: مالکیت channel تعریف کنیم — goroutineی که channel را instantiate، می‌نویسد و می‌بندد. مانند حافظه در زبان‌های بدون GC، مهم است بدانیم کدام goroutine مالک channel است. اعلام channel یک‌طرفه مالک (نمای write: chan یا chan<-) را از utilizer (فقط <-chan) جدا می‌کند.

مالک channel باید:

1. Channel را instantiate کند.
2. Write کند یا مالکیت را به goroutine دیگر بدهد.
3. Channel را ببندد.
4. سه مورد بالا را encapsulate کند و از طریق channel فقط خواندنی expose کند.

نتیجه:

• خطر deadlock با write به nil channel حذف می‌شود.
• خطر panic با close کردن nil حذف می‌شود.
• خطر panic با write به channel بسته حذف می‌شود.
• خطر panic با close دوباره حذف می‌شود.
• type checker در compile time از write نادرست جلوگیری می‌کند.

مصرف‌کننده فقط دو نگرانی دارد:

• دانستن زمان بسته شدن channel (boolean دوم read).
• مدیریت مسئولانه block به هر دلیل (timeout، توقف با سیگنال، یا block تا پایان process).

راه‌های reader در فصل بعد. مثال مالکیت واضح:

chanOwner := func() <-chan int {
    resultStream := make(chan int, 5) // Here we instantiate a buffered channel. Since we know we'll produce six results, we create a buffered channel of five so that the goroutine can complete as quickly as possible.
    go func() {
        defer close(resultStream) // Here we ensure resultStream is closed once we're finished with it. As the channel owner, this is our responsibility.
        for i := 0; i <= 5; i++ {
            resultStream <- i
        }
    }() // Here we start an anonymous goroutine that performs writes on resultStream. Notice that we've inverted how we create goroutines. It is now encapsulated within the surrounding function.
    return resultStream // Here we return the channel. Since the return value is declared as a read-only channel, resultStream will implicitly be converted to read-only for consumers.
}()

resultStream := chanOwner()
for result := range resultStream { // Here we range over resultStream. As a consumer, we are only concerned with blocking and closed channels.
    fmt.Printf("Received: %d\n", result)
}
fmt.Println("Done receiving!")

خروجی:

Received: 0
Received: 1
Received: 2
Received: 3
Received: 4
Received: 5
Done receiving!

چرخهٔ حیات resultStream در chanOwner encapsulate شده. write روی nil یا بسته نمی‌شود و close همیشه یک‌بار است. scope مالکیت channel را کوچک نگه دارید.

مصرف‌کننده فقط به channel خواندنی دسترسی دارد و فقط block و close را مدیریت می‌کند. در این مثال block تا بسته شدن channel پذیرفته شده.

اگر کد را بر این اصل بسازید، استدلال آسان‌تر و احتمال عملکرد مطابق انتظار بیشتر است. deadlock یا panic ممکن است اما احتمالاً scope مالکیت بزرگ یا مبهم شده.

Channelها یکی از دلایل جذب من به Go بود. با سادگی goroutine و closure، نوشتن کد همزمان تمیز و صحیح آسان بود. Channelها چسب goroutineها هستند. این فصل نمای کلی channel و استفاده داد. سرگرمی واقعی وقتی شروع می‌شود که channelها را برای الگوهای طراحی همزمانی سطح بالاتر ترکیب کنیم — فصل بعد.

دستور select

دستور select چسب channelهاست؛ نحوهٔ ترکیب channelها در برنامه برای abstractionهای بزرگ‌تر. اگر channelها goroutineها را به هم می‌چسبانند، select چه می‌گوید؟ کم‌گفتن نیست که select یکی از مهم‌ترین چیزها در برنامهٔ همزمان Go است. select channelها را محلی در یک تابع یا نوع، و سراسری در تقاطع اجزای سیستم به هم می‌چسباند. در این تقاطع‌ها، select می‌تواند لغو، timeout، انتظار و مقدار پیش‌فرض را با channelها ایمن ترکیب کند.

اگر select lingua franca برنامه باشد و فقط با channel کار کند، اجزا چگونه هماهنگ شوند؟ فصل ۵ (راهنما: ترجیح channel).

select چیست و چگونه کار می‌کند؟ مثال ساده:

var c1, c2 <-chan interface{}
var c3 chan<- interface{}
select {
case <- c1:
    // Do something
case <- c2:
    // Do something
case c3<- struct{}{}:
    // Do something
}

شبیه switch است: سری case که دستورات را محافظت می‌کنند؛ اما شباهت همین‌جا تمام می‌شود. در select، caseها به‌ترتیب تست نمی‌شوند و اگر هیچ‌کدام برقرار نبود اجرا به case بعدی نمی‌افتد.

همهٔ read و writeهای channel همزمان³ بررسی می‌شوند که آیا آماده‌اند: برای read، channel پر یا بسته؛ برای write، channel پر نیست. اگر هیچ channel آماده نبود، کل select block می‌شود. وقتی یکی آماده شد، آن عملیات و دستورات مربوط اجرا می‌شوند:

start := time.Now()
c := make(chan interface{})
go func() {
    time.Sleep(5*time.Second) // Here we close the channel after waiting five seconds.
    close(c)
}()

fmt.Println("Blocking on read...")
select {
case <-c: // Here we attempt a read on the channel. Note that as this code is written, we don't require a select statement—we could simply write <-c—but we'll expand on this example.
    fmt.Printf("Unblocked %v later.\n", time.Since(start))
}

خروجی:

Blocking on read...
Unblocked 5.000170047s later.

حدود پنج ثانیه بعد unblock — راه ساده و کارآمد برای block هنگام انتظار.

سؤالات:

• اگر چند channel همزمان چیزی برای read داشته باشند؟
• اگر هیچ channel آماده نشود؟
• اگر بخواهیم کاری انجام دهیم اما هیچ channel آماده نباشد؟

چند channel همزمان آماده:

c1 := make(chan interface{}); close(c1)
c2 := make(chan interface{}); close(c2)

var c1Count, c2Count int
for i := 1000; i >= 0; i-- {
    select {
    case <-c1:
        c1Count++
    case <-c2:
        c2Count++
    }
}

fmt.Printf("c1Count: %d\nc2Count: %d\n", c1Count, c2Count)

خروجی نمونه:

c1Count: 505
c2Count: 496

در هزار تکرار، تقریباً نصف از c1 و نصف از c2. تصادفی به نظر می‌رسد و هست! Runtime Go انتخاب شبه‌تصادفی یکنواخت روی مجموعهٔ caseها انجام می‌دهد — هر case شانس برابر.

Runtime نمی‌تواند intent select را بداند؛ بهترین کار عملکرد خوب در حالت میانگین است — متغیر تصادفی در معادله، یعنی انتخاب channel با وزن برابر.

اگر هیچ channel آماده نشود و نمی‌توانید برای همیشه block کنید، timeout. پکیج time با channel در paradigm select:

var c <-chan int
select {
case <-c: // This case statement will never become unblocked because we're reading from a nil channel.
case <-time.After(1 * time.Second):
    fmt.Println("Timed out.")
}

خروجی:

Timed out.

time.After یک time.Duration می‌گیرد و channelی برمی‌گرداند که پس از آن مدت زمان فعلی را می‌فرستد. الگوی timeout در select. در فصل ۴ راه‌حل مقاوم‌تر.

اگر هیچ channel آماده نبود و می‌خواهید در عین حال کاری انجام دهید، default:

start := time.Now()
var c1, c2 <-chan int
select {
case <-c1:
case <-c2:
default:
    fmt.Printf("In default after %v\n\n", time.Since(start))
}

خروجی:

In default after 1.421µs

تقریباً فوراً default اجرا شد — خروج از select بدون block. معمولاً default با حلقهٔ for-select برای پیشرفت کار در goroutine هنگام انتظار نتیجه:

done := make(chan interface{})
go func() {
    time.Sleep(5*time.Second)
    close(done)
}()

workCounter := 0
loop:
for {
    select {
    case <-done:
        break loop
    default:
    }

    // Simulate work
    workCounter++
    time.Sleep(1*time.Second)
}

fmt.Printf("Achieved %v cycles of work before signalled to stop.\n", workCounter)

خروجی:

Achieved 5 cycles of work before signalled to stop.

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

حالت خاص: select بدون case:

select {}

برای همیشه block می‌کند.

در فصل ۶ عمیق‌تر به نحوهٔ کار select می‌پردازیم. از دید سطح بالا، ترکیب مفاهیم و زیرسیستم‌ها را ایمن و کارآمد نشان می‌دهد.

اهرم GOMAXPROCS

در پکیج runtime تابعی به نام GOMAXPROCS هست. به نظرم نام گمراه‌کننده است: بسیاری فکر می‌کنند به تعداد logical processor روی ماشین میزبان مربوط است — و به‌نوعی چنین است — اما واقعاً تعداد threadهای OS را کنترل می‌کند که «صف‌های کار» را میزبانی می‌کنند. جزئیات در فصل ۶.

قبل از Go 1.5، GOMAXPROCS همیشه ۱ بود و این snippet در اکثر برنامه‌ها بود:

runtime.GOMAXPROCS(runtime.NumCPU())

تقریباً همه می‌خواستند از همهٔ coreها استفاده کنند. از Go 1.5 به بعد به‌طور خودکار برابر تعداد logical CPU روی ماشین میزبان تنظیم می‌شود.

چرا این مقدار را تغییر دهید؟ اغلب نمی‌خواهید. الگوریتم زمان‌بندی Go در بیشتر موارد کافی است و افزایش یا کاهش صف‌ها و threadها احتمالاً بیشتر ضرر می‌زند، اما گاهی مفید است.

مثلاً پروژه‌ای با test suite پر از race condition. با افزایش GOMAXPROCS فراتر از تعداد logical CPU، race conditionها بیشتر trigger شدند و سریع‌تر اصلاح شدند.

برخی با آزمایش عملکرد بهتر با تعداد خاصی از صف‌ها و threadها می‌یابند، اما احتیاط کنید. اگر با tweak کردن performance می‌گیرید، بعد از هر commit، سخت‌افزار مختلف و نسخه‌های مختلف Go دوباره بررسی کنید. این مقدار برنامه را به metal نزدیک‌تر می‌کند به هزینهٔ abstraction و پایداری عملکرد بلندمدت.

جمع‌بندی

در این فصل همهٔ primitiveهای پایهٔ همزمانی Go را پوشش دادیم. اگر خوانده و فهمیده‌اید، تبریک! در راه نوشتن برنامه‌های performant، خوانا و منطقاً صحیح هستید. می‌دانید چه زمانی primitiveهای همگام‌سازی دسترسی حافظه در پکیج sync مناسب‌اند و چه زمانی «share memory by communicating» با channelها و select.

برای نوشتن کد همزمان Go فقط مانده است این primitiveها را به‌صورت ساختاریافته ترکیب کنید که scale کنند و قابل فهم باشند. نیمهٔ دوم کتاب همین است. فصل بعد دربارهٔ ترکیب این primitiveها با الگوهایی است که جامعه کشف کرده است.

---

¹ کسانی که با C آشنا هستند ممکن است مقایسه با تابع fork را در نظر بگیرند. مدل fork-join مدل منطقی نحوهٔ انجام همزمانی است. برنامهٔ C که fork و سپس wait می‌زند را در سطح منطقی توصیف می‌کند، اما دربارهٔ مدیریت حافظه چیزی نمی‌گوید.

² شخصاً کتاب عالی Head First Design Patterns از O'Reilly را توصیه می‌کنم.

³ آنچه زیر پوست اتفاق می‌افتد پیچیده‌تر است، همان‌طور که در فصل ۶ می‌بینیم.