WaitGroup ב-Golang: Add, Done, Wait ו-worker pool

הדפוס הבסיסי

sync.WaitGroup סופר goroutines שרצות. Add מעלה את הספירה, Done מורידה אותה, ו-Wait נחסם עד שהיא אפס.

package main

import (
	"fmt"
	"sync"
	"time"
)

func download(name string) int {
	time.Sleep(10 * time.Millisecond) // pretend to work
	return len(name) * 100
}

func main() {
	files := []string{"a.txt", "report.pdf", "photo.jpg"}
	sizes := make([]int, len(files))

	var wg sync.WaitGroup
	for i, f := range files {
		wg.Add(1) // before the go statement
		go func() {
			defer wg.Done() // runs even if download panics
			sizes[i] = download(f)
		}()
	}
	wg.Wait()

	for i, f := range files {
		fmt.Printf("%-11s %d bytes\n", f, sizes[i])
	}
}

שלוש ההורדות רצות במקביל, כך שהתוכנית לוקחת בערך 10 ms במקום 30. התוצאות מודפסות לפי סדר הקלט כי כל goroutine כותבת רק לאינדקס שלה ב-sizes, ו-main קוראת אותן רק אחרי Wait.

ערך האפס של WaitGroup מוכן לשימוש. אין בנאי.

שלושת הכללים

קראו ל-Add לפני go, לא בתוך ה-goroutine. אם ה-goroutine קוראת ל-Add בעצמה, main יכולה להגיע ל-Wait לפני שאף goroutine התחילה, לראות ספירה של אפס ולחזור בזמן שהעבודה עוד לא התחילה. כשהספירה ידועה מראש, wg.Add(len(files)) פעם אחת לפני הלולאה שקול לזה.

קראו ל-Done עם defer כשורה הראשונה של ה-goroutine. goroutine שחוזרת מוקדם בגלל שגיאה, או גורמת ל-panic, עדיין מקטינה את המונה. Done חסר משאיר את Wait חסום לנצח. אם זו ה-goroutine היחידה שנשארה, סביבת הריצה מדווחת fatal error: all goroutines are asleep עם sync.WaitGroup.Wait ב-trace.

לעולם אל תעתיקו WaitGroup אחרי השימוש הראשון. העבירו *sync.WaitGroup לפונקציות, או לכדו את המשתנה ב-closure כמו למעלה.

העברת WaitGroup לפונקציה

כשגוף ה-goroutine הוא פונקציה עם שם, העבירו מצביע:

package main

import (
	"fmt"
	"sync"
)

func worker(id int, wg *sync.WaitGroup, out []string) {
	defer wg.Done()
	out[id] = fmt.Sprintf("worker %d done", id)
}

func main() {
	var wg sync.WaitGroup
	out := make([]string, 3)

	for id := range 3 {
		wg.Add(1)
		go worker(id, &wg, out)
	}
	wg.Wait()

	for _, line := range out {
		fmt.Println(line)
	}
}

עם wg sync.WaitGroup כפרמטר לפי ערך, כל worker היה קורא ל-Done על העותק שלו ו-main הייתה נחסמת ב-Wait לנצח. go vet תופס את זה לפני שמריצים משהו:

./main.go:8:24: worker passes lock by value: sync.WaitGroup contains sync.noCopy

תכנון נקי יותר משאיר את המקביליות מחוץ ל-worker לגמרי: תנו לה להיות פונקציה רגילה ובצעו את הניהול של Add/Done ב-closure של הקורא. אז קל לבדוק את worker ולקרוא לה באופן סינכרוני.

מונה שלילי

Done היא Add(-1). אם הספירה יורדת מתחת לאפס, התוכנית גורמת ל-panic:

package main

import (
	"fmt"
	"sync"
)

func main() {
	defer func() {
		fmt.Println("recovered:", recover())
	}()

	var wg sync.WaitGroup
	wg.Add(1)
	wg.Done()
	wg.Done() // one Done too many
}

הפלט הוא recovered: sync: negative WaitGroup counter. הסיבה הרגילה היא goroutine עם defer wg.Done() שקוראת גם ל-wg.Done() במפורש באחד המסלולים.

איסוף שגיאות

WaitGroup רק סופר. לשגיאות, תנו לכל goroutine מקום משלה ובדקו אותן אחרי Wait:

package main

import (
	"errors"
	"fmt"
	"sync"
)

func check(host string) error {
	if host == "db" {
		return errors.New("db: connection refused")
	}
	return nil
}

