priopool

README

Goroutines pool with priority queue buffer.

---

Overview

Package priopool provides goroutines pool based on panjf2000/ants library with priority queue buffer based on stdlib heap package.

Priority pool:

• is non-blocking,
• prioritizes tasks with higher priority value,
• can be configured with unlimited queue buffer.

Install

go get -u github.com/alexvanin/priopool

Example

package main

import (
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/alexvanin/priopool"
)

func main() {
	regularJob := func(i int) {
		time.Sleep(1 * time.Second)
		fmt.Printf("Job %d is done\n", i)
	}

	highPriorityJob := func() {
		fmt.Println("High priority job is done")
	}

	pool, err := priopool.New(2, -1) // pool for two parallel executions
	if err != nil {
		log.Fatal(err)
	}

	wg := new(sync.WaitGroup)
	wg.Add(5 + 1)
	for i := 0; i < 5; i++ {
		ind := i + 1
		// enqueue 5 regular jobs
		pool.Submit(1, func() { regularJob(ind); wg.Done() })
	}
	// after 5 regular jobs enqueue high priority job
	pool.Submit(10, func() { highPriorityJob(); wg.Done() })
	wg.Wait()

	/*
		Output:
		Job 2 is done
		Job 1 is done
		High priority job is done
		Job 4 is done
		Job 3 is done
		Job 5 is done
	*/
}

License

Source code is available under the MIT License.

Documentation

Index

• Variables
• type PriorityPool
  • func New(poolCapacity, queueCapacity int) (*PriorityPool, error)
  • func (p *PriorityPool) Submit(priority uint32, task func()) error

Variables

var (
	// ErrQueueOverload will be returned on submit operation
	// when both goroutines pool and priority queue are full.
	ErrQueueOverload = errors.New("pool and priority queue are full")

	// ErrPoolCapacitySize will be returned when constructor
	// provided with non-positive pool capacity.
	ErrPoolCapacitySize = errors.New("pool capacity must be positive")
)

Types

type PriorityPool

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

PriorityPool is a pool of goroutines with priority queue buffer. Based on panjf2000/ants and stdlib heap libraries.

func New

func New(poolCapacity, queueCapacity int) (*PriorityPool, error)

New creates instance of priority pool. Pool capacity must be positive. Zero queue capacity disables priority queue. Negative queue capacity disables priority queue length limit.

func (*PriorityPool) Submit

func (p *PriorityPool) Submit(priority uint32, task func()) error

Submit sends the task into priority pool. Non-blocking operation. If pool has available workers, then task executes immediately. If pool is full, then task is stored in priority queue. It will be executed when available worker pops the task from priority queue. Tasks from queue do not evict running tasks from pool. Tasks with bigger priority number are popped earlier. If queue is full, submit returns ErrQueueOverload error.