حالت تاریک
فصل ۵ — همزمانی در مقیاس
اکنون که برخی الگوهای رایج برای بهکارگیری همزمانی در Go را یاد گرفتهاید، توجه خود را به ترکیب این الگوها در مجموعهای از شیوهها معطوف میکنیم که به شما امکان میدهد سیستمهای بزرگ و قابلترکیب بسازید که مقیاسپذیر باشند.
در این فصل، راههای مقیاسدهی عملیات همزمان درون یک فرایند را بررسی میکنیم و همچنین به بررسی نقش همزمانی هنگام کار با بیش از یک فرایند میپردازیم.
Error Propagation
در کد همزمان، و بهویژه در سیستمهای توزیعشده، هم آسان است چیزی در سیستم شما خراب شود و هم دشوار است بفهمید چرا رخ داده است. با در نظر گرفتن دقیق نحوهٔ انتشار مسائل در سیستم و نحوهٔ نمایش آنها به کاربر، میتوانید درد زیادی را از خود، تیم و کاربرانتان دور کنید. در بخش «Error Handling» در صفحهٔ ۹۷، دربارهٔ انتشار خطا از goroutineها صحبت کردیم، اما زمانی صرف نکردیم که خطاها چه شکلی باید داشته باشند یا خطاها چگونه باید در یک سیستم بزرگ و پیچیده جریان یابند. اینجا کمی دربارهٔ فلسفهٔ انتشار خطا بحث میکنیم. آنچه در ادامه میآید چارچوبی نظرمحور برای مدیریت خطا در سیستمهای همزمان است.
بسیاری از توسعهدهندگان اشتباه میکنند و انتشار خطا را ثانویه یا «جدا» از جریان سیستم میپندارند. به نحوهٔ جریان داده در سیستم توجه دقیق میشود، اما خطاها چیزیاند که تحمل میشوند و بدون فکر زیاد به بالای stack منتقل میشوند و در نهایت جلوی کاربر ریخته میشوند. Go تلاش کرد این بدعهدی را با وادار کردن کاربران به مدیریت خطا در هر لایهٔ call stack اصلاح کند، اما هنوز رایج است خطاها را شهروند درجهٔ دوم نسبت به control flow سیستم ببینند. با کمی پیشاندیشی و سربار کم، میتوانید مدیریت خطا را به دارایی سیستم و لذتی برای کاربران تبدیل کنید.
ابتدا بررسی کنیم خطاها چیستند. چه زمانی رخ میدهند و چه سودی دارند؟
خطاها نشان میدهند سیستم شما وارد حالتی شده که نمیتواند عملیاتی را که کاربر صریحاً یا ضمنی درخواست کرده انجام دهد. به همین دلیل باید چند قطعه اطلاعات حیاتی را منتقل کند:
چه اتفاقی افتاد.
این بخشی از خطاست که اطلاعات دربارهٔ آنچه رخ داده را دارد، مثلاً «disk full»، «socket closed» یا «credentials expired». این اطلاعات احتمالاً بهطور ضمنی توسط آنچه خطا را تولید کرده ساخته میشود، اگرچه میتوانید آن را با contextی که به کاربر کمک میکند تزئین کنید.
چه زمانی و کجا رخ داد.
خطاها همیشه باید stack trace کاملی داشته باشند که از نحوهٔ آغاز call تا جایی که خطا instantiate شده ادامه یابد. stack trace نباید در پیام خطا باشد (بیشتر دربارهٔ این میگوییم)، اما هنگام مدیریت خطا در بالای stack باید بهراحتی در دسترس باشد.
علاوه بر این، خطا باید اطلاعاتی دربارهٔ contextی که در آن اجرا میشود داشته باشد. مثلاً در سیستم توزیعشده باید راهی برای شناسایی ماشینی که خطا روی آن رخ داده باشد. بعداً هنگام فهم آنچه در سیستم رخ داده، این اطلاعات بیقیمت است.
همچنین خطا باید زمان روی ماشینی که خطا روی آن instantiate شده را به UTC داشته باشد.
پیام کاربرپسند برای کاربر.
پیامی که به کاربر نمایش داده میشود باید متناسب با سیستم و کاربرانش سفارشی شود. فقط باید اطلاعات خلاصه و مرتبط از دو نکتهٔ قبلی را داشته باشد. پیام دوستانه انسانمحور است، نشانهای میدهد که آیا مسئله موقتی است و باید حدود یک خط متن باشد.
چگونه کاربر اطلاعات بیشتری بگیرد.
در نقطهای احتمالاً کسی میخواهد بداند هنگام خطا دقیقاً چه اتفاقی افتاده. خطاهای ارائهشده به کاربر باید IDای داشته باشند که با log متناظر cross-reference شود و اطلاعات کامل خطا را نشان دهد: زمان وقوع خطا (نه زمان log شدن)، stack trace — همه چیزهایی که هنگام ساخت خطا در آن گذاشتید. گنجاندن hash از stack trace برای تجمیع مسائل مشابه در bug tracker هم مفید است.
بهطور پیشفرض، هیچ خطایی بدون مداخلهٔ شما همهٔ این اطلاعات را ندارد. بنابراین میتوانید بگویید هر خطایی که بدون این اطلاعات به کاربر منتقل شود اشتباه است و در نتیجه bug است. این به چارچوب کلی برای فکر کردن به خطاها منجر میشود. میتوان همهٔ خطاها را در یکی از دو دسته قرار داد:
- Bugs
- Known edge cases (مثلاً قطع اتصال شبکه، شکست نوشتن روی disk و غیره)
Bugs خطاهاییاند که برای سیستم سفارشی نکردهاید، یا «raw» errors — edge caseهای شناختهشده. گاهی عمدی است؛ شاید موافق باشید خطاهای edge case به کاربر برسند تا چند iteration اول سیستم را تحویل دهید. گاهی تصادفی است. اما اگر با رویکردی که گفتم موافقید، raw errors همیشه bug هستند. این تمایز هنگام تعیین نحوهٔ انتشار خطا، رشد سیستم و آنچه در نهایت به کاربر نشان داده میشود مفید است.
سیستم بزرگی با چند ماژول را تصور کنید:
شکل: سیستم بزرگ با چند ماژول — Low Level Component، Intermediary Component و لایههای بالاتر.
فرض کنید خطایی در «Low Level Component» رخ میدهد و آنجا خطای well-formed ساختهایم تا به بالای stack پاس داده شود. در context «Low Level Component» این خطا well-formed بهنظر میرسد، اما در context سیستم ما شاید نباشد. بگیریم در مرزهای هر component، همهٔ خطاهای ورودی باید در خطای well-formed برای componentی که کد ما در آن است wrap شوند. مثلاً اگر در «Intermediary Component» هستیم و کدی از «Low Level Component» را صدا میزنیم که ممکن است خطا بدهد، میتوانیم این را داشته باشیم:
go
func PostReport(id string) error {
result, err := lowlevel.DoWork()
if err != nil {
if _, ok := err.(lowlevel.Error); ok {
err = WrapErr(err, "cannot post report with id %q", id)
}
return err
}
// ...
}اینجا بررسی میکنیم خطای well-formed دریافت میکنیم. اگر نه، خطای malformed را به بالای stack میفرستیم تا bug بودن را نشان دهیم.
اینجا از فراخوانی فرضی برای wrap کردن خطای ورودی با اطلاعات مرتبط برای ماژول و دادن type جدید استفاده میکنیم. توجه کنید wrap کردن ممکن است شامل پنهان کردن جزئیات سطح پایین باشد که در این context برای کاربر مهم نیستند.
جزئیات سطح پایین محل ریشهٔ خطا (مثلاً goroutine، ماشین، stack trace و غیره) هنگام instantiate اولیهٔ خطا پر میشوند، اما معماری ما میگوید در مرزهای ماژول خطا را به type خطای ماژول خود تبدیل کنیم — و احتمالاً اطلاعات مرتبط را پر کنیم. حالا هر خطایی که بدون type خطای ماژول ما از ماژول خارج شود malformed و bug محسوب میشود. توجه کنید فقط در مرزهای ماژول خود — توابع/متدهای public — یا وقتی کد میتواند context ارزشمند اضافه کند لازم است خطاها را اینگونه wrap کنید. معمولاً این نیاز به wrap در بیشتر کد را حذف میکند.
این رویکرد به سیستم اجازه میدهد بهصورت ارگانیک رشد کند. مطمئن میشویم خطاهای ورودی well-formed هستند و خودمان هم به نحوهٔ خروج خطا از ماژول فکر میکنیم. صحت خطا ویژگی emergent سیستم میشود. همچنین از ابتدا کمال را کنار میگذاریم با مدیریت صریح خطاهای malformed و چارچوبی برای اصلاح اشتباهات در طول زمان. خطاهای malformed هم با type و هم با آنچه به کاربر نشان داده میشود مشخصاند.
همانطور که گفتیم، همهٔ خطاها باید با حداکثر اطلاعات موجود log شوند. اما هنگام نمایش خطا به کاربر، تمایز بین bugs و known edge cases مطرح میشود.
وقتی کد user-facing خطای well-formed دریافت میکند، مطمئن میشویم در همهٔ سطوح کد به ساخت پیام خطا توجه شده و میتوانیم آن را log کنیم و به کاربر نشان دهیم. اطمینانی که از دیدن خطا با type صحیح میگیریم قابلکمارزشگذاری نیست.
وقتی خطاهای malformed یا bugs به کاربر میرسند، باز هم خطا را log میکنیم، اما پیام دوستانهای نشان میدهیم که چیز غیرمنتظرهای رخ داده. اگر گزارش خودکار خطا داریم، خطا بهعنوان bug گزارش میشود. اگر نه، شاید از کاربر بخواهیم bug report بدهد. توجه کنید خطای malformed ممکن است اطلاعات مفید داشته باشد، اما تضمین نمیکنیم؛ پس — چون تنها تضمین این است که خطا سفارشی نشده — پیام انسانمحور سادهای دربارهٔ آنچه رخ داده نشان میدهیم.
به یاد داشته باشید در هر دو حالت، با خطای well-formed یا malformed، log ID در پیام گنجاندهایم تا کاربر برای اطلاعات بیشتر مرجع داشته باشد. پس حتی اگر bugs اطلاعات مفید داشته باشند، کاربر کنجکاو راهی برای بررسی دارد.
بیایید مثال کاملی ببینیم. این مثال خیلی robust نیست (مثلاً type خطا شاید سادهانگارانه باشد) و call stack خطی است که پنهان میکند فقط در مرزهای ماژول لازم است خطا wrap شود. همچنین نمایش توابع در packageهای مختلف در کتاب دشوار است، پس وانمود میکنیم.
ابتدا type خطایی بسازیم که همهٔ جنبههای خطای well-formed را داشته باشد:
go
type MyError struct {
Inner error
Message string
StackTrace string
Misc map[string]interface{}
}
func wrapError(err error, messagef string, msgArgs ...interface{}) MyError {
return MyError{
Inner: err,
Message: fmt.Sprintf(messagef, msgArgs...),
StackTrace: string(debug.Stack()),
Misc: make(map[string]interface{}),
}
}
func (err MyError) Error() string {
return err.Message
}اینجا خطایی که wrap میکنیم را ذخیره میکنیم. همیشه میخواهیم به خطای سطح پایینتر برگردیم اگر نیاز به بررسی داشتیم.
این خط stack trace هنگام ساخت خطا را ثبت میکند. type خطای پیشرفتهتر ممکن است stack-frame از wrapError را حذف کند.
اینجا catch-all برای اطلاعات متفرقه میسازیم. اینجا ممکن است concurrent ID، hash از stack trace یا context دیگر برای تشخیص خطا ذخیره کنیم.
سپس ماژول lowlevel:
go
// "lowlevel" module
type LowLevelErr struct {
error
}
func isGloballyExec(path string) (bool, error) {
info, err := os.Stat(path)
if err != nil {
return false, LowLevelErr{(wrapError(err, err.Error()))}
}
return info.Mode().Perm()&0100 == 0100, nil
}اینجا raw error از os.Stat را با خطای سفارشی wrap میکنیم. در این مورد با پیام این خطا موافقیم و آن را mask نمیکنیم.
سپس ماژول intermediate که توابع lowlevel را صدا میزند:
go
// "intermediate" module
type IntermediateErr struct {
error
}
func runJob(id string) error {
const jobBinPath = "/bad/job/binary"
isExecutable, err := isGloballyExec(jobBinPath)
if err != nil {
return err
} else if isExecutable == false {
return wrapError(nil, "job binary is not executable")
}
return exec.Command(jobBinPath, "--id="+id).Run()
}اینجا خطاهای ماژول lowlevel را پاس میدهیم. بهخاطر تصمیم معماری که خطاهای پاسشده از ماژولهای دیگر بدون wrap در type خودمان bug هستند، بعداً مشکل ایجاد میکند.
در نهایت تابع main سطح بالا که توابع intermediate را صدا میزند — بخش user-facing برنامه:
go
func handleError(key int, err error, message string) {
log.SetPrefix(fmt.Sprintf("[logID: %v]: ", key))
log.Printf("%#v", err)
fmt.Printf("[%v] %v", key, message)
}
func main() {
log.SetOutput(os.Stdout)
log.SetFlags(log.Ltime|log.LUTC)
err := runJob("1")
if err != nil {
msg := "There was an unexpected issue; please report this as a bug."
if _, ok := err.(IntermediateErr); ok {
msg = err.Error()
}
handleError(1, err, msg)
}
}اینجا بررسی میکنیم خطا از type مورد انتظار است. اگر باشد، خطای well-crafted است و پیامش را به کاربر میدهیم.
در این خط log و پیام خطا را با ID برابر ۱ bind میکنیم. میتوان بهصورت monotonic افزایش داد یا GUID برای ID یکتا استفاده کرد.
اینجا خطای کامل را log میکنیم اگر کسی بخواهد بررسی کند.
وقتی اجرا میکنیم، log شامل این است:
[logID: 1]: 21:46:07 main.LowLevelErr{error:main.MyError{Inner:
(*os.PathError)(0xc4200123f0),
Message:"stat /bad/job/binary: no such file or directory",
StackTrace:"goroutine 1 [running]:
runtime/debug.Stack(0xc420012420, 0x2f, 0xc420045d80)
/home/kate/.guix-profile/src/runtime/debug/stack.go:24 +0x79
main.wrapError(0x530200, 0xc4200123f0, 0xc420012420, 0x2f, 0x0, 0x0,
0x0, 0x0, 0x0, 0x0, ...)
/tmp/babel-79540aE/go-src-7954NTK.go:22 +0x62
main.isGloballyExec(0x4d1313, 0xf, 0xc420045eb8, 0x487649, 0xc420056050)
/tmp/babel-79540aE/go-src-7954NTK.go:37 +0xaa
main.runJob(0x4cfada, 0x1, 0x4d4c35, 0x22)
/tmp/babel-79540aE/go-src-7954NTK.go:47 +0x48
main.main()
/tmp/babel-79540aE/go-src-7954NTK.go:67 +0x63
", Misc:map[string]interface {}{}}}و پیام stdout:
[1] There was an unexpected issue; please report this as a bug.میبینیم جایی در مسیر این خطا درست مدیریت نشده و چون نمیتوانیم مطمئن باشیم پیام خطا برای انسان مناسب است، خطای سادهای چاپ میکنیم که چیز غیرمنتظرهای رخ داده (درست اگر این methodology را دنبال کنیم). اگر به ماژول intermediate نگاه کنیم، دلیل را یادآوری میکنیم: خطاهای lowlevel را wrap نکردیم. بیایید اصلاح کنیم:
go
// "intermediate" module
type IntermediateErr struct {
error
}
func runJob(id string) error {
const jobBinPath = "/bad/job/binary"
isExecutable, err := isGloballyExec(jobBinPath)
if err != nil {
return IntermediateErr{wrapError(
err,
"cannot run job %q: requisite binaries not available",
id,
)}
} else if isExecutable == false {
return wrapError(
nil,
"cannot run job %q: requisite binaries are not executable",
id,
)
}
return exec.Command(jobBinPath, "--id="+id).Run()
}اینجا خطا را با پیام ساختهشده سفارشی میکنیم. میخواهیم جزئیات سطح پایین دلیل اجرا نشدن job را پنهان کنیم چون برای مصرفکنندگان ماژول مهم نیست.
go
func handleError(key int, err error, message string) {
log.SetPrefix(fmt.Sprintf("[logID: %v]: ", key))
log.Printf("%#v", err)
fmt.Printf("[%v] %v", key, message)
}
func main() {
log.SetOutput(os.Stdout)
log.SetFlags(log.Ltime|log.LUTC)
err := runJob("1")
if err != nil {
msg := "There was an unexpected issue; please report this as a bug."
if _, ok := err.(IntermediateErr); ok {
msg = err.Error()
}
handleError(1, err, msg)
}
}با کد بهروزشده log مشابهی میگیریم و پیام خطا دقیقاً همان چیزی است که میخواهیم کاربر ببیند:
[1] cannot run job "1": requisite binaries not availablepackageهای خطا¹ وجود دارند که با این رویکرد سازگارند، اما پیادهسازی با package خطای انتخابی شماست. خبر خوب این است که این تکنیک ارگانیک است؛ میتوانید error handling سطح بالا را بررسی کنید و بین bugs و خطاهای well-crafted تمایز بگذارید و تدریجاً همهٔ خطاهای ساختهشده را well-crafted بدانید.
¹ من http://github.com/pkg/errors را توصیه میکنم.
Timeouts and Cancellation
هنگام کار با کد همزمان، timeoutها و cancellation زیاد پیش میآیند. همانطور که در این بخش میبینیم، timeoutها برای ساخت سیستمی با رفتار قابلفهم حیاتیاند. cancellation پاسخ طبیعی به timeout است. دلایل دیگر لغو فرایند همزمان را هم بررسی میکنیم.
چرا ممکن است بخواهیم فرایندهای همزمان timeout را پشتیبانی کنند؟ چند مورد:
System saturation
همانطور که در بخش «Queuing» در صفحهٔ ۱۲۴ گفتیم، اگر سیستم اشباع باشد (یعنی ظرفیت پردازش درخواستها پر باشد)، شاید بخواهیم درخواستهای لبهٔ سیستم timeout شوند بهجای اینکه مدت زیادی طول بکشد. مسیر به problem space بستگی دارد، اما راهنماهای کلی برای زمان timeout:
- اگر احتمال تکرار درخواست پس از timeout کم است.
- اگر منابع ذخیرهٔ درخواستها را ندارید (مثلاً حافظه برای صف in-memory، فضای disk برای صف persisted).
- اگر نیاز به درخواست یا دادهٔ ارسالی کهنه میشود (بعداً میگوییم). اگر درخواست احتمال تکرار دارد، سیستم سربار پذیرش و timeout درخواستها را میگیرد که میتواند به death-spiral منجر شود اگر سربار از ظرفیت سیستم بیشتر شود. اما اگر منابع ذخیرهٔ درخواست را نداریم بیمعناست. حتی با رعایت این دو، enqueue کردن درخواستی که تا زمان پردازش منقضی میشود فایدهٔ کمی دارد. این به دلیل بعدی timeout میرسد.
Stale data
گاهی داده پنجرهای دارد که باید در آن پردازش شود قبل از دادهٔ مرتبطتر یا قبل از انقضای نیاز به پردازش. اگر فرایند همزمان بیش از این پنجره طول بکشد، timeout و لغو میخواهیم. مثلاً اگر پس از انتظار طولانی dequeue میکند، درخواست یا داده در صف کهنه شده.
اگر پنجره از قبل مشخص است، منطقی است به فرایند همزمان context.Context ساخته با context.WithDeadline یا context.WithTimeout بدهیم. اگر از قبل مشخص نیست، والد باید بتواند فرایند را لغو کند وقتی نیاز به درخواست نیست. context.WithCancel برای این عالی است.
Attempting to prevent deadlocks
در سیستم بزرگ — بهویژه توزیعشده — گاهی جریان داده یا edge caseها را سخت است فهمید. منطقی و حتی توصیهشده است روی همهٔ عملیات همزمان timeout بگذارید تا deadlock نشود. دورهٔ timeout لازم نیست نزدیک زمان واقعی عملیات باشد؛ فقط برای جلوگیری از deadlock است و باید کوتاهتر از زمانی باشد که سیستم deadlockشده در use case شما معقولاً unblock شود.
از بخش «Deadlocks, Livelocks, and Starvation» در صفحهٔ ۱۰ به یاد آورید که timeout برای اجتناب از deadlock ممکن است مشکل را از deadlock به livelock تبدیل کند. اما در سیستمهای بزرگ قطعات بیشتری هست و احتمال timing profile متفاوت از آخرین deadlock بیشتر است. پس ترجیح میدهیم livelock را بپذیریم و بعداً اصلاح کنیم تا deadlock و بازیابی فقط با restart.
این توصیه برای ساخت صحیح سیستم نیست؛ پیشنهاد برای سیستمی tolerant به خطاهای timing است که در توسعه و تست ندیدهاید. timeoutها را نگه دارید، اما هدف سیستمی بدون deadlock است که timeoutها هرگز trigger نشوند.
اکنون که میدانیم چه زمانی timeout، به علل cancellation و ساخت فرایند همزمان برای مدیریت graceful آن میپردازیم. دلایل لغو فرایند همزمان:
Timeouts
timeout لغو ضمنی است.
User intervention
برای UX خوب، معمولاً فرایندهای طولانی را همزمان شروع و وضعیت را در فواصل polling یا query کاربر گزارش میدهیم. برای عملیات user-facing گاهی لازم است کاربر عملیات را لغو کند.
Parent cancellation
هر والد — انسان یا غیره — که متوقف شود، بهعنوان فرزند لغو میشویم.
Replicated requests
شاید داده را به چند فرایند همزمان بفرستیم تا پاسخ سریعتر از یکی بیاید. با اولین پاسخ بقیه را لغو میکنیم. در بخش «Replicated Requests» جزئیات میآید.
احتمالاً دلایل دیگر هم هست. اما «چرا» به سختی «چگونه» جالب است. در فصل ۴ دو راه لغو دیدیم: done channel و context.Context. بخش سخت: وقتی فرایند لغو میشود برای الگوریتم و مصرفکنندگان downstream چه معنایی دارد؟ هنگام نوشتن کد قابلقطع در هر لحظه چه باید در نظر گرفت؟
اول preemptability فرایند همزمان را بررسی کنیم. کد زیر را فرض کنید در goroutine خودش اجرا میشود:
go
var value interface{}
select {
case <-done:
return
case value = <-valueStream:
}
result := reallyLongCalculation(value)
select {
case <-done:
return
case resultStream<-result:
}خواندن از valueStream و نوشتن به resultStream را با بررسی done couple کردهایم، اما مشکل داریم. reallyLongCalculation preemptable بهنظر نمیرسد و شاید خیلی طول بکشد. اگر در حین اجرای آن لغو شود، ممکن است مدت زیادی طول بکشد تا cancellation را بپذیریم و متوقف شویم.
بیایید reallyLongCalculation را preemptable کنیم:
go
reallyLongCalculation := func(
done <-chan interface{},
value interface{},
) interface{} {
intermediateResult := longCalculation(value)
select {
case <-done:
return nil
default:
}
return longCaluclation(intermediateResult)
}پیشرفت کردیم: reallyLongCalculation preemptable است، اما مشکل را نصف کردیم — فقط بین فراخوانیهای طولانی preempt میکنیم. برای حل، longCalculation هم preemptable شود:
go
reallyLongCalculation := func(
done <-chan interface{},
value interface{},
) interface{} {
intermediateResult := longCalculation(done, value)
return longCaluclation(done, intermediateResult)
}اگر این استدلال را به انتها ببریم، دو کار لازم است: دورهای که فرایند preemptable است تعریف کنیم و هر عملیاتی که بیش از این دوره طول میکشد خود preemptable باشد. راه ساده: goroutine را به قطعات کوچکتر بشکنید. همهٔ عملیات atomic غیرقابلقطع باید کمتر از دورهٔ قابلقبول تمام شوند.
مشکل دیگر: اگر goroutine state مشترک — database، فایل، ساختار in-memory — را تغییر دهد و لغو شود چه؟ rollback کار میانی؟ چقدر وقت دارد؟ چیزی گفته متوقف شوید، پس rollback نباید طولانی باشد، درست؟
توصیهٔ عمومی سخت است چون ماهیت الگوریتم تعیینکننده است؛ اما اگر تغییرات state مشترک در scope تنگ باشند یا rollback آسان باشد، cancellation را خوب مدیریت میکنید. اگر ممکن است، نتایج میانی را in-memory بسازید و سپس state را سریع تغییر دهید. مثال روش اشتباه:
go
result := add(1, 2, 3)
writeTallyToState(result)
result = add(result, 4, 5, 6)
writeTallyToState(result)
result = add(result, 7, 8, 9)
writeTallyToState(result)سه بار به state مینویسیم. اگر قبل از نوشتن نهایی لغو شود، باید دو writeTallyToState قبلی را rollback کنیم. در مقابل:
go
result := add(1, 2, 3, 4, 5, 6, 7, 8, 9)
writeTallyToState(result)سطح rollback کوچکتر است. اگر پس از writeToState لغو شود، باز هم rollback لازم است، اما احتمال کمتر است چون فقط یک بار state را تغییر دادیم.
مسئلهٔ دیگر پیامهای تکراری است. pipeline با سه مرحله: generator، stage A و stage B. generator stage A را با زمان از آخرین خواندن از channel رصد میکند و اگر instance فعلی کند شد، A2 جدید میآورد. ممکن است stage B پیام تکراری بگیرد (شکل ۵-۱).
شکل ۵-۱. مثالی از چگونگی رخ دادن پیام تکراری
اگر پیام لغو پس از ارسال نتیجهٔ stage A به stage B برسد، stage B ممکن است duplicate بگیرد.
راههای جلوگیری از ارسال duplicate:
اول یا آخرین نتیجه را بپذیرید
اگر الگوریتم اجازه دهد یا فرایند idempotent باشد، duplicate را در downstream بپذیرید و اولین یا آخرین پیام را انتخاب کنید.
از والد اجازه بگیرید
با ارتباط دوطرفه صریحاً اجازهٔ ارسال بخواهید. شبیه heartbeats است (شکل ۵-۲).
شکل ۵-۲. مثالی از polling والد goroutine
چون صریحاً اجازهٔ نوشتن به channel B میخواهیم، امنتر از heartbeats است؛ اما در عمل کمتر لازم است و پیچیدهتر است. heartbeats عمومیترند، پس heartbeats را پیشنهاد میکنم.
هنگام طراحی فرایندهای همزمان timeout و cancellation را در نظر بگیرید. مثل بسیاری موضوعات مهندسی نرمافزار، نادیده گرفتن از ابتدا و اضافه کردن بعداً شبیه افزودن تخممرغ به کیک پختهشده است.
Heartbeats
Heartbeats راهی برای سیگنال زنده بودن فرایندهای همزمان به طرفهای بیرونی است. نامشان از آناتومی انسان است که heartbeat به ناظر زندگی را نشان میدهد. قبل از Go هم بودهاند و در Go مفید میمانند.
چند دلیل جالب برای کد همزمان: بینش به سیستم و deterministic کردن تست وقتی در غیر این صورت نبود.
دو نوع heartbeat در این بخش:
- Heartbeatهایی که در بازهٔ زمانی رخ میدهند.
- Heartbeatهایی که در ابتدای واحد کار رخ میدهند.
Heartbeatهای بازهای برای کدی مفیدند که شاید منتظر رویداد دیگری برای پردازش واحد کار باشد. چون نمیدانید کار چه زمانی میآید، goroutine ممکن است مدتی منتظر بماند. heartbeat راهی است برای گفتن به listenerها که همهچیز خوب است و سکوت انتظار است.
کد زیر goroutine با heartbeat را نشان میدهد:
go
doWork := func(
done <-chan interface{},
pulseInterval time.Duration,
) (<-chan interface{}, <-chan time.Time) {
heartbeat := make(chan interface{})
results := make(chan time.Time)
go func() {
defer close(heartbeat)
defer close(results)
pulse := time.Tick(pulseInterval)
workGen := time.Tick(2*pulseInterval)
sendPulse := func() {
select {
case heartbeat <-struct{}{}:
default:
}
}
sendResult := func(r time.Time) {
for {
select {
case <-done:
return
case <-pulse:
sendPulse()
case results <- r:
return
}
}
}
for {
select {
case <-done:
return
case <-pulse:
sendPulse()
case r := <-workGen:
sendResult(r)
}
}
}()
return heartbeat, results
}اینجا channel برای ارسال heartbeat میسازیم و از
doWorkبرمیگردانیم.
heartbeat را روی
pulseIntervalتنظیم میکنیم. هرpulseIntervalچیزی برای خواندن روی این channel هست.
ticker دیگر برای شبیهسازی ورود کار. مدت بیشتر از
pulseIntervalتا heartbeatها را ببینیم.
defaultclause داریم. باید همیشه در برابر اینکه کسی heartbeat را گوش ندهد محافظت کنیم. نتایج حیاتیاند، pulseها نیستند.
مثل done channel، در هر send یا receive باید case برای pulse heartbeat هم باشد.
چون ممکن است چند pulse هنگام انتظار برای ورودی یا ارسال نتیجه بفرستیم، همهٔ selectها باید در for loop باشند. نحوهٔ مصرف:
go
done := make(chan interface{})
time.AfterFunc(10*time.Second, func() { close(done) })
const timeout = 2*time.Second
heartbeat, results := doWork(done, timeout/2)
for {
select {
case _, ok := <-heartbeat:
if ok == false {
return
}
fmt.Println("pulse")
case r, ok := <-results:
if ok == false {
return
}
fmt.Printf("results %v\n", r.Second())
case <-time.After(timeout):
return
}
}done channel استاندارد را میسازیم و پس از ۱۰ ثانیه میبندیم تا goroutine وقت کار داشته باشد.
دورهٔ timeout را تنظیم میکنیم و heartbeat interval را به آن couple میکنیم.
timeout/2میدهیم تا heartbeat یک tick اضافه برای پاسخ داشته باشد و timeout خیلی حساس نباشد.
روی heartbeat select میکنیم. بدون نتیجه، حداقل هر
timeout/2از channel heartbeat پیام داریم. اگر نگیریم، مشکلی در goroutine است.
از results channel select میکنیم؛ ساده است.
اگر نه heartbeat نه نتیجهٔ جدید نگیریم timeout میکنیم.
خروجی:
pulse
pulse
results 52
pulse
pulse
results 54
pulse
pulse
results 56
pulse
pulse
results 58
pulseحدود دو pulse به ازای هر نتیجه دریافت میکنیم.
در سیستم سالم، heartbeatهای بازهای خیلی جالب نیستند. شاید برای آمار idle time استفاده شوند، اما وقتی goroutine طبق انتظار رفتار نمیکند واقعاً میدرخشند.
مثال بعدی goroutine نادرست با panic را شبیهسازی میکند — پس از دو iteration متوقف میشود و channelها را نمیبندد:
go
doWork := func(
done <-chan interface{},
pulseInterval time.Duration,
) (<-chan interface{}, <-chan time.Time) {
heartbeat := make(chan interface{})
results := make(chan time.Time)
go func() {
pulse := time.Tick(pulseInterval)
workGen := time.Tick(2*pulseInterval)
sendPulse := func() {
select {
case heartbeat <-struct{}{}:
default:
}
}
sendResult := func(r time.Time) {
for {
select {
case <-pulse:
sendPulse()
case results <- r:
return
}
}
}
for i := 0; i < 2; i++ {
select {
case <-done:
return
case <-pulse:
sendPulse()
case r := <-workGen:
sendResult(r)
}
}
}()
return heartbeat, results
}
done := make(chan interface{})
time.AfterFunc(10*time.Second, func() { close(done) })
const timeout = 2 * time.Second
heartbeat, results := doWork(done, timeout/2)
for {
select {
case _, ok := <-heartbeat:
if ok == false {
return
}
fmt.Println("pulse")
case r, ok := <-results:
if ok == false {
return
}
fmt.Printf("results %v\n", r)
case <-time.After(timeout):
fmt.Println("worker goroutine is not healthy!")
return
}
}panic شبیهسازیشده. بهجای حلقهٔ بینهایت تا توقف، فقط دو بار loop میکنیم.
خروجی:
pulse
pulse
worker goroutine is not healthy!در دو ثانیه سیستم میفهمد goroutine مشکل دارد و حلقهٔ for-select را میشکند. با heartbeat از deadlock اجتناب و deterministic میمانیم بدون اتکا به timeout طولانیتر. در «Healing Unhealthy Goroutines» بیشتر میرویم.
heartbeatها حالت مخالف را هم پوشش میدهند: goroutineهای طولانی زندهاند اما فقط برای تولید مقدار روی values channel طول میکشد.
اکنون heartbeatهای ابتدای واحد کار — بسیار مفید برای تستها:
go
doWork := func(done <-chan interface{}) (<-chan interface{}, <-chan int) {
heartbeatStream := make(chan interface{}, 1)
workStream := make(chan int)
go func () {
defer close(heartbeatStream)
defer close(workStream)
for i := 0; i < 10; i++ {
select {
case heartbeatStream <- struct{}{}:
default:
}
select {
case <-done:
return
case workStream <- rand.Intn(10):
}
}
}()
return heartbeatStream, workStream
}
done := make(chan interface{})
defer close(done)
heartbeat, results := doWork(done)
for {
select {
case _, ok := <-heartbeat:
if ok {
fmt.Println("pulse")
} else {
return
}
case r, ok := <-results:
if ok {
fmt.Printf("results %v\n", r)
} else {
return
}
}
}heartbeat channel با buffer یک ساخته میشود تا حداقل یک pulse حتی اگر کسی بهموقع گوش ندهد ارسال شود.
select جدا برای heartbeat. نمیخواهیم با send روی results در همان select باشد چون اگر receiver آماده نباشد pulse میگیرد و مقدار نتیجه از دست میرود. case برای done نیست چون
defaultاز آن عبور میکند.
باز هم در برابر عدم شنونده محافظت میکنیم. با buffer یک، اگر کسی گوش دهد اما برای اولین pulse دیر برسد، باز هم pulse میبیند.
خروجی: یک pulse به ازای هر نتیجه. این تکنیک در تستها میدرخشد. heartbeatهای بازهای هم قابل استفادهاند، اما اگر فقط شروع کار goroutine مهم است این سبک سادهتر است.
go
func DoWork(
done <-chan interface{},
nums ...int,
) (<-chan interface{}, <-chan int) {
heartbeat := make(chan interface{}, 1)
intStream := make(chan int)
go func() {
defer close(heartbeat)
defer close(intStream)
time.Sleep(2*time.Second)
for _, n := range nums {
select {
case heartbeat <- struct{}{}:
default:
}
select {
case <-done:
return
case intStream <- n:
}
}
}()
return heartbeat, intStream
}تاخیر قبل از شروع کار goroutine شبیهسازی میشود. در عمل nondeterministic است — CPU، disk، شبکه، goblins.
DoWork generator سادهای است که اعداد را به stream تبدیل میکند. تست بد:
go
func TestDoWork_GeneratesAllNumbers(t *testing.T) {
done := make(chan interface{})
defer close(done)
intSlice := []int{0, 1, 2, 3, 5}
_, results := DoWork(done, intSlice...)
for i, expected := range intSlice {
select {
case r := <-results:
if r != expected {
t.Errorf(
"index %v: expected %v, but received %v,",
i,
expected,
r,
)
}
case <-time.After(1 * time.Second):
t.Fatal("test timed out")
}
}
}پس از مدت معقول timeout میکنیم تا goroutine خراب تست را deadlock نکند.
خروجی تست:
go test ./bad_concurrent_test.go
--- FAIL: TestDoWork_GeneratesAllNumbers (1.00s)
bad_concurrent_test.go:46: test timed out
FAIL
FAIL command-line-arguments 1.002sتست nondeterministic است. با time.Sleep همیشه fail میشود؛ بدون آن گاهی pass گاهی fail — بدتر. عوامل خارجی باعث تاخیر iteration اول میشوند. حتی schedule شدن goroutine نگرانی است. نمیتوانیم تضمین کنیم iteration اول قبل از timeout باشد — فکر به احتمالات. timeout را زیاد کنیم، failها کند میشوند. موقعیت بدی است؛ تیم به fail اعتماد نمیکند.
با heartbeat بهراحتی حل میشود. تست deterministic:
go
func TestDoWork_GeneratesAllNumbers(t *testing.T) {
done := make(chan interface{})
defer close(done)
intSlice := []int{0, 1, 2, 3, 5}
heartbeat, results := DoWork(done, intSlice...)
<-heartbeat
i := 0
for r := range results {
if expected := intSlice[i]; r != expected {
t.Errorf("index %v: expected %v, but received %v,", i, expected, r)
}
i++
}
}منتظر سیگنال شروع پردازش iteration میمانیم.
ok command-line-arguments 2.002sبا heartbeat تست بدون timeout مینویسیم. تنها ریسک iteration بیشازحد طولانی است. اگر مهم است heartbeatهای بازهای امنتر استفاده کنید.
تست با heartbeat بازهای:
go
func DoWork(
done <-chan interface{},
pulseInterval time.Duration,
nums ...int,
) (<-chan interface{}, <-chan int) {
heartbeat := make(chan interface{}, 1)
intStream := make(chan int)
go func() {
defer close(heartbeat)
defer close(intStream)
time.Sleep(2*time.Second)
pulse := time.Tick(pulseInterval)
numLoop:
for _, n := range nums {
for {
select {
case <-done:
return
case <-pulse:
select {
case heartbeat <- struct{}{}:
default:
}
case intStream <- n:
continue numLoop
}
}
}
}()
return heartbeat, intStream
}
func TestDoWork_GeneratesAllNumbers(t *testing.T) {
done := make(chan interface{})
defer close(done)
intSlice := []int{0, 1, 2, 3, 5}
const timeout = 2*time.Second
heartbeat, results := DoWork(done, timeout/2, intSlice...)
<-heartbeat
i := 0
for {
select {
case r, ok := <-results:
if ok == false {
return
} else if expected := intSlice[i]; r != expected {
t.Errorf(
"index %v: expected %v, but received %v,",
i,
expected,
r,
)
}
i++
case <-heartbeat:
case <-time.After(timeout):
t.Fatal("test timed out")
}
}
}دو loop لازم است: یکی روی لیست اعداد، این inner loop تا ارسال موفق روی
intStream.
از label برای
continueاز inner loop استفاده میکنیم.
outer loop ادامه مییابد.
منتظر اولین heartbeat برای ورود به loop goroutine.
روی heartbeat هم select میکنیم تا timeout رخ ندهد.
ok command-line-arguments 3.002sاین نسخه کمتر واضح است؛ منطق تست کمی گلآلود. اگر مطمئنید loop goroutine پس از شروع متوقف نمیشود، فقط روی اولین heartbeat block کنید و سپس range ساده. تستهای جدا برای بستن channel، iteration طولانی و مسائل timing بنویسید.
Heartbeatها برای کد همزمان ضروری نیستند، اما utility آنها را نشان دادیم. برای goroutineهای طولانی یا نیازمند تست، این الگو را توصیه میکنم.
Replicated Requests
برای برخی برنامهها دریافت پاسخ هرچه سریعتر اولویت است — مثلاً پاسخ به HTTP request کاربر یا واکشی blob تکراری. میتوانید trade-off کنید: درخواست را به چند handler (goroutine، process یا server) replicate کنید؛ یکی زودتر از بقیه برمیگردد و فوراً نتیجه را میدهید. هزینه: منابع برای نگهداشتن چند نسخهٔ handler.
اگر replication در حافظه باشد شاید گران نباشد، اما اگر process، server یا data center replicate شود پرهزینه میشود. تصمیم: آیا سود ارزش هزینه را دارد؟
replication درون یک process: چند goroutine بهعنوان handler؛ بین یک تا شش نانوثانیه sleep تصادفی برای شبیهسازی بار. مثال با ۱۰ handler:
go
doWork := func(
done <-chan interface{},
id int,
wg *sync.WaitGroup,
result chan<- int,
) {
started := time.Now()
defer wg.Done()
// Simulate random load
simulatedLoadTime := time.Duration(1+rand.Intn(5))*time.Second
select {
case <-done:
case <-time.After(simulatedLoadTime):
}
select {
case <-done:
case result <- id:
}
took := time.Since(started)
// Display how long handlers would have taken
if took < simulatedLoadTime {
took = simulatedLoadTime
}
fmt.Printf("%v took %v\n", id, took)
}
done := make(chan interface{})
result := make(chan int)
var wg sync.WaitGroup
wg.Add(10)
for i:=0; i < 10; i++ {
go doWork(done, i, &wg, result)
}
firstReturned := <-result
close(done)
wg.Wait()
fmt.Printf("Received an answer from #%v\n", firstReturned)۱۰ handler برای درخواستها شروع میشوند.
اولین مقدار برگشتی از گروه گرفته میشود.
بقیهٔ handlerها لغو میشوند تا کار اضافی نکنند.
خروجی:
8 took 1.000211046s
4 took 3s
9 took 2s
1 took 1.000568933s
7 took 2s
3 took 1.000590992s
5 took 5s
0 took 3s
6 took 4s
2 took 2s
Received an answer from #8handler #8 سریعترین بود. زمان هر handler را نشان میدهیم تا صرفهجویی این تکنیک را ببینید. با یک handler و #5 باید پنج ثانیه منتظر میماندید بهجای کمی بیش از یک ثانیه.
محدودیت: همهٔ handlerها باید فرصت برابر سرویسدهی داشته باشند. handlerای که نتواند درخواست را سرویس کند سریعترین زمان را نمیدهد. منابع handlerها هم replicate شوند.
علامت دیگر همان مسئله uniformity است. اگر handlerها خیلی شبیه باشند، outlier کمتر است. فقط به handlerهایی با شرایط runtime متفاوت replicate کنید: process، ماشین، مسیر data store یا data store متفاوت.
گران برای راهاندازی و نگهداری است، اما اگر سرعت هدف است تکنیک ارزشمند است. fault tolerance و scalability را هم طبیعی فراهم میکند.
Rate Limiting
اگر با API سرویس کار کردهاید، احتمالاً با rate limiting مواجه شدهاید — محدود کردن تعداد دسترسی به منبع به عدد محدود در واحد زمان. منبع میتواند هر چیزی باشد: اتصال API، خواندن/نوشتن disk، packet شبکه، خطا.
چرا سرویسها rate limit میگذارند؟ چرا دسترسی نامحدود ندهند؟ پاسخ واضح: با rate limiting از کلاسهایی از بردارهای حمله جلوگیری میکنید. کاربر مخرب با منابع خودش سریع به سیستم دسترسی پیدا کند میتواند کارهای زیادی بکند.
مثلاً disk سرویس را با log یا درخواستهای معتبر پر کند. اگر log rotation اشتباه باشد، فعالیت مخرب و درخواستهای زیاد رکورد را از log به /dev/null بچرخاند. brute-force به منبع، یا DDoS. بدون rate limiting سیستم را بهراحتی امن نمیکنید.
استفادهٔ مخرب تنها دلیل نیست. در سیستم توزیعشده، کاربر مشروع با حجم بالا یا کد buggy عملکرد را برای دیگران خراب میکند و death-spiralهای قبلی را ممکن میکند. از نظر محصول وحشتناک است — معمولاً تضمینی به کاربران دربارهٔ عملکرد پایدار میدهید. اگر یک کاربر آن را بشکند وضعیت بد است. مدل ذهنی کاربر معمولاً sandboxed بودن دسترسی است — نه تأثیر و نه تأثر از فعالیت دیگران. شکستن این مدل سیستم را ناکارآمد نشان میدهد.
حتی با یک کاربر، rate limit مفید است. سیستمها برای use case رایج خوب کار میکنند اما تحت شرایط متفاوت رفتار عوض میشود. در سیستم توزیعشده این cascade میکند — packet drop، از دست رفتن quorum database، شکست درخواستها… سیستم گاهی DDoS علیه خودش میکند!
داستانی از میدان
روی سیستم توزیعشدهای کار میکردیم که با process جدید موازی scale میکرد (افقی روی چند ماشین). هر process اتصال database باز میکرد، داده میخواند و محاسبه میکرد. مدتی موفق بودیم. بعد utilization رشد کرد و خواندنها timeout میشدند.
DBAها log را بررسی کردند. چون rate limit نبود، processها روی هم میپریدند. disk contention به ۱۰۰٪ میرسید و میماند. timeout-retry حلقهای sadistic. jobها تمام نمیشدند.
محدودیت تعداد اتصال database و rate limit بر bit بر ثانیه خواندن گذاشتیم و مشکل رفت. مشتریان بیشتر منتظر ماندند اما job تمام شد و capacity planning ساختاریافته شد.
Rate limit به شما امکان استدلال دربارهٔ عملکرد و پایداری با جلوگیری از خروج از مرزهای بررسیشده را میدهد. با تست و قهوه مرزها را کنترلشده گسترش دهید.
در مدل پولی، rate limit رابطهٔ سالم با مشتریان را نگه میدارد — امتحان با محدودیت شدید (مثل Google Cloud). برای مشتریان پولی، از باگ runaway در دسترسی برنامهای جلوگیری میکند — اشتباه پرهزینه و تصمیم سخت: صاحب سرویس هزینه را میخورد یا مشتری پرداخت میکند و رابطه خراب میشود.
Rate limit اغلب از دید سازندهٔ منبع دیده میشود، اما کاربران هم میتوانند استفاده کنند. هنگام یادگیری API، محدود کردن شدید rate آرامش میدهد که پای خود را نشانید.
امیدوارم دلیل کافی دادم که rate limit خوب است حتی اگر فکر میکنید هرگز به حد نمیرسید. سادهاند و مشکلات زیادی حل میکنند.
پیادهسازی در Go؟ بیشتر با الگوریتم token bucket. تئوری:
برای استفاده از منبع باید access token داشته باشید. بدون token درخواست رد میشود. tokenها در سطل (bucket) منتظرند. عمق سطل d — حداکثر d token. مثلاً عمق ۵ یعنی پنج token.
هر بار به منبع دسترسی میخواهید، از سطل یک token برمیدارید. پنج token، پنج بار موفق؛ بار ششم token نیست — صف یا رد.
جدول زمان (درخواستها آنی فرض میشوند):
| time | bucket | request |
|---|---|---|
| 0 | 5 | tok |
| 0 | 4 | tok |
| 0 | 3 | tok |
| 0 | 2 | tok |
| 0 | 1 | tok |
| 0 | 0 | |
| 1 | 0 | |
| 0 |
پنج درخواست قبل از ثانیهٔ اول، سپس block.
تکمیل token: نرخ r افزودن token به سطل — یک در نانوثانیه یا یک در دقیقه. این همان rate limit است — تا token جدید صبر میکنیم.
سطل عمق ۱، نرخ ۱ token/ثانیه:
| time | bucket | request |
|---|---|---|
| 0 | 1 | |
| 0 | 0 | tok |
| 1 | 0 | |
| 2 | 1 | |
| 2 | 0 | tok |
| 3 | 0 | |
| 4 | 1 | |
| 4 | 0 | tok |
فوراً یک درخواست، سپس یک درخواست هر دو ثانیه.
دو پارامتر: d (عمق — tokenهای فوری) و r (نرخ تکمیل). burstiness یعنی چند درخواست وقتی سطل پر است.
سطل عمق ۵، نرخ ۰.۵ token/ثانیه:
| time | bucket | request |
|---|---|---|
| 0 | 5 | |
| 0 | 4 | tok |
| 0 | 3 | tok |
| 0 | 2 | tok |
| 0 | 1 | tok |
| 0 | 0 | tok |
| 1 | 0 (0.5) | |
| 2 | 1 | |
| 2 | 0 | tok |
| 3 | 0 (0.5) | |
| 4 | 1 | |
| 4 | 0 | tok |
پنج درخواست فوری، سپس هر دو ثانیه یکی. burst در ابتدا.
کاربر ممکن است کل سطل را یکجا نخورد — d فقط ظرفیت است. burst دو، چهار ثانیه بعد burst پنج:
| time | bucket | request |
|---|---|---|
| 0 | 5 | |
| 0 | 4 | tok |
| 0 | 3 | tok |
| 1 | 3 | |
| 2 | 4 | |
| 3 | 5 | |
| 4 | 5 | |
| 5 | 4 | tok |
| 5 | 3 | tok |
| 5 | 2 | tok |
| 5 | 1 | tok |
| 5 | 0 | tok |
با token موجود، دسترسی فقط با قابلیت caller محدود میشود. برای کاربران متناوب که round-trip سریع میخواهند burst خوب است. یا سیستم همه burst همزمان را تحمل کند یا آماری بعید باشد همه همزمان burst کنند. rate limit ریسک محاسبهشده میدهد.
الگوریتم را در Go بهکار ببریم. API با دو endpoint: خواندن فایل و resolve دامنه به IP. client ساده:
go
func Open() *APIConnection {
return &APIConnection{}
}
type APIConnection struct {}
func (a *APIConnection) ReadFile(ctx context.Context) error {
// Pretend we do work here
return nil
}
func (a *APIConnection) ResolveAddress(ctx context.Context) error {
// Pretend we do work here
return nil
}چون درخواست روی wire است، context.Context بهعنوان آرگومان اول برای لغو یا pass value به server میگیریم.
driver ساده: ۱۰ فایل و ۱۰ آدرس — بدون رابطه، concurrent. بعداً rate limiter را تحت فشار میگذارد:
go
func main() {
defer log.Printf("Done.")
log.SetOutput(os.Stdout)
log.SetFlags(log.Ltime | log.LUTC)
apiConnection := Open()
var wg sync.WaitGroup
wg.Add(20)
for i := 0; i < 10; i++ {
go func() {
defer wg.Done()
err := apiConnection.ReadFile(context.Background())
if err != nil {
log.Printf("cannot ReadFile: %v", err)
}
log.Printf("ReadFile")
}()
}
for i := 0; i < 10; i++ {
go func() {
defer wg.Done()
err := apiConnection.ResolveAddress(context.Background())
if err != nil {
log.Printf("cannot ResolveAddress: %v", err)
}
log.Printf("ResolveAddress")
}()
}
wg.Wait()
}خروجی — همه تقریباً همزمان:
20:13:13 ResolveAddress
20:13:13 ReadFile
...
20:13:13 Done.بدون rate limit. باگ infinite loop در driver بدون rate limit میتواند فاجعهبار باشد.
rate limiter در APIConnection — معمولاً روی server است تا bypass نشود. client-side هم برای جلوگیری از callهای بیهوده optimization است؛ اینجا client-side ساده نگه میداریم.
از golang.org/x/time/rate — نزدیکترین به standard library.
Limit و NewLimiter:
go
// Limit defines the maximum frequency of some events. Limit is
// represented as number of events per second. A zero Limit allows no
// events.
type Limit float64
// NewLimiter returns a new Limiter that allows events up to rate r
// and permits bursts of at most b tokens.
func NewLimiter(r Limit, b int) *Limiterدر NewLimiter، r نرخ و b عمق سطل است.
Every برای تبدیل time.Duration به Limit:
go
// Every converts a minimum time interval between events to a Limit.
func Every(interval time.Duration) Limitترجیح میدهیم عملیات per time period بیان شود:
go
rate.Limit(events/timePeriod.Seconds())اما helper با Every — اگر interval صفر باشد rate.Inf (بدون limit):
go
func Per(eventCount int, duration time.Duration) rate.Limit {
return rate.Every(duration/time.Duration(eventCount))
}پس از rate.Limiter، Wait تا token:
go
// Wait is shorthand for WaitN(ctx, 1).
func (lim *Limiter) Wait(ctx context.Context)
// WaitN blocks until lim permits n events to happen.
// It returns an error if n exceeds the Limiter's burst size, the Context is
// canceled, or the expected wait time exceeds the Context's Deadline.
func (lim *Limiter) WaitN(ctx context.Context, n int) (err error)APIConnection را تغییر میدهیم:
go
func Open() *APIConnection {
return &APIConnection{
rateLimiter: rate.NewLimiter(rate.Limit(1), 1),
}
}
type APIConnection struct {
rateLimiter *rate.Limiter
}
func (a *APIConnection) ReadFile(ctx context.Context) error {
if err := a.rateLimiter.Wait(ctx); err != nil {
return err
}
// Pretend we do work here
return nil
}
func (a *APIConnection) ResolveAddress(ctx context.Context) error {
if err := a.rateLimiter.Wait(ctx); err != nil {
return err
}
// Pretend we do work here
return nil
}rate limit همهٔ اتصالات API: یک event در ثانیه.
منتظر token کافی در rate limiter.
خروجی — یک درخواست در ثانیه:
22:08:30 ResolveAddress
22:08:31 ReadFile
...
22:08:49 Done.در production چند لایه limit میخواهیم: per second و per minute/hour/day. گاهی یک limiter کافی است، گاهی نه؛ ادغام در یک لایه intent را از دست میدهد. limiterها جدا و multiLimiter برای ترکیب:
go
type RateLimiter interface {
Wait(context.Context) error
Limit() rate.Limit
}
func MultiLimiter(limiters ...RateLimiter) *multiLimiter {
byLimit := func(i, j int) bool {
return limiters[i].Limit() < limiters[j].Limit()
}
sort.Slice(limiters, byLimit)
return &multiLimiter{limiters: limiters}
}
type multiLimiter struct {
limiters []RateLimiter
}
func (l *multiLimiter) Wait(ctx context.Context) error {
for _, l := range l.limiters {
if err := l.Wait(ctx); err != nil {
return err
}
}
return nil
}
func (l *multiLimiter) Limit() rate.Limit {
return l.limiters[0].Limit()
}
RateLimiterinterface تاMultiLimiterبتواندMultiLimiterدیگر داشته باشد.
مرتبسازی بر اساس
Limit().
محدودترین limit اولین عنصر slice است.
Wait روی هر child صدا زده میشود — ممکن است block کند؛ باید هر limiter از درخواست مطلع شود. با انتظار برای هر کدام، دقیقاً به اندازهٔ طولانیترین انتظار میمانیم. waitهای کوتاهتر بخشی از طولانیترین بودند؛ wait طولانی باقیمانده محاسبه میشود چون در waitهای قبلی bucketهای بعدی پر شدهاند؛ waitهای بعدی فوری برمیگردند.
limit per second و per minute:
go
func Open() *APIConnection {
secondLimit := rate.NewLimiter(Per(2, time.Second), 1)
minuteLimit := rate.NewLimiter(Per(10, time.Minute), 10)
return &APIConnection{
rateLimiter: MultiLimiter(secondLimit, minuteLimit),
}
}
type APIConnection struct {
rateLimiter RateLimiter
}
func (a *APIConnection) ReadFile(ctx context.Context) error {
if err := a.rateLimiter.Wait(ctx); err != nil {
return err
}
// Pretend we do work here
return nil
}
func (a *APIConnection) ResolveAddress(ctx context.Context) error {
if err := a.rateLimiter.Wait(ctx); err != nil {
return err
}
// Pretend we do work here
return nil
}limit per second بدون burst.
limit per minute با burst ۱۰ برای pool اولیه. per second از overload جلوگیری میکند.
دو limit ترکیب و master limiter
APIConnection.
خروجی — دو درخواست در ثانیه تا #۱۱، سپس هر شش ثانیه (pool per-minute خالی شد):
22:46:10 ResolveAddress
22:46:10 ReadFile
...
22:47:10 Done.درخواست #۱۱ پس از دو ثانیه نه شش — پنجرهٔ sliding per-minute، limiter یک token دیگر جمع کرده.
این تکنیک limitهای coarse-grained را صریح بیان میکند و درخواستها را در سطح ریز محدود میکند.
بعد از زمان، dimensionهای دیگر: disk، شبکه:
go
func Open() *APIConnection {
return &APIConnection{
apiLimit: MultiLimiter(
rate.NewLimiter(Per(2, time.Second), 2),
rate.NewLimiter(Per(10, time.Minute), 10),
),
diskLimit: MultiLimiter(
rate.NewLimiter(rate.Limit(1), 1),
),
networkLimit: MultiLimiter(
rate.NewLimiter(Per(3, time.Second), 3),
),
}
}
type APIConnection struct {
networkLimit,
diskLimit,
apiLimit RateLimiter
}
func (a *APIConnection) ReadFile(ctx context.Context) error {
err := MultiLimiter(a.apiLimit, a.diskLimit).Wait(ctx)
if err != nil {
return err
}
// Pretend we do work here
return nil
}
func (a *APIConnection) ResolveAddress(ctx context.Context) error {
err := MultiLimiter(a.apiLimit, a.networkLimit).Wait(ctx)
if err != nil {
return err
}
// Pretend we do work here
return nil
}limit API: per second و per minute.
disk: یک خواندن در ثانیه.
شبکه: سه درخواست در ثانیه.
ReadFile: api + disk.
ResolveAddress: api + network.
خروجی نشان میدهد limiterهای منطقی ترکیب شدهاند و APIClient درست عمل میکند. callهای شبکه منظمتر و در دو سوم اول تمام میشوند — احتمالاً rate limiterها کار میکنند.
rate.Limiter قابلیتهای دیگر هم دارد؛ اینجا فقط انتظار برای token را دیدیم.
در این بخش justification، الگوریتم token bucket، پیادهسازی Go و ترکیب limiterها را دیدیم — نمای کلی برای شروع در میدان.
Healing Unhealthy Goroutines
در فرایندهای long-lived مثل daemon، مجموعهای از goroutineهای long-lived رایج است — معمولاً block و منتظر داده، کار، pass به جلو. گاهی به منبعی وابستهاند که کنترل کمی دارید — web service، فایل ephemeral. goroutine بهراحتی در حالت بد گیر میکند و بدون کمک خارجی recover نمیشود. با جداسازی concerns، شاید healing از concern کار goroutine نباشد. در فرایند long-running مکانیزمی که goroutineها سالم بمانند و در صورت unhealthy restart شوند مفید است — «healing».²
برای healing از الگوی heartbeat برای liveliness استفاده میکنیم. نوع heartbeat به monitoring بستگی دارد؛ اگر livelock ممکن است، heartbeat باید نشان دهد goroutine نه فقط up بلکه کار مفید میکند. اینجا فقط live یا dead.
منطق monitoring را steward و goroutine تحت نظارت را ward مینامیم. steward restart ward را هم بر عهده دارد — به تابعی برای شروع goroutine نیاز دارد:
go
type startGoroutineFn func(
done <-chan interface{},
pulseInterval time.Duration,
) (heartbeat <-chan interface{})
newSteward := func(
timeout time.Duration,
startGoroutine startGoroutineFn,
) startGoroutineFn {
return func(
done <-chan interface{},
pulseInterval time.Duration,
) (<-chan interface{}) {
heartbeat := make(chan interface{})
go func() {
defer close(heartbeat)
var wardDone chan interface{}
var wardHeartbeat <-chan interface{}
startWard := func() {
wardDone = make(chan interface{})
wardHeartbeat = startGoroutine(or(wardDone, done), timeout/2)
}
startWard()
pulse := time.Tick(pulseInterval)
monitorLoop:
for {
timeoutSignal := time.After(timeout)
for {
select {
case <-pulse:
select {
case heartbeat <- struct{}{}:
default:
}
case <-wardHeartbeat:
continue monitorLoop
case <-timeoutSignal:
log.Println("steward: ward unhealthy; restarting")
close(wardDone)
startWard()
continue monitorLoop
case <-done:
return
}
}
}
}()
return heartbeat
}
}امضای goroutine قابلmonitor و restart —
done،pulseInterval،heartbeat.
steward
timeoutوstartGoroutineمیگیرد و خودstartGoroutineFnبرمیگرداند — steward هم monitorable است.
closure برای شروع یکنواخت goroutine تحت نظارت.
channel جدید برای سیگنال توقف ward.
ward شروع میشود؛ halt اگر steward یا steward بخواهد ward را متوقف کند —
or(wardDone, done).pulseIntervalنصف timeout (قابل tweak طبق Heartbeats).
inner loop برای pulseهای steward.
pulse ward → ادامهٔ monitoring.
بدون pulse ward در timeout → halt ward و ward جدید → ادامهٔ monitoring.
for loop شلوغ اما با آشنایی با الگوها خوانا است. تست با ward misbehaving:
go
log.SetOutput(os.Stdout)
log.SetFlags(log.Ltime | log.LUTC)
doWork := func(done <-chan interface{}, _ time.Duration) <-chan interface{} {
log.Println("ward: Hello, I'm irresponsible!")
go func() {
<-done
log.Println("ward: I am halting.")
}()
return nil
}
doWorkWithSteward := newSteward(4*time.Second, doWork)
done := make(chan interface{})
time.AfterFunc(9*time.Second, func() {
log.Println("main: halting steward and ward.")
close(done)
})
for range doWorkWithSteward(done, 4*time.Second) {}
log.Println("Done")goroutine فقط منتظر cancel — بدون pulse.
steward با timeout چهار ثانیه برای
doWork.
پس از نه ثانیه halt.
steward شروع و range روی pulseها.
خروجی:
18:28:07 ward: Hello, I'm irresponsible!
18:28:11 steward: ward unhealthy; restarting
18:28:11 ward: Hello, I'm irresponsible!
18:28:11 ward: I am halting.
18:28:15 steward: ward unhealthy; restarting
18:28:15 ward: Hello, I'm irresponsible!
18:28:15 ward: I am halting.
18:28:16 main: halting steward and ward.
18:28:16 ward: I am halting.
18:28:16 Doneward ساده است — فقط cancellation و heartbeat. ward پیچیدهتر با closure:
go
doWorkFn := func(
done <-chan interface{},
intList ...int,
) (startGoroutineFn, <-chan interface{}) {
intChanStream := make(chan (<-chan interface{}))
intStream := bridge(done, intChanStream)
doWork := func(
done <-chan interface{},
pulseInterval time.Duration,
) <-chan interface{} {
intStream := make(chan interface{})
heartbeat := make(chan interface{})
go func() {
defer close(intStream)
select {
case intChanStream <- intStream:
case <-done:
return
}
pulse := time.Tick(pulseInterval)
for {
valueLoop:
for _, intVal := range intList {
if intVal < 0 {
log.Printf("negative value: %v\n", intVal)
return
}
for {
select {
case <-pulse:
select {
case heartbeat <- struct{}{}:
default:
}
case intStream <- intVal:
continue valueLoop
case <-done:
return
}
}
}
}
}()
return heartbeat
}
return doWork, intStream
}مقادیر برای closure ward و channelهای ارتباط برگشتی.
channel of channels برای bridge pattern.
closure که steward شروع و monitor میکند.
channel ارتباط در این instance ward.
bridge از channel جدید مطلع میشود.
ward unhealthy با log و return روی عدد منفی.
چند نسخه ward ممکن است — bridge channel (بخش «The bridge-channel» صفحهٔ ۱۲۲) یک channel بدون وقفه به consumer میدهد. wardها با ترکیب الگوها پیچیده میشوند:
go
log.SetFlags(log.Ltime | log.LUTC)
log.SetOutput(os.Stdout)
done := make(chan interface{})
defer close(done)
doWork, intStream := doWorkFn(done, 1, 2, -1, 3, 4, 5)
doWorkWithSteward := newSteward(1*time.Millisecond, doWork)
doWorkWithSteward(done, 1*time.Hour)
for intVal := range take(done, intStream, 6) {
fmt.Printf("Received: %v\n", intVal)
}ward با slice اعداد variadic و stream برگشتی.
steward با monitoring یک میلیثانیه — انتظار fail سریع.
steward ward را شروع و monitor میکند.
take— شش مقدار اول ازintStream.
خروجی:
Received: 1
23:25:33 negative value: -1
Received: 2
23:25:33 steward: ward unhealthy; restarting
Received: 1
23:25:33 negative value: -1
Received: 2
...خطاهای ward و restart steward. فقط ۱ و ۲ — ward هر بار از صفر شروع میکند. اگر سیستم به duplicate حساس است در نظر بگیرید. steward پس از تعداد fail مشخص exit کند. generator stateful با بهروزرسانی intList در هر iteration:
بهجای:
go
valueLoop:
for _, intVal := range intList {
// ...
}میتوان:
go
valueLoop:
for {
intVal := intList[0]
intList = intList[1:]
// ...
}جای restart حفظ میشود اما روی عدد منفی گیر میکنیم و ward fail میکند.
این الگو goroutineهای long-lived را up و healthy نگه میدارد.
² آشنایان Erlang این مفهوم را میشناسند — supervisorهای Erlang همین کار را میکنند.
Summary
در این فصل راههایی برای پایداری و قابلفهم ماندن سیستمها با رشد problem domain و سیستمهای بزرگ و شاید توزیعشده را پوشش دادیم. همچنین نشان دادیم primitiveهای همزمانی Go با abstractionهای سطح بالاتر چگونه scale میکنند. بدون زبان طراحیشده حول همزمانی، این الگوها احتمالاً خیلی سختتر و کمتر robust بودند.
در فصل پایانی internals بخشی از runtime Go را برای درک عمیقتر بررسی میکنیم و ابزارهای مفید برای توسعه و debug نرمافزار Go را میبینیم.