README
¶
taskgroup
A *taskgroup.Group represents a group of goroutines working on related tasks.
New tasks can be added to the group at will, and the caller can wait until all
tasks are complete. Errors are automatically collected and delivered
synchronously to a user-provided callback. This does not replace the full
generality of Go's built-in features, but it simplifies some of the plumbing
for common concurrent tasks.
Here is a working example in the Go Playground.
Rationale
Go provides powerful concurrency primitives, including goroutines, channels, select, and the standard library's sync package. In some common situations, however, managing goroutine lifetimes can become unwieldy using only what is built in.
For example, consider the case of copying a large directory tree: Walk through a source directory recursively, creating a parallel target directory structure and starting a goroutine to copy each of the files concurrently. In outline:
func copyTree(source, target string) error {
err := filepath.Walk(source, func(path string, fi os.FileInfo, err error) error {
adjusted := adjustPath(path)
if fi.IsDir() {
return os.MkdirAll(adjusted, 0755)
}
go copyFile(adjusted, target)
return nil
})
if err != nil {
// ... clean up the output directory ...
}
return err
}
This solution is deficient, however, as it does not provide any way to detect
when all the file copies are finished. To do that we will typically use a
sync.WaitGroup:
var wg sync.WaitGroup
...
wg.Add(1)
go func() {
defer wg.Done()
copyFile(adjusted, target)
}()
...
wg.Wait() // block until all the tasks signal done
In addition, we need to handle errors. Copies might fail (the disk may fill, or there might be a permissions error). For some applications it might suffice to log the error and continue, but usually in case of error we should back out and clean up the partial state.
To do that, we need to capture the return value from the function inside the goroutine―and that will require us either to add a lock or plumb in another channel:
errs := make(chan error)
...
go copyFile(adjusted, target, errs)
Since multiple operations can be running in parallel, we will also need another goroutine to drain the errors channel and accumulate the results somewhere:
var failures []error
go func() {
for e := range errs {
failures = append(failures, e)
}
}()
...
wg.Wait()
close(errs)
Once the work is finished, we must also detect when the error collector is
done, so we can examine the failures without a data race. We'll need another
channel or wait group to signal for this:
var failures []error
edone := make(chan struct{})
go func() {
defer close(edone)
for e := range errs {
failures = append(failures, e)
}
}()
...
wg.Wait() // all the workers are done
close(errs) // signal the error collector to stop
<-edone // wait for the error collector to be done
Another issue is, if one of the file copies fails, we don't necessarily want to
wait around for all the copies to finish before reporting the error―we want to
stop everything and clean up the whole operation. Typically we would do this
using a context.Context:
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
...
copyFile(ctx, adjusted, target, errs)
Now copyFile will have to check for ctx to be finished:
func copyFile(ctx context.Context, source, target string, errs chan<- error) {
if ctx.Err() != nil {
return
}
// ... do the copy as normal, or propagate an error
}
Finally, we want the ability to to limit the number of concurrent copies. Even if the host has plenty of memory and CPU, unbounded concurrency is likely to run us out of file descriptors. To handle this we might use a semaphore or a throttling channel:
throttle := make(chan struct{}, 64) // allow up to 64 concurrent copies
go func() {
throttle <- struct{}{} // block until the throttle has a free slot
defer func() { wg.Done(); <-throttle }()
copyFile(ctx, adjusted, target, errs)
}()
So far, we're up to four channels (errs, edone, context, and throttle) plus a wait group. The point to note is that while these tools are easy to express what we want, it can be tedious to wire them all together and keep track of the current state of the system.
The taskgroup package exists to handle the plumbing for the common case of a
group of tasks that are all working on a related outcome (e.g., copying a
directory structure), and where an error on the part of any single task may
be grounds for terminating the work as a whole.
The package provides a taskgroup.Group type that has built-in support for
some of these concerns:
- Limiting the number of active goroutines.
- Collecting and filtering errors.
- Waiting for completion and delivering status.
A taskgroup.Group collects error values from each task and can deliver them
to a user-provided callback. The callback can filter them or take other actions
(such as cancellation). Invocations of the callback are all done from a single
goroutine so it is safe to manipulate local resources without a lock.
A group does not directly support cancellation, but integrates cleanly with the
standard context package. A context.CancelFunc
can be used as a trigger to signal the whole group when an error occurs.
A task is expressed as a func() error, and is added to a group using the Go
method:
g := taskgroup.New(nil).Go(myTask)
Any number of tasks may be added, and it is safe to do so from multiple goroutines concurrently. To wait for the tasks to finish, use:
err := g.Wait()
Wait blocks until all the tasks in the group have returned, and then reports
the first non-nil error returned by any of the worker tasks.
An implementation of this example can be found in examples/copytree/copytree.go.
Filtering Errors
The taskgroup.New function takes an optional callback to be invoked for each
non-nil error reported by a task in the group. The callback may choose to
propagate, replace, or discard the error. For example, suppose we want to
ignore "file not found" errors from a copy operation:
g := taskgroup.New(func(err error) error {
if os.IsNotExist(err) {
return nil // ignore files that do not exist
}
return err
})
Controlling Concurrency
The Limit method supports limiting the number of concurrently active
goroutines in the group. It returns a start function that adds goroutines to
the group, but will will block when the limit of goroutines is reached until
some of the goroutines already running have finished.
For example:
// Allow at most 3 concurrently-active goroutines in the group.
g, start := taskgroup.New(nil).Limit(3)
// Start tasks by calling the function returned by taskgroup.Limit:
start(task1)
start(task2)
start(task3)
start(task4) // blocks until one of the previous tasks is finished
// ...
Solo Tasks
In some cases it is useful to start a single background task to handle an
isolated concern. For example, suppose we want to read a file into a buffer
while we take care of some other work. Rather than creating a whole group for
a single goroutine, we can create a solo task using the Go constructor.
var data []byte
s := taskgroup.Go(func() error {
f, err := os.Open(filePath)
if err != nil {
return err
}
defer f.Close()
data, err = io.ReadAll(f)
return err
})
When we're ready, we call Wait to receive the result of the task:
if err := s.Wait(); err != nil {
log.Fatalf("Loading config failed: %v", err)
}
doThingsWith(data)
Collecting Results
One common use for a background task is accumulating the results from a batch
of concurrent workers. This can be handled by a solo task, as described above,
and it is a common enough case that the library provides a Collector type to
handle it specifically.
To use it, pass a function to NewCollector to receive the values:
var sum int
c := taskgroup.NewCollector(func(v int) { sum += v })
The Task, NoError, and Report methods of c wrap a function that yields
a value into a task. If the function reports an error, that error is returned
from the task as usual. Otherwise, its non-error value is given to the
accumulator callback. As in the above example, calls to the function are
serialized so that it is safe to access state without additional locking:
// Report an error, no value for the collector.
g.Go(c.Task(func() (int, error) {
return -1, errors.New("bad")
}))
// Report the value 25 to the collector.
g.Go(c.Task(func() (int, error) {
return 25, nil
}))
// Report a random integer to the collector.
g.Go(c.NoError(func() int { return rand.Intn(1000) })
// Report multiple values to the collector.
g.Go(c.Report(func(report func(int)) error {
report(10)
report(20)
report(30)
return nil
}))
Once all the tasks derived from the collector are done, it is safe to access the values accumulated by the callback:
g.Wait() // wait for tasks to finish
// Now you can access the values accumulated by c.
fmt.Println(sum)
Documentation
¶
Overview ¶
Package taskgroup manages collections of cooperating goroutines. It defines a Group that handles waiting for goroutine termination and the propagation of error values. The caller may provide a callback to filter and respond to task errors.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Collector ¶ added in v0.7.0
type Collector[T any] struct { // contains filtered or unexported fields }
A Collector collects values reported by task functions and delivers them to an accumulator function.
Example ¶
package main
import (
"errors"
"fmt"
"math/rand"
"github.com/creachadair/taskgroup"
)
func shuffled(n int) []int {
vs := make([]int, n)
for i := range vs {
vs[i] = i + 1
}
rand.Shuffle(n, func(i, j int) {
vs[i], vs[j] = vs[j], vs[i]
})
return vs
}
func main() {
var total int
c := taskgroup.NewCollector(func(v int) {
total += v
})
const numTasks = 25
input := shuffled(500)
// Start a bunch of tasks to find elements in the input...
g := taskgroup.New(nil)
for i := 0; i < numTasks; i++ {
target := i + 1
g.Go(c.Task(func() (int, error) {
for _, v := range input {
if v == target {
return v, nil
}
}
return 0, errors.New("not found")
}))
}
// Wait for the searchers to finish, then signal the collector to stop.
g.Wait()
// Now get the final result.
fmt.Println(total)
}
Output: 325
func NewCollector ¶ added in v0.7.0
NewCollector creates a new collector that delivers task values to the specified accumulator function. The collector serializes calls to value, so that it is safe for the function to access shared state without a lock.
The tasks created from a collector do not return until all the values reported by the underlying function have been processed by the accumulator.
func (*Collector[T]) NoError ¶ added in v0.7.0
NoError returns a Task wrapping a call to f. The resulting task reports a nil error for all calls.
func (*Collector[T]) Report ¶ added in v0.8.0
Report returns a task wrapping a call to f, which is passed a function that sends results to the accumulator. The report function does not return until the accumulator has finished processing the value.
Example ¶
package main
import (
"errors"
"fmt"
"log"
"github.com/creachadair/taskgroup"
)
func main() {
type val struct {
who string
v int
}
c := taskgroup.NewCollector(func(z val) { fmt.Println(z.who, z.v) })
err := taskgroup.New(nil).
// The Report method passes its argument a function to report multiple
// values to the collector.
Go(c.Report(func(report func(v val)) error {
for i := 0; i < 3; i++ {
report(val{"even", 2 * i})
}
return nil
})).
// Multiple reporters are fine.
Go(c.Report(func(report func(v val)) error {
for i := 0; i < 3; i++ {
report(val{"odd", 2*i + 1})
}
// An error from a reporter is propagated like any other task error.
return errors.New("no bueno")
})).
Wait()
if err == nil || err.Error() != "no bueno" {
log.Fatalf("Unexpected error: %v", err)
}
}
Output: even 0 odd 1 even 2 odd 3 even 4 odd 5
func (*Collector[T]) Stream
deprecated
added in
v0.7.1
Stream returns a task wrapping a call to f, which is passed a channel on which results can be sent to the accumulator. Each call to Stream starts a goroutine to process the values on the channel.
Deprecated: Tasks that wish to deliver multiple values should use Report instead, which does not spawn a goroutine. This method may be removed in a future version.
func (*Collector[T]) Task ¶ added in v0.7.0
Task returns a Task wrapping a call to f. If f reports an error, that error is propagated as the return value of the task; otherwise, the non-error value reported by f is passed to the value callback.
func (*Collector[T]) Wait
deprecated
added in
v0.7.0
func (c *Collector[T]) Wait()
Wait waits until the collector has finished processing.
Deprecated: This method is now a noop; it is safe but unnecessary to call it. Once all the tasks created from c have returned, any state accessed by the accumulator is settled. Wait may be removed in a future version.
type ErrorFunc ¶
An ErrorFunc is called by a group each time a task reports an error. Its return value replaces the reported error, so the ErrorFunc can filter or suppress errors by modifying or discarding the input error.
func Listen ¶
Listen creates an ErrorFunc that reports each non-nil task error to f. The resulting ErrorFunc returns task errors unmodified.
Example ¶
package main
import (
"errors"
"fmt"
"log"
"strings"
"github.com/creachadair/taskgroup"
)
func main() {
// The taskgroup itself will only report the first non-nil task error, but
// you can use an error listener used to accumulate all of them.
var all []error
g := taskgroup.New(taskgroup.Listen(func(e error) {
all = append(all, e)
}))
g.Go(func() error { return errors.New("badness 1") })
g.Go(func() error { return errors.New("badness 2") })
g.Go(func() error { return errors.New("badness 3") })
if err := g.Wait(); err == nil || !strings.Contains(err.Error(), "badness") {
log.Fatalf("Unexpected error: %v", err)
}
fmt.Println(errors.Join(all...))
}
Output: badness 1 badness 2 badness 3
func Trigger ¶
func Trigger(f func()) ErrorFunc
Trigger creates an ErrorFunc that calls f each time a task reports an error. The resulting ErrorFunc returns task errors unmodified.
Example ¶
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/creachadair/taskgroup"
)
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
const badTask = 5
// Construct a group in which any task error cancels the context.
g := taskgroup.New(taskgroup.Trigger(cancel))
for i := 0; i < 10; i++ {
i := i
g.Go(func() error {
if i == badTask {
return fmt.Errorf("task %d failed", i)
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(10 * time.Second):
return nil
}
})
}
if err := g.Wait(); err == nil {
log.Fatal("I expected an error here")
} else {
fmt.Println(err.Error())
}
}
Output: task 5 failed
type Group ¶
type Group struct {
// contains filtered or unexported fields
}
A Group manages a collection of cooperating goroutines. Add new tasks to the group with the Go method. Call the Wait method to wait for the tasks to complete. A zero value is ready for use, but must not be copied after its first use.
The group collects any errors returned by the tasks in the group. The first non-nil error reported by any task (and not otherwise filtered) is returned from the Wait method.
Example ¶
package main
import (
"fmt"
"github.com/creachadair/taskgroup"
)
func main() {
msg := make(chan string)
g := taskgroup.New(nil)
g.Go(func() error {
msg <- "ping"
fmt.Println(<-msg)
return nil
})
g.Go(func() error {
fmt.Println(<-msg)
msg <- "pong"
return nil
})
g.Wait()
fmt.Println("<done>")
}
Output: ping pong <done>
func New ¶
New constructs a new empty group. If ef != nil, it is called for each error reported by a task running in the group. The value returned by ef replaces the task's error. If ef == nil, errors are not filtered.
Calls to ef are issued by a single goroutine, so it is safe for ef to manipulate local data structures without additional locking.
func (*Group) Limit ¶
Limit returns g and a function that starts each task passed to it in g, allowing no more than n tasks to be active concurrently. If n ≤ 0, the start function is equivalent to g.Go, which enforces no limit.
The limiting mechanism is optional, and the underlying group is not restricted. A call to the start function will block until a slot is available, but calling g.Go directly will add a task unconditionally and will not take up a limiter slot.
Example ¶
var p peakValue
g, start := taskgroup.New(nil).Limit(4)
for i := 0; i < 100; i++ {
start(func() error {
p.inc()
defer p.dec()
time.Sleep(1 * time.Microsecond)
return nil
})
}
g.Wait()
fmt.Printf("Max active ≤ 4: %v\n", p.max <= 4)
Output: Max active ≤ 4: true
func (*Group) Wait ¶
Wait blocks until all the goroutines currently active in the group have returned, and all reported errors have been delivered to the callback. It returns the first non-nil error reported by any of the goroutines in the group and not filtered by an ErrorFunc.
As with sync.WaitGroup, new tasks can be added to g during a call to Wait only if there was already at least one task active when Wait was called. After Wait has returned, the group is ready for reuse.
Wait may be called from at most one goroutine at a time.
type Single ¶ added in v0.4.0
type Single[T any] struct { // contains filtered or unexported fields }
A single manages a single background goroutine. The task is started when the value is first created, and the caller can use the Wait method to block until it has exited.
Example ¶
package main
import (
"fmt"
"io"
"log"
"time"
"github.com/creachadair/taskgroup"
)
type slowReader struct {
n int
d time.Duration
}
func (s *slowReader) Read(data []byte) (int, error) {
if s.n == 0 {
return 0, io.EOF
}
time.Sleep(s.d)
nr := min(len(data), s.n)
s.n -= nr
for i := 0; i < nr; i++ {
data[i] = 'x'
}
return nr, nil
}
func main() {
// A fake reader to simulate a slow file read.
// 2500 bytes and each read takes 50ms.
sr := &slowReader{2500, 50 * time.Millisecond}
var data []byte
// Start a task to read te "file" in the background.
fmt.Println("start")
s := taskgroup.Go(func() error {
var err error
data, err = io.ReadAll(sr)
return err
})
fmt.Println("work, work")
if err := s.Wait(); err != nil {
log.Fatalf("Read failed: %v", err)
}
fmt.Println("done")
fmt.Println(len(data), "bytes")
}
Output: start work, work done 2500 bytes
func Call ¶ added in v0.6.0
Call starts task in a new goroutine. The caller must call Wait to wait for the task to return and collect its result.
Source Files
Directories