tsk
Tsk is a Go library for handling generic concurrent tasks.
It provides multiple packages that each implement a single pattern.
go get -u github.com/abevier/tsk
Packages
futures
The futures package provides a wrapper around an asynchronous computation.
A channel and a goroutine are often used to achieve the same result, but there are differences.
- A Future can be read by any number of go routines, a channel only provides the value once
- A Future can be safely completed multiple times: only the first completion is used, subsequent completions are ignored silently, a channel will panic if written to when closed.
Basic Future Example:
fut := futures.New[Response]()
go func() {
bgCtx := context.WithTimeout(context.Background(), 45 * time.Second)
resp, err := slowServiceClient.MakeRequest(bgCtx, req)
if err != nil {
fut.Fail(err)
return
}
fut.Complete(resp)
}()
// This will block until the slowServiceClient returns a response and the future is completed
resp, err := fut.Get(ctx)Another basic Future example using FromFunc
fut := futures.FromFunc(func() (Response, error) {
bgCtx := context.WithTimeout(context.Background(), 45 * time.Second)
resp, err := slowServiceClient.MakeRequest(bgCtx, req)
return resp, err
})
// This will block until the slowServiceClient returns a response and the future is completed
resp, err := fut.Get(ctx)Simple Cache using Future Example
type SimpleCache struct {
m sync.Mutex{}
values map[string]*futures.Future[Response]
}
func NewCache() *SimpleCache {
return &SimpleCache{
m: sync.Mutex{},
values: make(map[string]*futures.Future[Response])
}
}
// Get will return a future if it exists, if it does not then a new future will
// be returned after it is added to the cache
func (c *SimpleCache) Get(key string) *futures.Future[Response] {
c.m.Lock()
defer c.m.Unlock()
f, ok := c.values[key]
if !ok {
f = futures.FromFunc(func() (Response, error) {
bgCtx := context.WithTimeout(context.Background(), 45 * time.Second)
resp, err := slowServiceClient.MakeRequest(bgCtx, Request{Key: key})
return resp, err
})
c.values[key] = f
}
return f
}
func main() {
wg := sync.WaitGroup{}
cache := NewCache()
// All go routines read the same result - only one slow request is made despite 100
// concurrent requests for the same key
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
defer wg.Done()
futureResp := cache.Get("the-key")
resp, _ := futureResp.Get(context.TODO())
log.Printf("response is: %v", resp)
}()
}
wg.Wait()
}batch
The batch package provides a batch executor implementation that allows multiple go routines to seamlessly batch tasks which are then flushed to a user defined executor function.
A common use case for this is batching multiple http requests that wish to write to a database into a single update to the database. Another common use is multiple http requests that should publish a message to AWS SQS which allows for batching of up to 10 messages in a single request.
Batch Example
be := batch.New(batch.Opts{MaxSize: 10, MaxLinger:250 * time.Millisecond}, runBatch)
func runBatch(events []string) ([]results.Result[string], error) {
serviceRequest := myservice.Request{items: events}
resp, err := myservice.Write(serviceRequest)
if err != nil {
// All items in the batch will fail with this error
return nil, err
}
// The run batch function must return an result for every item passed into the function
var res []results.Result[string]
for _, r := range resp.items {
if r.Err != nil {
res = append(res, results.Fail[string](err))
continue
}
res = append(res, results.Success(item.ID))
}
return res, nil
}
func eventHandler(w http.ResponseWriter, req *http.Request) {
b, _ := ioutil.ReadAll(req.body)
event := string(b)
// Each request will submit individual items to the batch which will flush after 10 items or once the batch is 250ms old
// This will not return until the batch flushes and a result is returned from runBatch
id, err := be.Submit(req.Context(), event)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
return
}
w.Write([]byte(id))
}
func main() {
http.HandleFunc("/event", eventHandler)
http.ListenAndServe(":8080", nil)
}ratelimiter
The ratelimiter package provides a rate limiter that utilizes the Token Bucket algorithm to limit the rate that a function will be invoked. A common use case is to prevent a service from overwhelming other systems.
Rate Limiter Example
opts := ratelimiter.Opts{
Limit: 10,
Burst: 1,
MaxQueueDepth: 100,
FullQueueStategy: ratelimiter.ErrorWhenFull,
}
rl := ratelimiter.New(opts, do)
// This function will be called up to 10 times per second
func do(ctx context.Context, request string) (string, error) {
resp, err := myservice.RateLimitedQuery(ctx, request)
return resp, err
}
func requestHandler(w http.ResponseWriter, req *http.Request) {
b, _ := ioutil.ReadAll(req.body)
request := string(b)
// Each call to submit will block until the rate limited function is invoked and returns a result.
// If the number pending request is too high ErrQueueFull is returned and the server will respond
// to the request with an HTTP 429
resp, err := rl.Submit(req.Context(), event)
if err != nil {
if errors.Is(err, ratelimiter.ErrQueueFull) {
w.WriteHeader(http.StatusTooManyRequests)
} else {
w.WriteHeader(http.StatusInternalServerError)
}
return
}
w.Write([]byte(resp))
}
func main() {
http.HandleFunc("/request", requestHandler)
http.ListenAndServe(":8080", nil)
}taskqueue
The taskqueue package provides a task queue that schedules work on a pool of goroutines. It is used to limit concurrent invocations to a function. Much like a rate limiter a common use case is to prevent overwhelming other services.
Task Queue Example
opts := taskqueue.Opts{
MaxWorkers: 3,
MaxQueueDepth: 100,
FullQueueStategy: taskqueue.ErrorWhenFull,
}
tq := taskqueue.New(opts, do)
// This function will never be invoked more than 3 times concurrently while behind the TaskQueue
func do(ctx context.Context, request string) (string, error) {
resp, err := myservice.ConcurrencyLimitedQuery(ctx, request)
return resp, err
}
func requestHandler(w http.ResponseWriter, req *http.Request) {
b, _ := ioutil.ReadAll(req.body)
request := string(b)
// Each call to submit will block until the concurrency limited function is invoked and returns a result.
// If the number pending request is too high ErrQueueFull is returned and the server will respond
// to the request with an HTTP 429
resp, err := tq.Submit(req.Context(), event)
if err != nil {
if errors.Is(err, taskqueue.ErrQueueFull) {
w.WriteHeader(http.StatusTooManyRequests)
} else {
w.WriteHeader(http.StatusInternalServerError)
}
return
}
w.Write([]byte(resp))
}
func main() {
http.HandleFunc("/request", requestHandler)
http.ListenAndServe(":8080", nil)
}Motivation
I wrote this library because I found that I was repeating these patterns in multiple places. My preference is to keep application business logic and task plumbing logic separate when at all possible. This means I try to avoid exposing channels and other concurrency primitives to business services. Many concurrency bugs I've encountered in Go have been due to improperly sending/receiving to unbuffered channels and forgetting to cancel when a context is cancelled.