src

errgroup.go (4033B)

---

 // Copyright 2016 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
 // Package errgroup provides synchronization, error propagation, and Context
 // cancellation for groups of goroutines working on subtasks of a common task.
 //
 // [errgroup.Group] is related to [sync.WaitGroup] but adds handling of tasks
 // returning errors.
 package errgroup
 
 import (
 	"context"
 	"fmt"
 	"sync"
 )
 
 type token struct{}
 
 // A Group is a collection of goroutines working on subtasks that are part of
 // the same overall task. A Group should not be reused for different tasks.
 //
 // A zero Group is valid, has no limit on the number of active goroutines,
 // and does not cancel on error.
 type Group struct {
 	cancel func(error)
 
 	wg sync.WaitGroup
 
 	sem chan token
 
 	errOnce sync.Once
 	err     error
 }
 
 func (g *Group) done() {
 	if g.sem != nil {
 		<-g.sem
 	}
 	g.wg.Done()
 }
 
 // WithContext returns a new Group and an associated Context derived from ctx.
 //
 // The derived Context is canceled the first time a function passed to Go
 // returns a non-nil error or the first time Wait returns, whichever occurs
 // first.
 func WithContext(ctx context.Context) (*Group, context.Context) {
 	ctx, cancel := context.WithCancelCause(ctx)
 	return &Group{cancel: cancel}, ctx
 }
 
 // Wait blocks until all function calls from the Go method have returned, then
 // returns the first non-nil error (if any) from them.
 func (g *Group) Wait() error {
 	g.wg.Wait()
 	if g.cancel != nil {
 		g.cancel(g.err)
 	}
 	return g.err
 }
 
 // Go calls the given function in a new goroutine.
 //
 // The first call to Go must happen before a Wait.
 // It blocks until the new goroutine can be added without the number of
 // goroutines in the group exceeding the configured limit.
 //
 // The first goroutine in the group that returns a non-nil error will
 // cancel the associated Context, if any. The error will be returned
 // by Wait.
 func (g *Group) Go(f func() error) {
 	if g.sem != nil {
 		g.sem <- token{}
 	}
 
 	g.wg.Add(1)
 	go func() {
 		defer g.done()
 
 		// It is tempting to propagate panics from f()
 		// up to the goroutine that calls Wait, but
 		// it creates more problems than it solves:
 		// - it delays panics arbitrarily,
 		//   making bugs harder to detect;
 		// - it turns f's panic stack into a mere value,
 		//   hiding it from crash-monitoring tools;
 		// - it risks deadlocks that hide the panic entirely,
 		//   if f's panic leaves the program in a state
 		//   that prevents the Wait call from being reached.
 		// See #53757, #74275, #74304, #74306.
 
 		if err := f(); err != nil {
 			g.errOnce.Do(func() {
 				g.err = err
 				if g.cancel != nil {
 					g.cancel(g.err)
 				}
 			})
 		}
 	}()
 }
 
 // TryGo calls the given function in a new goroutine only if the number of
 // active goroutines in the group is currently below the configured limit.
 //
 // The return value reports whether the goroutine was started.
 func (g *Group) TryGo(f func() error) bool {
 	if g.sem != nil {
 		select {
 		case g.sem <- token{}:
 			// Note: this allows barging if and only if channels in general allow barging.
 		default:
 			return false
 		}
 	}
 
 	g.wg.Add(1)
 	go func() {
 		defer g.done()
 
 		if err := f(); err != nil {
 			g.errOnce.Do(func() {
 				g.err = err
 				if g.cancel != nil {
 					g.cancel(g.err)
 				}
 			})
 		}
 	}()
 	return true
 }
 
 // SetLimit limits the number of active goroutines in this group to at most n.
 // A negative value indicates no limit.
 // A limit of zero will prevent any new goroutines from being added.
 //
 // Any subsequent call to the Go method will block until it can add an active
 // goroutine without exceeding the configured limit.
 //
 // The limit must not be modified while any goroutines in the group are active.
 func (g *Group) SetLimit(n int) {
 	if n < 0 {
 		g.sem = nil
 		return
 	}
 	if active := len(g.sem); active != 0 {
 		panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", active))
 	}
 	g.sem = make(chan token, n)
 }