workerpoolxt
Worker pool library that extends https://github.com/gammazero/workerpool
Note
If you are using context, please use with caution in production!! See here for more info
-
Go Playground which shows the area of concern
-
Wonderful explanation from the author of
workerpool:
Once a goroutine is started, there no way to kill it unless there is something inside that goroutine that is looking for a signal, waiting on a context or channel, etc. In the case of workerpool, and workerpoolxt, the pool is providing the goroutine and calling someone else's function in that goroutine. There is no telling if that function will ever return and allow your goroutine to finish or run another task. The only thing that workerpoolxt can do is to run the task function in another goroutine and wait for the task to finish or for a timer to expire or some other signal to give up waiting (such as context being canceled). When this signal to give up happens, the worker goroutine can return and report an error, but the task goroutine is still running somewhere in the background and may or may not ever finish. All you have done is given up waiting for it to do so.
- @gammazero github.com/gammazero
Synopsis
- Allows you to retain access to underlying
*WorkerPoolobject as if you importedworkerpooldirectly - Hello, world!
- How we extend
workerpool- Results
- Job results are captured so you can work with them later
- How to handle errors?
- Context
- Supply your own context
- "Default" context required when calling
workerpoolxt.New(...) - You can override the default context on a per job basis
- This allows you to do things like custom job timeouts
- Retry
intthat defines N number of retries- Can only supply retry on a per job basis
- Options
- Options are optional
- Provide either global/default options or per job options
- Options are nothing more than
map[string]interface{}so that you may supply anything you wish - Job options override default options, we do NOT merge options
- Runtime duration
- Access a job's runtime duration via it's result
- e.g.
howLongItTook := someResultFromSomeJob.Duration time.Duration
- Results
Hello World
- Obligatory "as simple as it gets" example
package main
import (
"context"
"fmt"
wpxt "github.com/oze4/workerpoolxt"
)
func main() {
ctx := context.Background()
numWorkers := 10
wp := wpxt.New(ctx, numWorkers)
wp.SubmitXT(wpxt.Job{
Name: "My first job",
Task: func(o wpxt.Options) wpxt.Result {
return wpxt.Result{Data: "Hello, world!"}
},
})
jobResults := wp.StopWaitXT()
for _, jobresult := range jobResults {
fmt.Println(jobresult)
}
}How we extend workerpool
Results
// ...
// ... pretend we submitted jobs here
// ...
results := wp.StopWaitXT() // -> []wpxt.Result
for _, result := range results {
// If job failed, `result.Error != nil`
}Error Handling
- What if I encounter an error in one of my jobs?
- How can I handle or check for errors/timeout?
Return Error From Job
// Just set the `Error` field on the `wpxt.Result` you return
wp.SubmitXT(wpxt.Job{
Name: "How to handle errors",
Task: func(o wpxt.Options) wpxt.Result {
// Pretend we got an error doing something
if theError != nil {
return wpxt.Result{Error: theError}
}
},
})Check For Errors In Result
// ... pretend we submitted a bunch of jobs
//
// StopWaitXT() returns []wpxt.Result
// Each result has an `Error` field
// Whether a timeout, or an error you set
// Check for it like
if someResultFromSomeJob.Error != nil {
// ....
}Context
- Required default context when creating new
workerpoolxt - You can override default context per job
Default Context
myctx := context.Background() // Any `context.Context`
numWorkers := 10
wp := wpxt.New(myctx, numWorkers)Per Job Context
Timeouts
defaultCtx := context.Background()
numWorkers := 10
wp := wpxt.New(defaultCtx, numWorkers)
timeout := time.Duration(time.Millisecond)
myCtx, done := context.WithTimeout(context.Background(), timeout)
defer done()
wp.SubmitXT(wpxt.Job{
Name: "my ctx job",
Context: myCtx,
Task: func(o wpxt.Options) wpxt.Result {
// Simulate long running task
time.Sleep(time.Second*10)
return wpxt.Result{Data: "I could be anything"}
},
})
// > `Result.Error` will be `context.DeadlineExceeded`Retry
- Optional
- Seamlessly retry failed jobs
wp.SubmitXT(wpxt.Job{
// This job is configured to fail immediately,
// therefore it will retry 5 times
// (as long as we have not exceeded our job timeout)
timeoutctx, _ := context.WithTimeout(context.Background(), time.Duration(time.Millisecond*500))
Retry: 5,
// ^^^^^^
Name: "I will retry 5 times",
// Set timeout field on job
Context: timeoutctx,
Task: func(o wpxt.Options) wpxt.Result {
return wpxt.Result{Error: errors.New("some_err")}
},
})Options
- Help make jobs flexible
Default Options
myopts := map[string]interface{}{
"myclient": &http.Client{},
}
wp := wpxt.New(context.Background(), 10)
wp.WithOptions(myopts)
wp.SubmitXT(wpxt.Job{
Name: "myjob",
Task: func(o wpxt.Options) wpxt.Result {
// access options here
client := o["myclient"]
},
})Per Job Options
myhttpclient := &http.Client{}
myk8sclient := kubernetes.Clientset{}
// This Job Only Needs an HTTP Client
wp.SubmitXT(wpxt.Job{
Name: "This Job Only Needs an HTTP Client",
Options: map[string]interface{}{
"http": myhttpclient,
},
Task: func(o wpxt.Options) wpxt.Result {
// access options here
httpclient := o["http"]
// ... do work with `httpclient`
},
})
// This Job Only Needs Kubernetes Clientset
wp.SubmitXT(wpxt.Job{
Name: "This Job Only Needs Kubernetes Clientset",
Options: map[string]interface{}{
"kube": myk8sclient,
},
Task: func(o wpxt.Options) wpxt.Result {
// access options here
kubernetesclient := o["kube"]
// ... do work with `kubernetesclient`
},
})