typestar

pipeline.go in Go

A cancellable three-stage pipeline with bounded concurrency.

// Command pipeline fans work out across workers and merges the results.
package main

import (
    "fmt"
    "strings"
    "sync"
)

func generate(done <-chan struct{}, words ...string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for _, word := range words {
            select {
            case out <- word:
            case <-done:
                return
            }
        }
    }()
    return out
}

func score(done <-chan struct{}, in <-chan string) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for word := range in {
            value := len(word) + strings.Count(word, "e")
            select {
            case out <- value:
            case <-done:
                return
            }
        }
    }()
    return out
}

func merge(done <-chan struct{}, inputs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, input := range inputs {
        wg.Go(func() {
            for value := range input {
                select {
                case out <- value:
                case <-done:
                    return
                }
            }
        })
    }
    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

func main() {
    done := make(chan struct{})
    defer close(done)

    words := generate(done, "typestar", "practice", "goroutine", "channel",
        "select", "context")

    const workers = 3
    stages := make([]<-chan int, workers)
    for i := range stages {
        stages[i] = score(done, words)
    }

    total, count := 0, 0
    for value := range merge(done, stages...) {
        total += value
        count++
    }

    fmt.Printf("%d words scored, total %d\n", count, total)
    fmt.Printf("mean %.2f across %d workers\n", float64(total)/float64(count),
        workers)
}

How it works

  1. Each stage owns and closes the channel it returns.
  2. Every stage selects on the done channel, so cancellation propagates.
  3. The fan-out stage runs a fixed number of workers.

Keywords and builtins used here

The run, in numbers

Lines
85
Characters to type
1392
Tokens
365
Three-star pace
110 tpm

At the three-star pace of 110 tokens a minute, this run takes about 199 seconds.

Type this snippet

Step 2 of 2 in Encore, step 25 of 25 in Concurrency.

← Previous