threadpool

README

Golang Threadpool implementation

Scalable threadpool implementation using Go to handle the huge network trafic.

Install

go get github.com/shettyh/threadpool

Usage

Threadpool

• Implement Runnable interface for tha task that needs to be executed. For example

  type MyTask struct { }
  
  func (t *MyTask) Run(){
    // Do your task here
  }
  
  

• Create instance of ThreadPool with number of workers required and the task queue size

  pool := threadpool.NewThreadPool(200,1000000)
  

• Create Task and execute

  task:=&MyTask{}
  err := pool.Execute(task)
  

• Using Callable task

  type MyTaskCallable struct { }
  
  func (c *MyTaskCallable) Call() interface{} {
    //Do task 
    return result
  }
  
  //Execute callable task
  task := &MyTaskCallable{}
  future, err := pool.ExecuteFuture(task)
  
  //Check if the task is done
  isDone := future.IsDone() // true/false
  
  //Get response , blocking call
  result := future.Get()
  
  

• Close the pool

  pool.Close()
  

Scheduled threadpool

• Create instance of ScheduledThreadPool with number of workers required

  schedulerPool:= threadpool.NewScheduledThreadPool(10)
  

• Create Task and schedule

  task:=&MyTask{}
  pool.ScheduleOnce(task, time.Second*20) // Time delay is in seconds only as of now
  

• Close the pool

  pool.Close()
  

Documentation

Index

• Variables
• type Callable
• type Future
  • func (f *Future) Get() interface{}
  • func (f *Future) IsDone() bool
• type Runnable
• type ScheduledThreadPool
  • func NewScheduledThreadPool(noOfWorkers int) *ScheduledThreadPool
  • func (stf *ScheduledThreadPool) Close()
  • func (stf *ScheduledThreadPool) ScheduleOnce(task Runnable, delay time.Duration)
• type ThreadPool
  • func NewThreadPool(noOfWorkers int, queueSize int64) *ThreadPool
  • func (t *ThreadPool) Close()
  • func (t *ThreadPool) Execute(task Runnable) error
  • func (t *ThreadPool) ExecuteFuture(task Callable) (*Future, error)
• type Worker
  • func NewWorker(workerPool chan chan interface{}, closeHandle chan bool) *Worker
  • func (w Worker) Start()

Variables

var (
	ErrQueueFull = fmt.Errorf("queue is full, not able add the task")
)

Types

type Callable

type Callable interface {
	Call() interface{}
}

Callable the tasks which returns the output after exit should implement this interface

type Future

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

Future is the handle returned after submitting a callable task to the thread threadpool

func (*Future) Get

func (f *Future) Get() interface{}

Get returns the response of the Callable task when done Is is the blocking call it waits for the execution to complete

func (*Future) IsDone

func (f *Future) IsDone() bool

IsDone returns true if the execution is already done

type Runnable

type Runnable interface {
	Run()
}

Runnable is interface for the jobs that will be executed by the threadpool

type ScheduledThreadPool

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

ScheduledThreadPool Schedules the task with the given delay

func NewScheduledThreadPool

func NewScheduledThreadPool(noOfWorkers int) *ScheduledThreadPool

NewScheduledThreadPool creates new scheduler thread threadpool with given number of workers

func (*ScheduledThreadPool) Close

func (stf *ScheduledThreadPool) Close()

Close will close the thread threadpool TODO: check the existing task before closing

func (*ScheduledThreadPool) ScheduleOnce

func (stf *ScheduledThreadPool) ScheduleOnce(task Runnable, delay time.Duration)

ScheduleOnce the task with given delay

type ThreadPool

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

ThreadPool type for holding the workers and handle the job requests

func NewThreadPool

func NewThreadPool(noOfWorkers int, queueSize int64) *ThreadPool

NewThreadPool creates thread threadpool

func (*ThreadPool) Close

func (t *ThreadPool) Close()

Close will close the threadpool It sends the stop signal to all the worker that are running TODO: need to check the existing /running task before closing the threadpool

func (*ThreadPool) Execute

func (t *ThreadPool) Execute(task Runnable) error

Execute submits the job to available worker

func (*ThreadPool) ExecuteFuture

func (t *ThreadPool) ExecuteFuture(task Callable) (*Future, error)

ExecuteFuture will submit the task to the threadpool and return the response handle

type Worker

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

Worker type holds the job channel and passed worker threadpool

func NewWorker

func NewWorker(workerPool chan chan interface{}, closeHandle chan bool) *Worker

NewWorker creates the new worker

func (Worker) Start

func (w Worker) Start()

Start starts the worker by listening to the job channel