Concurrency With Channels: Safe Communication Between Goroutines

Overview

This chapter delves into Go’s powerful concurrency mechanism using channels to enable safe communication between goroutines. You will learn how to implement basic and advanced channel usage, including buffered and unbuffered channels, how to signal completion, and gracefully shut down applications. The chapter will guide you through practical examples and best practices for avoiding common concurrency pitfalls, making it easier to build robust, concurrent applications that scale efficiently.

Concurrency With Channels

Channels

Channels are a typed conduit through which you can send and receive values.

The key to understanding how channels work is quite simple:

  • When does a channel block?
  • When does a channel unblock?

Simple Channel Example


package main

import "fmt"

func main() {

	// make a new channel of type string
	messages := make(chan string)

	// this goroutine launches immediately
	go func() {

		// This line blocks until the channel is read from
		fmt.Println("waiting to write to messages")
		messages <- "hello!"
		fmt.Println("wrote to messages")
	}()

	// this line blocks until someone writes to the channel
	fmt.Println("waiting to read from messages")
	msg := <-messages
	fmt.Println("read from messages")

	fmt.Println(msg)
}

Understanding The Channel Arrows

In Go we use the <- operator to indicate sending or receiving information on a channel. At first it can be difficult to remember where the arrow goes.

To help remember where the arrow goes you just need to remember that the arrow points in the direction the data is traveling in regards to the channel.

ch <- // data is going into the channel
<- ch // data is coming out of the channel

Buffered Channels

By default channels are unbuffered – sends on a channel will block until there is something ready to receive from it.

You can create a buffered channel - which allows you to control how many items can sit in a channel unread.


package main

import "fmt"

func main() {
	// Adding a second argument to the make function creates a buffered channel
	messages := make(chan string, 2)

	// The program is no longer blocked on writing to a channel,
	// as it has capacity to write
	messages <- "hello!"
	messages <- "hello again!"

	// Reads are no longer blocked as there is already something to read from
	fmt.Println(<-messages)
	fmt.Println(<-messages)
}

Buffered Channel And Delivery

Use buffered channels cautiously. They do not guarantee delivery of the message. It is your responsibility to ensure a channel is drained before exiting a routine.


messages := make(chan string, 10)

go func() {
	for i := 0; i < 10; i++ {
		msg := fmt.Sprintf("message %d", i)
		messages <- msg
		fmt.Printf("sent: %s\n", msg)
	}
}()

m := <-messages
fmt.Printf("received: %s\n", m)
os.Exit(0)
// sent: message 0
// sent: message 1
// sent: message 2
// received: message 0
// sent: message 3
// sent: message 4

Channels Are Not Message Queues

In systems like RabbitMQ or Kafka, a message queue delivers each message to a specific consumer, and can guarantee fair distribution, redelivery on failure, and persistence. Developers coming from these systems sometimes expect Go channels to work the same way.

Go channels are simpler. They are an in-memory synchronization primitive, not a distributed messaging system:

  • Only one goroutine receives each value sent on a channel
  • There is no guaranteed fair distribution. One goroutine may receive more messages than another
  • There is no redelivery. If a goroutine crashes after receiving, the value is gone

ch := make(chan int)
for i := 0; i < N; i++ {
	go func(i int) {
		for m := range ch {
			fmt.Printf("routine %d received %d\n", i, m)
		}
	}(i)
}

for i := 0; i < N; i++ {
	ch <- i
}
time.Sleep(time.Second)

Output:


// Output (order will vary):
// routine 0 received 0
// routine 1 received 18
// routine 3 received 7
// routine 7 received 3
// routine 13 received 10
// routine 11 received 1
// routine 9 received 5
// routine 2 received 19
// routine 6 received 2
// routine 16 received 8
// ...

Notice each value is received by exactly one goroutine, but the distribution across goroutines is unpredictable.

Closing Channels

You can signal that there are no more values in a channel by closing it. Receivers can test to see if a channel is closed during a read.


package main

import "fmt"

func echo(s string, c chan string) {
	for i := 0; i < cap(c); i++ {
		// write the string down the channel
		c <- s
	}
	close(c)
}

func main() {
	// make a channel with a capacity of 10
	c := make(chan string, 10)

	// launch the goroutine
	go echo("hello!", c)

	for s := range c {
		fmt.Println(s)
	}
}

Detecting Closed Channels On Read

This can be done, as with type detection and maps, by using the optional, boolean, second argument that comes with reading from a channel.


package main

import "fmt"

func main() {
	c := make(chan int)
	go count(c)
	for {
		i, ok := <-c
		if !ok {
			fmt.Println("closed channel")
			return
		}
		fmt.Printf("read %d from channel\n", i)
	}
}

func count(c chan int) {
	for i := 0; i < 10; i++ {
		c <- i
	}
	close(c)
}

Zero Value On Closed Read

It is important to ignore the value returned on a closed read from a channel, as the runtime will explicitly set your values back to the zero values for the type defined in the channel:


package main

import "fmt"

type User struct {
	ID   int
	Name string
}

func main() {
	c := make(chan User, 1)
	u := User{ID: 1, Name: "Rob"}

	// Write the value into the channel
	c <- u

	// Read the value off the channel
	u1 := <-c
	fmt.Printf("read successful: %+v\n", u1)

	// Close the channel
	close(c)

	// Read off the channel again...
	u1 = <-c
	fmt.Printf("read closed: %+v\n", u1)

	var ok bool

	// Proper way to read
	if u1, ok = <-c; ok {
		fmt.Printf("read successful: %+v\n", u1)
	} else {
		fmt.Println("attempted read of closed channel, ignore value returned")
	}
}

Ranging Through Channels

You can use the range keyword to read channels. When using range, it is a shortcut for using a for loop with on conditionals.


func rangeLoop(messages chan string) {
	// loop until the channel is closed
	for m := range messages {
		fmt.Println(m)
	}
}

The range keyword makes it very simple to drain a channel. Below is functionally equivalent code without using a range directive.


func forLoop(messages chan string) {
	// loop until the channel is closed
	for {
		m, ok := <-messages
		if !ok {
			// channel was closed
			break
		}
		fmt.Println(m)
	}
}

Reading From Closed Buffered Channels

Even if a buffered channel is closed, you can still read all the data written to it.


func main() {
	c := make(chan int, 10)

	// Write values into a channel
	for i := 0; i < cap(c); i++ {
		c <- i
	}

	// Close the channel
	close(c)

	// You can still read all the data in the channel
	for i := range c {
		fmt.Println(i)
	}

	// but you can no longer write to the channel as it's closed
	c <- 11
}

Channels As Signals

You can also use a channel as a signal only used to block a process.


func main() {
	quit := make(chan struct{})
	go monitor()
	<-quit
}

func monitor() {
	for {
		log.Println("monitoring...")
		time.Sleep(time.Second)
	}
}

This program will block until you terminate the program.

Signal Vs. WaitGroup

For a simple program, you could use a channel in place of a sync.WaitGroup.

Given the following program with a WaitGroup:


func main() {
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go count(5, wg)
	wg.Wait()
}

func count(i int, wg *sync.WaitGroup) {
	defer wg.Done()
	for j := 0; j < i; j++ {
		fmt.Println("processing...", j)
	}
}

It could be re-written with a channel:


func main() {
	quit := make(chan struct{})
	go count(5, quit)
	<-quit
}

func count(i int, quit chan struct{}) {
	defer close(quit)
	for j := 0; j < i; j++ {
		fmt.Println("processing...", j)
	}
}

Output:


$ go run ./main.go
processing... 0
processing... 1
processing... 2
processing... 3
processing... 4

Channel Exercise (15 Min)

Create a program that does the following:

  • has a buffered channel of type string with a capacity of 5
  • add 5 messages to the channel
  • have a function that ranges through the channel printing them out
  • the function should close a signal channel to allow the program to exit
  • close the messages channel to end the goroutine previously launched

package main

import "fmt"

func process(messages chan string, quit chan struct{}) {
	// Using the `range` operator, loop through messages
	// until the channel is closed
	for m := range messages {
		_ = m // TODO: print the message
	}

	// TODO: close the `quit` channel
}

func main() {
	// TODO: declare the messages channel of type string and capacity of 5

	// TODO: declare a signal channel

	// TODO: launch the `process` function in a goroutine

	// declare 5 fruits in a []string
	fruits := []string{
		// TODO: add 5 fruits
	}

	// loop through the fruits and send them to the messages channel
	for _, f := range fruits {
		_ = f // TODO: send f to the messages channel
	}

	// TODO: close the messages channel

	// TODO: wait for everything to finish (hint, block on the quit channel)

	fmt.Println("done")
}

Channel Solution


package main

import (
	"fmt"
)

func process(messages chan string, quit chan struct{}) {
	// loop until the channel is closed
	for m := range messages {
		fmt.Println(m)
	}

	close(quit)
}

func main() {
	messages := make(chan string, 5)
	quit := make(chan struct{})

	go process(messages, quit)

	fruits := []string{"apple", "plum", "peach", "pear", "grape"}
	for _, s := range fruits {
		messages <- s
	}
	// closing the messages channel will end the for loop
	// in the processes goroutine
	close(messages)

	<-quit
	fmt.Println("done")
}

Using Select

You can use a select statement to allow a goroutine to operate on multiple communication operations:


func main() {
	c := make(chan string)
	quit := make(chan struct{})

	go func(messages []string) {
		for _, s := range messages {
			c <- s
		}
		close(quit)
	}([]string{"hi", "bye"})

	for {
		select {
		case message := <-c:
			fmt.Println(message)
		case <-quit:
			fmt.Println("shutting down")
			os.Exit(0)
		}
	}
}

Select With Default

If no operations inside a select statement can process, the select will block. To work around this problem, you can add a default statement to your select:


package main

import "time"

func main() {
	tick := time.Tick(100 * time.Millisecond)
	done := time.After(500 * time.Millisecond)
	for {
		select {
		case <-tick:
			print("tick")
		case <-done:
			print(".done")
			return
		default:
			print(".")
			time.Sleep(10 * time.Millisecond)
		}
	}
}

Break In Select

A common bug when using select inside a for loop is attempting to use break to exit the loop. According to the Go specification:

A “break” statement terminates execution of the innermost “for”, “switch”, or “select” statement.

This means break inside a select only breaks out of the select, not the surrounding for loop.

Break In Select: Common Mistake

The following code will run forever because the break statement only exits the select, not the for loop:


package main

import (
	"fmt"
	"time"
)

func main() {
	ticker := time.NewTicker(500 * time.Millisecond)
	defer ticker.Stop()

	done := time.After(3 * time.Second)

	for {
		select {
		case <-ticker.C:
			fmt.Println("tick")
		case <-done:
			fmt.Println("done")
			break // MISTAKE: this only breaks out of select, not the for loop!
		}
	}

	// This line is never reached
	fmt.Println("exiting")
}

Break In Select: The Fix

Wrap the for/select in an anonymous function. A return exits the function, which breaks the loop:


func main() {
	ticker := time.NewTicker(500 * time.Millisecond)
	defer ticker.Stop()

	done := time.After(3 * time.Second)

	func() {
		for {
			select {
			case <-ticker.C:
				fmt.Println("tick")
			case <-done:
				fmt.Println("done")
				return // exits the anonymous function, breaking the loop
			}
		}
	}()

	fmt.Println("exiting")
}

Output:


$ go run main.go
tick
tick
tick
tick
tick
tick
done
exiting

Leaked Go Routines

