README
¶
group
group is a library for managing pools of goroutines. It has been adapted from errgoup with small improvements to the API. It has zero dependencies and is intentionally very simple.
Usage
Simple
This will create a new group and start 10 goroutines:
g := group.New(ctx)
for i := range 10 {
g.Go(func(ctx context.Context) error {
fmt.Printf("loop %d\n", i)
return nil
})
}
if err := g.Wait(); err != nil {
log.Fatal(err)
}
The parent context provided is used to create a child context that group uses internally, which is passed through to each goroutine. If any goroutine produces an error, this child context is canceled, allowing the other goroutines to stop/cleanup:
Bounded concurrency
The option WithLimit can be passed to the group constructor to establish a bound on concurrency:
g := group.New(ctx, group.WithLimit(2))
for i := range 10 {
g.Go(func(ctx context.Context) error {
fmt.Printf("loop %d\n", i)
return nil
})
}
if err := g.Wait(); err != nil {
log.Fatal(err)
}
Here, only 2 goroutines will ever be running at a given time.
Method chaining
A group can also be setup using method chaining:
if err := group.New(ctx).Go(func(ctx context.Context) error {
fmt.Printf("goroutine 1\n")
time.Sleep(1 * time.Second)
return nil
}).Go(func(ctx context.Context) error {
fmt.Printf("goroutine 2\n")
time.Sleep(5 * time.Second)
return nil
}).Wait(); err != nil {
log.Fatal(err)
}
Results
g := group.NewResultGroup[int](ctx)
for i := range 10 {
g.Go(func(ctx context.Context) (int, error) {
return i, nil
})
}
for v, err := range g.Get() {
if err != nil {
log.Fatal(err)
}
fmt.Println(v)
}
Getting a future result
g := group.NewResultGroup[string](ctx)
result := g.Go(func(ctx context.Context) (string, error) {
time.Sleep(500 * time.Millisecond)
return "future value", nil
})
v, err := result.Get()
if err != nil {
log.Fatal(err)
}
if err := g.Wait(); err != nil {
log.Fatal(err)
}
Notes
Documentation
¶
Overview ¶
Package group provides group.Group for managing pools of goroutines.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Future ¶ added in v0.3.0
type Future[T any] interface { // Get blocks until the result function is complete. Get() (T, error) }
Future is a value that may not yet be ready.
type Group ¶
type Group struct {
// contains filtered or unexported fields
}
Group manages a pool of goroutines.
Example ¶
package main
import (
"context"
"fmt"
"log"
"go.chrisrx.dev/group"
)
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
g := group.New(ctx)
for range 10 {
g.Go(func(ctx context.Context) error {
fmt.Printf("starting goroutine ...\n")
return nil
})
}
if err := g.Wait(); err != nil {
log.Fatal(err)
}
fmt.Printf("done\n")
}
Output: starting goroutine ... starting goroutine ... starting goroutine ... starting goroutine ... starting goroutine ... starting goroutine ... starting goroutine ... starting goroutine ... starting goroutine ... starting goroutine ... done
func New ¶
func New(ctx context.Context, opts ...GroupOption) *Group
New constructs a new group using the provided options.
func (*Group) Done ¶ added in v0.3.0
Done blocks until all the goroutines in this group have returned. If any errors occur, the first error encountered is sent on the returned channel, otherwise the channel is closed.
type GroupOption ¶
type GroupOption func(*Group)
func WithLimit ¶
func WithLimit(n int) GroupOption
WithLimit sets the bounded concurrency for a pool of goroutines.
type ResultGroup ¶ added in v0.3.0
type ResultGroup[T any] struct { // contains filtered or unexported fields }
ResultGroup manages a pool of goroutines that return a result value.
Example ¶
package main
import (
"context"
"fmt"
"log"
"go.chrisrx.dev/group"
)
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
g := group.NewResultGroup[string](ctx)
for range 5 {
g.Go(func(ctx context.Context) (string, error) {
return "result", nil
})
}
for v, err := range g.Get() {
if err != nil {
log.Fatal(err)
}
fmt.Println(v)
}
fmt.Printf("done\n")
}
Output: result result result result result done
func NewResultGroup ¶ added in v0.3.0
func NewResultGroup[T any](ctx context.Context, opts ...GroupOption) *ResultGroup[T]
NewResultGroup constructs a new result group using the provided options.
func (*ResultGroup[T]) Get ¶ added in v0.3.0
func (r *ResultGroup[T]) Get() iter.Seq2[T, error]
Get returns an iterator of result/error pairs. It blocks until all results are read or the group context is done.
func (*ResultGroup[T]) Go ¶ added in v0.3.0
func (r *ResultGroup[T]) Go(fn func(context.Context) (T, error)) Future[T]
Go runs the provided function in a goroutine and returns a future containing a result value or error.
If an error is encountered, the context for the group is canceled. This happens regardless if the error is checked on the future.
If a concurrency limit is set, calls to Go will block once the number of running goroutines is reached and will continue blocking until a running goroutine returns.
func (*ResultGroup[T]) Wait ¶ added in v0.3.0
func (r *ResultGroup[T]) Wait() error
Wait blocks until all the goroutines in this group have returned. If any errors occur, the first error encountered will be returned. It will also block until at least one goroutine is scheduled.
Source Files