workerpool

package module

v2.0.0 Latest Latest Go to latest Published: Mar 26, 2020 License: MIT Imports: 3 Imported by: 3

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/vardius/worker-pool

Links

Open Source Insights

README ¶

👨‍🔧 worker-pool

Go simple async worker pool.

📖 ABOUT

Contributors:

Rafał Lorenz

Want to contribute ? Feel free to send pull requests!

Have problems, bugs, feature ideas? We are using the github issue tracker to manage them.

📚 Documentation

For examples visit godoc#pkg-examples

For GoDoc reference, visit pkg.go.dev

🚏 HOW TO USE

🚅 Benchmark

CPU: 3,3 GHz Intel Core i7

RAM: 16 GB 2133 MHz LPDDR3

➜  worker-pool git:(master) ✗ go test -bench=. -cpu=4 -benchmem
goos: darwin
goarch: amd64
pkg: github.com/vardius/worker-pool/v2
BenchmarkWorker1-4                	 3435846	       358 ns/op	     152 B/op	       4 allocs/op
BenchmarkWorker1Parallel-4        	 2993271	       403 ns/op	     144 B/op	       3 allocs/op
BenchmarkWorker100-4              	 2140670	       543 ns/op	     152 B/op	       4 allocs/op
BenchmarkWorker100Parallel-4      	 3379311	       332 ns/op	     144 B/op	       3 allocs/op
BenchmarkWorkerNumCPU-4           	 2536502	       438 ns/op	     152 B/op	       4 allocs/op
BenchmarkWorkerNumCPUParallel-4   	 3061671	       353 ns/op	     144 B/op	       3 allocs/op
PASS
ok  	github.com/vardius/worker-pool/v2	9.590s

🏫 Basic example

package main

import (
    "fmt"
    "sync"

    "github.com/vardius/worker-pool/v2"
)

func main() {
	var wg sync.WaitGroup

	poolSize := 1
	jobsAmount := 3
	workersAmount := 2

	// create new pool
	pool := workerpool.New(poolSize)
	out := make(chan int, jobsAmount)
	worker := func(i int) {
        defer wg.Done()
        out <- i
    }

	for i := 1; i <= workersAmount; i++ {
		if err := pool.AddWorker(worker); err != nil {
			panic(err)
		}
	}

	wg.Add(jobsAmount)

	for i := 0; i < jobsAmount; i++ {
		pool.Delegate(i)
	}

	go func() {
		// stop all workers after jobs are done
		wg.Wait()
		close(out)
		pool.Stop()
	}()

	sum := 0
	for n := range out {
		sum += n
	}

	fmt.Println(sum)
	// Output:
	// 3
}

📜 License

This package is released under the MIT license. See the complete license in the package

Documentation ¶

Overview ¶

Package workerpool provides simple async workers

Example ¶

package main

import (
	"fmt"
	"sync"

	workerpool "github.com/vardius/worker-pool/v2"
)

func main() {
	var wg sync.WaitGroup

	poolSize := 1
	jobsAmount := 3
	workersAmount := 2

	// create new pool
	pool := workerpool.New(poolSize)
	out := make(chan int, jobsAmount)
	worker := func(i int) {
		defer wg.Done()
		out <- i
	}

	for i := 1; i <= workersAmount; i++ {
		if err := pool.AddWorker(worker); err != nil {
			panic(err)
		}
	}

	wg.Add(jobsAmount)

	for i := 0; i < jobsAmount; i++ {
		pool.Delegate(i)
	}

	go func() {
		// stop all workers after jobs are done
		wg.Wait()
		close(out)
		pool.Stop()
	}()

	sum := 0
	for n := range out {
		sum += n
	}

	fmt.Println(sum)
}

Output:

3

Example (Second) ¶

package main

import (
	"fmt"

	workerpool "github.com/vardius/worker-pool/v2"
)

func main() {
	poolSize := 2
	jobsAmount := 8
	workersAmount := 3

	ch := make(chan int, jobsAmount)
	defer close(ch)

	// create new pool
	pool := workerpool.New(poolSize)
	defer pool.Stop()

	worker := func(i int, out chan<- int) { out <- i }

	for i := 1; i <= workersAmount; i++ {
		if err := pool.AddWorker(worker); err != nil {
			panic(err)
		}
	}

	go func() {
		for n := 0; n < jobsAmount; n++ {
			pool.Delegate(n, ch)
		}
	}()

	var sum = 0
	for sum < jobsAmount {
		select {
		case <-ch:
			sum++
		}
	}

	fmt.Println(sum)
}

Output:

8

Example (Third) ¶

package main

import (
	"fmt"
	"strconv"
	"sync"

	workerpool "github.com/vardius/worker-pool/v2"
)

func main() {
	poolSize := 2
	jobsAmount := 8
	workersAmount := 3

	var wg sync.WaitGroup
	wg.Add(jobsAmount)

	// allocate queue
	pool := workerpool.New(poolSize)

	// moc arg
	argx := make([]string, jobsAmount)
	for j := 0; j < jobsAmount; j++ {
		argx[j] = "_" + strconv.Itoa(j) + "_"
	}

	// assign job
	for i := 0; i < jobsAmount; i++ {
		go func(i int) {
			pool.Delegate(argx[i])
		}(i)
	}

	worker := func(s string) {
		defer wg.Done()
		defer fmt.Println("job " + s + " is done !")
		fmt.Println("job " + s + " is running ..")
	}

	// start workers
	for i := 1; i <= workersAmount; i++ {
		if err := pool.AddWorker(worker); err != nil {
			panic(err)
		}
	}

	// clean up
	wg.Wait()
	pool.Stop()

	// fmt.Println("# hi: ok?")

Examples ¶

Package
Package (Second)
Package (Third)

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type Pool ¶

type Pool interface {
	// Delegate job to a workers
	// will block if channel is full, you might want to wrap it with goroutine to avoid it
	// will panic if called after Stop()
	Delegate(args ...interface{})

	// AddWorker adds worker to the pool
	AddWorker(fn interface{}) error
	// RemoveWorker removes worker from the pool
	RemoveWorker(fn interface{}) error

	// WorkersNum returns number of workers in the pool
	WorkersNum() int
	// Stop all workers
	Stop()
}

Pool implements worker pool

func New ¶

func New(queueLength int) Pool

New creates new worker pool with a given job queue length

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL