Merge pull request #116 from yarpc/dev

Release latest yab changes as 0.6.2

.travis.yml

@@ -9,6 +9,7 @@ go:
 env:
     global:
         - GO15VENDOREXPERIMENT=1
+        - TEST_TIMEOUT_SCALE=10
 
 cache:
   directories:

CHANGELOG.md

@@ -1,6 +1,13 @@
 Changelog
 =========
 
+# 0.6.2
+
+* Add error rate to benchmarking output.
+* Improve boolean parsing for Thrift input.
+* Check the system XDG directory for a config file.
+* Allow unlimited duration or requests when benchmarking. (#105)
+
 # 0.6.1
 
 * Improve default format detection:

Makefile

@@ -2,11 +2,13 @@ PACKAGES := $(shell glide novendor)
 
 export GO15VENDOREXPERIMENT=1
 
+.DEFAULT_GOAL:=build
+
 
 .PHONY: build
 build:
 	go build -i $(PACKAGES)
-
+	go build -i .
 
 .PHONY: install
 install:

README.md

@@ -16,6 +16,8 @@ go get -u -f github.com/yarpc/yab
 
 This will install `yab` to `$GOPATH/bin/yab`.
 
+Optionally, you can get precompiled binaries from [Releases][releases].
+
 ### Usage
 
 ```
@@ -136,6 +138,7 @@ connection (`--concurrency`).
 yab -t ~/keyvalue.thrift -p localhost:12345 keyvalue KeyValue::get -r '{"key": "hello"}' -d 5s --rps 100 --connections 4
 ```
 
+[releases]: https://github.com/yarpc/yab/releases
 [ci-img]: https://travis-ci.org/yarpc/yab.svg?branch=master
 [ci]: https://travis-ci.org/yarpc/yab
 [cov-img]: https://coveralls.io/repos/github/yarpc/yab/badge.svg?branch=master

bench_state.go

@@ -30,9 +30,12 @@ import (
 )
 
 type benchmarkState struct {
-	statter   statsd.Client
-	errors    map[string]int
-	latencies []time.Duration
+	statter       statsd.Client
+	errors        map[string]int
+	totalErrors   int
+	totalSuccess  int
+	totalRequests int
+	latencies     []time.Duration
 }
 
 func newBenchmarkState(statter statsd.Client) *benchmarkState {
@@ -42,13 +45,19 @@ func newBenchmarkState(statter statsd.Client) *benchmarkState {
 	}
 }
 
+func (s *benchmarkState) recordRequest() {
+	s.totalRequests++
+}
+
 func (s *benchmarkState) recordError(err error) {
 	if err == nil {
 		panic("recordError not passed error")
 	}
+	s.recordRequest()
 
 	msg := errorToMessage(err)
 	s.errors[msg]++
+	s.totalErrors++
 	s.statter.Inc("error")
 }
 
@@ -57,10 +66,15 @@ func (s *benchmarkState) merge(other *benchmarkState) {
 		s.errors[k] += v
 	}
 	s.latencies = append(s.latencies, other.latencies...)
+	s.totalErrors += other.totalErrors
+	s.totalSuccess += other.totalSuccess
+	s.totalRequests += other.totalRequests
 }
 
 func (s *benchmarkState) recordLatency(d time.Duration) {
+	s.recordRequest()
 	s.latencies = append(s.latencies, d)
+	s.totalSuccess++
 	s.statter.Inc("success")
 	s.statter.Timing("latency", d)
 }
@@ -79,13 +93,12 @@ func (s *benchmarkState) printErrors(out output) {
 		return
 	}
 	out.Printf("Errors:\n")
-	total := 0
 	for _, k := range sorted.MapKeys(s.errors) {
 		v := s.errors[k]
 		out.Printf("  %4d: %v\n", v, k)
-		total += v
 	}
-	out.Printf("Total errors: %v\n", total)
+	out.Printf("Total errors: %v\n", s.totalErrors)
+	out.Printf("Error rate: %.4f%%\n", 100*float32(s.totalErrors)/float32(s.totalRequests))
 }
 
 func (s *benchmarkState) getQuantile(q float64) time.Duration {

bench_state_test.go

@@ -49,7 +49,19 @@ func TestBenchmarkStateErrors(t *testing.T) {
 	})
 
 	buf, out := getOutput(t)
