Channels, and the select statement

Goroutines run independently. Channels are how they talk. The slogan Go is famous for is *do not communicate by sharing memory; share memory by communicating*, which sounds like a fortune cookie until you have written both versions of the same program.

A channel is a typed pipe

ch := make(chan int)    // unbuffered
ch <- 42                // send
v := <-ch               // receive

The arrow always points in the direction the value travels. ch <- v puts a value in, <-ch takes one out.

func main() {
    ch := make(chan string)

    go func() {
        ch <- "hello from the goroutine"
    }()

    msg := <-ch          // blocks until something arrives
    fmt.Println(msg)
}

No time.Sleep this time. The receive blocks until the send happens, which is synchronisation and message passing in one operation.

Unbuffered channels are a handshake

An unbuffered channel has no room. A send blocks until a receiver is ready, and a receive blocks until a sender is ready. The two goroutines meet.

ch := make(chan int)
ch <- 1                 // deadlock: nobody is receiving

fatal error: all goroutines are asleep - deadlock!

Go detects when every goroutine is blocked and stops the program rather than hanging forever. This is a feature, and it is how you will find most channel bugs.

Buffered channels have room

ch := make(chan int, 3)  // capacity 3
ch <- 1                  // does not block
ch <- 2
ch <- 3
ch <- 4                  // now it blocks: the buffer is full

A send blocks only when the buffer is full; a receive blocks only when it is empty.

Closing, and ranging

The sender closes a channel to say "no more values".

func produce(ch chan<- int) {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch)           // only the sender closes
}

func main() {
    ch := make(chan int)
    go produce(ch)

    for v := range ch {  // ends when the channel is closed
        fmt.Println(v)
    }
}

Receiving from a closed channel returns the zero value immediately. The two-value form tells you which happened:

v, ok := <-ch
if !ok {
    // the channel is closed and drained
}

Note chan<- int in that signature: a directional channel type, meaning send-only. <-chan int is receive-only. Using them in function parameters makes the data flow obvious and lets the compiler stop a receiver from closing.

select: waiting on several channels

select blocks until one of its cases can proceed.

select {
case msg := <-ch1:
    fmt.Println("from ch1:", msg)
case msg := <-ch2:
    fmt.Println("from ch2:", msg)
case ch3 <- "value":
    fmt.Println("sent to ch3")
}

If several are ready it picks one at random, which prevents one busy channel starving another.

A default case makes it non-blocking:

select {
case msg := <-ch:
    fmt.Println(msg)
default:
    fmt.Println("nothing waiting")   // does not block
}

Timeouts, the most common select

select {
case result := <-work:
    fmt.Println("finished:", result)
case <-time.After(2 * time.Second):
    fmt.Println("timed out")
}

time.After returns a channel that delivers a value after the duration. This is the idiomatic timeout, and it is three lines.

Signalling with a done channel

A channel carrying no data, used only to signal, is a common shape:

func worker(done <-chan struct{}) {
    for {
        select {
        case <-done:
            fmt.Println("stopping")
            return
        default:
            // do a unit of work
        }
    }
}

func main() {
    done := make(chan struct{})
    go worker(done)

    time.Sleep(time.Second)
    close(done)          // closing broadcasts to every receiver
    time.Sleep(10 * time.Millisecond)
}

Closing is the broadcast mechanism: every goroutine blocked on <-donewakes up at once, because a closed channel always yields immediately. Sending a value would wake only one.

This pattern is so common that the standard library wrapped it, and that wrapper is context, which is chapter sixteen. Understand the done channel first and context becomes obvious rather than magical.

The nil channel, and why it is useful

Sending or receiving on a nil channel blocks forever. That sounds like a pure bug and it has one real use: disabling a select case.

for {
    select {
    case v, ok := <-in:
        if !ok {
            in = nil        // disable this case, keep the loop alive
            continue
        }
        process(v)
    case <-done:
        return
    }
}

Once in is nil its case can never be chosen, so the loop keeps serving donewithout spinning on a closed channel returning zeros forever.

Worker pools

The canonical channel program: a fixed number of workers pulling from a jobs channel.

func worker(id int, jobs <-chan int, results chan<- string,
    wg *sync.WaitGroup) {
    defer wg.Done()
    for j := range jobs {           // ends when jobs is closed
        results <- fmt.Sprintf("worker %d did job %d", id, j)
    }
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan string, 100)
    var wg sync.WaitGroup

    for w := 1; w <= 3; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }

    for j := 1; j <= 9; j++ {
        jobs <- j
    }
    close(jobs)                     // tells the workers no more are coming

    wg.Wait()                       // wait for them to finish
    close(results)                  // now safe to close

    for r := range results {
        fmt.Println(r)
    }
}

The ordering is the whole lesson. Close jobs so the workers' range ends.Wait for the workers. Then close results, because closing it earlier would panic a worker still writing.

package main

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

// Stage 1: produce lines. Returns a channel rather than taking one,
// which is the convention: the producer owns and closes what it
// creates.
func generate(done <-chan struct{}, lines []string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for _, l := range lines {
            select {
            case out <- l:
            case <-done:
                return          // stop early without leaking
            }
        }
    }()
    return out
}

// Stage 2: normalise.
func normalise(done <-chan struct{}, in <-chan string) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for l := range in {
            select {
            case out <- strings.ToLower(strings.TrimSpace(l)):
            case <-done:
                return
            }
        }
    }()
    return out
}

// Stage 3: count words, fanned out across several goroutines.
func count(done <-chan struct{}, in <-chan string,
    workers int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for i := 0; i < workers; i++ {
        wg.Go(func() {
            for l := range in {
                time.Sleep(10 * time.Millisecond)   // pretend it is slow
                select {
                case out <- len(strings.Fields(l)):
                case <-done:
                    return
                }
            }
        })
    }

    // One goroutine whose only job is to close out once every worker is
    // finished. Without this, nothing ever closes out and main hangs.
    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    lines := []string{
        "  The quick brown fox  ", "jumps over", "THE LAZY DOG",
        "and then", "runs away quickly", "into the woods",
    }

    done := make(chan struct{})
    defer close(done)            // every stage stops when main returns

    start := time.Now()
    total := 0
    pipeline := count(done, normalise(done, generate(done, lines)), 3)
    for n := range pipeline {
        total += n
    }

    fmt.Printf("%d words in %v\n", total,
        time.Since(start).Round(time.Millisecond))
}

n := 0
for range count(done, normalise(done, generate(done, lines)), 3) {
    n++
    if n == 2 {
        break
    }
}

Assignment

Commit your work with the message Channels - Assignment One, exactly as written, so the leaderboard can score it.

fmt.Println(runtime.NumGoroutine())