taskq

README

TaskQ

Simple and powerful goroutine manager.

---

Installing

go get github.com/antonmashko/taskq

Purpose

Usually, you need TaskQ for managing your goroutines. This will allow you to control service resource consumption and graceful shutdown.

Example

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/antonmashko/taskq"
)

type Task struct{}

func (Task) Do(ctx context.Context) error {
	fmt.Println("hello world")
	return nil
}

func main() {
	// Initializing new TaskQ instance with a limit of 10 max active goroutines
	// Use `limit=0` for not limiting goroutines number.
	tq := taskq.New(10)

	// Starting reading and executing tasks from queue
	err := tq.Start()
	if err != nil {
		log.Fatal(err)
	}

	// Enqueue new task for execution.
	// TaskQ will run Do method of Task when it will have an available worker (goroutine)
	_, err = tq.Enqueue(context.Background(), Task{})
	if err != nil {
		log.Fatal(err)
	}

	// Gracefully shutting down
	err = tq.Close()
	if err != nil {
		log.Fatal(err)
	}
}

Graceful shutdown

Shutdown and Close gracefully shuts down the TaskQ without interrupting any active tasks. If TaskQ need to finish all tasks in queue, use context ContextWithWait as Shutdown method argument.

Benchmark results

Benchmarks

Documentation

Index

• Variables
• func ContextWithWait(ctx context.Context) context.Context
• type ConcurrentQueue
  • func NewConcurrentQueue() *ConcurrentQueue
  • func (q *ConcurrentQueue) Dequeue(_ context.Context) (Task, error)
  • func (q *ConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)
  • func (q *ConcurrentQueue) Len(_ context.Context) int
• type LimitedConcurrentQueue
  • func NewLimitedConcurrentQueue(size int) *LimitedConcurrentQueue
  • func (q *LimitedConcurrentQueue) Dequeue(ctx context.Context) (Task, error)
  • func (q *LimitedConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)
  • func (q *LimitedConcurrentQueue) Len(_ context.Context) int
• type Queue
• type RetryableTask
  • func (t *RetryableTask) Do(ctx context.Context) error
• type Task
  • func NewRetryableTask(task Task, maxRetries int) Task
• type TaskDone
• type TaskFunc
  • func (t TaskFunc) Do(ctx context.Context) error
• type TaskOnError
• type TaskQ
  • func New(limit int) *TaskQ
  • func NewWithQueue(limit int, q Queue) *TaskQ
  • func Pool(size int) *TaskQ
  • func PoolWithQueue(size int, q Queue) *TaskQ
  • func (t *TaskQ) Close() error
  • func (t *TaskQ) Enqueue(ctx context.Context, task Task) (int64, error)
  • func (t *TaskQ) Shutdown(ctx context.Context) error
  • func (t *TaskQ) Start() error
• type WaitGroup
  • func NewWaitGroup(size int) *WaitGroup
  • func (wg *WaitGroup) Enqueue(ctx context.Context, t Task) (int64, error)
  • func (wg *WaitGroup) Wait()

Variables

var (
	ErrClosed  = errors.New("taskq closed")
	ErrNilTask = errors.New("nil task")
)

var (
	EmptyQueue = errors.New("empty queue")
)

var ErrRetryTask = errors.New("error occurred during task execution. retry task")

Functions

func ContextWithWait

func ContextWithWait(ctx context.Context) context.Context

Types

type ConcurrentQueue

type ConcurrentQueue struct {
	// contains filtered or unexported fields
}

func NewConcurrentQueue

func NewConcurrentQueue() *ConcurrentQueue

func (*ConcurrentQueue) Dequeue

func (q *ConcurrentQueue) Dequeue(_ context.Context) (Task, error)

func (*ConcurrentQueue) Enqueue

func (q *ConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)

func (*ConcurrentQueue) Len

func (q *ConcurrentQueue) Len(_ context.Context) int

type LimitedConcurrentQueue

type LimitedConcurrentQueue struct {
	// contains filtered or unexported fields
}

func NewLimitedConcurrentQueue

func NewLimitedConcurrentQueue(size int) *LimitedConcurrentQueue

func (*LimitedConcurrentQueue) Dequeue

func (q *LimitedConcurrentQueue) Dequeue(ctx context.Context) (Task, error)

func (*LimitedConcurrentQueue) Enqueue

func (q *LimitedConcurrentQueue) Enqueue(_ context.Context, t Task) (int64, error)

func (*LimitedConcurrentQueue) Len

func (q *LimitedConcurrentQueue) Len(_ context.Context) int

type Queue

type Queue interface {
	Enqueue(context.Context, Task) (int64, error)
	// Dequeue Task from queue
	// if queue empty return `EmptyQueue` as error
	Dequeue(context.Context) (Task, error)
}

type RetryableTask

type RetryableTask struct {
	// contains filtered or unexported fields
}

func (*RetryableTask) Do

func (t *RetryableTask) Do(ctx context.Context) error

type Task

type Task interface {
	Do(ctx context.Context) error
}

Task for TaskQ

func NewRetryableTask

func NewRetryableTask(task Task, maxRetries int) Task

type TaskDone

type TaskDone interface {
	Done(context.Context)
}

type TaskFunc

type TaskFunc func(ctx context.Context) error

func (TaskFunc) Do

func (t TaskFunc) Do(ctx context.Context) error

type TaskOnError

type TaskOnError interface {
	OnError(context.Context, error)
}

type TaskQ

type TaskQ struct {
	OnDequeueError func(ctx context.Context, workerID uint64, err error)
	// contains filtered or unexported fields
}

func New

func New(limit int) *TaskQ

func NewWithQueue

func NewWithQueue(limit int, q Queue) *TaskQ

func Pool

func Pool(size int) *TaskQ

func PoolWithQueue

func PoolWithQueue(size int, q Queue) *TaskQ

func (*TaskQ) Close

func (t *TaskQ) Close() error

func (*TaskQ) Enqueue

func (t *TaskQ) Enqueue(ctx context.Context, task Task) (int64, error)

func (*TaskQ) Shutdown

func (t *TaskQ) Shutdown(ctx context.Context) error

func (*TaskQ) Start

func (t *TaskQ) Start() error

type WaitGroup

type WaitGroup struct {
	// contains filtered or unexported fields
}

func NewWaitGroup

func NewWaitGroup(size int) *WaitGroup

func (*WaitGroup) Enqueue

func (wg *WaitGroup) Enqueue(ctx context.Context, t Task) (int64, error)

func (*WaitGroup) Wait

func (wg *WaitGroup) Wait()