A common bug that occurs when using select and channels as signals is a blocked (and therefor leaked) go routine.

Look at the following code:


package main

import (
	"fmt"
	"os"
	"os/signal"
	"time"
)

func main() {

	go process()

	// wait to exit based on user input
	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt)
	<-c
}

func process() {
	// create a timeout.
	timeout := time.NewTimer(1 * time.Second)
	// this ensures we don't leak a go routine
	defer timeout.Stop()

	// create a channel to send our result
	res := make(chan int)

	// create another go routine to do some work
	go func() {
		time.Sleep(2 * time.Second)
		res <- 1
		fmt.Println("wrote to result channel")
	}()

	select {
	case <-res:
		fmt.Println("sub routine executed first")
	case <-timeout.C:
		fmt.Println("timeout executed first")
	}
}

The Problem

In the previous code, we see that we created a res channel in our process function to receive the output of our local anonymous function. We use this in combination with a timeout in our select statement.

The problem is that if the timeout occurs first, their is no process left to receive from the res channel. Because the res channel is unbuffered, there is no way for it to write.


func process() {
	// create a timeout.
	timeout := time.NewTimer(1 * time.Second)
	// this ensures we don't leak a go routine
	defer timeout.Stop()

	// create a channel to send our result
	res := make(chan int)

	// create another go routine to do some work
	go func() {
		time.Sleep(2 * time.Second)
		res <- 1 // <-- If timeout occurs first, this channel is blocked forever
		fmt.Println("wrote to result channel")
	}()

	select {
	case <-res:
		fmt.Println("sub routine executed first")
	case <-timeout.C:
		fmt.Println("timeout executed first")
	}
}

The Fix

Using a buffered channel will fix the bug:


func process() {
	// create a timeout.
	timeout := time.NewTimer(1 * time.Second)
	// this ensures we don't leak a go routine
	defer timeout.Stop()

	// create a channel to send our result
	res := make(chan int, 1)

	// create another go routine to do some work
	go func() {
		time.Sleep(2 * time.Second)
		res <- 1
		fmt.Println("wrote to result channel")
	}()

	select {
	case <-res:
		fmt.Println("sub routine executed first")
	case <-timeout.C:
		fmt.Println("timeout executed first")
	}
}

Common Goroutine Leak Patterns

Five Ways To Leak A Goroutine

Based on real bugs from production systems (CockroachDB, Kubernetes, etcd, gRPC):

  1. Early Return - Error handling orphans a sender
  2. Unbuffered Channel Worker - Parallel processing meets early exit
  3. No-Close Range - Worker pools that never end
  4. Context Timeout - When cancellation wins the race (covered above)
  5. Double Send - Missing return statement

These patterns all passed code review, testing, and shipped to production before being discovered.

Pattern: Unbuffered Channel Worker

A common pattern in parallel processing that leaks when you return early on error:


func processWorkItems(ws []workItem) ([]workResult, error) {
	ch := make(chan result) // UNBUFFERED - problem!

	for _, w := range ws {
		go func(w workItem) {
			res, err := processWorkItem(w)
			ch <- result{res, err} // BLOCKS if early return happens
		}(w)
	}

	var results []workResult
	for i := 0; i < len(ws); i++ {
		r := <-ch
		if r.err != nil {
			return nil, r.err // EARLY RETURN - remaining workers leak!
		}
		results = append(results, r.res)
	}
	return results, nil
}

Why It Leaks

When item 3 fails and we return the error:

  • Workers for items 4 and 5 are still running
  • They try to send on the unbuffered channel
  • No one is receiving anymore
  • They block forever

The Fix: Buffered Channel

Using a buffered channel allows all workers to complete their send:


func processWorkItems(ws []workItem) ([]workResult, error) {
	ch := make(chan result, len(ws)) // BUFFERED - workers can always send

	for _, w := range ws {
		go func(w workItem) {
			res, err := processWorkItem(w)
			ch <- result{res, err} // Never blocks
		}(w)
	}

	var results []workResult
	for i := 0; i < len(ws); i++ {
		r := <-ch
		if r.err != nil {
			// Workers complete even though we return early
			return nil, r.err
		}
		results = append(results, r.res)
	}
	return results, nil
}

Pattern: No-Close Range

Worker pools using range that forget to close the channel:


func processWithWorkers(items []string, workers int) {
	ch := make(chan string)

	for i := 0; i < workers; i++ {
		go func(id int) {
			for item := range ch { // Blocks forever - ch never closed!
				fmt.Printf("Worker %d processing: %s\n", id, item)
				time.Sleep(50 * time.Millisecond)
			}
			fmt.Printf("Worker %d exiting\n", id) // NEVER PRINTED
		}(i)
	}

	for _, item := range items {
		ch <- item
	}
	// Returns without closing ch - workers stuck in range forever!
}

Why It Leaks

The for item := range ch loop only exits when the channel is closed.

If you never close the channel:

  • Workers block forever in the range loop
  • Each worker becomes a permanent goroutine leak
  • Memory grows until the process crashes

The Fix: Always Close


func processWithWorkers(items []string, workers int) {
	ch := make(chan string)
	var wg sync.WaitGroup

	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for item := range ch {
				fmt.Printf("Worker %d processing: %s\n", id, item)
				time.Sleep(50 * time.Millisecond)
			}
			fmt.Printf("Worker %d exiting\n", id) // NOW PRINTED
		}(i)
	}

	for _, item := range items {
		ch <- item
	}
	close(ch) // Signal workers to exit
	wg.Wait() // Wait for all workers to finish
}

Key points:

  • Always close channels when done sending
  • Use sync.WaitGroup to wait for workers
  • Consider defer close(ch) to handle error paths

Pattern: Double Send

A missing return statement causes two sends on a channel:


