errgroup.go

     1  // Copyright 2016 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  // Package errgroup provides synchronization, error propagation, and Context
     6  // cancelation for groups of goroutines working on subtasks of a common task.
     7  //
     8  // [errgroup.Group] is related to [sync.WaitGroup] but adds handling of tasks
     9  // returning errors.
    10  package errgroup
    11  
    12  import (
    13  	"context"
    14  	"fmt"
    15  	"sync"
    16  )
    17  
    18  type token struct{}
    19  
    20  // A Group is a collection of goroutines working on subtasks that are part of
    21  // the same overall task.
    22  //
    23  // A zero Group is valid, has no limit on the number of active goroutines,
    24  // and does not cancel on error.
    25  type Group struct {
    26  	cancel func(error)
    27  
    28  	wg sync.WaitGroup
    29  
    30  	sem chan token
    31  
    32  	errOnce sync.Once
    33  	err     error
    34  }
    35  
    36  func (g *Group) done() {
    37  	if g.sem != nil {
    38  		<-g.sem
    39  	}
    40  	g.wg.Done()
    41  }
    42  
    43  // WithContext returns a new Group and an associated Context derived from ctx.
    44  //
    45  // The derived Context is canceled the first time a function passed to Go
    46  // returns a non-nil error or the first time Wait returns, whichever occurs
    47  // first.
    48  func WithContext(ctx context.Context) (*Group, context.Context) {
    49  	ctx, cancel := withCancelCause(ctx)
    50  	return &Group{cancel: cancel}, ctx
    51  }
    52  
    53  // Wait blocks until all function calls from the Go method have returned, then
    54  // returns the first non-nil error (if any) from them.
    55  func (g *Group) Wait() error {
    56  	g.wg.Wait()
    57  	if g.cancel != nil {
    58  		g.cancel(g.err)
    59  	}
    60  	return g.err
    61  }
    62  
    63  // Go calls the given function in a new goroutine.
    64  // It blocks until the new goroutine can be added without the number of
    65  // active goroutines in the group exceeding the configured limit.
    66  //
    67  // The first call to return a non-nil error cancels the group's context, if the
    68  // group was created by calling WithContext. The error will be returned by Wait.
    69  func (g *Group) Go(f func() error) {
    70  	if g.sem != nil {
    71  		g.sem <- token{}
    72  	}
    73  
    74  	g.wg.Add(1)
    75  	go func() {
    76  		defer g.done()
    77  
    78  		if err := f(); err != nil {
    79  			g.errOnce.Do(func() {
    80  				g.err = err
    81  				if g.cancel != nil {
    82  					g.cancel(g.err)
    83  				}
    84  			})
    85  		}
    86  	}()
    87  }
    88  
    89  // TryGo calls the given function in a new goroutine only if the number of
    90  // active goroutines in the group is currently below the configured limit.
    91  //
    92  // The return value reports whether the goroutine was started.
    93  func (g *Group) TryGo(f func() error) bool {
    94  	if g.sem != nil {
    95  		select {
    96  		case g.sem <- token{}:
    97  			// Note: this allows barging iff channels in general allow barging.
    98  		default:
    99  			return false
   100  		}
   101  	}
   102  
   103  	g.wg.Add(1)
   104  	go func() {
   105  		defer g.done()
   106  
   107  		if err := f(); err != nil {
   108  			g.errOnce.Do(func() {
   109  				g.err = err
   110  				if g.cancel != nil {
   111  					g.cancel(g.err)
   112  				}
   113  			})
   114  		}
   115  	}()
   116  	return true
   117  }
   118  
   119  // SetLimit limits the number of active goroutines in this group to at most n.
   120  // A negative value indicates no limit.
   121  //
   122  // Any subsequent call to the Go method will block until it can add an active
   123  // goroutine without exceeding the configured limit.
   124  //
   125  // The limit must not be modified while any goroutines in the group are active.
   126  func (g *Group) SetLimit(n int) {
   127  	if n < 0 {
   128  		g.sem = nil
   129  		return
   130  	}
   131  	if len(g.sem) != 0 {
   132  		panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", len(g.sem)))
   133  	}
   134  	g.sem = make(chan token, n)
   135  }
   136
View as plain text