README
¶
taskgroup
http://godoc.org/github.com/creachadair/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 to a
user-provided callback in a single goroutine. This does not replace the full
generality of Go's built-in features, but it simplifies some of the plumbing
for common concurrent tasks.
Rationale
Go provides excellent and powerful concurrency primitives, in the form of 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 throught the source directory recursively, creating the parallel target directory structure and spinning up a goroutine to copy each of the files concurrently. This part is simple:
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
}
Unfortunately, this isn't quite sufficient: How will we detect when all the
file copies are finished? Typically we do this with a sync.WaitGroup:
var wg sync.WaitGroup
...
wg.Add(1)
go func() {
defer wg.Done()
copyFile(adjusted, target)
}()
...
wg.Wait() // blocks until all the tasks signal done
In addition, we need to cope with errors: Copies might fail―the disk might fill up, or there might be a permissions error, say. For some applications we might be content to log the error and continue, but often in case of error you'd like to back out and clean up your mess. To do this, however, we need to capture the return value from the function inside the goroutine―and that will require us to plumb in another channel:
errs := make(chan error)
...
go copyFile(adjusted, target, errs)
Moreover, we may get more than one error, so we will need another goroutine to clear it out:
var failures []error
go func() {
for e := range errs {
failures = append(failures, e)
}
}()
...
wg.Wait()
close(errs)
But now, how do we know when the error collector is finished so that we can
safely examine the failures? We need another channel or sync.WaitGroup to
signal for that:
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
That's tedious, but works. But now, in this example, if something fails we really don't want to wait around for all the copies to finish―if one of the file copies fails, we want to stop what you're doing and clean up. So now we nee another channel or a context to signal cancellation:
ctx, cancel := context.WithCancel(context.Background())
...
copyFile(ctx, adjusted, target, errs)
and then copyFile will have to check for that:
func copyFile(ctx context.Context, source, target string, errs chan<- error) {
select {
case <-ctx.Done():
return
default:
// ... do the copy as normal, or propagate an error
}
}
Finally, the way this is set up right now there is no limit to the number of copies that can be concurrently in flight. Even if we have plenty of RAM, it is quite likely we may hit the limit of open file descriptors our process is allowed. Ideally, we should set some kind of bound on the active concurrency. 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)
}()
In case you've lost count, we're up to four channels (including the context).
The lesson here is that, while Go's concurrency primitives are more than powerful enough to express these relationships, it can be tedious to wire them all together.
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.
The API is straightforward: 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()
This blocks until all the tasks in the group have returned (either
successfully, or with an error). Wait returns the first non-nil error
returned by any of the worker tasks.
An example program can be found in examples/copytree/copytree.go.
Filtering Errors
The taskgroup.New function takes an optional callback that is 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:
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 of a supports limiting the number of concurrently active
goroutines in the group. For example:
// Allow at most 25 concurrently-active goroutines in the group.
g, start := taskgroup.New(nil).Limit(25)
// Start tasks by calling the function returned by taskgroup.Limit:
start(task1)
start(task2)
// ...
Documentation
¶
Overview ¶
Package taskgroup manages a collection of cooperating goroutines. It simplifies common concerns of waiting for goroutine termination and collecting errors.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
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 adapts f to an ErrorFunc. The resulting function returns task errors unmodified.
Example ¶
package main
import (
"errors"
"fmt"
"github.com/creachadair/taskgroup"
)
func main() {
g := taskgroup.New(taskgroup.Listen(func(e error) {
fmt.Println(e)
}))
g.Go(func() error { return errors.New("heard you") })
fmt.Println(g.Wait()) // the error was preserved
}
Output: heard you heard you
func Trigger ¶
func Trigger(f func()) ErrorFunc
Trigger adapts f to an ErrorFunc. The resulting function 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
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 nil
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. New tasks can be added to the group via the Go method. The caller can wait for the tasks to complete by calling the Wait method.
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. Add tasks to the group using g.Go. Wait for the group to complete with g.Wait. If ef != nil, it is called each time a task reports an error. The error returned by the ErrorFunc replaces the task's error.
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.
Example ¶
package main
import (
"fmt"
"sync"
"time"
"github.com/creachadair/taskgroup"
)
func main() {
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)
}
type peakValue struct {
μ sync.Mutex
cur, max int
}
func (p *peakValue) inc() {
p.μ.Lock()
p.cur++
if p.cur > p.max {
p.max = p.cur
}
p.μ.Unlock()
}
func (p *peakValue) dec() {
p.μ.Lock()
p.cur--
p.μ.Unlock()
}
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. Wait returns the first non-nil error returned by any of the goroutines in the group and not filtered by an ErrorFunc.
After a call to Wait returns, the group is ready for reuse.
Source Files
Directories