A Minimal 70-Line Go Coroutine Pool: Simple and Elegant

The author offers a lightweight coroutine (goroutine) pool written in Go, aiming to avoid complex wrappers while still providing essential features such as bounded concurrency, task queuing, and optional timeout handling.

The core data structure is

type Pool struct { sem chan struct{} // limits the number of active goroutines taskQueue chan func() // buffered queue of pending tasks }

. The semaphore channel sem enforces the maximum number of concurrent workers, while taskQueue stores tasks awaiting execution. func NewPool(capacity int, queueSize int) *Pool validates the parameters, creates the two channels, and, if queueSize is greater than zero, immediately launches one worker goroutine to start consuming tasks from the queue. This pre‑start step ensures that a buffered queue can be drained as soon as tasks arrive.

The public methods Exec(task func()) and ExecTimeout(task func(), duration time.Duration) error are thin wrappers around the internal exec function. ExecTimeout supplies a timeout channel generated by time.After(duration) so that the caller can be notified when the pool cannot accept more work.

The exec method contains the main scheduling logic using a select statement with three cases:

Attempt to send the task to taskQueue. If the queue is not full, the task is accepted and the method returns nil.

If the queue is full, the pool tries to acquire a semaphore slot, launches a new worker via go p.goWorker(task), and then returns nil. This dynamically increases concurrency when the backlog grows.

If a timeout channel fires first, the method returns the predefined error ErrExecTimeOut, indicating that both the queue and the worker limit are saturated.

The worker routine func (p *Pool) goWorker(task func()) defers a release of the semaphore slot, executes the initial task, and then continuously receives further tasks from taskQueue in a for range loop, processing each until the channel is closed.

Full source code (70 lines):

package tinygpool

import (
    "errors"
    "time"
)

var (
    ErrExecTimeOut = errors.New("exec time out")
)

type Pool struct {
    sem       chan struct{} // maximum number of goroutines > 0
    taskQueue chan func()   // task queue >= 0
}

func NewPool(capacity int, queueSize int) *Pool {
    if capacity <= 0 || queueSize < 0 {
        panic("capacity or queueSize is invalid")
    }
    p := Pool{
        sem:       make(chan struct{}, capacity),
        taskQueue: make(chan func(), queueSize),
    }
    // If the queue has buffer, start one goroutine immediately to wait for tasks
    if queueSize > 0 {
        p.sem <- struct{}{}
        go p.goWorker(func() {})
    }
    return &p
}

func (p *Pool) Exec(task func()) {
    p.exec(task, nil)
}

func (p *Pool) ExecTimeout(task func(), duration time.Duration) error {
    return p.exec(task, time.After(duration))
}

func (p *Pool) exec(task func(), timeout <-chan time.Time) error {
    select {
    case p.taskQueue <- task: // queue not full, accept task
        return nil
    case p.sem <- struct{}{}: // queue full, start a new goroutine
        go p.goWorker(task)
    case <-timeout: // both queue and workers saturated
        return ErrExecTimeOut
    }
    return nil
}

func (p *Pool) goWorker(task func()) {
    defer func() { <-p.sem }()
    // execute initial task
    task()
    // listen for more tasks
    for task := range p.taskQueue {
        task()
    }
}