Worker pools, semaphores and bounded parallelism

// pool.go: 200 images to read from a store that serves at most 8 reads at a time and refuses a read beyond that, as
// a storage service with a concurrency limit does. One goroutine per image sends all 200 reads at once; a pool of 8
// workers ranging over one jobs channel never has more than 8 in flight, and the producer waits while all 8 are busy.
package main

import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

var errBusy = errors.New("store busy")

// store serves at most limit reads at once, each taking 20 ms, and refuses a read while limit are in flight.
type store struct {
	limit        int64
	active, most atomic.Int64
}

func (s *store) read(id int) (int, error) {
	n := s.active.Add(1)
	if n > s.limit {
		s.active.Add(-1)
		return 0, errBusy
	}
	for m := s.most.Load(); n > m && !s.most.CompareAndSwap(m, n); m = s.most.Load() {
	}
	defer s.active.Add(-1)
	time.Sleep(20 * time.Millisecond)
	return 1000 + id, nil
}

// perImage starts one goroutine for each image, all at once.
func perImage(s *store, ids []int) (done, refused int) {
	var mu sync.Mutex
	var wg sync.WaitGroup
	for _, id := range ids {
		wg.Go(func() {
			_, err := s.read(id)
			mu.Lock()
			defer mu.Unlock()
			if err != nil {
				refused++
			} else {
				done++
			}
		})
	}
	wg.Wait()
	return done, refused
}

// pool starts n workers that range over one jobs channel. It shuts down in order: the producer closes jobs, each
// worker's loop ends, and a closer closes results once wg.Wait says that every worker has returned.
func pool(s *store, ids []int, n int) (done, refused int, took, waited time.Duration) {
	start := time.Now()
	jobs := make(chan int)
	results := make(chan error)
	var wg sync.WaitGroup
	for range n {
		wg.Go(func() {
			for id := range jobs {
				_, err := s.read(id)
				results <- err
			}
		})
	}
	go func() {
		for _, id := range ids {
			t := time.Now()
			jobs <- id // waits while all n workers are busy
			waited += time.Since(t)
		}
		close(jobs) // written after waited, so main reads waited only once results is closed
	}()
	go func() {
		wg.Wait()
		close(results)
	}()
	for err := range results {
		if err != nil {
			refused++
		} else {
			done++
		}
	}
	return done, refused, time.Since(start), waited
}

// about rounds a time to the nearest 100 ms: 25 rounds of 20 ms reads gather a little scheduling delay.
func about(d time.Duration) string { return "about " + d.Round(100*time.Millisecond).String() }

func main() {
	ids := make([]int, 200)
	for i := range ids {
		ids[i] = i
	}
	done, refused := perImage(&store{limit: 8}, ids)
	fmt.Printf("one goroutine per image: %3d reads done, %3d refused\n", done, refused)
	s := &store{limit: 8}
	done, refused, took, waited := pool(s, ids, 8)
	fmt.Printf("a pool of 8 workers:     %3d reads done, %3d refused\n", done, refused)
	fmt.Printf("the pool kept at most %d reads in flight and took %s\n", s.most.Load(), about(took))
	fmt.Printf("the producer waited %s in all for a free worker\n", about(waited))
}

one goroutine per image:   8 reads done, 192 refused
a pool of 8 workers:     200 reads done,   0 refused
the pool kept at most 8 reads in flight and took about 500ms
the producer waited about 500ms in all for a free worker

// sem.go: the same 200 reads, bounded by a buffered channel of struct{} used as a counting semaphore: a send takes
// one of its 8 slots and blocks while all 8 are taken, a receive gives a slot back. Taking the slot inside each
// goroutine bounds the reads; taking it before the go statement bounds the goroutines too.
package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

// read stands for one read from the store of pool.go: 20 ms of waiting.
func read(id int) int {
	time.Sleep(20 * time.Millisecond)
	return 1000 + id
}

// peak counts what runs now and the most that ever ran at once.
type peak struct{ now, most atomic.Int64 }

func (p *peak) up() {
	n := p.now.Add(1)
	for m := p.most.Load(); n > m && !p.most.CompareAndSwap(m, n); m = p.most.Load() {
	}
}

func (p *peak) down() { p.now.Add(-1) }

// slotInside starts every goroutine at once; each one waits for a slot before it reads.
func slotInside(ids []int) (reads, goroutines int64) {
	sem := make(chan struct{}, 8)
	var r, g peak
	var wg sync.WaitGroup
	for _, id := range ids {
		wg.Go(func() {
			g.up()
			sem <- struct{}{} // take a slot: blocks while all 8 are taken
			r.up()
			read(id)
			r.down()
			g.down()
			<-sem // give the slot back
		})
	}
	wg.Wait()
	return r.most.Load(), g.most.Load()
}

// slotBefore takes the slot in the loop, before the go statement, so a goroutine starts only when it can read.
func slotBefore(ids []int) (reads, goroutines int64) {
	sem := make(chan struct{}, 8)
	var r, g peak
	var wg sync.WaitGroup
	for _, id := range ids {
		sem <- struct{}{} // the loop itself waits here while all 8 are taken
		wg.Go(func() {
			g.up()
			r.up()
			read(id)
			r.down()
			g.down()
			<-sem
		})
	}
	wg.Wait()
	return r.most.Load(), g.most.Load()
}

func main() {
	ids := make([]int, 200)
	for i := range ids {
		ids[i] = i
	}
	r, g := slotInside(ids)
	fmt.Printf("slot taken inside each goroutine: %d reads at once, %3d goroutines alive at once\n", r, g)
	r, g = slotBefore(ids)
	fmt.Printf("slot taken before each go:        %d reads at once, %3d goroutines alive at once\n", r, g)
}

slot taken inside each goroutine: 8 reads at once, 200 goroutines alive at once
slot taken before each go:        8 reads at once,   8 goroutines alive at once

"""The starter ResizeAll of practice/resize.go against its tests: all three fail until the pool is written."""

import session

session.run(
    """
cd practice && go test .
""",
    expect_failure=["cd practice && go test ."],
)

$ cd practice && go test .
--- FAIL: TestOrderKept (…s)
    resize_test.go:50: got 0 thumbnails; want 8
--- FAIL: TestAtMostN (…s)
    resize_test.go:68: 0 thumbnails, at most 0 resizes at once, in 0s; want 10, 3 and 40ms
--- FAIL: TestOneAtATime (…s)
    resize_test.go:80: 0 thumbnails, at most 0 resizes at once, in 0s; want 5, 1 and 50ms
FAIL
FAIL	example.com/practice	…s
FAIL
[exit status 1]