README
¶
dowork 
dowork is a general purpose task queueing system for Go programs. It queues, executes, and reschedules tasks in a Goroutine in-process.
A global task queue is provided for simple use-cases. To use it:
import (
"context"
"git.sr.ht/~sircmpwn/dowork"
)
work.Submit(func(ctx context.Context) error {
// ...do work...
return nil
})
This task will be executed in the background. The first time a task is submitted to the global queue, it will be initialized and start running in the background.
To customize options like maximum retries and timeouts, use work.Enqueue:
task := work.NewTask(func(ctx context.Context) error {
// ...
}).Retries(5).MaxTimeout(10 * time.Minute)
work.Enqueue(task)
task := work.NewTask(func(ctx context.Context) error {
// ...
}).
Retries(5). // Maximum number of attempts
MaxTimeout(10 * time.Minute). // Maximum timeout between attempts
Within(10 * time.Second). // Deadline for each attempt
After(func(ctx context.Context, task *work.Task) {
// Executed once the task completes, successful or not
})
work.Enqueue(task)
Retries are conducted with an exponential backoff.
You may also manage your own work queues. Use NewQueue() to obtain a queue,
(*Queue).Dispatch() to execute all overdue tasks, and (*Queue).Start() to
spin up a goroutine and start dispatching tasks automatically.
Use work.Shutdown() or (*Queue).Shutdown() to perform a soft shutdown of the
queue, which will stop accepting new tasks and block until all already-queued
tasks complete.
Distributed task queues
No such functionality is provided OOTB, but you may find this package useful in doing the actual queueing work for your own distributed work queue. Such an integeration is left as an exercise to the reader.
Instrumentation
Instrumentation is provided via Prometheus and the
client_golang library. Relevant metrics are prefixed with
queue_ in the metric name.
A common pattern for soft restarting web servers is to shut down the http.Server, allowing the new web server process to start up and begin accepting new connections, then allow the queue to finish executing any pending tasks before terminating the process. If this describes your program, note that you may want to provide Prometheus metrics on a secondary http.Server on a random port, so that you may monitor the queue shutdown. Something similar to the following will set up a secondary HTTP server for this purpose:
import (
"log"
"net"
"net/http"
"github.com/prometheus/client_golang/prometheus/promhttp"
)
mux := &http.ServeMux{}
mux.Handle("/metrics", promhttp.Handler())
server := &http.Server{Handler: mux}
listen, err := net.Listen("tcp", ":0")
if err != nil {
panic(err)
}
log.Printf("Prometheus listening on :%d", listen.Addr().(*net.TCPAddr).Port)
go server.Serve(listen)
Documentation
¶
Overview ¶
dowork is a generic task queueing system for Go programs. It queues, executes, and reschedules tasks in Goroutine in-process.
A global task queue is provided for simple use-cases. To use it:
import (
"context"
"git.sr.ht/~sircmpwn/dowork"
)
work.Submit(func(ctx context.Context) error {
// ...do work...
return nil
})
This task will be executed in the background. The first time a task is submitted to the global queue, it will be initialized and start running in the background.
To customize options like maximum retries and timeouts, use work.Enqueue:
task := work.NewTask(func(ctx context.Context) error {
// ...
}).
Retries(5). // Maximum number of attempts
MaxTimeout(10 * time.Minute). // Maximum timeout between attempts
Within(10 * time.Second). // Deadline for each attempt
After(func(ctx context.Context, task *work.Task) {
// Executed once the task completes, successful or not
})
work.Enqueue(task)
Retries are conducted with an exponential backoff.
You may also manage your own work queues. Use NewQueue() to obtain a queue, (*Queue).Dispatch() to execute all overdue tasks, and (*Queue).Start() to spin up a goroutine and start dispatching tasks automatically.
Use work.Shutdown() or (*Queue).Shutdown() to perform a soft shutdown of the queue, which will stop accepting new tasks and block until all already-queued tasks complete.
Index ¶
- Variables
- func Enqueue(t *Task)
- func Join(queues ...*Queue)
- func Shutdown()
- func Start()
- type Queue
- func (q *Queue) Dispatch(ctx context.Context) bool
- func (q *Queue) Enqueue(t *Task) error
- func (q *Queue) Name() string
- func (q *Queue) Now(now func() time.Time)
- func (q *Queue) Run(ctx context.Context)
- func (q *Queue) Shutdown()
- func (q *Queue) Start(ctx context.Context)
- func (q *Queue) Submit(fn TaskFunc) (*Task, error)
- type Task
- func (t *Task) After(fn func(ctx context.Context, task *Task)) *Task
- func (t *Task) Attempt(ctx context.Context) (time.Time, error)
- func (t *Task) Attempts() int
- func (t *Task) Before(fn func(ctx context.Context, task *Task)) *Task
- func (t *Task) Done() bool
- func (t *Task) MaxTimeout(d time.Duration) *Task
- func (t *Task) NextAttempt() time.Time
- func (t *Task) NoJitter() *Task
- func (t *Task) NotBefore(date time.Time) *Task
- func (t *Task) Result() error
- func (t *Task) Retries(n int) *Task
- func (t *Task) Within(deadline time.Duration) *Task
- type TaskFunc
Constants ¶
This section is empty.
Variables ¶
var ( // Returned when a task is attempted which was already successfully completed. ErrAlreadyComplete = errors.New("This task was already successfully completed once") // If this is returned from a task function, the task shall not be re-attempted. ErrDoNotReattempt = errors.New("This task should not be re-attempted") // This task has been attempted too many times. ErrMaxRetriesExceeded = errors.New("The maximum retries for this task has been exceeded") // Set this function to influence the clock that will be used for // scheduling re-attempts. Now = func() time.Time { return time.Now().UTC() } )
var (
ErrQueueShuttingDown = errors.New("Queue is shutting down; new tasks are not being accepted")
)
Functions ¶
func Join ¶
func Join(queues ...*Queue)
Shuts down any number of work queues in parallel and blocks until they're all finished.
Types ¶
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
func NewQueue ¶
Creates a new task queue. The name of the task queue is used in Prometheus label names and must match [a-zA-Z0-9:_] (snake case is used by convention).
func (*Queue) Dispatch ¶
Attempts any tasks which are due and updates the task schedule. Returns true if there is more work to do.
func (*Queue) Enqueue ¶
Enqueues a task.
An error will only be returned if the queue has been shut down.
func (*Queue) Shutdown ¶
func (q *Queue) Shutdown()
Stops accepting new tasks and blocks until all already-queued tasks are complete. The queue must have been started with Start, not Run.
func (*Queue) Start ¶
Starts the task queue in the background. If you wish to use the warm shutdown feature, you must use Start, not Run.
func (*Queue) Submit ¶
Creates and enqueues a new task, returning the new task. Note that the caller cannot customize settings on the task without creating a race condition; so attempting to will panic. See NewTask and (*Queue).Enqueue to create tasks with customized options.
An error will only be returned if the queue has been shut down.
type Task ¶
type Task struct {
Metadata map[string]interface{}
// contains filtered or unexported fields
}
Stores state for a task which shall be or has been executed. Each task may only be executed successfully once.
func (*Task) After ¶
Sets a function which will be executed once the task is completed, successfully or not. The final result (nil or an error) is passed to the callee.
func (*Task) Attempt ¶
Attempts to execute this task.
If successful, the zero time and nil are returned.
Otherwise, the error returned from the task function is returned to the caller. If an error is returned for which errors.Is(err, ErrDoNotReattempt) is true, the caller should not call Attempt again.
func (*Task) Before ¶
Before Sets a function which will be executed before a task is attempted to run. This will be called before every attempt, including retries.
func (*Task) MaxTimeout ¶
Sets the maximum timeout between retries, or zero to exponentially increase the timeout indefinitely. Defaults to 30 minutes.
func (*Task) NextAttempt ¶
Returns the time the next attempt is scheduled for, or the zero value if it has not been attempted before.
func (*Task) NoJitter ¶
Specifies that randomness should not be introduced into the exponential backoff algorithm.
func (*Task) Result ¶
Returns the result of the task. The task must have been completed for this to be valid.
Source Files
Directories