move benchmarking into tests

balance_test.go

@@ -0,0 +1,56 @@
+package main
+
+import "testing"
+
+func BenchmarkFastBalancer(b *testing.B) {
+	normalBalancer(b, fastJob)
+}
+
+func BenchmarkFastSingleThread(b *testing.B) {
+	workSingleThread(b, fastJob)
+}
+
+func BenchmarkSlowBalancer(b *testing.B) {
+	normalBalancer(b, constantJob)
+}
+
+func BenchmarkSlowSingleThread(b *testing.B) {
+	workSingleThread(b, constantJob)
+}
+
+func normalBalancer(b *testing.B, job func() int) {
+	work := make(chan Request)
+	done := make(chan *Worker)
+	pool := NewPool(4, done)
+	balancer := Balancer{*pool, done, false}
+	b.ResetTimer()
+	go balancer.balance(work)
+	for n := 0; n < b.N; n++ {
+		FiniteRequests(work, job, 20)
+	}
+}
+
+func workSingleThread(b *testing.B, job func() int) {
+	jobsRepeats := 20
+	singleWorker := make(chan Request, 20)
+	for n := 0; n < b.N; n++ {
+		go func() {
+			for i := 0; i < jobsRepeats; i++ {
+				req := <-singleWorker
+				req.result <- req.job()
+			}
+		}()
+		FiniteRequests(singleWorker, job, jobsRepeats)
+	}
+}
+
+func FiniteRequests(work chan<- Request, job func() int, size int) {
+	c := make(chan int, 5)
+	e := make(chan error, 5)
+	for i := 0; i < size; i++ {
+		work <- Request{job, c, e}
+	}
+	for i := 0; i < size; i++ {
+		<-c
+	}
+}

balancer.go

@@ -9,6 +9,7 @@ import (
 type Balancer struct {
 	pool Pool
 	done chan *Worker
+	log  bool
 }
 
 func (b *Balancer) balance(work chan Request) {
@@ -28,13 +29,17 @@ func (b *Balancer) dispatch(req Request) {
 	w.requests <- req
 	w.pending++
 	heap.Push(&b.pool, w)
-	fmt.Println(b.pool, "| DISPATCHED")
+	if b.log {
+		fmt.Println(b.pool, "| DISPATCHED")
+	}
 }
 
 // Job is complete; update heap
 func (b *Balancer) completed(w *Worker) {
 	w.pending--
 	heap.Remove(&b.pool, w.index)
 	heap.Push(&b.pool, w)
-	fmt.Println(b.pool, "| COMPLETED")
+	if b.log {
+		fmt.Println(b.pool, "| COMPLETED")
+	}
 }

main.go

@@ -8,27 +8,17 @@ import (
 )
 
 func main() {
-	compare := flag.Bool("compare", false, " compare load balancer vs single thread execution speed")
+	poolSize := flag.Int("poolSize", runtime.GOMAXPROCS(0), " number of workers in the worker pool")
 	flag.Parse()
 
-	poolSize := runtime.GOMAXPROCS(0)
-	fmt.Printf("We got %d workers!!\n", poolSize)
+	fmt.Printf("We got %d workers!!\n", *poolSize)
 
 	work := make(chan Request)
 	done := make(chan *Worker)
-	pool := NewPool(poolSize, done)
-	balancer := Balancer{*pool, done}
+	pool := NewPool(*poolSize, done)
+	balancer := Balancer{*pool, done, true}
 	go balancer.balance(work)
-
-	switch *compare {
-	case true:
-		distributedTime := Benchmark(work)
-		synchronousTime := benchmarkSingleThread()
-		fmt.Printf("Worker pool of %d workers took %s to complete\n", poolSize, distributedTime)
-		fmt.Printf("Single thread took %s to complete\n", synchronousTime)
-	case false:
-		InfiniteRequester(work)
-	}
+	InfiniteRequester(work)
 }
 
 func benchmarkSingleThread() time.Duration {

request.go

@@ -54,6 +54,12 @@ func Benchmark(work chan<- Request) time.Duration {
 }
 
 func constantJob() int {
-	time.Sleep(200 * time.Millisecond)
+	time.Sleep(100 * time.Millisecond)
 	return 1
 }
+
+func fastJob() int {
+	y := rand.Intn(4000) / 5
+	x := 2 + 5/4 + y
+	return x
+}

worker.go

@@ -22,6 +22,6 @@ func (w *Worker) work(done chan *Worker) {
 		req = <-w.requests      // get Request from balancer
 		req.result <- req.job() // call fn and send result
 		done <- w               // we've finished this request
-
 	}
+
 }