func sendResult(ch chan string, err error) {
	if err != nil {
		ch <- "error occurred"
		// Missing: return <- OOPS!
	}
	// Still executing here if err was not nil!
	ch <- "success" // BLOCKS FOREVER if receiver already read
}

Why It Leaks

“One missing line. One missing return. Infinite goroutine leak.”

  • If err != nil, we send the error message
  • Without return, we continue to the next line
  • The second send blocks if the receiver already read
  • The goroutine is stuck forever

This often happens during refactoring when converting if/else to just if.

The Fix: Return After Send


func sendResult(ch chan string, err error) {
	if err != nil {
		ch <- "error occurred"
		return // FIXED: explicit return after send
	}
	ch <- "success"
}

Leak Pattern Summary

Pattern Cause Fix
Early Return Error path orphans senders Buffered channel or drain
No-Close Range Forgot to close channel Always close(ch) when done
Double Send Missing return statement Explicit return after send
Context Timeout Race with cancellation Buffered channel

Remember: All of these passed code review and testing before reaching production.

Knowing the patterns helps, but you don’t have to rely on code review alone. In tests, uber-go/goleak will fail any test that finishes with unexpected goroutines still running. Go 1.26 also adds a runtime goroutineleak pprof profile for catching leaks in production (experimental, behind GOEXPERIMENT=goroutineleakprofile, and expected to be generally available in Go 1.27). Both tools are covered in depth in the development workflow module.

Uni-Directional Channels

By default channels are bi-directional, meaning you can both send and receive data from the channel.

A common use for uni-directional channels is when you are passing a channel as an argument or receiving a channel as return value. This allows for control of the channel for the function/method and prevents outside callers from polluting the channel.

The standard library does this in the time package with methods like time.Tick.

What is the point of one-way-channels

Receive Only Channels

Here is an example of using a read only channel as an argument to a function:


package main

import "fmt"

func main() {
	// make a full directional channel
	c := make(chan string)
	quit := make(chan struct{})

	go receive(c, quit)

	// Write into the channel
	for _, s := range []string{"one", "two", "three"} {
		c <- s
	}
	close(c)
	// wait for quit channel to close
	<-quit
}

// receive takes the first argument as a read only channel
func receive(c <-chan string, quit chan struct{}) {
	// read from the channel
	for s := range c {
		fmt.Println(s)
	}
	// You can't write to this channel:
	//c <- "foo"

	// You also can't close a read only channel.  This is to protect the sender from you also polluting this channel:
	//close(c)

	close(quit)
}

Receive Only Channel Limitations

You can’t write to a receive only channel, that will result in a compile time error

receive.go:28:4: invalid operation: c <- "foo" (send to receive-only type <-chan string)

You can’t close a receive only channel, that will also result in a compile time error

/receive.go:31:7: invalid operation: close(c) (cannot close receive-only channel)

Send Only Channels

Here is an example of using a send only channel as an argument to a function:


package main

import "fmt"

func main() {
	// make a full directional channel
	c := make(chan string)

	go send(c)

	// read from the channel
	for s := range c {
		fmt.Println(s)
	}

}

// send takes the first argument as a send only channel
func send(c chan<- string) {
	// Write into the channel
	for _, s := range []string{"one", "two", "three"} {
		c <- s
	}
	// You can't read from this channel, that is a compile time error:
	//s := <-c

	// it is ok to close this channel
	close(c)
}

Send Only Channel Limitations

You can’t read from a send only channel, that will result in a compile time error

send.go:25:7: invalid operation: <-c (receive from send-only type chan<- string)

Panics

When using channels, it is important to understand when a panic can occur.

Here are the four ways you can encounter a panic with channels:


c1 := make(chan int)
close(c1)
c1 <- 1 // panic: send on a closed channel

c2 := make(chan int)
close(c2)
close(c2) // panic: closeing a channel that is already closed

var c3 chan int
c3 <- 1 // panic: send on a nil channel

var c4 chan bool
<-c4 // panic: read from a nil channel: all goroutines are asleep

Dealing With Back Pressure

There are times you may want to “skip” an operation due to back pressure in the system.

A common example would be to not log “info” type messages if it would result in slowing or blocking normal operation of your program.

You can use a buffered channel and a select with default to accomplish this.


package main

import (
	"fmt"
	"time"
)

func main() {
	logChan := make(chan string, 5)
	quit := make(chan struct{})

	// Logger reading fro the logChan
	go func() {
		for msg := range logChan {
			fmt.Println("delivered message: ", msg)
			// create some back pressure
			time.Sleep(10 * time.Millisecond)
		}
		close(quit)
	}()

	// simulate some load on the logger
	go func() {
		for i := 0; i < 20; i++ {
			log(i, logChan)
		}
		// this will close the logger
		close(logChan)
	}()

	<-quit

}

func log(i int, c chan string) {
	select {
	// write the message to the logger
	case c <- fmt.Sprintf("this is log message %d", i):
	default:
		fmt.Printf("log channel buffer full: dropped message due to back pressure %d\n", i)
	}
}

Code Walkthrough (Optional)

Taking the example from the previous chapter to retrieve a list of websites, let’s re-write it using channels instead:


package main

import (
	"fmt"
	"log"
	"net/http"
	"time"
)

