Mastering Go Pipelines: Concurrency, Cancellation, and Efficient MD5 Processing

Introduction

Go's concurrency primitives make it easy to build streaming data pipelines that efficiently use I/O and multiple CPUs. This article shows pipeline examples, highlights subtle issues that arise when operations fail, and presents clean techniques for handling failures.

What Is a Pipeline?

A pipeline in Go is an informal concept consisting of a series of stages. Each stage runs a group of goroutines executing the same function. In each stage the goroutines:

receive data from an inbound channel

process the data, usually producing a new value

send the result downstream via an outbound channel

All stages have inbound and outbound channels except the first (Source/Producer) and the last (Sink/Consumer).

Squaring Numbers

Consider a three‑stage pipeline that squares numbers.

func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out)
    }()
    return out
}

func sq(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * n
        }
        close(out)
    }()
    return out
}

func main() {
    // Set up the pipeline
    c := gen(2, 3)
    out := sq(c)
    // Consume the output
    fmt.Println(<-out) // 4
    fmt.Println(<-out) // 9
}

The pipeline can be rewritten as a loop to compose stages arbitrarily.

func main() {
    for n := range sq(sq(gen(2, 3))) {
        fmt.Println(n) // 16 then 81
    }
}

Fan‑in and Fan‑out

Fan‑out distributes work among multiple workers by having several goroutines read from the same input channel. Fan‑in merges multiple input channels into a single output channel.

func main() {
    in := gen(2, 3)
    // Distribute the sq work across two goroutines that both read from in.
    c1 := sq(in)
    c2 := sq(in)
    // Consume the merged output from c1 and c2.
    for n := range merge(c1, c2) {
        fmt.Println(n) // 4 then 9, or 9 then 4
    }
}

func merge(cs ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)
    output := func(c <-chan int) {
        for n := range c {
            out <- n
        }
        wg.Done()
    }
    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Brief Pause

Each stage typically closes its outbound channel after sending all values and keeps receiving from its inbound channel until it is closed. When a stage fails to consume all inbound values, upstream goroutines can block, leading to resource leaks.

c := make(chan int, 2) // buffer size 2
c <- 1  // succeeds immediately
c <- 2  // succeeds immediately
c <- 3  // blocks until a receiver reads a value

Using a buffered channel or explicit cancellation can prevent these blocks.

Explicit Cancellation

The main function can signal upstream stages to stop by closing a shared done channel. Each sending goroutine selects between sending a value and receiving from done.

func main() {
    in := gen(2, 3)
    c1 := sq(in)
    c2 := sq(in)
    done := make(chan struct{})
    out := merge(done, c1, c2)
    fmt.Println(<-out) // 4 or 9
    close(done) // broadcast cancellation
}

func merge(done <-chan struct{}, cs ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)
    output := func(c <-chan int) {
        defer wg.Done()
        for n := range c {
            select {
            case out <- n:
            case <-done:
                return
            }
        }
    }
    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Digesting a Tree

The article then applies pipelines to compute MD5 digests of all regular files under a directory.

func MD5All(root string) (map[string][md5.Size]byte, error) {
    m := make(map[string][md5.Size]byte)
    err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
        if err != nil {
            return err
        }
        if !info.Mode().IsRegular() {
            return nil
        }
        data, err := ioutil.ReadFile(path)
        if err != nil {
            return err
        }
        m[path] = md5.Sum(data)
        return nil
    })
    if err != nil {
        return nil, err
    }
    return m, nil
}

Parallel Digest Computation

A parallel version splits the work into two stages: sumFiles walks the file tree and launches a goroutine per file to compute its MD5, sending result values on a channel; MD5All collects the results.

type result struct {
    path string
    sum  [md5.Size]byte
    err  error
}

func sumFiles(done <-chan struct{}, root string) (<-chan result, <-chan error) {
    c := make(chan result)
    errc := make(chan error, 1)
    go func() {
        var wg sync.WaitGroup
        err := filepath.Walk(root, func(path string, info fs.FileInfo, err error) error {
            if err != nil {
                return err
            }
            if !info.Mode().IsRegular() {
                return nil
            }
            wg.Add(1)
            go func() {
                data, err := ioutil.ReadFile(path)
                select {
                case c <- result{path, md5.Sum(data), err}:
                case <-done:
                }
                wg.Done()
            }()
            select {
            case <-done:
                return errors.New("walk canceled")
            default:
                return nil
            }
        })
        errc <- err
    }()
    return c, errc
}

func MD5All(root string) (map[string][md5.Size]byte, error) {
    done := make(chan struct{})
    defer close(done)
    c, errc := sumFiles(done, root)
    m := make(map[string][md5.Size]byte)
    for r := range c {
        if r.err != nil {
            return nil, r.err
        }
        m[r.path] = r.sum
    }
    if err := <-errc; err != nil {
        return nil, err
    }
    return m, nil
}

Bounded Parallelism

To avoid spawning an unbounded number of goroutines, a fixed pool of workers is used. The pipeline now has three stages: walkFiles emits file paths, a pool of digester workers reads each file and computes its MD5, and the collector gathers results.

func walkFiles(done <-chan struct{}, root string) (<-chan string, <-chan error) {
    paths := make(chan string)
    errc := make(chan error, 1)
    go func() {
        defer close(paths)
        errc <- filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
            if err != nil {
                return err
            }
            if !info.Mode().IsRegular() {
                return nil
            }
            select {
            case paths <- path:
            case <-done:
                return errors.New("walk canceled")
            }
            return nil
        })
    }()
    return paths, errc
}

func digester(done <-chan struct{}, paths <-chan string, c chan<- result) {
    for path := range paths {
        data, err := ioutil.ReadFile(path)
        select {
        case c <- result{path, md5.Sum(data), err}:
        case <-done:
            return
        }
    }
}

func MD5All(root string) (map[string][md5.Size]byte, error) {
    done := make(chan struct{})
    defer close(done)
    paths, errc := walkFiles(done, root)
    c := make(chan result)
    var wg sync.WaitGroup
    const numDigesters = 20
    wg.Add(numDigesters)
    for i := 0; i < numDigesters; i++ {
        go func() {
            digester(done, paths, c)
            wg.Done()
        }()
    }
    go func() {
        wg.Wait()
        close(c)
    }()
    m := make(map[string][md5.Size]byte)
    for r := range c {
        if r.err != nil {
            return nil, r.err
        }
        m[r.path] = r.sum
    }
    if err := <-errc; err != nil {
        return nil, err
    }
    return m, nil
}

Conclusion

The article presented techniques for building robust Go pipelines, handling failures with channel closure and explicit cancellation, and demonstrated these patterns with a practical MD5‑digest example. Proper use of done channels and careful coordination of goroutine lifecycles are essential for avoiding deadlocks and resource leaks.