Browse Source

simplified according to pipelines article

https://blog.golang.org/pipelines
master
Stephen McQuay 5 years ago
parent
commit
129663bc89
1 changed files with 51 additions and 35 deletions
  1. +51
    -35
      main.go

+ 51
- 35
main.go View File

@ -3,59 +3,75 @@ package main
import (
"bufio"
"fmt"
"io"
"os"
"sync"
"time"
)
func expensive(worker int, task string) error {
time.Sleep(250 * time.Millisecond)
return fmt.Errorf("worker %d is done with %q", worker, task)
}
func worker(id int, work chan string, results chan error, wg *sync.WaitGroup) {
defer wg.Done()
for {
i, ok := <-work
if !ok {
return
func compute(id int, work <-chan string) <-chan error {
out := make(chan error)
go func() {
for w := range work {
time.Sleep(1 * time.Second)
out <- fmt.Errorf("worker %d is done with %q", id, w)
}
results <- expensive(id, i)
}
close(out)
}()
return out
}
func main() {
work := make(chan string)
results := make(chan error)
numWorkers := 4
wwg := &sync.WaitGroup{}
wg := &sync.WaitGroup{}
for w := 0; w < numWorkers; w++ {
wwg.Add(1)
go worker(w, work, results, wwg)
}
s := bufio.NewScanner(os.Stdin)
go func(wg *sync.WaitGroup) {
func source(in io.Reader) <-chan string {
out := make(chan string)
s := bufio.NewScanner(in)
go func() {
for s.Scan() {
wg.Add(1)
work <- s.Text()
out <- s.Text()
}
if err := s.Err(); err != nil {
panic(err)
}
close(work)
}(wg)
close(out)
}()
return out
}
func merge(cs []<-chan error) <-chan error {
out := make(chan error)
var wg sync.WaitGroup
output := func(c <-chan error) {
for n := range c {
out <- n
}
wg.Done()
}
wg.Add(len(cs))
for _, c := range cs {
go output(c)
}
go func() {
wwg.Wait()
wg.Wait()
close(results)
close(out)
}()
return out
}
for e := range results {
func main() {
work := source(os.Stdin)
results := []<-chan error{}
workers := 4
for w := 0; w < workers; w++ {
results = append(results, compute(w, work))
}
out := merge(results)
for e := range out {
fmt.Printf("received: %+v\n", e)
wg.Done()
}
}

Loading…
Cancel
Save