package worker
import (
"context"
"errors"
"log"
"sync"
)
var ErrPoolClosed = errors.New("worker: pool is shutting down")
type Job func()
type Pool struct {
jobs chan Job
wg sync.WaitGroup
}
func NewPool(ctx context.Context, size int) *Pool {
p := &Pool{jobs: make(chan Job)}
for i := 0; i < size; i++ {
p.wg.Add(1)
go p.worker(ctx, i)
}
return p
}
func (p *Pool) worker(ctx context.Context, id int) {
defer p.wg.Done()
for {
select {
case <-ctx.Done():
log.Printf("worker %d: stopping (%v)", id, ctx.Err())
return
case job := <-p.jobs:
job()
}
}
}
func (p *Pool) Submit(ctx context.Context, job Job) error {
select {
case <-ctx.Done():
return ErrPoolClosed
case p.jobs <- job:
return nil
}
}
func (p *Pool) Wait() {
p.wg.Wait()
}
package main
import (
"context"
"fmt"
"os/signal"
"syscall"
"time"
"example.com/app/worker"
)
func main() {
ctx, stop := signal.NotifyContext(
context.Background(),
syscall.SIGINT, syscall.SIGTERM,
)
defer stop()
pool := worker.NewPool(ctx, 4)
// Block on drain only after cancellation has been requested.
defer pool.Wait()
defer stop()
for i := 0; i < 20; i++ {
n := i
err := pool.Submit(ctx, func() {
time.Sleep(200 * time.Millisecond)
fmt.Printf("processed job %d\n", n)
})
if err != nil {
fmt.Printf("submit %d aborted: %v\n", n, err)
break
}
}
<-ctx.Done()
fmt.Println("shutdown signal received, draining workers...")
}
This snippet demonstrates the canonical Go pattern for broadcasting a shutdown signal to a set of goroutines using a closed channel, wrapped in the more modern context.Context machinery. The core idea is that a chan struct{} (or the Done() channel of a context) can be observed by any number of goroutines simultaneously: closing a channel unblocks every pending receive, so one close fans out a stop signal to the whole pool without needing to know how many listeners exist.
In pool.go, the Pool type spins up size workers, each running worker in its own goroutine. Every worker uses a select that races real work (jobs channel) against ctx.Done(). When the context is cancelled, ctx.Done() becomes readable and the worker returns cleanly. This is why cancellation is checked inside the loop rather than only at the top — a worker blocked on a slow job or a receive must still be able to wake up and exit. The sync.WaitGroup guarantees Wait does not return until every goroutine has actually finished draining, which is the difference between signalling shutdown and completing it.
The subtlety in Submit is that sending on jobs is itself guarded by a select on ctx.Done(). Without that guard, a caller could block forever trying to enqueue work into a pool that is already shutting down, since no worker would ever receive it. Returning ErrPoolClosed surfaces that race explicitly instead of deadlocking.
In main.go, signal.NotifyContext ties the whole tree to SIGINT/SIGTERM, so pressing Ctrl-C cancels the root context, which cascades to every worker's ctx.Done(). The defer stop() and defer pool.Wait() ordering ensures the program requests shutdown, then blocks until in-flight jobs drain. This pattern is the standard way to wire OS signals to goroutine lifecycles, and it scales from two workers to thousands because the cost of the broadcast is a single channel close regardless of listener count. The main pitfall to avoid is closing the same channel twice, which panics — using a context sidesteps that entirely since context is safe to cancel repeatedly.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
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.
