README
¶
ctxerrgroup
Create a group of a given number of worker goroutines to behave as a worker pool. The group will do work that is aware
of
context.Context and error handling.
Full example
This example will use a worker group to HTTP GET https://golang.org 16 times and print the status codes with a logger.
package main
import (
"bytes"
"context"
"log"
"net/http"
"os"
"time"
"github.com/MicahParks/ctxerrgroup"
)
func main() {
// Create an error handler that logs all errors.
var errorHandler ctxerrgroup.ErrorHandler
errorHandler = func(group ctxerrgroup.Group, err error) {
log.Printf("An error occurred. Error: \"%s\".\n", err.Error())
}
// Create a worker group with 4 workers.
group := ctxerrgroup.New(4, errorHandler)
// Create some variables to inherit through a closure.
httpClient := &http.Client{}
u := "https://golang.org"
logger := log.New(os.Stdout, "status codes: ", 0)
// Create the worker function.
var work ctxerrgroup.Work
work = func(ctx context.Context) (err error) {
// Create the HTTP request.
var req *http.Request
if req, err = http.NewRequestWithContext(ctx, http.MethodGet, u, bytes.NewReader([]byte{})); err != nil {
return err
}
// Do the HTTP request.
var resp *http.Response
if resp, err = httpClient.Do(req); err != nil {
return err
}
// Log the status code.
logger.Println(resp.StatusCode)
return nil
}
// Do the work 16 times.
for i := 0; i < 16; i++ {
// Create a context for the work.
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// Send the work to the group.
group.AddWorkItem(ctx, cancel, work)
}
// Wait for the group to finish.
group.Wait()
}
Terminology
| Term | Description |
|---|---|
worker |
A goroutine dedicated to completing work items. |
worker group |
A number of workers who all consume work items from the same set and report errors via a common handler. |
worker function |
A function matching a specific signature that can be run by a worker. |
work item |
A worker function plus a unique context.Context and context.CancelFunc pair that will be run once by a worker. |
Differences between golang.org/x/sync/errgroup
This package github.com/MicahParks/ctxerrgroup and
golang.org/x/sync/errgroup are similar, but serve different use
cases.
The current overview for errgroup is:
Package errgroup provides synchronization, error propagation, and Context cancelation for groups of goroutines working on subtasks of a common task.
In terms of Context, this is to say that each group is associated to one context. Another key point is: all
tasks (work items) for a group are subtasks of a common task.
In contrast, ctxerrgroup makes work items and Contexts have a one to one relationship. worker functions do not
need to be subtasks of a common task as one of the primary features of the group is to behave as a worker pool. I find
worker pools helpful in performing work asynchronously without worrying about creating too many goroutines.
| Purpose | one to one with context.Context |
errors | |
|---|---|---|---|
errgroup |
subtasks of a common task | errgroup.Group |
Propagation |
ctxerrgroup |
goroutines as worker pools | work item |
Handler function |
Benefits
- Async error handling simplified.
- Familiar methods.
Donemethod mimicscontext.Context's.Waitmethod mimicssync.WaitGroup's.
- Flat and simple.
- Only exported struct is
ctxerrgroup.Group.
- Only exported struct is
- Apache 2.0 License.
- No dependencies outside of the packages included with the Golang compiler.
- Small code base.
- Three source files with less than 350 lines of code including lots of comments.
- Test coverage is greater than 90%.
- The group and its workers will all be cleaned up with
group.Kill(). (All work sent to the group should exit as well, if it respects its own context.)
Usage
Basic Workflow
Create an error handler
The first step to using a worker group is creating an error handler. The worker group is expecting
all worker functions to match the ctxerrgroup.Work function
signature: type Work func(ctx context.Context) (err error).
Error handlers have the function signature of type ErrorHandler func(group Group, err error) where the first argument
is the ctxerrgroup.Group that the error handler is handling errors for and the second argument is the current error
reported from a worker.
The example error handler below logs all errors with the build in logger.
// Create an error handler that logs all errors.
var errorHandler ctxerrgroup.ErrorHandler
errorHandler = func(group ctxerrgroup.Group, err error) {
log.Printf("An error occurred. Error: \"%s\".\n", err.Error())
}
Create a worker group
After the error handler has been created, the worker group can be created.
// Create a worker group with 4 workers.
group := ctxerrgroup.New(4, errorHandler)
The first argument is the number of workers. The number of workers is the maximum number of goroutines that can be
working on a work item at any one time. If the number of workers is 0, the worker group will be useless.
The second argument is the error handler created in the previous step. All errors will be sent to the error handler asynchronously (in a separate goroutine).
Create worker functions
worker functions sent to the worker group must match the ctxerrgroup.Work function signature: type Work func(workCtx context.Context) (err error) and is expected to respect its given context, workCtx. If the context is not respected
and the worker group is killed, the goroutine performing the work will leak.
Here is an example of a worker function that respects its context:
// Create the worker function.
var work ctxerrgroup.Work
work = func(ctx context.Context) (err error) {
// Create the HTTP request.
var req *http.Request
if req, err = http.NewRequestWithContext(ctx, http.MethodGet, u, bytes.NewReader([]byte{})); err != nil {
return err
}
// Do the HTTP request.
var resp *http.Response
if resp, err = httpClient.Do(req); err != nil {
return err
}
// Log the status code.
logger.Println(resp.StatusCode)
return nil
}
Here is an example of a worker function that does the same thing without respecting its own context:
// Create the worker function.
var work ctxerrgroup.Work
work = func(ctx context.Context) (err error) {
// Create the HTTP request.
var req *http.Request
if req, err = http.NewRequest(http.MethodGet, u, bytes.NewReader([]byte{})); err != nil {
return err
}
// Do the HTTP request.
var resp *http.Response
if resp, err = httpClient.Do(req); err != nil {
return err
}
// Log the status code.
logger.Println(resp.StatusCode)
return nil
}
Since most functions do not match the ctxerrgroup.Work signature,
function closures are typically a convenient way to create worker functions.
// Create some variables to inherit through a closure.
httpClient := &http.Client{}
u := "https://golang.org"
logger := log.New(os.Stdout, "status codes: ", 0)
// Create the worker function.
var work ctxerrgroup.Work
work = func(ctx context.Context) (err error) {
// Create the HTTP request.
var req *http.Request
if req, err = http.NewRequestWithContext(ctx, http.MethodGet, u, bytes.NewReader([]byte{})); err != nil {
return err
}
// Do the HTTP request.
var resp *http.Response
if resp, err = httpClient.Do(req); err != nil {
return err
}
// Log the status code.
logger.Println(resp.StatusCode)
return nil
}
Create a context
work items and contexts have a 1:1 relationship. Each context's cancel function is called on the completion of its
work item completion. Never send a work item with a context that has been used before.
Here is an example of what not to do:
// This context will cause up to 4 jobs to fail.
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// Do the work 16 times.
for i := 0; i < 16; i++ {
// Send the work to the group.
group.AddWorkItem(ctx, cancel, work)
}
Instead, create the context in the loop and do not shadow a context variable from out of scope.
// Do the work 16 times.
for i := 0; i < 16; i++ {
// Create a context for the work.
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// Send the work to the group.
group.AddWorkItem(ctx, cancel, work)
}
Adding work items
Adding, in its simplest form. Was addressed above on these lines.
// Send the work to the group.
group.AddWorkItem(ctx, cancel, work)
Any time the AddWorkItem method is called, a new work item will be taken and performed by the worker pool.
Remember that worker functions can also be created via function closures. This allows access to variables that are
needed but do not match the function signature: ctxerrgroup.Work.
The first and second arguments are the unique context and cancellation functions for the given work item.
The third argument is the work function created in a previous step.
Let all work items finish
group.Wait()
or
<-group.Done()
Both statements will block until the worker group has completed all given work items. The Done method is idea for
select statements.
Clean up the worker group
group.Kill()
Killing the worker groups isn't required, but if the worker group is no longer being used it's best to tell all its
goroutines to return to reclaim their resources. If the program's main function is about to end, group.Kill() will
be accomplished regardless.
A worker group can be killed before all work items finish. Outstanding work items' context.CancelFuncs will be
called.
Test Coverage
Testing coverage for this repository is currently greater than 90%. Depending on how Go runtime schedules things, certain paths may or may not be executed. This is based on a sample of 1000 tests with coverage and race detection. All tests pass without any detected race conditions.
If you are interested in test coverage, please view the cmd/coverage.go tool and the cmd/profiles directory. The
directory has log output from the tool and testing coverage profiles for multiple samples of the 1000 tests.
In my experience, test coverage that counts the number of lines executed can be a misleading number. While I believe the current written tests adequately cover the code base, if there are test cases that are not covered by the current testing files, please feel free to add an issue requesting the test case or create a PR that adds tests.
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrCantDo indicates that there was a failure to send the function to work on to a worker before the context // expired. ErrCantDo = errors.New("failed to send work item to a worker before the context expired") )
Functions ¶
This section is empty.
Types ¶
type ErrorHandler ¶
ErrorHandler is a function that receives an error and handles it.
type Group ¶
type Group struct {
// contains filtered or unexported fields
}
Group is the way to control a group of worker goroutines that understand context.Context and error handling.
func (Group) AddWorkItem ¶
AddWorkItem takes in context information and a Work function and gives it to a worker. This can block if all workers are busy and the work item buffer is full. This function will block if no workers are ready. Call with the go keyword to launch it in another goroutine to guarantee no blocking.
func (Group) Death ¶
func (g Group) Death() <-chan struct{}
Death returns a channel that will close when the Group has died.
func (Group) Done ¶
func (g Group) Done() <-chan struct{}
Done mimics the functionality of the context.Context Done method. It returns a channel that will close when all given work has been completed or when the group dies.
Source Files
Directories