func main() {
	args := []string{"http://youtube.com", "http://en.wikipedia.org", "http://facebook.com"}

	now := time.Now()

	for _, a := range args {
		get(a)
	}

	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(s string) {
	now := time.Now()
	_, err := http.Get(s)
	if err != nil {
		log.Printf("failed to retrieve %s: %s\n", s, err)
		return
	}
	log.Printf("Retrieved site %s in %s\n", s, time.Since(now))
}

Let's Add Channels

To give this basic concurrency, let’s rewrite the solution with channels:


package main

import (
	"fmt"
	"log"
	"net/http"
	"time"
)

// create a custom type to receive our responses
type resp struct {
	site     string
	duration time.Duration
	err      error
}

func main() {
	args := []string{"http://youtube.com", "http://en.wikipedia.org", "http://facebook.com"}

	responses := make(chan resp, len(args))

	now := time.Now()

	for _, a := range args {
		go get(a, responses)
	}

	// read back our responses
	for i := 0; i < cap(responses); i++ {
		r := <-responses
		if r.err != nil {
			log.Printf("failed to retrieve %s: %s\n", r.site, r.err)
			continue
		}
		log.Printf("Retrieved site %s in %s\n", r.site, r.duration)
	}

	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(s string, responses chan<- resp) {
	now := time.Now()

	_, err := http.Get(s)
	if err != nil {
		responses <- resp{site: s, err: err}
		return
	}
	responses <- resp{site: s, duration: time.Since(now)}
}

Better With Channels

With the new code, we are now able to communicate the errors and responses back upstream to process were desired. However, we still need to address the following problems:

  • Need to be able to scale and process millions of sites
  • Need to set a reasonable timeout for each site retrieval

Addressing Timeout

To address the timeout, we can use time.NewTimer and another channel. We only need to change the get function to address the timeout.


func get(s string, responses chan<- resp) {
	now := time.Now()

	// pick a magic timeout duration
	td := 2 * time.Second
	// create a timeout.
	timeout := time.NewTimer(td)
	// this ensures we don't leak a go routine
	defer timeout.Stop()

	// create a channel to to receive the response on locally
	// make sure it's buffered so you don't leak a goroutine
	work := make(chan resp, 1)

	// retrieve the site in a goroutine to avoid blocking
	go func() {
		// retrieve the website
		_, err := http.Get(s)
		work <- resp{site: s, duration: time.Since(now), err: err}
	}()

	select {
	case r := <-work:
		// retrieve the value from our local function and pass it through to our send channel
		responses <- r
	case <-timeout.C:
		responses <- resp{site: s, err: fmt.Errorf("timed out after %s", td)}
	}
}

We purposefully used our own timeout to show how we can control our operation. In production, you would set the timeout for the http client as even though we exited our goroutine, the client is still open in the background using up resources.

Scaling

Finally, we need to make this scale to millions of records. We can do this by using a buffered channel and control how large it is, thus throttling how many resources are in play at any given time

We can run the following code and pass in all the sites with the following command:

go run main.go "$(< sites.txt)"


package main

import (
	"fmt"
	"log"
	"net/http"
	"os"
	"strings"
	"time"
)

// create a custom type to receive our responses
type resp struct {
	site     string
	duration time.Duration
	err      error
}

func main() {
	// Load sites from a file
	args := os.Args[1:]
	// Args might be a single string with line returns to delimit sites, break them apart
	if len(args) == 1 {
		args = strings.Split(args[0], "\n")
	}

	// Create a channel to send all sites to
	// We'll create it buffered so that we throttle how much many go routines will be processing at any given time
	sites := make(chan string, 30)

	// create an unbuffered channel to receive on
	// because we don't have contention or latency on the end of our
	// process, there is no need to buffer this channel
	responses := make(chan resp)

	now := time.Now()

	// launch the same number of go routines that we have a buffer size for
	for i := 0; i < cap(sites); i++ {
		go get(sites, responses)
	}

	// feed all of our data into the send channel
	// this channel will block when it's full, and unblock
	// when space becomes available
	// We will do this in a goroutine so we don't block execution
	go func() {
		for _, s := range args {
			sites <- s
		}
		// ensure that any open go routines close down
		close(sites)
	}()

	// read back our responses
	for i := 0; i < len(args); i++ {
		r := <-responses
		if r.err != nil {
			log.Printf("failed to retrieve %s: %s\n", r.site, r.err)
			continue
		}
		log.Printf("Retrieved site %s in %s\n", r.site, r.duration)
	}
	fmt.Printf("It took %s for the entire process to finish\n", time.Since(now))
}

func get(sites <-chan string, responses chan<- resp) {
	// range through the receiving channel.  When this channel is closed
	// all go routines will terminate
	for s := range sites {
		now := time.Now()

		// pick a magic timeout duration
		td := 2 * time.Second
		// create a timeout.
		timeout := time.NewTimer(td)
		// this ensures we don't leak a go routine
		defer timeout.Stop()

		// create a channel to to receive the response on locally
		// make sure it's buffered so you don't leak a goroutine
		work := make(chan resp, 1)

		// retrieve the site in a goroutine to avoid blocking
		go func() {
			// retrieve the website
			_, err := http.Get(s)
			work <- resp{site: s, duration: time.Since(now), err: err}
		}()

		select {
		case r := <-work:
			// retrieve the value from our local function and pass it through to our send channel
			responses <- r
		case <-timeout.C:
			responses <- resp{site: s, err: fmt.Errorf("timed out after %s", td)}
		}
	}
}

Here is a sample of the output of all 100 sites:


2019/10/09 19:04:46 Retrieved site http://google.com in 271.050112ms
2019/10/09 19:04:47 Retrieved site http://nytimes.com in 475.879293ms
2019/10/09 19:04:47 Retrieved site http://forbes.com in 490.292424ms
2019/10/09 19:04:47 Retrieved site http://play.google.com in 495.126496ms
2019/10/09 19:04:47 Retrieved site http://apple.com in 500.157454ms

2019/10/09 19:04:50 Retrieved site http://npr.org in 987.585514ms
2019/10/09 19:04:50 Retrieved site http://goodreads.com in 1.254808047s
2019/10/09 19:04:50 Retrieved site http://theguardian.com in 1.499847207s
2019/10/09 19:04:50 Retrieved site http://wowhead.com in 1.154202684s
2019/10/09 19:04:50 Retrieved site http://groupon.com in 1.85877198s
2019/10/09 19:04:51 failed to retrieve http://apartments.com: timed out after 2s
It took 4.753549602s for the entire process to finish

Understanding What We Just Built

The buffered channel pattern we used here is actually implementing a semaphore pattern!

sites := make(chan string, 30)  // Buffer size throttles to 30 concurrent workers
for i := 0; i < cap(sites); i++ {
    go get(sites, responses)
}

While this works great for worker pools, semaphores are more explicit when your primary goal is throttling concurrency rather than building a worker pool.

We’ll look at this throttle pattern on its own next. When you want explicit throttling without channels, the sync package module covers golang.org/x/sync/semaphore.

A Buffered Channel As A Semaphore

Here’s that throttle pattern on its own, without the rest of the fetcher. Send an empty struct into the channel to claim a slot before starting work, and receive from it to release the slot when you’re done. The buffer size is the concurrency limit:


func main() {
	maxConcurrent := 3

	// Create semaphore with max concurrency
	sem := make(chan struct{}, maxConcurrent)

	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()

			// Acquire slot (blocks if full)
			sem <- struct{}{}

			fmt.Printf("worker %d running\n", id)
			time.Sleep(100 * time.Millisecond)

			// Release slot
			<-sem
		}(i)
	}

	wg.Wait()
	fmt.Println("all workers done")
}

