small adjustments

Author: steevehook

patterns/fan-in-out/efficient/main.go

@@ -0,0 +1,64 @@
+package main
+
+import (
+	"concurrency/patterns/fanin"
+	"fmt"
+	"math/rand"
+	"runtime"
+	"time"
+
+	"concurrency/patterns/generators"
+)
+
+var (
+	take     = generators.Take
+	toInt    = generators.ToInt
+	repeatFn = generators.RepeatFn
+)
+
+func main() {
+	done := make(chan struct{})
+	defer close(done)
+	random := func() interface{} {
+		return rand.Intn(50000000)
+	}
+	start := time.Now()
+	randIntStream := toInt(done, repeatFn(done, random))
+	numFinders := runtime.NumCPU()
+	finders := make([]<-chan int, numFinders)
+	for i := 0; i < numFinders; i++ {
+		finders[i] = toInt(done, primeFinder(done, randIntStream))
+	}
+
+	fmt.Println("primes:")
+	for prime := range take(done, fanin.FanIn(done, finders...), 10) {
+		fmt.Println("prime:", prime)
+	}
+	fmt.Printf("search took: %v", time.Since(start))
+}
+
+func primeFinder(done <-chan struct{}, intStream <-chan int) <-chan interface{} {
+	primeStream := make(chan interface{})
+	go func() {
+		defer close(primeStream)
+		for integer := range intStream {
+			integer -= 1
+			prime := true
+			for divisor := integer - 1; divisor > 1; divisor-- {
+				if integer%divisor == 0 {
+					prime = false
+					break
+				}
+			}
+
+			if prime {
+				select {
+				case <-done:
+					return
+				case primeStream <- integer:
+				}
+			}
+		}
+	}()
+	return primeStream
+}

patterns/fan-in-out/inefficient/main.go

@@ -0,0 +1,57 @@
+package main
+
+import (
+	"fmt"
+	"math/rand"
+	"time"
+
+	"concurrency/patterns/generators"
+)
+
+var (
+	take     = generators.Take
+	toInt    = generators.ToInt
+	repeatFn = generators.RepeatFn
+)
+
+func main() {
+	done := make(chan struct{})
+	defer close(done)
+	random := func() interface{} {
+		return rand.Intn(50000000)
+	}
+	start := time.Now()
+
+	randIntStream := toInt(done, repeatFn(done, random))
+	fmt.Println("primes:")
+	for prime := range take(done, primeFinder(done, randIntStream), 10) {
+		fmt.Println("prime:", prime)
+	}
+	fmt.Printf("search took: %v", time.Since(start))
+}
+
+func primeFinder(done <-chan struct{}, intStream <-chan int) <-chan interface{} {
+	primeStream := make(chan interface{})
+	go func() {
+		defer close(primeStream)
+		for integer := range intStream {
+			integer -= 1
+			prime := true
+			for divisor := integer - 1; divisor > 1; divisor-- {
+				if integer%divisor == 0 {
+					prime = false
+					break
+				}
+			}
+
+			if prime {
+				select {
+				case <-done:
+					return
+				case primeStream <- integer:
+				}
+			}
+		}
+	}()
+	return primeStream
+}

patterns/fan-in-out/main.go

