Select from a dynamic list of channels

dynamic_select_test.go

package scratch

import (
	"reflect"
	"testing"
)

/*

Results with Go1.5 on a 6 core Xeon CPU E5-1650 0 @ 3.20GHz

$ go test dynamic_select_test.go -bench=. -cpu=1,4,12 -benchtime=20s
testing: warning: no tests to run
PASS
BenchmarkReflectSelect   	       5	1510021901 ns/op
BenchmarkReflectSelect-4 	       3	2327781407 ns/op
BenchmarkReflectSelect-12	       2	2558476135 ns/op
BenchmarkGoSelect        	     100	  81484274 ns/op
BenchmarkGoSelect-4      	     100	  59641844 ns/op
BenchmarkGoSelect-12     	     100	 109973957 ns/op
BenchmarkHybrid          	      10	 807781023 ns/op
BenchmarkHybrid-4        	     200	  46523190 ns/op
BenchmarkHybrid-12       	     200	  49585614 ns/op
ok  	command-line-arguments	99.169s

*/

const numProduce = 1000
const numChannels = 100
const expectedResult int64 = numChannels * numProduce * (numProduce - 1) / 2

func BenchmarkReflectSelect(b *testing.B) {
	for i := 0; i < b.N; i++ {
		checkResult(b, benchmarkReflectSelect(b))
	}
}

func BenchmarkGoSelect(b *testing.B) {
	for i := 0; i < b.N; i++ {
		checkResult(b, benchmarkGoSelect(b))
	}
}

func BenchmarkHybrid(b *testing.B) {
	for i := 0; i < b.N; i++ {
		checkResult(b, benchmarkHybrid(b))
	}
}

/*
func BenchmarkSanity(b *testing.B) {
	for i := 0; i < b.N; i++ {
		checkResult(b, benchmarkSanity())
	}
}*/

// Use reflection to select from a slice of channels.
func benchmarkReflectSelect(b *testing.B) int64 {
	b.StopTimer()
	channels := makeChannels(numChannels)
	var result int64 = 0

	cases := make([]reflect.SelectCase, numChannels)
	for i, ch := range channels {
		cases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)}
	}

	b.StartTimer()
	for finished := 0; finished < numChannels; {
		i, value, ok := reflect.Select(cases)
		if !ok {
			cases = append(cases[:i], cases[i+1:]...)
			finished++
		} else {
			result += value.Int()
		}
	}

	return result
}

// Use an aggregate channel to select from a slice of channels
func benchmarkGoSelect(b *testing.B) int64 {
	b.StopTimer()
	channels := makeChannels(numChannels)
	var result int64 = 0

	done := make(chan struct{})
	combinedChannel := make(chan int)
	for i := 0; i < numChannels; i++ {
		go func(c chan int) {
			for v := range c {
				combinedChannel <- v
			}
			done <- struct{}{}
		}(channels[i])
	}

	b.StartTimer()
	finished := 0
	for finished < numChannels {
		select {
		case i := <-combinedChannel:
			result += int64(i)
		case <-done:
			finished++
		}
	}
	close(combinedChannel)
	close(done)

	return result
}

func benchmarkHybrid(b *testing.B) int64 {
	b.StopTimer()
	channels := makeChannels(numChannels)
	var result int64 = 0

	cases := make([]reflect.SelectCase, numChannels)
	for i, ch := range channels {
		cases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)}
	}

	removeChannel := func(i int) {
		cases = append(cases[:i], cases[i+1:]...)
		channels = append(channels[:i], channels[i+1:]...)
	}

	b.StartTimer()

outer:
	for len(channels) > 0 {

		// First try to pull from each channel.
		for i, ch := range channels {
			select {
			case v, ok := <-ch:
				if !ok {
					removeChannel(i)
				} else {
					result += int64(v)
				}
				continue outer
			default:
				// don't block
			}
		}

		// No channels were ready to receive, fallback to reflection.
		i, value, ok := reflect.Select(cases)
		if !ok {
			removeChannel(i)
		} else {
			result += value.Int()
		}
	}

	return result
}

// Sanity check.
func benchmarkSanity() int64 {
	var result int64 = 0
	for i := 0; i < numChannels; i++ {
		for j := 0; j < numProduce; j++ {
			result += int64(j)
		}
	}
	return result
}

func checkResult(b *testing.B, result int64) {
	if result != expectedResult {
		b.Fatalf("Fail! Expected %v but got %v", expectedResult, result)
	}
}

func produce(c chan int, n int) {
	for i := 0; i < n; i++ {
		c <- i
	}
	close(c)
}

func makeChannels(n int) []chan int {
	chans := make([]chan int, n)
	for i := 0; i < n; i++ {
		c := make(chan int)
		go produce(c, numProduce)
		chans[i] = c
	}
	return chans
}