koi

README

Typed worker pools for Go, with generic result mapping

Koi turns an ordinary function into a named pool of goroutines. You register a worker on a pond, push inputs at it, and read typed results back — no channel plumbing, no any, no hand-rolled sync.WaitGroup, and no goroutines left running when you are done.

pond := koi.NewPond[int, int]()
pond.MustRegisterWorker("square", koi.MustNewWorker(func(i int) int { return i * i }, 10, 4))

pond.AddWork("square", 7)
fmt.Println(<-pond.ResultChan("square")) // 49

pond.Close()

Why koi

• Typed end to end. Pond[T, E] and Worker[T, E] carry your request and result types. No any, no type assertions, no casting results back.
• Named workers. Register as many workers as you like on one pond and reach each by id, instead of juggling a channel pair per job kind.
• Per-worker concurrency. Every worker sets its own queue size and how many goroutines drain that queue.
• Graceful shutdown. Close stops the workers, waits for in-flight work, and closes the result channels. It is idempotent, and calls that arrive after it fail with ErrPondClosed instead of panicking on a closed channel.
• Fire and forget. Type a worker's result as koi.NoReturn and results are dropped at the source — no channel to drain, no goroutine blocked on a receiver that never comes.
• Generic result mapping. MapResults is a Go 1.27 generic method: it introduces its own type parameter, so a Pond[int, int] can hand you a <-chan string without the pond ever knowing about strings.

Installation

Koi needs Go 1.27 or newer — it uses generic methods, which landed in 1.27.

go get github.com/1995parham/koi

Usage

Fire and forget

Use koi.NoReturn as the result type when the work is its own reward. Results are never published, so there is nothing to drain.

package main

import (
	"log"
	"sync"
	"time"

	"github.com/1995parham/koi"
)

func main() {
	pond := koi.NewPond[int, koi.NoReturn]()

	var wg sync.WaitGroup

	printer := func(a int) koi.NoReturn {
		time.Sleep(1 * time.Second)
		log.Println(a)

		wg.Done()

		return koi.None
	}

	//nolint:mnd
	printWorker := koi.MustNewWorker(printer, 2, 10)

	pond.MustRegisterWorker("printer", printWorker)

	for i := range 10 {
		wg.Add(1)

		if _, err := pond.AddWork("printer", i); err != nil {
			log.Printf("error while adding job: %s\n", err)
		}
	}

	wg.Wait()

	// stop the workers and release their goroutines.
	pond.Close()

	log.Println("all jobs done")
}

Collecting results

When the worker returns something, read it from ResultChan, or transform it on the way out with MapResults — the generic method that gives this library its Go 1.27 requirement.

package main

import (
	"log"
	"strconv"

	"github.com/1995parham/koi"
)

const jobs = 5

func main() {
	pond := koi.NewPond[int, int]()
	defer pond.Close()

	square := func(i int) int {
		return i * i
	}

	pond.MustRegisterWorker("square", koi.MustNewWorker(square, jobs, jobs))

	// MapResults is a go1.27 generic method: U is inferred from the function,
	// so a Pond[int, int] can hand back a <-chan string without the pond ever
	// knowing about strings.
	labels := pond.MapResults("square", strconv.Itoa)

	for i := range jobs {
		if _, err := pond.AddWork("square", i); err != nil {
			log.Printf("error while adding job: %s\n", err)
		}
	}

	for range jobs {
		log.Println(<-labels)
	}
}

Both programs live in example/ and are built on every CI run.

API

NewPond[T, E]()	create an empty pond
RegisterWorker(id, w) · MustRegisterWorker	validate and start a worker, addressable by id
AddWork(id, req)	enqueue a request; returns that worker's result channel
ResultChan(id)	the worker's result channel, or nil if unknown
MapResults[U](id, fn)	a <-chan U of results passed through fn (generic method)
Close()	stop every worker, drain in-flight work, close result channels
NewWorker[T, E](work, queueSize, concurrentCount) · MustNewWorker	build a worker from a plain func(T) E
koi.NoReturn · koi.None	result type and value for workers that produce nothing

Failures surface as ErrWorkerNotFound, ErrPondClosed, and ErrMinConcurrentCount.

Semantics worth knowing

• AddWork is non-blocking unless the worker's queue is full — the queue size is your backpressure knob.
• Read a worker's output through either ResultChan or MapResults, not both: they consume from the same channel.
• MapResults closes its channel once the worker's results drain, i.e. after Close, so range over it terminates cleanly.
• A pond is safe for concurrent use.

Terminology

• Koi: an informal name for the colored variants of C. rubrofuscus kept for ornamental purposes.
• Pond: an area of water smaller than a lake, often artificially made.

Credits

Koi began as a fork of mehditeymorian/koi and keeps its name and its spirit.

License