+
+	// before merge
+	assert.Equal(t, state1.totalErrors, 4, "Error count mismatch")
+	assert.Equal(t, state1.totalSuccess, 0, "Success count mismatch")
+	assert.Equal(t, state1.totalRequests, 4, "Request count mismatch")
+
 	state1.merge(state2)
+
+	// after merge
+	assert.Equal(t, state1.totalErrors, 7, "Error count mismatch")
+	assert.Equal(t, state1.totalSuccess, 0, "Success count mismatch")
+	assert.Equal(t, state1.totalRequests, 7, "Request count mismatch")
+
 	state1.printErrors(out)
 
 	expected := map[string]int{
@@ -88,6 +100,11 @@ func TestBenchmarkStateLatencies(t *testing.T) {
 	}
 
 	buf, out := getOutput(t)
+
+	assert.Equal(t, state.totalErrors, 0, "Error count mismatch")
+	assert.Equal(t, state.totalSuccess, 10001, "Success count mismatch")
+	assert.Equal(t, state.totalRequests, 10001, "Request count mismatch")
+
 	state.printLatencies(out)
 
 	expected := []string{
@@ -122,8 +139,16 @@ func TestBenchmarkStateMergeLatencies(t *testing.T) {
 			state2.recordLatency(time.Duration(i) * time.Microsecond)
 		}
 	}
+	assert.Equal(t, state1.totalErrors, 0, "Error count mismatch")
+	assert.Equal(t, state1.totalSuccess, 5001, "Success count mismatch")
+	assert.Equal(t, state1.totalRequests, 5001, "Request count mismatch")
+
 	state1.merge(state2)
 
+	assert.Equal(t, state1.totalErrors, 0, "Error count mismatch")
+	assert.Equal(t, state1.totalSuccess, 10001, "Success count mismatch")
+	assert.Equal(t, state1.totalRequests, 10001, "Request count mismatch")
+
 	buf, out := getOutput(t)
 	state1.printLatencies(out)
 

benchmark.go

@@ -65,16 +65,23 @@ func runWorker(t transport.Transport, m benchmarkMethod, s *benchmarkState, run
 func runBenchmark(out output, allOpts Options, m benchmarkMethod) {
 	opts := allOpts.BOpts
 
-	// By default, benchmarks are disabled. At least MaxDuration needs to
-	// be set to enable them.
-	if opts.MaxDuration == 0 {
+	// By default, benchmarks are disabled. At least MaxDuration or MaxRequests
+	// should be > 0 for the benchmark to start.
+	if opts.MaxDuration == 0 && opts.MaxRequests == 0 {
 		return
 	}
 
-	if opts.RPS > 0 {
+	if opts.MaxDuration < 0 {
+		out.Fatalf("Benchmark duration cannot be negative")
+	}
+	if opts.MaxRequests < 0 {
+		out.Fatalf("Benchmark max requests cannot be negative")
+	}
+
+	if opts.RPS > 0 && opts.MaxDuration > 0 {
 		// The RPS * duration in seconds may cap opts.MaxRequests.
 		rpsMax := int(float64(opts.RPS) * opts.MaxDuration.Seconds())
-		if rpsMax < opts.MaxRequests {
+		if rpsMax < opts.MaxRequests || opts.MaxRequests == 0 {
 			opts.MaxRequests = rpsMax
 		}
 	}
@@ -138,8 +145,8 @@ func runBenchmark(out output, allOpts Options, m benchmarkMethod) {
 	overall.printLatencies(out)
 
 	out.Printf("Elapsed time:      %v\n", (total / time.Millisecond * time.Millisecond))
-	out.Printf("Total requests:    %v\n", len(overall.latencies))
-	out.Printf("RPS:               %.2f\n", float64(len(overall.latencies))/total.Seconds())
+	out.Printf("Total requests:    %v\n", overall.totalRequests)
+	out.Printf("RPS:               %.2f\n", float64(overall.totalRequests)/total.Seconds())
 }
 
 // stopOnInterrupt sets up a signal that will trigger the run to stop.

benchmark_test.go

@@ -21,16 +21,46 @@
 package main
 
 import (
+	"fmt"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/yarpc/yab/transport"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/uber-go/atomic"
+	"github.com/uber/tchannel-go/testutils"
 )
 
 func TestBenchmark(t *testing.T) {
+	tests := []struct {
+		msg          string
+		n            int
+		d            time.Duration
+		rps          int
+		want         int
+		wantDuration time.Duration
+	}{
+		{
+			msg:  "Capped by max requests",
+			n:    100,
+			d:    100 * time.Second,
+			want: 100,
+		},
+		{
+			msg:          "Capped by RPS * duration",
+			d:            500 * time.Millisecond,
+			rps:          120,
+			want:         60,
+			wantDuration: 500 * time.Millisecond,
+		},
+		{
+			msg:          "Capped by duration",
+			d:            500 * time.Millisecond,
+			wantDuration: 500 * time.Millisecond,
+		},
+	}
 
 	var requests atomic.Int32
 	s := newServer(t)
@@ -40,24 +70,83 @@ func TestBenchmark(t *testing.T) {
 	}))
 
 	m := benchmarkMethodForTest(t, fooMethod, transport.TChannel)
-	buf, out := getOutput(t)
-
-	runBenchmark(out, Options{
-		BOpts: BenchmarkOptions{
-			MaxRequests:    1000,
-			MaxDuration:    time.Second,
-			Connections:    50,
-			WarmupRequests: 10,
-			Concurrency:    2,
+
+	for _, tt := range tests {
+		requests.Store(0)
+
+		start := time.Now()
+		buf, out := getOutput(t)
+		runBenchmark(out, Options{
+			BOpts: BenchmarkOptions{
+				MaxRequests:    tt.n,
+				MaxDuration:    tt.d,
+				RPS:            tt.rps,
+				Connections:    50,
+				WarmupRequests: 10,
+				Concurrency:    2,
+			},
+			TOpts: s.transportOpts(),
+		}, m)
+
+		bufStr := buf.String()
+		assert.Contains(t, bufStr, "Max RPS")
+		assert.NotContains(t, bufStr, "Errors")
+
+		warmupExtra := 10 * 50 // warmup requests * connections
+		if tt.want != 0 {
+			assert.EqualValues(t, tt.want+warmupExtra, requests.Load(),
+				"%v: Invalid number of requests", tt.msg)
+		}
+
+		if tt.wantDuration != 0 {
+			// Make sure the total duration is within a delta.
+			slack := testutils.Timeout(500 * time.Millisecond)
+			duration := time.Since(start)
+			assert.True(t, duration <= tt.wantDuration+slack && duration >= tt.wantDuration-slack,
+				"%v: Took %v, wanted duration %v", tt.msg, duration, tt.wantDuration)
+		}
+	}
+}
+
+func TestRunBenchmarkErrors(t *testing.T) {
+	tests := []struct {
+		opts    BenchmarkOptions
+		wantErr string
+	}{
+		{
+			opts: BenchmarkOptions{
+				MaxRequests: -1,
+			},
+			wantErr: "max requests cannot be negative",
+		},
+		{
+			opts: BenchmarkOptions{
+				MaxDuration: -time.Second,
+			},
+			wantErr: "duration cannot be negative",
 		},
-		TOpts: s.transportOpts(),
-	}, m)
+	}
+
+	for _, tt := range tests {
+		var fatalMessage string
+		out := &testOutput{
+			fatalf: func(msg string, args ...interface{}) {
+				fatalMessage = fmt.Sprintf(msg, args...)
+			},
+		}
+		m := benchmarkMethodForTest(t, fooMethod, transport.TChannel)
+		opts := Options{BOpts: tt.opts}
 
-	bufStr := buf.String()
-	assert.Contains(t, bufStr, "Max RPS")
-	assert.NotContains(t, bufStr, "Errors")
+		var wg sync.WaitGroup
+		wg.Add(1)
+		// Since the benchmark calls Fatalf which kills the current goroutine, we
+		// need to run the benchmark in a separate goroutine.
+		go func() {
+			defer wg.Done()
+			runBenchmark(out, opts, m)
+		}()
 
-	// Due to warm up, we make:
-	// 10 * Connections extra requests
-	assert.EqualValues(t, 1000+10*50, requests.Load(), "Invalid number of requests")
+		wg.Wait()
+		assert.Contains(t, fatalMessage, tt.wantErr, "Missing error for %+v", tt.opts)
+	}
 }