bounded parallel pattern inspired by http://blog.golang.org/pipelines and http://d.hatena.ne.jp/sugyan/20140717/1405606334

bounded-sleep.go

// +build OMIT

package main

import (
	"errors"
	"fmt"
	"math/rand"

	"sort"
	"sync"
	"time"
)

func countNums(done <-chan struct{}, max int) (<-chan int, <-chan error) {
	nums := make(chan int)
	errc := make(chan error, 1)
	go func() { // HL
		// Close the paths channel after Walk returns.
		defer close(nums) // HL
		// No select needed for this send, since errc is buffered.
		errc <- func() error {
			for num := 0; num < max; num++ {
				select {
				case nums <- num:
				case <-done:
					return errors.New("walk canceled")
				}
			}
			return nil
		}()
	}()
	return nums, errc
}

type result struct {
	index    int
	sleepSec int
	err      error
}

func randomSleep() (int, error) {
	timeout := time.After(5 * time.Second)

	sec := rand.Intn(10)
	c1 := make(chan int, 1)
	go func() {
		time.Sleep(time.Duration(sec) * time.Second)
		c1 <- 0
	}()

	select {
	case <-c1:
	case <-timeout:
		fmt.Println("timeout")
		return -1, errors.New("Timeout")
	}

	return sec, nil
}

func worker(done <-chan struct{}, nums <-chan int, c chan<- result) {
	for index := range nums { // HLpaths
		sleepSec, err := randomSleep()
		fmt.Printf("%d: %d sec sleeped\n", index, sleepSec)
		select {
		case c <- result{index, sleepSec, err}:
		case <-done:
			return
		}
	}
}

func All() (map[int]result, error) {
	// MD5All closes the done channel when it returns; it may do so before
	// receiving all the values from c and errc.
	done := make(chan struct{})
	defer close(done)

	maxIndex := 100
	nums, errc := countNums(done, maxIndex)

	// Start a fixed number of goroutines to read and digest files.
	c := make(chan result) // HLc
	var wg sync.WaitGroup
	const numWorkers = 20
	wg.Add(numWorkers)
	for i := 0; i < numWorkers; i++ {
		go func() {
			defer wg.Done()
			worker(done, nums, c) // HLc
		}()
	}
	go func() {
		wg.Wait()
		close(c) // HLc
	}()
	// End of pipeline. OMIT

	m := make(map[int]result, maxIndex)
	for r := range c {
		m[r.index] = r
	}
	// Check whether the Walk failed.
	if err := <-errc; err != nil { // HLerrc
		return nil, err
	}
	return m, nil
}

func main() {
	m, err := All()
	if err != nil {
		fmt.Println(err)
		return
	}

	var indexes []int
	for index := range m {
		indexes = append(indexes, index)
	}

	sort.Ints(indexes)

	for _, i := range indexes {
		r := m[i]
		if r.err != nil {
			fmt.Printf("%04d: %s\n", r.index, r.err)
		} else {
			fmt.Printf("%04d: %d sec sleeped\n", r.index, r.sleepSec)
		}
	}
}