Sending blocks once the buffer is full, so only cap(sem) goroutines ever run at the same time. struct{}{} carries no data and takes no memory - we only care about the slot itself.

Worker Pool Vs. Semaphore

Both bound concurrency, but they’re shaped differently:

| | Semaphore | Worker Pool | |–|———–|————-| | Goroutines | One per task | Fixed set of workers | | Shape | Launch all tasks, cap how many run | Workers pull tasks off a channel | | Best for | Short, one-off tasks | Long-running workers, steady streams |

For a one-time batch like fetching a list of sites, the semaphore (throttle) shape is usually the cleaner fit. A worker pool earns its keep when you have long-lived workers pulling from a queue.

Work Pools And Rate Limiting

Channels work great for worker pools. You can find examples of both here:

Channel Exercise (20 Min)

The following program currently works but could be improved by adding concurrency.

Make the following changes in the program:

  • Launch charCount in a go routine.
  • Use a buffered channel to retrieve back the results.
  • Create a struct type to receive the count and error from the channel
  • Make sure you don’t exit the program until each go routine has finished.

NOTE: You must download the files using the Download Source button on the left and run this from the src/exercises/02-concurrency directory to ensure you have the local files to work with


package main

import (
	"bytes"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"time"
)

func main() {
	start := time.Now()

	paths := []string{}

	root := "./files"
	err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
		if !info.IsDir() {
			paths = append(paths, path)
		}
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	for _, p := range paths {
		charCount(p)
	}

	fmt.Println("process took", time.Since(start))
}

func charCount(path string) {
	f, err := os.Open(path)
	if err != nil {
		fmt.Printf("error opening %s: %s\n", path, err)
		return
	}
	defer f.Close()
	var count int
	buff := make([]byte, bytes.MinRead)
	for {
		i, err := f.Read(buff)
		count += i
		if err == io.EOF {
			fmt.Printf("%s has %d characters\n", path, count)
			return
		}
		if err != nil {
			fmt.Printf("error reading file %s: %s\n", path, err)
			return
		}
	}
}

Channel Exercise - Extra

Using a mix of channels and waitgroups, design a solution that immediately launches a go routine from the filepath.Walk function.

Things to consider:

  • How will you use the waitgroup to know when all go routines have finished processing.
  • How do you know when you have processed everything on the channel (hint, use an unbuffered channel)
  • You can start processing the channel in a go routine before you start putting items in the channel…

Channel Solution


package main

import (
	"bytes"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"time"
)

// <-- add a struct type to send responses back on the channel
type result struct {
	path  string
	count int
	err   error
}

func main() {
	start := time.Now()

	paths := []string{}

	root := "./files"
	err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
		if !info.IsDir() {
			paths = append(paths, path)
		}
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	ch := make(chan result, len(paths)) // <-- create a buffered channel to start execution right away.

	// <-- launch all the processes, use an anonymous go routine to not block execution
	go func() {
		for _, p := range paths {
			go charCount(p, ch) // <-- Launch in a go routine, pass the channel
		}
	}()

	// <-- Create a loop to read back the size of the buffered channel
	for i := 0; i < cap(ch); i++ {
		r := <-ch // <-- get the next available result from the channel
		if r.err != nil {
			fmt.Println(r.err)
			continue
		}
		fmt.Printf("%s has %d characters\n", r.path, r.count)
	}

	fmt.Println("process took", time.Since(start))
}

func charCount(path string, ch chan result) { // <-- Accept the channel as an argument
	f, err := os.Open(path)
	if err != nil {
		// <-- send back the error on a channel
		ch <- result{path: path, count: 0, err: fmt.Errorf("error opening %s: %s\n", path, err)}
		return
	}
	defer f.Close()
	var count int
	buff := make([]byte, bytes.MinRead)
	for {
		i, err := f.Read(buff)
		count += i
		if err == io.EOF {
			// <-- send back the result on a channel
			ch <- result{path: path, count: count}
			return
		}
		if err != nil {
			ch <- result{path: path, count: 0, err: fmt.Errorf("error reading file %s: %s\n", path, err)}
			return
		}
	}
}

Channel Solution - Extra


package main

import (
	"bytes"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"sync"
	"time"
)

// <-- add a struct type to send responses back on the channel
type result struct {
	path  string
	count int
	err   error
}

