curlew

README

Curlew

Curlew is a job pool based on a local machine.

Feature

• Automatically scale the number of workers up or down.
• Maximum job execution time.
• Don't ensure the job execution order.

Usage

package main

import (
	"context"
	"errors"
	"fmt"
	"github.com/xiaojiaoyu100/curlew"
)

func monitor(err error) {
	fmt.Println(err)
}

func main() {
	d, err := curlew.New(curlew.WithMonitor(monitor))
	if err != nil {
		fmt.Println(err)
		return
	}
	j := curlew.NewJob()
	j.Arg = 3
	j.Fn = func(ctx context.Context, arg interface{}) error {
		fmt.Println("I'm done")
		return nil
	}
	d.Submit(j)
	select {}
}

Documentation

Index

• type Dispatcher
  • func New(setters ...Setter) (*Dispatcher, error)
  • func (d *Dispatcher) RunningWorkerNum() int
  • func (d *Dispatcher) Submit(j *Job)
• type Handler
• type Job
  • func NewJob() *Job
• type Monitor
• type Setter
  • func WithJobSize(size int) Setter
  • func WithMaxWorkerNum(num int) Setter
  • func WithMonitor(monitor Monitor) Setter
  • func WithWorkerIdleTimeout(t time.Duration) Setter
• type Worker
  • func NewWorker(d *Dispatcher) *Worker
  • func (w *Worker) IsClosed() bool
  • func (w *Worker) LastBusyTime() time.Time
  • func (w *Worker) Running() bool
  • func (w *Worker) SetLastBusyTime()
  • func (w *Worker) SetRunning(b bool)

Types

type Dispatcher

type Dispatcher struct {
	MaxWorkerNum         int           // maximum  worker num in the pool
	JobSize              int           // job buffer size
	WorkerIdleTimeout    time.Duration // worker
	MaxJobRunningTimeout time.Duration // job execution timeout
	WorkerPool           chan *Worker  // worker pool
	// contains filtered or unexported fields
}

Dispatcher takes the responsibility of dispatching jobs to available workers.

func New

func New(setters ...Setter) (*Dispatcher, error)

New creates a dispatcher instance.

func (*Dispatcher) RunningWorkerNum

func (d *Dispatcher) RunningWorkerNum() int

RunningWorkerNum returns the current running worker num.

func (*Dispatcher) Submit

func (d *Dispatcher) Submit(j *Job)

Submit submits a job.

type Handler

type Handler func(ctx context.Context, arg interface{}) error

Handler provides the job function signature.

type Job

type Job struct {
	Fn  Handler
	Arg interface{}
}

Job defines a job.

func NewJob

func NewJob() *Job

NewJob creates a job instance.

type Monitor

type Monitor func(err error)

type Setter

type Setter func(d *Dispatcher) error

Setter configures a Dispatcher.

func WithJobSize

func WithJobSize(size int) Setter

WithJobSize configures job buffer size.

func WithMaxWorkerNum

func WithMaxWorkerNum(num int) Setter

WithMaxWorkerNum configures maximum number of workers in the pool.

func WithMonitor

func WithMonitor(monitor Monitor) Setter

WithMonitor configures a monitor.

func WithWorkerIdleTimeout

func WithWorkerIdleTimeout(t time.Duration) Setter

WithWorkerIdleTimeout configures wrker idle timeout.

type Worker

type Worker struct {
	Jobs chan *Job
	// contains filtered or unexported fields
}

func NewWorker

func NewWorker(d *Dispatcher) *Worker

func (*Worker) IsClosed

func (w *Worker) IsClosed() bool

func (*Worker) LastBusyTime

func (w *Worker) LastBusyTime() time.Time

func (*Worker) Running

func (w *Worker) Running() bool

func (*Worker) SetLastBusyTime

func (w *Worker) SetLastBusyTime()

func (*Worker) SetRunning

func (w *Worker) SetRunning(b bool)