func main() {
	hosts := []string{"web", "db", "cache"}
	errs := make([]error, len(hosts))

	var wg sync.WaitGroup
	for i, h := range hosts {
		wg.Add(1)
		go func() {
			defer wg.Done()
			errs[i] = check(h)
		}()
	}
	wg.Wait()

	if err := errors.Join(errs...); err != nil {
		fmt.Println("health check failed:", err)
		return
	}
	fmt.Println("all healthy")
}

errors.Join (Go 1.20) מדלגת על ערכי nil ומחזירה nil אם כולם nil, כך שהיא משלבת "שגיאה אחת לכל goroutine" בלי שום ניהול נוסף.

אם רוצים לעצור את שאר העבודה ברגע ש-goroutine אחת נכשלת, השתמשו במקום זה ב-golang.org/x/sync/errgroup. הוא WaitGroup ועוד השגיאה הראשונה ועוד context שמבוטל בכישלון, ו-g.SetLimit(n) מגביל את המקביליות. הוא נמצא מחוץ לספרייה הסטנדרטית, ולכן לא יכול לרוץ בעורך של הדף הזה:

g, ctx := errgroup.WithContext(ctx)
for _, h := range hosts {
	g.Go(func() error { return checkCtx(ctx, h) })
}
if err := g.Wait(); err != nil {
	return err // the first error; ctx was cancelled for the others
}

Worker pool

מספר קבוע של goroutines שקוראות משימות מ-channel שומר על מקביליות מוגבלת, לא משנה כמה משימות יש. ה-WaitGroup אומר מתי כל ה-workers סיימו, וזה הרגע שבו אפשר לסגור את ה-channel של התוצאות.

package main

import (
	"fmt"
	"sort"
	"sync"
)

type result struct {
	job, value, worker int
}

func main() {
	jobs := make(chan int)
	results := make(chan result)

	const workers = 3
	var wg sync.WaitGroup
	for w := 1; w <= workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := range jobs { // ends when jobs is closed
				results <- result{job: j, value: j * j, worker: w}
			}
		}()
	}

	// Close results once every worker has exited.
	go func() {
		wg.Wait()
		close(results)
	}()

	// Feed the jobs, then close the channel so workers stop.
	go func() {
		for j := 1; j <= 8; j++ {
			jobs <- j
		}
		close(jobs)
	}()

	var all []result
	for r := range results {
		all = append(all, r)
	}
	sort.Slice(all, func(a, b int) bool { return all[a].job < all[b].job })

	for _, r := range all {
		fmt.Printf("job %d -> %d\n", r.job, r.value)
	}
	fmt.Println("jobs done:", len(all))
}

הסדר של שלושת החלקים חשוב:

• main חייבת לקבל תוצאות בזמן שה-workers רצים. אם main הייתה קוראת ל-wg.Wait() ישירות לפני הקריאה, ה-workers היו נחסמים בשליחה ל-results, אף פעם לא היו מגיעים ל-Done, והכול היה נתקע ב-deadlock. לכן Wait רץ ב-goroutine משלו.
• close(results) קורה רק אחרי Wait, כך שאף worker לא יכול לשלוח ל-channel סגור.
• מזין המשימות רץ גם הוא ב-goroutine, כך שההזנה והאיסוף חופפים.

איזה worker טיפל באיזו משימה משתנה מהרצה להרצה, ולכן התוכנית ממיינת לפי משימה לפני ההדפסה. כל מה שהיא מדפיסה דטרמיניסטי.

WaitGroup, channel או errgroup

Go 1.25 מוסיפה את wg.Go(func() { ... }), שעושה בשבילכם את ה-Add(1) ואת ה-Done שב-defer. קוד ל-Go 1.24 ולפני כן, כולל העורך של הדף הזה, משתמש בצורה המפורשת שמוצגת למעלה.

טעויות נפוצות

• wg.Add(1) בתוך ה-goroutine. Wait יכול לחזור לפני שהיא רצה.
• שכחה של Done בחזרה מוקדמת. תמיד defer wg.Done().
• העברת ה-WaitGroup לפי ערך. השתמשו במצביע; go vet מסמן את ההעתקה.
• המתנה באותה goroutine שצריכה לרוקן channel. העבירו את wg.Wait() ואת ה-close ל-goroutine נפרדת.
• שימוש חוזר ב-WaitGroup לפני שה-Wait הקודם חזר. התחילו מחזור חדש של קריאות Add רק אחרי ש-Wait הסתיים.