package workpool
import (
"context"
"sync"
)
type Pool struct {
jobs chan Job
results chan Result
wg sync.WaitGroup
size int
}
func New(size, buffer int) *Pool {
return &Pool{
jobs: make(chan Job, buffer),
results: make(chan Result, buffer),
size: size,
}
}
func (p *Pool) Start(ctx context.Context) {
for i := 0; i < p.size; i++ {
p.wg.Add(1)
go p.worker(ctx, i)
}
}
func (p *Pool) worker(ctx context.Context, id int) {
defer p.wg.Done()
for job := range p.jobs {
select {
case <-ctx.Done():
return
default:
p.results <- process(id, job)
}
}
}
func (p *Pool) Submit(ctx context.Context, job Job) error {
select {
case p.jobs <- job:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (p *Pool) Results() <-chan Result {
return p.results
}
func (p *Pool) Close() {
close(p.jobs)
p.wg.Wait()
close(p.results)
}
package workpool
import (
"fmt"
"time"
)
type Job struct {
ID int
Payload string
}
type Result struct {
JobID int
WorkerID int
Output string
Err error
}
func process(workerID int, job Job) Result {
time.Sleep(50 * time.Millisecond) // simulate real work
if job.Payload == "" {
return Result{JobID: job.ID, WorkerID: workerID, Err: fmt.Errorf("empty payload for job %d", job.ID)}
}
return Result{
JobID: job.ID,
WorkerID: workerID,
Output: fmt.Sprintf("processed %q", job.Payload),
}
}
package main
import (
"context"
"fmt"
"log"
"time"
"example.com/app/workpool"
)
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
pool := workpool.New(4, 16)
pool.Start(ctx)
go func() {
defer pool.Close()
for i := 1; i <= 20; i++ {
job := workpool.Job{ID: i, Payload: fmt.Sprintf("task-%d", i)}
if err := pool.Submit(ctx, job); err != nil {
log.Printf("submit aborted: %v", err)
return
}
}
}()
for res := range pool.Results() {
if res.Err != nil {
log.Printf("job %d failed: %v", res.JobID, res.Err)
continue
}
fmt.Printf("worker %d -> %s\n", res.WorkerID, res.Output)
}
}
This snippet shows the classic Go fan-out worker pool: a fixed number of goroutines drain a shared buffered channel of jobs while a single sync.WaitGroup coordinates shutdown. The pattern is useful whenever there is more work than it makes sense to run concurrently — HTTP fetches, image resizes, database writes — and unbounded goroutine spawning would exhaust memory or overwhelm a downstream service. Bounding the pool size caps concurrency, and the buffered channel provides backpressure so producers block once the queue is full instead of piling up work indefinitely.
In pool.go, the Pool owns a jobs channel sized by buffer and a results channel. Start launches size goroutines, each running worker, and increments the WaitGroup once per goroutine. Each worker ranges over jobs, which is the idiomatic way to consume a channel until it is closed — the loop exits automatically when Submit/Close closes p.jobs, so no sentinel value is needed. The select inside the loop also watches ctx.Done(), allowing an in-flight batch to abort early on cancellation rather than finishing the entire queue.
Submit is deliberately guarded by select against ctx.Done() so a caller isn't blocked forever pushing into a full channel during shutdown; it returns ctx.Err() instead. Close closes p.jobs to signal there is no more work, waits for every worker to finish via wg.Wait(), and only then closes results. Closing results after the wait is critical: closing it while a worker might still send would panic. This ordering — close input, wait, close output — is the safe teardown recipe for any pipeline stage.
In job.go, Job and Result are plain structs; process simulates real work and returns an error that is threaded into Result so failures are observable per job rather than crashing a worker. main.go wires it together: it feeds jobs in a separate goroutine so the producer and the result-draining loop run concurrently, avoiding a deadlock where a full results channel would stall the workers. A context.WithTimeout bounds total runtime, demonstrating cooperative cancellation. The trade-off is that jobs already dequeued when the deadline hits still attempt to run unless process itself checks the context.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler
from sklearn.linear_model import LogisticRegression
standard_pipeline = Pipeline([
('scaler', StandardScaler()),Scaling and normalization choices for different model families
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
