package pipeline
import "context"
func generate(ctx context.Context, nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
select {
case out <- n:
case <-ctx.Done():
return
}
}
}()
return out
}
func square(ctx context.Context, in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
select {
case out <- n * n:
case <-ctx.Done():
return
}
}
}()
return out
}package pipeline
import (
"context"
"sync"
)
func fanOut(ctx context.Context, in <-chan int, workers int) []<-chan int {
outs := make([]<-chan int, workers)
for i := 0; i < workers; i++ {
outs[i] = square(ctx, in)
}
return outs
}
func fanIn(ctx context.Context, ins ...<-chan int) <-chan int {
var wg sync.WaitGroup
merged := make(chan int)
forward := func(c <-chan int) {
defer wg.Done()
for v := range c {
select {
case merged <- v:
case <-ctx.Done():
return
}
}
}
wg.Add(len(ins))
for _, c := range ins {
go forward(c)
}
go func() {
wg.Wait()
close(merged)
}()
return merged
}package pipeline
import (
"context"
"fmt"
"time"
)
func Run() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
source := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8)
workers := fanOut(ctx, source, 3)
results := fanIn(ctx, workers...)
sum := 0
for v := range results {
sum += v
fmt.Printf("received: %d\n", v)
}
if err := ctx.Err(); err != nil {
fmt.Printf("pipeline halted early: %v\n", err)
}
fmt.Printf("total: %d\n", sum)
}This snippet builds a three-stage concurrent pipeline in Go where each stage runs in its own goroutine (or pool of goroutines) and communicates only through channels. The pipeline pattern decomposes a data-processing job into independent stages — generate, transform, collect — so each stage can run concurrently and back-pressure flows naturally through unbuffered or small-buffered channels. Because each stage reads from an input channel and writes to an output channel, stages are decoupled and composable, and the runtime schedules them across available cores.
In stages.go, generate is the source stage: it emits work items onto a channel and closes it when done, which signals completion downstream. The critical detail is the select on ctx.Done() alongside the send — without it, a blocked send on a stalled channel would leak the goroutine forever if a consumer stops reading. square is a stateless transform stage that ranges over its input until the channel is closed, then closes its own output, propagating the end-of-stream signal.
The fan-out/fan-in mechanism lives in fanin.go. fanOut starts n identical worker goroutines all reading from the same input channel, letting the runtime distribute items to whichever worker is free — a simple load-balancing worker pool. fanIn merges the multiple worker outputs back into one channel using a sync.WaitGroup to know when every source is drained, then closes the merged channel exactly once. This close-after-Wait idiom is the canonical way to avoid closing a channel twice or closing it while a sender is still active.
In main.go, context.WithTimeout provides a single cancellation signal threaded through every stage, so a timeout or an early return unwinds the whole pipeline cleanly rather than leaking goroutines. The stages are wired together as generate into fanOut into fanIn, and the final for range drives the pipeline by consuming results. The key trade-offs: unbuffered channels give tight back-pressure but more context switches, while the explicit ctx plumbing adds boilerplate but guarantees no goroutine outlives the work. This structure is worth reaching for when a CPU- or IO-bound job has clear sequential phases that benefit from parallelism within a phase.
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
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
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