Apache 2.0 — see LICENSE.

Documentation

Overview

Package koi is a generic goroutine and worker manager. A Pond owns a set of named Workers, routes requests to them, and shuts them down gracefully.

Index

• Variables
• type NoReturn
• type Pond
  • func NewPond[T any, E any]() *Pond[T, E]
  • func (p *Pond[T, E]) AddWork(workerID string, request T) (<-chan E, error)
  • func (p *Pond[T, E]) Close()
  • func (p *Pond[T, E]) MapResults[U any](workerID string, fn func(E) U) <-chan U
  • func (p *Pond[T, E]) MustRegisterWorker(id string, worker *Worker[T, E])
  • func (p *Pond[T, E]) RegisterWorker(id string, worker *Worker[T, E]) error
  • func (p *Pond[T, E]) ResultChan(workerID string) <-chan E
• type Worker
  • func MustNewWorker[T any, E any](work func(T) E, queueSize uint, concurrentCount int) *Worker[T, E]
  • func NewWorker[T any, E any](work func(T) E, queueSize uint, concurrentCount int) (*Worker[T, E], error)
  • func (w *Worker[T, E]) Validate() error

Variables

var (
	// ErrWorkerNotFound is returned when no worker is registered under the given id.
	ErrWorkerNotFound = errors.New("worker not found")
	// ErrMinConcurrentCount is returned when a worker would run with fewer than one goroutine.
	ErrMinConcurrentCount = errors.New("concurrent count must be at least 1")
	// ErrPondClosed is returned by operations on a pond that is already closed.
	ErrPondClosed = errors.New("pond is closed")
)

Types

type NoReturn

type NoReturn int

NoReturn is the result type for fire-and-forget workers that produce no meaningful output. Use it as the E type parameter together with None.

const None NoReturn = 0

None is the canonical NoReturn value returned by workers without a result.

type Pond

type Pond[T any, E any] struct {
	// contains filtered or unexported fields
}

Pond owns a set of named workers and routes work to them. It is safe for concurrent use.

func NewPond

func NewPond[T any, E any]() *Pond[T, E]

NewPond creates an empty pond ready to accept workers.

func (*Pond[T, E]) AddWork

func (p *Pond[T, E]) AddWork(workerID string, request T) (<-chan E, error)

AddWork enqueues request for the worker registered under workerID and returns that worker's result channel.

func (*Pond[T, E]) Close

func (p *Pond[T, E]) Close()

Close stops every worker, waits for in-flight work to finish, and closes each worker's result channel. After Close returns, AddWork and RegisterWorker fail with ErrPondClosed. Close is idempotent.

func (*Pond[T, E]) MapResults

func (p *Pond[T, E]) MapResults[U any](workerID string, fn func(E) U) <-chan U

MapResults returns a channel that yields fn applied to each result produced by the worker registered under workerID, or nil if no such worker exists.

The result type U is chosen per call and is independent of the pond's own result type E: MapResults is a Go 1.27 generic method, so U lives in the method's scope rather than the package's. The returned channel is closed once the worker's result channel drains, i.e. after Close.

MapResults consumes from the worker's result channel, so a given worker's results should be read either through MapResults or through ResultChan, not both.

func (*Pond[T, E]) MustRegisterWorker

func (p *Pond[T, E]) MustRegisterWorker(id string, worker *Worker[T, E])

MustRegisterWorker is like RegisterWorker but panics on error.

func (*Pond[T, E]) RegisterWorker

func (p *Pond[T, E]) RegisterWorker(id string, worker *Worker[T, E]) error

RegisterWorker validates and starts the worker, making it addressable by id.

func (*Pond[T, E]) ResultChan

func (p *Pond[T, E]) ResultChan(workerID string) <-chan E

ResultChan returns the result channel of the worker registered under workerID, or nil if no such worker exists.

type Worker

type Worker[T any, E any] struct {
	QueueSize       uint
	ConcurrentCount int
	Work            func(T) E

	ResultChan  chan E
	RequestChan chan T
	// contains filtered or unexported fields
}

Worker runs Work concurrently over requests received on RequestChan and, unless its result type is NoReturn, publishes results on ResultChan.

func MustNewWorker

func MustNewWorker[T any, E any](work func(T) E, queueSize uint, concurrentCount int) *Worker[T, E]

MustNewWorker is like NewWorker but panics on a validation error.

func NewWorker

func NewWorker[T any, E any](work func(T) E, queueSize uint, concurrentCount int) (*Worker[T, E], error)

NewWorker creates and validates a Worker. queueSize sets the buffer of both the request and result channels; concurrentCount sets how many goroutines process requests in parallel and must be at least 1.

func (*Worker[T, E]) Validate

func (w *Worker[T, E]) Validate() error

Validate reports whether the worker is configured correctly.