@@ -15,9 +15,15 @@ var (
 )
 
 func main() {
+	done := make(chan struct{})
+	defer close(done)
+
 	odd := intGen(20)
-	c1, c2, c3 := fanOut(odd), fanOut(odd), fanOut(odd)
-	for v := range fanIn(c1, c2, c3) {
+	c1 := fanOut(done, odd)
+	c2 := fanOut(done, odd)
+	c3 := fanOut(done, odd)
+
+	for v := range fanIn(done, c1, c2, c3) {
 		fmt.Println("value:", v)
 	}
 }

patterns/fanin/cmd/main.go

@@ -8,10 +8,14 @@ import (
 )
 
 func main() {
+	done := make(chan struct{})
+	defer close(done)
+
 	odd := generator.OddIntGen(5)
 	even := generator.EvenIntGen(5)
 	hex := generator.HexIntGen(5)
-	out := fanin.FanIn(odd, even, hex)
+	out := fanin.FanIn(done, odd, even, hex)
+
 	for n := range out {
 		fmt.Println("fanned number:", n)
 	}

patterns/fanin/fanin.go

@@ -1,25 +1,33 @@
 package fanin
 
+import (
+	"sync"
+)
+
 // FanIn reads from multiple channels and writes into 1 final channel
 // The FAN-IN aka Multiplexing pattern states that a function receives multiple channels as inputs
 // It reads each input and sends all the values into 1 final output channel
-func FanIn(inputs ...<-chan int) <-chan int {
+func FanIn(done chan struct{}, inputs ...<-chan int) <-chan int {
 	out := make(chan int)
-	done := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(len(inputs))
+
 	for _, in := range inputs {
 		go func(numbers <-chan int) {
+			defer wg.Done()
 			for n := range numbers {
-				out <- n
+				select {
+				case <-done:
+					return
+				case out <- n:
+				}
 			}
-			done <- struct{}{}
 		}(in)
 	}
 	go func() {
-		for i := 0; i < len(inputs); i++ {
-			<-done
-		}
-		close(done)
+		wg.Wait()
 		close(out)
 	}()
+
 	return out
 }

patterns/fanout/cmd/main.go

@@ -8,11 +8,15 @@ import (
 )
 
 func main() {
+	done := make(chan struct{})
+	defer close(done)
+
 	odd := generator.OddIntGen(10)
-	c1 := fanout.FanOut(odd)
-	c2 := fanout.FanOut(odd)
-	c3 := fanout.FanOut(odd)
-	c4 := fanout.FanOut(odd)
+	c1 := fanout.FanOut(done, odd)
+	c2 := fanout.FanOut(done, odd)
+	c3 := fanout.FanOut(done, odd)
+	c4 := fanout.FanOut(done, odd)
+
 	display(c1, "c1")
 	display(c2, "c2")
 	display(c3, "c3")

patterns/fanout/fanout.go

@@ -4,11 +4,15 @@ package fanout
 // The FAN-OUT pattern states that multiple invocation of this function
 // will generate multiple go routines to read from the same channel till the input channel is closed
 // This allows for better work distribution
-func FanOut(input <-chan int) <-chan int {
+func FanOut(done chan struct{}, input <-chan int) <-chan int {
 	out := make(chan int)
 	go func() {
 		for v := range input {
-			out <- v
+			select {
+			case <-done:
+				return
+			case out <- v:
+			}
 		}
 		close(out)
 	}()

patterns/pipeline/cmd/main.go

@@ -14,20 +14,23 @@ var (
 )
 
 func main() {
+	done := make(chan struct{})
+	defer close(done)
+
 	// 1, 2, 3
-	nums := gen(1, 2, 3)
+	nums := gen(done, 1, 2, 3)
 	// 2, 3, 4
-	incremented := inc(nums)
+	incremented := inc(done, nums)
 	// 4, 9, 16
-	squared := sq(incremented)
+	squared := sq(done, incremented)
 	// 3, 8, 15
-	res := dec(squared)
+	res := dec(done, squared)
 	for n := range res {
 		fmt.Println(n)
 	}
 
 	fmt.Println("the same exact result using nested calls")
-	for n := range dec(sq(inc(gen(1, 2, 3)))) {
+	for n := range dec(done, sq(done, inc(done, gen(done, 1, 2, 3)))) {
 		fmt.Println(n)
 	}
 

patterns/pipeline/pipeline.go

@@ -44,49 +44,65 @@ func (p *IntPipeline) Square() *IntPipeline {
 	return p
 }
 
-func (p *IntPipeline) Result() chan int {
+func (p *IntPipeline) Result() <-chan int {
 	close(p.out)
 	return p.out
 }
 
-func Gen(vs ...int) chan int {
+func Gen(done chan struct{}, vs ...int) <-chan int {
 	out := make(chan int)
 	go func() {
 		for _, n := range vs {
-			out <- n
+			select {
+			case <-done:
+				return
+			case out <- n:
+			}
 		}
 		close(out)
 	}()
 	return out
 }
 
-func Inc(in <-chan int) chan int {
+func Inc(done chan struct{}, in <-chan int) <-chan int {
 	out := make(chan int)
 	go func() {
 		for i := range in {
-			out <- i + 1
+			select {
+			case <-done:
+				return
+			case out <- i + 1:
+			}
 		}
 		close(out)
 	}()
 	return out
 }
 
-func Dec(in <-chan int) chan int {
+func Dec(done chan struct{}, in <-chan int) <-chan int {
 	out := make(chan int)
 	go func() {
 		for i := range in {
-			out <- i - 1
+			select {
+			case <-done:
+				return
+			case out <- i - 1:
+			}
 		}
 		close(out)
 	}()
 	return out
 }
 
-func Sq(in <-chan int) chan int {
+func Sq(done chan struct{}, in <-chan int) <-chan int {
 	out := make(chan int)
 	go func() {
 		for i := range in {
-			out <- i * i
+			select {
+			case <-done:
+				return
+			case out <- i * i:
+			}
 		}
 		close(out)
 	}()

patterns/tee-channel/main.go

@@ -16,16 +16,16 @@ func main() {
 	done := make(chan struct{})
 	defer close(done)
 
-	nums := pipeline.Gen(1, 2, 3, 4)
+	nums := pipeline.Gen(done, 1, 2, 3, 4)
 	out := tee(done, nums, 3)
 
-	for v := range pipeline.Inc(out[0]) {
+	for v := range pipeline.Inc(done, out[0]) {
 		fmt.Println("incremented:", v)
 	}
-	for v := range pipeline.Dec(out[1]) {
+	for v := range pipeline.Dec(done, out[1]) {
 		fmt.Println("decremented:", v)
 	}
-	for v := range pipeline.Sq(out[2]) {
+	for v := range pipeline.Sq(done, out[2]) {
 		fmt.Println("squared:", v)
 	}
 }