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
	}

	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.