Goroutines and Channels

The pipeline pattern

Pipelines chain goroutines together via channels — each stage receives from an upstream channel, processes the data, and sends to a downstream channel. This is the Go idiom for streaming data processing. Each stage can run in parallel with others.

package main

import "fmt"

// generate: produces integers on a channel
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)    // close signals: no more values to send
    }()
    return out
}

// square: reads from in, sends squares to out
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {    // range on channel reads until close
            out <- n * n
        }
        close(out)
    }()
    return out
}

func main() {
    // Set up the pipeline: generate -> square -> print
    c := generate(2, 3, 4, 5)
    out := square(c)

    for n := range out {
        fmt.Println(n)    // 4, 9, 16, 25
    }
}

Fan-out and fan-in

Fan-out: distribute work from one channel to multiple goroutines (parallelism). Fan-in: merge results from multiple channels into one. Together they build work-stealing-style worker pools.

package main

import (
    "fmt"
    "sync"
)

// worker receives jobs from a channel and sends results
func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {
        results <- j * j    // compute square
        fmt.Printf("worker %d processed job %d\n", id, j)
    }
}

func main() {
    const numWorkers = 3
    jobs := make(chan int, 10)
    results := make(chan int, 10)

    var wg sync.WaitGroup

    // Fan-out: start 3 workers reading from the same jobs channel
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    // Send 9 jobs
    for j := 1; j <= 9; j++ {
        jobs <- j
    }
    close(jobs)    // closing jobs signals workers to stop ranging

    // Close results once all workers are done
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    for r := range results {
        fmt.Println("result:", r)
    }
}

Context: cancellation and deadlines

The context package propagates cancellation, deadlines, and request-scoped values through goroutine trees. Every goroutine doing work on behalf of a request should accept a context.Context as its first argument and check ctx.Done() to know when to stop. This prevents goroutine leaks when requests are cancelled.

package main

import (
    "context"
    "fmt"
    "time"
)

func doWork(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("worker %d stopping: %v\n", id, ctx.Err())
            return
        default:
            fmt.Printf("worker %d doing work...\n", id)
            time.Sleep(300 * time.Millisecond)
        }
    }
}

func main() {
    // WithTimeout: automatically cancels after 1 second
    ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
    defer cancel()    // always call cancel to free resources

    go doWork(ctx, 1)
    go doWork(ctx, 2)

    <-ctx.Done()    // block until context is cancelled/timed out
    fmt.Println("main: context done:", ctx.Err())  // context deadline exceeded
    time.Sleep(100 * time.Millisecond)              // let goroutines print stop messages
}

The Go scheduler: GMP model

The Go runtime scheduler uses the GMP model: G (goroutine), M (machine = OS thread), and P (processor = a scheduling context that holds a local run queue). The number of Ps defaults to GOMAXPROCS, which defaults to the number of logical CPU cores. Each P has a local run queue of goroutines waiting to be scheduled onto an M. When a P's local queue is empty, it performs work stealing — taking goroutines from another P's queue. This achieves close-to-optimal CPU utilisation without programmer intervention.

Goroutines are preempted at function call sites (cooperative preemption) and, since Go 1.14, via asynchronous preemption using OS signals — a goroutine in a tight CPU-bound loop can be preempted so other goroutines get CPU time. Stack growth is handled automatically: goroutines start with a small stack (2KB in current versions) that is grown by copying when needed (segmented stacks were replaced by copying stacks in Go 1.4).

Channel implementation: hchan and the lock-free path

Internally, a channel is an hchan struct containing: a ring buffer (for buffered channels), send and receive queues of blocked goroutines (sudog structs), and a mutex for synchronisation. When a goroutine sends on a full buffered channel or an unbuffered channel with no receiver, its sudog is enqueued on the channel's send queue and the goroutine is parked (removed from the run queue). When a receiver arrives, it directly copies data from the parked sender's stack, places the sender back on the run queue, and returns — this is the direct send optimisation that avoids copying through the buffer. This means channel communication can be O(1) in the common case and avoids unnecessary memory allocations.

Select: runtime randomisation and netpoller integration

The select statement is compiled into a call to runtime.selectgo, which shuffles the case order randomly before evaluating — this is what produces the "if multiple are ready, one is chosen uniformly at random" guarantee in the specification. Channels involved in select are locked in a consistent order (by channel address) to prevent deadlock from concurrent select statements racing on overlapping channels. The Go runtime's netpoller (based on epoll/kqueue/IOCP) integrates with the scheduler: I/O operations that would block an OS thread instead park the goroutine and wake it via the netpoller when the I/O is ready, allowing one OS thread to service many concurrent network connections.