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