func main() {
	start := time.Now()

	wg := sync.WaitGroup{}
	ch := make(chan result)

	// begin watching the channel to process results.
	// Do this in a go routine so it doesn't block program execution
	go func() {
		for r := range ch {
			func() {
				defer wg.Done()
				if r.err != nil {
					fmt.Println(r.err)
					return
				}
				fmt.Printf("%s has %d characters\n", r.path, r.count)
			}()
		}
	}()

	root := "./files"
	err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
		if !info.IsDir() {
			wg.Add(1) // make sure to track each go routine you launch in the wait group
			go charCount(path, ch)
		}
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	wg.Wait() // wait for all go routines to finish
	close(ch) // close the channel to exit the range go routine

	fmt.Println("process took", time.Since(start))
}

func charCount(path string, ch chan result) { // <-- Accept the channel as an argument
	f, err := os.Open(path)
	if err != nil {
		// <-- send back the error on a channel
		ch <- result{path: path, count: 0, err: fmt.Errorf("error opening %s: %s\n", path, err)}
		return
	}
	defer f.Close()
	var count int
	buff := make([]byte, bytes.MinRead)
	for {
		i, err := f.Read(buff)
		count += i
		if err == io.EOF {
			// <-- send back the result on a channel
			ch <- result{path: path, count: count}
			return
		}
		if err != nil {
			ch <- result{path: path, count: 0, err: fmt.Errorf("error reading file %s: %s\n", path, err)}
			return
		}
	}
}

Graceful Shutdown

All programs should attempt a graceful shutdown. Graceful shutdown includes:

  • Detecting that the program was requested to shut down
  • Shut down all internal processes, including long running go routines
  • Have a reasonable timeout in the event that internal processes are taking to long to shut down or are deadlocked
  • Respond to an actual user request for immediate hard shutdown
  • Record the result of the shutdown (success, timeout, user intervention)

The `os/signals` Package

You can use channels and the os/signal package to shut down your program:


func main() {
	// Set up channel on which to send signal notifications.
	// We must use a buffered channel or risk missing the signal
	// if we're not ready to receive when the signal is sent.
	c := make(chan os.Signal, 1)

	// Wire up the channel to an `os.Signal`
	signal.Notify(c, os.Interrupt)

	fmt.Println("awaiting signal...")

	// Block until a signal is received.
	s := <-c

	fmt.Println("Got signal:", s)
	// do my code shutdown....
	// then exit
}

Program Shutdown

Given an existing basic startup:


func main() {
	c := cmd.New()
	if err := c.Open(); err != nil {
		log.Fatal(c)
	}

	wait := make(chan struct{})
	<-wait

	if err := c.Close(); err != nil {
		log.Fatal(err)
	}
	log.Println("successfuly shutdown")
}

There is no way to gracefully shutdown this program. Any interrupt will result in an immediate exit, and not actually finish proper shutdown.

Output:


2020/04/20 11:09:09 monitor check
2020/04/20 11:09:10 monitor check
2020/04/20 11:09:11 monitor check
^Csignal: interrupt

Notice that we don’t see any output from the actual shutdown. This is because once the interrupt was detected, it immediately exited the program.

Add Shutdown Detection

Using the os.Signals package, we can re-write this code to allow for a better shutdown:


func main() {

	c := cmd.New()
	if err := c.Open(); err != nil {
		log.Fatal(c)
	}

	quit := make(chan os.Signal, 1)

	// Wire up the channel to an `os.Signal`
	signal.Notify(quit, os.Interrupt)

	log.Println("program started")

	// Block until a signal is received.
	s := <-quit

	log.Printf("received %s signal for shutdown", s)

	if err := c.Close(); err != nil {
		log.Fatal(err)
	}
	log.Println("successfuly shutdown")
}

Now when we send in an interrupt, the program will intercept it and allow us to shut down gracefully


2020/04/20 11:17:03 program started
2020/04/20 11:17:04 monitor check
2020/04/20 11:17:05 monitor check
2020/04/20 11:17:06 monitor check
^C2020/04/20 11:17:06 received interrupt signal for shutdown
2020/04/20 11:17:06 shutting down monitor
2020/04/20 11:17:06 successfuly shutdown

Note: This program will hang indefinitely if the Close routine deadlocks for any reason. To solve this, we need to add a timeout to our shutdown routine.

Shutdown With Timeout

A well behaved program should attempt to shut down gracefully. The following code sample is one possible template to do so.


func main() {

	c := cmd.New()
	if err := c.Open(); err != nil {
		log.Fatal(c)
	}

	sigCh := make(chan os.Signal, 1)

	// Wire up the channel to an `os.Signal`
	signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)

	log.Println("Listening for signals")

	// launch all program code...

	// Block until one of the signals above is received
	<-sigCh
	log.Println("Signal received, initializing clean shutdown...")

	// do graceful shutdown
	// make sure not to block for the timeout added later
	closed := make(chan struct{})
	go func() {
		// do all graceful shutdown of program
		if err := c.Close(); err != nil {
			log.Fatal(err)
		}

		// signal you've completed your shutdown
		close(closed)
	}()

	// Block again until another signal is received, a shutdown timeout elapses,
	// or the Command is gracefully closed
	log.Println("Waiting for clean shutdown...")
	select {
	case <-sigCh:
		log.Println("second signal received, initializing hard shutdown")
	case <-time.After(time.Second * 30):
		log.Println("time limit reached, initializing hard shutdown")
	case <-closed:
		log.Println("server shutdown completed")
	}

	// goodbye.
}

Output:


2020/04/20 11:22:35 Listening for signals
2020/04/20 11:22:36 monitor check
2020/04/20 11:22:37 monitor check
^C2020/04/20 11:22:37 Signal received, initializing clean shutdown...
2020/04/20 11:22:37 Waiting for clean shutdown...
2020/04/20 11:22:37 shutting down monitor
2020/04/20 11:22:42 server shutdown completed

Hot Reload (HUP)

You can also use signals to perform live reloads. Here is a psuedo example of how to accomplish that task:


func main() {
	// wire up a quit channel to monitor
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, os.Interrupt)

	// create a hup channel
	hup := make(chan os.Signal, 1)
	signal.Notify(hup, os.Interrupt, syscall.SIGHUP)

	wg := &sync.WaitGroup{}
	wg.Add(1)
	go hotReload(hup, wg)

	<-quit
	fmt.Println("waiting for clean shutdown")
	wg.Wait()
	fmt.Println("clean shutdown")
}

func hotReload(hup chan os.Signal, wg *sync.WaitGroup) {
	defer wg.Done()
	for {
		s := <-hup
		switch s {
		case syscall.SIGHUP:
			fmt.Println("reloading config...")
		case os.Interrupt:
			fmt.Println("closing config reload")
			return
		}
	}
}

Because we are using the kill command, we need to compile and run the binary. Using go run will intercept the SIGHUP and the program won’t perform as expected.

Commands:


$ ps | grep hup
46490 ttys006    0:00.01 ./hup

$ kill -s SIGHUP 46490

$ kill -s SIGINT 46490

Output:


$ ./hup
reloading config...
waiting for clean shutdown
closing config reload
clean shutdown

Channels - Some Wisdom

I can’t tell you how many times I start with channels, and by the time I’m done, I’ve completely optimized them out. – Cory LaNou

When I first learned about channels, I wanted use them everywhere. Now, I rarely use them at all. – Mat Ryer

These quotes are not intended to steer you away from using channels, but more to think about do you in fact need a channel? It is common for developers new to Go to overuse channels, which leads to unecessary code complexity with no benefit to program performance.

Exercise (15 Min)

Given the following program, add the ability for intercepting a os.Interrupt signal and wait for the program to shut down gracefully.

Note: You only have to make modifications to the main function.

For extra credit, add in a timeout for shutdown as well, as well as monitor for a user sending in a second shutdown request.


package main

import (
	"log"
	"sync"
	"time"
)

func main() {
	c := NewCommand()

	c.Open()

	wait := make(chan struct{})
	<-wait

	c.Close()
}

func NewCommand() *command {
	wg := sync.WaitGroup{}
	return &command{
		quit: make(chan struct{}),
		wg:   &wg,
	}
}

type command struct {
	quit chan struct{}
	wg   *sync.WaitGroup
}

func (c *command) Open() {
	c.wg.Add(1)
	go c.monitor()
}

func (c *command) Close() {
	close(c.quit)
	c.wg.Wait()
}

func (c *command) monitor() {
	defer c.wg.Done()
	ticker := time.NewTicker(time.Second)
	for {
		select {
		case <-c.quit:
			log.Println("shutting down monitor")
			return
		case <-ticker.C:
			log.Println("monitor check")
		}
	}
}

Solution


package main

import (
	"log"
	"os"
	"os/signal"
	"sync"
	"time"
)

func main() {
	c := NewCommand()

	c.Open()

	sigCh := make(chan os.Signal, 1)

	// Wire up the channel to an `os.Signal`
	signal.Notify(sigCh, os.Interrupt)

	log.Println("Listening for signals")

	<-sigCh
	log.Println("Signal received, initializing clean shutdown...")

	// do graceful shutdown
	// make sure not to block for the timeout added later
	closed := make(chan struct{})
	go func() {
		// do all graceful shutdown of program
		c.Close()

		// signal you've completed your shutdown
		close(closed)
	}()

	log.Println("Waiting for clean shutdown...")

	<-closed
	log.Println("server shutdown completed")

}

func NewCommand() *command {
	wg := sync.WaitGroup{}
	return &command{
		quit: make(chan struct{}),
		wg:   &wg,
	}
}

type command struct {
	quit chan struct{}
	wg   *sync.WaitGroup
}

func (c *command) Open() {
	c.wg.Add(1)
	go c.monitor()
}

func (c *command) Close() {
	// add artificial time to simulate the real world
	time.Sleep(time.Second * 5)
	close(c.quit)
	c.wg.Wait()
}

func (c *command) monitor() {
	defer c.wg.Done()
	ticker := time.NewTicker(time.Second)
	for {
		select {
		case <-c.quit:
			log.Println("shutting down monitor")
			return
		case <-ticker.C:
			log.Println("monitor check")
		}
	}
}

Solution (Extra Credit)


package main

import (
	"log"
	"os"
	"os/signal"
	"sync"
	"time"
)

func main() {
	c := NewCommand()

	c.Open()

	sigCh := make(chan os.Signal, 1)

	// Wire up the channel to an `os.Signal`
	signal.Notify(sigCh, os.Interrupt)

	log.Println("Listening for signals")

	<-sigCh
	log.Println("Signal received, initializing clean shutdown...")

	// do graceful shutdown
	// make sure not to block for the timeout added later
	closed := make(chan struct{})
	go func() {
		// do all graceful shutdown of program
		c.Close()

		// signal you've completed your shutdown
		close(closed)
	}()

	log.Println("Waiting for clean shutdown...")

	select {
	case <-sigCh:
		log.Println("second signal received, initializing hard shutdown")
	case <-time.After(time.Second * 30):
		log.Println("time limit reached, initializing hard shutdown")
	case <-closed:
		log.Println("server shutdown completed")
	}

}

func NewCommand() *command {
	wg := sync.WaitGroup{}
	return &command{
		quit: make(chan struct{}),
		wg:   &wg,
	}
}

type command struct {
	quit chan struct{}
	wg   *sync.WaitGroup
}

func (c *command) Open() {
	c.wg.Add(1)
	go c.monitor()
}

func (c *command) Close() {
	// add artificial time to simulate the real world
	time.Sleep(time.Second * 5)
	close(c.quit)
	c.wg.Wait()
}

func (c *command) monitor() {
	defer c.wg.Done()
	ticker := time.NewTicker(time.Second)
	for {
		select {
		case <-c.quit:
			log.Println("shutting down monitor")
			return
		case <-ticker.C:
			log.Println("monitor check")
		}
	}
}