dynamic_workers.go

package main

import (
	"log"

	"golang.org/x/sync/errgroup"
)

func main() {
	var g errgroup.Group
	data := make(chan int)

	g.Go(func() error {
		produce(100, data)
		return nil
	})
	g.Go(func() error {
		consume(data)
		return nil
	})

	if err := g.Wait(); err != nil {
		log.Fatal(err)
	}
}

func produce(N int, data chan<- int) {
	log.Print("Producer: started")
	for ; N > 0; N-- {
		data <- N
		log.Printf("Producer: sent %d out", N)
	}
	close(data)
	log.Printf("Producer: done")
}

func consume(data <-chan int) {
	log.Printf("Consumer: started")
	var g errgroup.Group
	N := int64(0)
	for datum := range data {
		id := N
		g.Go(func() error {
			process(id, datum)
			return nil
		})
		N++
	}
	err := g.Wait()
	log.Printf("Consumer: done, processed %d items", N)
	if err != nil {
		log.Fatalf("error consuming data: %s", err)
	}
}

func process(id int64, datum int) {
	log.Printf("Worker %d: started", id)
	log.Printf("Worker %d: processed %d", id, datum)
	log.Printf("Worker %d: done", id)
}

fixed_workers_dynamic_data.go

package main

import (
	"log"
	"runtime"

	"golang.org/x/sync/errgroup"
)

func main() {
	var g errgroup.Group
	data := make(chan int)

	g.Go(func() error {
		source(10, data)
		return nil
	})

	maxworkers := runtime.GOMAXPROCS(0)
	for i := 0; i < maxworkers; i++ {
		i := i // without this, all workers would report the same ID
		g.Go(func() error {
			worker(i, data)
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		log.Fatal(err)
	}
}

func source(N int, data chan<- int) {
	for i := 0; i < N; i++ {
		data <- i
	}
	close(data)
}

func worker(i int, data <-chan int) {
	N := 0
	log.Printf("Worker %d: starting", i)
	for datum := range data {
		N++
		log.Printf("Worker %d: processed %d", i, datum)
	}
	log.Printf("Worker %d: done after processing %d requests", i, N)
}