Worker pool library that extends https://github.com/gammazero/workerpool
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
- 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
- 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)
}
}// ...
// ... pretend we submitted jobs here
// ...
results := wp.StopWaitXT() // -> []wpxt.Result
for _, result := range results {
// If job failed, `result.Error != nil`
}- What if I encounter an error in one of my jobs?
- How can I handle or check for errors/timeout?
// 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}
}
},
})// ... 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 {
// ....
}- Required default context when creating new
workerpoolxt - You can override default context per job
myctx := context.Background() // Any `context.Context`
numWorkers := 10
wp := wpxt.New(myctx, numWorkers)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`- 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")}
},
})- Help make jobs flexible
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"]
},
})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`
},
})
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent