Project: pipelines and workers

Lesson 24 of 24 · Part 6 · Project · about 20 minutes

This lesson puts the concurrency lessons together and builds, step by step, two common shapes of program:

• A pipeline: data moves along from stage to stage. Each stage is a goroutine, and channels connect one stage to the next.
• A worker pool: lots of jobs go into one channel, a few worker goroutines each take the next job when they're free, and hand back their results.

We'll be careful about one thing all the way: whichever goroutine goes first, the output must be the same.

Stage one: generate

The first stage puts a list of prices into a channel. It's a function that starts a goroutine to do the sending and returns the channel right away:

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func main() {17	for p := range generate(120, 45, 80) {18		fmt.Println(p)19	}20}

120
45
80

The return type <-chan int is a receive-only channel (lesson 19): whoever gets it can receive, but can't send or close. The goroutine inside generate does the sending, so it does the closing too: whoever sends, closes.

Stage two: discount

The second stage receives prices from the stage before, takes 10% off, and sends them on. Its parameter is a receive-only channel, and it returns a new one:

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func discount(in <-chan int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for p := range in {21			out <- p * 9 / 1022		}23	}()24	return out25}26 27func main() {28	for p := range discount(generate(120, 45, 80)) {29		fmt.Println(p)30	}31}

108
40
72

Press Step: three lanes, generate, discount and main. A price leaves the first lane, discount receives it, works it out and sends it on, and main receives and shows it. When the stage before closes its channel, discount's range ends and it closes its own channel too, and so on down the line.

Each stage is a single goroutine, and a channel delivers values in the order they were sent, so the prices stay in order all the way.

Stage three: sum

The last stage adds up all the prices. It sends just one value: the total.

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func discount(in <-chan int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for p := range in {21			out <- p * 9 / 1022		}23	}()24	return out25}26 27func sum(in <-chan int) <-chan int {28	out := make(chan int)29	go func() {30		total := 031		for p := range in {32			total += p33		}34		out <- total35	}()36	return out37}38 39func main() {40	fmt.Println("total", <-sum(discount(generate(120, 45, 80))))41}

total 220

Read sum(discount(generate(…))) from the inside out: generate, discount, sum. main does just one thing: receive the total from the last stage.

One more stage

Pipelines make it easy to add a stage. Say only items of 50 or more get the discount, and cheaper ones don't count: add a filter stage, atLeast, between generate and discount, and not one line of the other stages changes.

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func atLeast(in <-chan int, limit int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for p := range in {21			if p >= limit {22				out <- p23			}24		}25	}()26	return out27}28 29func discount(in <-chan int) <-chan int {30	out := make(chan int)31	go func() {32		defer close(out)33		for p := range in {34			out <- p * 9 / 1035		}36	}()37	return out38}39 40func main() {41	for p := range discount(atLeast(generate(120, 45, 80, 30), 50)) {42		fmt.Println(p)43	}44}

108
72

1package main2 3import "fmt"4 5func generate(nums ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, n := range nums {10			out <- n11		}12	}()13	return out14}15 16func double(in <-chan int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for n := range in {21			out <- n * 222		}23	}()24	return out25}26 27func main() {28	for n := range double(double(generate(1, 2, 3))) {29		fmt.Print(n, " ")30	}31	fmt.Println()32}

A pool of workers

Each stage of a pipeline is one goroutine. When one stage is slow (making thumbnails, say), you can have several workers share it. Below, 3 workers take images from jobs and put the finished thumbnails into results:

1package main2 3import (4	"fmt"5	"sort"6	"sync"7)8 9type Image struct {10	ID   int11	Size int12}13 14func worker(jobs <-chan Image, results chan<- Image, wg *sync.WaitGroup) {15	defer wg.Done()16	for img := range jobs {17		results <- Image{ID: img.ID, Size: img.Size / 10}18	}19}20 21func main() {22	sizes := []int{1200, 800, 2000, 640, 1500}23	jobs := make(chan Image)24	results := make(chan Image)25	var wg sync.WaitGroup26	for range 3 {27		wg.Add(1)28		go worker(jobs, results, &wg)29	}30	go func() {31		defer close(jobs)32		for i, s := range sizes {33			jobs <- Image{ID: i + 1, Size: s}34		}35	}()36	go func() {37		wg.Wait()38		close(results)39	}()40	var thumbs []Image41	for t := range results {42		thumbs = append(thumbs, t)43	}44	sort.Slice(thumbs, func(a, b int) bool { return thumbs[a].ID < thumbs[b].ID })45	for _, t := range thumbs {46		fmt.Printf("image %d: %d KB\n", t.ID, t.Size)47	}48}

image 1: 120 KB
image 2: 80 KB
image 3: 200 KB
image 4: 64 KB
image 5: 150 KB

Press Step: watch how the three worker lanes take jobs from jobs, and which channel main is waiting on.

Why sort? Nobody can say which worker finishes first. If you showed each result as it arrived:

1for t := range results {2	fmt.Printf("image %d: %d KB\n", t.ID, t.Size)3}

on a real machine you might get 1, 3, 2, 4, 5 this time and 2, 1, 3, 5, 4 the next. This page runs one goroutine at a time, so the order is the same every time here, but don't rely on it. Collect everything first, then sort, and the result no longer depends on who went first. For the same reason, don't show which worker made which image: that can change every run.

Any number of workers, same result

There's another way to collect results: no results channel at all, and each worker writes into the slot for its image, thumbs[i] (lesson 22). Below, the whole pool is one function, run with 1, 2 and 5 workers:

1package main2 3import (4	"fmt"5	"sync"6)7 8func makeThumbs(sizes []int, workers int) []int {9	thumbs := make([]int, len(sizes))10	jobs := make(chan int)11	var wg sync.WaitGroup12	for range workers {13		wg.Add(1)14		go func() {15			defer wg.Done()16			for i := range jobs {17				thumbs[i] = sizes[i] / 1018			}19		}()20	}21	for i := range sizes {22		jobs <- i23	}24	close(jobs)25	wg.Wait()26	return thumbs27}28 29func main() {30	sizes := []int{1200, 800, 2000, 640, 1500}31	for _, n := range []int{1, 2, 5} {32		fmt.Printf("workers %d: %v\n", n, makeThumbs(sizes, n))33	}34}

workers 1: [120 80 200 64 150]
workers 2: [120 80 200 64 150]
workers 5: [120 80 200 64 150]

jobs carries indexes. Each index goes to exactly one worker, so each slot has exactly one writer, and no lock is needed. The number of workers changes who makes which thumbnail, not the result.

Putting it together

Finally, connect the two shapes: generate → 3 workers sharing the discount → sum. With several workers in the discount stage, the order coming out is anyone's guess; but the sum only adds, and order doesn't change a total.

1package main2 3import (4	"fmt"5	"sync"6)7 8func generate(prices ...int) <-chan int {9	out := make(chan int)10	go func() {11		defer close(out)12		for _, p := range prices {13			out <- p14		}15	}()16	return out17}18 19func discountWorkers(in <-chan int, n int) <-chan int {20	out := make(chan int)21	var wg sync.WaitGroup22	for range n {23		wg.Add(1)24		go func() {25			defer wg.Done()26			for p := range in {27				out <- p * 9 / 1028			}29		}()30	}31	go func() {32		wg.Wait()33		close(out)34	}()35	return out36}37 38func sum(in <-chan int) <-chan int {39	out := make(chan int)40	go func() {41		total := 042		for p := range in {43			total += p44		}45		out <- total46	}()47	return out48}49 50func main() {51	prices := discountWorkers(generate(120, 45, 80, 300, 15), 3)52	fmt.Println("total", <-sum(prices))53}

total 503

Several workers send on the same out, so none of them may close it on their own: once wg.Wait() confirms they've all ended, a single goroutine closes it. That's the same trick as with results above.

Practice

Write the code, then press Check. A hint costs nothing; the answer is there when you need it.

195
217

1package main2 3import "fmt"4 5func generate(prices ...int) <-chan int {6	out := make(chan int)7	go func() {8		defer close(out)9		for _, p := range prices {10			out <- p11		}12	}()13	return out14}15 16func discount(in <-chan int) <-chan int {17	out := make(chan int)18	go func() {19		defer close(out)20		for p := range in {21			out <- p * 9 / 1022		}23	}()24	return out25}26 27// write shipping here28 29func sum(in <-chan int) <-chan int {30	out := make(chan int)31	go func() {32		total := 033		for p := range in {34			total += p35		}36		out <- total37	}()38	return out39}40 41func main() {42	fmt.Println(<-sum(discount(generate(100, 40, 60))))43	fmt.Println(<-sum(discount(generate(200, 30))))44}

[110 275 44 77]
[110 275 44 77]

1package main2 3import (4	"fmt"5	"sync"6)7 8func withTax(prices []int, workers int) []int {9	out := make([]int, len(prices))10	jobs := make(chan int)11	var wg sync.WaitGroup12	// start the workers here13 14	for i := range prices {15		jobs <- i16	}17	close(jobs)18	wg.Wait()19	return out20}21 22func main() {23	prices := []int{100, 250, 40, 70}24	fmt.Println(withTax(prices, 2))25	fmt.Println(withTax(prices, 4))26}

order 1, boxes: 3
order 2, boxes: 1
order 3, boxes: 1
order 4, boxes: 5
order 5, boxes: 6
order 6, boxes: 1
order 7, boxes: 2

1package main2 3import (4	"fmt"5	"sort"6	"sync"7)8 9type Order struct {10	ID    int11	Items int12}13 14func pack(orders []Order) {15	jobs := make(chan Order)16	results := make(chan Order)17	var wg sync.WaitGroup18	for range 3 {19		wg.Add(1)20		go func() {21			defer wg.Done()22			for o := range jobs {23				results <- Order{ID: o.ID, Items: (o.Items + 9) / 10}24			}25		}()26	}27	go func() {28		defer close(jobs)29		for _, o := range orders {30			jobs <- o31		}32	}()33	wg.Wait()34	close(results)35	var boxes []Order36	for r := range results {37		boxes = append(boxes, r)38	}39	sort.Slice(boxes, func(a, b int) bool { return boxes[a].ID < boxes[b].ID })40	for _, b := range boxes {41		fmt.Printf("order %d, boxes: %d\n", b.ID, b.Items)42	}43}44 45func main() {46	pack([]Order{47		{ID: 1, Items: 25},48		{ID: 2, Items: 10},49		{ID: 3, Items: 3},50		{ID: 4, Items: 41},51	})52	pack([]Order{53		{ID: 5, Items: 60},54		{ID: 6, Items: 9},55		{ID: 7, Items: 11},56	})57}

What you learned

• A pipeline: each stage is a function that takes a <-chan, starts a goroutine to do its work, and returns a new <-chan; function calls connect the stages.
• Each stage closes its channel when it's done, with defer close(out), so the next stage's range can end.
• With one goroutine per stage, channels deliver in the order sent, so the order holds; adding a stage doesn't change the others.
• A worker pool: jobs go into jobs, several workers range jobs; once every worker has ended (wg.Wait()), one separate goroutine closes results.
• Nobody knows which worker finishes first: collect everything and sort, or have each worker write its own slot, and the result doesn't depend on scheduling.
• A summary such as a sum or a count doesn't care about order, so it can come straight after the workers.