First commit:Benchmarks that runs server and client and separate processes.

Author: MakMukhi

benchmark/benchmark.go

@@ -227,9 +227,14 @@ func DoByteBufStreamingRoundTrip(stream testpb.BenchmarkService_StreamingCallCli
 
 // NewClientConn creates a gRPC client connection to addr.
 func NewClientConn(addr string, opts ...grpc.DialOption) *grpc.ClientConn {
+	return NewClientConnWithContext(context.Background(), addr, opts...)
+}
+
+// NewClientConnWithContext creates a gRPC client connection to addr using ctx.
+func NewClientConnWithContext(ctx context.Context, addr string, opts ...grpc.DialOption) *grpc.ClientConn {
 	opts = append(opts, grpc.WithWriteBufferSize(128*1024))
 	opts = append(opts, grpc.WithReadBufferSize(128*1024))
-	conn, err := grpc.Dial(addr, opts...)
+	conn, err := grpc.DialContext(ctx, addr, opts...)
 	if err != nil {
 		grpclog.Fatalf("NewClientConn(%q) failed to create a ClientConn %v", addr, err)
 	}

benchmark/client/main.go

@@ -20,14 +20,15 @@ package main
 
 import (
 	"flag"
-	"math"
-	"net"
-	"net/http"
-	_ "net/http/pprof"
+	"fmt"
+	"os"
+	"runtime"
+	"runtime/pprof"
 	"sync"
 	"time"
 
 	"golang.org/x/net/context"
+	"golang.org/x/sys/unix"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/benchmark"
 	testpb "google.golang.org/grpc/benchmark/grpc_testing"
@@ -36,145 +37,160 @@ import (
 )
 
 var (
-	server            = flag.String("server", "", "The server address")
-	maxConcurrentRPCs = flag.Int("max_concurrent_rpcs", 1, "The max number of concurrent RPCs")
-	duration          = flag.Int("duration", math.MaxInt32, "The duration in seconds to run the benchmark client")
-	trace             = flag.Bool("trace", true, "Whether tracing is on")
-	rpcType           = flag.Int("rpc_type", 0,
+	port    = flag.String("port", "50051", "Localhost port to connect to.")
+	r       = flag.Int("r", 1, "The number of concurrent RPCs on each connection.")
+	c       = flag.Int("c", 1, "The number of parallel connections.")
+	w       = flag.Int("w", 10, "Warm-up duration in seconds")
+	d       = flag.Int("d", 60, "Benchmark duration in seconds")
+	rqSize  = flag.Int("req", 1, "Request message size in bytes.")
+	rspSize = flag.Int("resp", 1, "Response message size in bytes.")
+	rpcType = flag.String("rpc_type", "unary",
 		`Configure different client rpc type. Valid options are:
-		   0 : unary call;
-		   1 : streaming call.`)
+		   unary;
+		   streaming.`)
+	testName = flag.String("test_name", "", "Name of the test used for creating profiles.")
+	wg       sync.WaitGroup
+	hopts    = stats.HistogramOptions{
+		NumBuckets:   2495,
+		GrowthFactor: .01,
+	}
+	mu    sync.Mutex
+	hists []*stats.Histogram
 )
 
-func unaryCaller(client testpb.BenchmarkServiceClient) {
-	benchmark.DoUnaryCall(client, 1, 1)
-}
-
-func streamCaller(stream testpb.BenchmarkService_StreamingCallClient) {
-	benchmark.DoStreamingRoundTrip(stream, 1, 1)
-}
-
-func buildConnection() (s *stats.Stats, conn *grpc.ClientConn, tc testpb.BenchmarkServiceClient) {
-	s = stats.NewStats(256)
-	conn = benchmark.NewClientConn(*server)
-	tc = testpb.NewBenchmarkServiceClient(conn)
-	return s, conn, tc
+func main() {
+	flag.Parse()
+	if *testName == "" {
+		grpclog.Fatalf("test_name not set")
+	}
+	req := &testpb.SimpleRequest{
+		ResponseType: testpb.PayloadType_COMPRESSABLE,
+		ResponseSize: int32(*rspSize),
+		Payload: &testpb.Payload{
+			Type: testpb.PayloadType_COMPRESSABLE,
+			Body: make([]byte, *rqSize),
+		},
+	}
+	connectCtx, connectCancel := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second))
+	defer connectCancel()
+	ccs := buildConnections(connectCtx)
+	warmDeadline := time.Now().Add(time.Duration(*w) * time.Second)
+	endDeadline := warmDeadline.Add(time.Duration(*d) * time.Second)
+	cf, err := os.Create("/tmp/" + *testName + ".cpu")
+	if err != nil {
+		grpclog.Fatalf("Error creating file: %v", err)
+	}
+	defer cf.Close()
+	pprof.StartCPUProfile(cf)
+	cpuBeg := getCPUTime()
+	for _, cc := range ccs {
+		runWithConn(cc, req, warmDeadline, endDeadline)
+	}
+	wg.Wait()
+	cpu := time.Duration(getCPUTime() - cpuBeg)
+	pprof.StopCPUProfile()
+	mf, err := os.Create("/tmp/" + *testName + ".mem")
+	if err != nil {
+		grpclog.Fatalf("Error creating file: %v", err)
+	}
+	defer mf.Close()
+	runtime.GC() // materialize all statistics
+	if err := pprof.WriteHeapProfile(mf); err != nil {
+		grpclog.Fatalf("Error writing memory profile: %v", err)
+	}
+	hist := stats.NewHistogram(hopts)
+	for _, h := range hists {
+		hist.Merge(h)
+	}
+	parseHist(hist)
+	fmt.Println("Client CPU utilization:", cpu)
+	fmt.Println("Client CPU profile:", cf.Name())
+	fmt.Println("Client Mem Profile:", mf.Name())
 }
 
-func closeLoopUnary() {
-	s, conn, tc := buildConnection()
-
-	for i := 0; i < 100; i++ {
-		unaryCaller(tc)
+func buildConnections(ctx context.Context) []*grpc.ClientConn {
+	ccs := make([]*grpc.ClientConn, *c)
+	for i := range ccs {
+		ccs[i] = benchmark.NewClientConnWithContext(ctx, "localhost:"+*port, grpc.WithInsecure(), grpc.WithBlock())
 	}
-	ch := make(chan int, *maxConcurrentRPCs*4)
-	var (
-		mu sync.Mutex
-		wg sync.WaitGroup
-	)
-	wg.Add(*maxConcurrentRPCs)
+	return ccs
+}
 
-	for i := 0; i < *maxConcurrentRPCs; i++ {
+func runWithConn(cc *grpc.ClientConn, req *testpb.SimpleRequest, warmDeadline, endDeadline time.Time) {
+	for i := 0; i < *r; i++ {
+		wg.Add(1)
 		go func() {
-			for range ch {
+			defer wg.Done()
+			caller := makeCaller(cc, req)
+			hist := stats.NewHistogram(hopts)
+			for {
 				start := time.Now()
-				unaryCaller(tc)
-				elapse := time.Since(start)
-				mu.Lock()
-				s.Add(elapse)
-				mu.Unlock()
+				if start.After(endDeadline) {
+					mu.Lock()
+					hists = append(hists, hist)
+					mu.Unlock()
+					return
+				}
+				caller()
+				elapsed := time.Since(start)
+				if start.After(warmDeadline) {
+					hist.Add(elapsed.Nanoseconds())
+				}
 			}
-			wg.Done()
 		}()
 	}
-	// Stop the client when time is up.
-	done := make(chan struct{})
-	go func() {
-		<-time.After(time.Duration(*duration) * time.Second)
-		close(done)
-	}()
-	ok := true
-	for ok {
-		select {
-		case ch <- 0:
-		case <-done:
-			ok = false
-		}
-	}
-	close(ch)
-	wg.Wait()
-	conn.Close()
-	grpclog.Println(s.String())
-
 }
 
-func closeLoopStream() {
-	s, conn, tc := buildConnection()
-	ch := make(chan int, *maxConcurrentRPCs*4)
-	var (
-		mu sync.Mutex
-		wg sync.WaitGroup
-	)
-	wg.Add(*maxConcurrentRPCs)
-	// Distribute RPCs over maxConcurrentCalls workers.
-	for i := 0; i < *maxConcurrentRPCs; i++ {
-		go func() {
-			stream, err := tc.StreamingCall(context.Background())
-			if err != nil {
-				grpclog.Fatalf("%v.StreamingCall(_) = _, %v", tc, err)
-			}
-			// Do some warm up.
-			for i := 0; i < 100; i++ {
-				streamCaller(stream)
-			}
-			for range ch {
-				start := time.Now()
-				streamCaller(stream)
-				elapse := time.Since(start)
-				mu.Lock()
-				s.Add(elapse)
-				mu.Unlock()
+func makeCaller(cc *grpc.ClientConn, req *testpb.SimpleRequest) func() {
+	client := testpb.NewBenchmarkServiceClient(cc)
+	if *rpcType == "unary" {
+		return func() {
+			if _, err := client.UnaryCall(context.Background(), req); err != nil {
+				grpclog.Fatalf("RPC failed: %v", err)
 			}
-			wg.Done()
-		}()
+		}
+	}
+	stream, err := client.StreamingCall(context.Background())
+	if err != nil {
+		grpclog.Fatalf("RPC failed: %v", err)
 	}
-	// Stop the client when time is up.
-	done := make(chan struct{})
-	go func() {
-		<-time.After(time.Duration(*duration) * time.Second)
-		close(done)
-	}()
-	ok := true
-	for ok {
-		select {
-		case ch <- 0:
-		case <-done:
-			ok = false
+	return func() {
+		if err := stream.Send(req); err != nil {
+			grpclog.Fatalf("Streaming RPC failed to send: %v", err)
+		}
+		if _, err := stream.Recv(); err != nil {
+			grpclog.Fatalf("Streaming RPC failed to read: %v", err)
 		}
 	}
-	close(ch)
-	wg.Wait()
-	conn.Close()
-	grpclog.Println(s.String())
+
 }
 
-func main() {
-	flag.Parse()
-	grpc.EnableTracing = *trace
-	go func() {
-		lis, err := net.Listen("tcp", ":0")
-		if err != nil {
-			grpclog.Fatalf("Failed to listen: %v", err)
-		}
-		grpclog.Println("Client profiling address: ", lis.Addr().String())
-		if err := http.Serve(lis, nil); err != nil {
-			grpclog.Fatalf("Failed to serve: %v", err)
+func parseHist(hist *stats.Histogram) {
+	fmt.Println("qps:", float64(hist.Count)/float64(*d))
+	fmt.Printf("Latency: (50/90/99 %%ile): %v/%v/%v\n",
+		time.Duration(median(.5, hist)),
+		time.Duration(median(.9, hist)),
+		time.Duration(median(.99, hist)))
+}
+
+func median(percentile float64, h *stats.Histogram) int64 {
+	need := int64(float64(h.Count) * percentile)
+	have := int64(0)
+	for _, bucket := range h.Buckets {
+		count := bucket.Count
+		if have+count >= need {
+			percent := float64(need-have) / float64(count)
+			return int64((1.0-percent)*bucket.LowBound + percent*bucket.LowBound*(1.0+hopts.GrowthFactor))
 		}
-	}()
-	switch *rpcType {
-	case 0:
-		closeLoopUnary()
-	case 1:
-		closeLoopStream()
+		have += bucket.Count
+	}
+	panic("should have found a bound")
+}
+
+func getCPUTime() int64 {
+	var ts unix.Timespec
+	if err := unix.ClockGettime(unix.CLOCK_PROCESS_CPUTIME_ID, &ts); err != nil {
+		grpclog.Fatal(err)
 	}
+	return ts.Nano()
 }