Address review comments

Change Abs to abs
Remove unimplemented distribution
Name change
Get server port from config or cmd line option

benchmark/worker/benchmark_client.go

@@ -80,7 +80,7 @@ func startBenchmarkClient(config *testpb.ClientConfig) (*benchmarkClient, error)
 
 	grpclog.Printf(" - security params: %v", config.SecurityParams)
 	if config.SecurityParams != nil {
-		creds, err := credentials.NewClientTLSFromFile(Abs(caFile), config.SecurityParams.ServerHostOverride)
+		creds, err := credentials.NewClientTLSFromFile(abs(caFile), config.SecurityParams.ServerHostOverride)
 		if err != nil {
 			return nil, grpc.Errorf(codes.InvalidArgument, "failed to create TLS credentials %v", err)
 		}
@@ -90,7 +90,7 @@ func startBenchmarkClient(config *testpb.ClientConfig) (*benchmarkClient, error)
 	}
 
 	grpclog.Printf(" - core limit: %v", config.CoreLimit)
-	// Use one cpu core by default
+	// Use one cpu core by default.
 	// TODO: change default number of cores used if 1 is not fastest.
 	if config.CoreLimit > 1 {
 		runtime.GOMAXPROCS(int(config.CoreLimit))
@@ -125,13 +125,11 @@ func startBenchmarkClient(config *testpb.ClientConfig) (*benchmarkClient, error)
 	rpcCountPerConn, connCount := int(config.OutstandingRpcsPerChannel), int(config.ClientChannels)
 
 	grpclog.Printf(" - load params: %v", config.LoadParams)
-	var dist *int
-	switch lp := config.LoadParams.Load.(type) {
+	// TODO add open loop distribution.
+	switch config.LoadParams.Load.(type) {
 	case *testpb.LoadParams_ClosedLoop:
 	case *testpb.LoadParams_Poisson:
-		grpclog.Printf("   - %v", lp.Poisson)
 		return nil, grpc.Errorf(codes.Unimplemented, "unsupported load params: %v", config.LoadParams)
-		// TODO poisson
 	case *testpb.LoadParams_Uniform:
 		return nil, grpc.Errorf(codes.Unimplemented, "unsupported load params: %v", config.LoadParams)
 	case *testpb.LoadParams_Determ:
@@ -175,21 +173,17 @@ func startBenchmarkClient(config *testpb.ClientConfig) (*benchmarkClient, error)
 
 	switch rpcType {
 	case "unary":
-		if dist == nil {
-			bc.doCloseLoopUnaryBenchmark(conns, rpcCountPerConn, payloadReqSize, payloadRespSize)
-		}
-		// TODO else do open loop
+		bc.doCloseLoopUnary(conns, rpcCountPerConn, payloadReqSize, payloadRespSize)
+		// TODO open loop.
 	case "streaming":
-		if dist == nil {
-			bc.doCloseLoopStreamingBenchmark(conns, rpcCountPerConn, payloadReqSize, payloadRespSize, payloadType)
-		}
-		// TODO else do open loop
+		bc.doCloseLoopStreaming(conns, rpcCountPerConn, payloadReqSize, payloadRespSize, payloadType)
+		// TODO open loop.
 	}
 
 	return &bc, nil
 }
 
-func (bc *benchmarkClient) doCloseLoopUnaryBenchmark(conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int) {
+func (bc *benchmarkClient) doCloseLoopUnary(conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int) {
 	clients := make([]testpb.BenchmarkServiceClient, len(conns))
 	for ic, conn := range conns {
 		clients[ic] = testpb.NewBenchmarkServiceClient(conn)
@@ -235,7 +229,7 @@ func (bc *benchmarkClient) doCloseLoopUnaryBenchmark(conns []*grpc.ClientConn, r
 	}
 }
 
-func (bc *benchmarkClient) doCloseLoopStreamingBenchmark(conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int, payloadType string) {
+func (bc *benchmarkClient) doCloseLoopStreaming(conns []*grpc.ClientConn, rpcCountPerConn int, reqSize int, respSize int, payloadType string) {
 	var doRPC func(testpb.BenchmarkService_StreamingCallClient, int, int) error
 	if payloadType == "bytebuf" {
 		doRPC = benchmark.DoByteBufStreamingRoundTrip
@@ -319,6 +313,8 @@ func (bc *benchmarkClient) getStats() *testpb.ClientStats {
 	}
 }
 
+// reset clears the contents for histogram and set lastResetTime to Now().
+// It is often called to get ready for benchmark runs.
 func (bc *benchmarkClient) reset() {
 	bc.mu.Lock()
 	defer bc.mu.Unlock()

benchmark/worker/benchmark_server.go

@@ -57,7 +57,7 @@ var (
 type benchmarkServer struct {
 	port          int
 	cores         int
-	close         func()
+	closeFunc     func()
 	mu            sync.RWMutex
 	lastResetTime time.Time
 }
@@ -83,7 +83,7 @@ func startBenchmarkServer(config *testpb.ServerConfig, serverPort int) (*benchma
 
 	grpclog.Printf(" - security params: %v", config.SecurityParams)
 	if config.SecurityParams != nil {
-		creds, err := credentials.NewServerTLSFromFile(Abs(certFile), Abs(keyFile))
+		creds, err := credentials.NewServerTLSFromFile(abs(certFile), abs(keyFile))
 		if err != nil {
 			grpclog.Fatalf("failed to generate credentials %v", err)
 		}
@@ -95,29 +95,27 @@ func startBenchmarkServer(config *testpb.ServerConfig, serverPort int) (*benchma
 	numOfCores := 1
 	if config.CoreLimit > 1 {
 		numOfCores = int(config.CoreLimit)
+		runtime.GOMAXPROCS(numOfCores)
 	}
-	runtime.GOMAXPROCS(numOfCores)
 
 	grpclog.Printf(" - port: %v", config.Port)
-	var port int
 	// Priority: config.Port > serverPort > default (0).
-	if config.Port != 0 {
-		port = int(config.Port)
-	} else if serverPort != 0 {
+	port := int(config.Port)
+	if port == 0 {
 		port = serverPort
 	}
 
 	grpclog.Printf(" - payload config: %v", config.PayloadConfig)
 	var (
-		addr  string
-		close func()
-		err   error
+		addr      string
+		closeFunc func()
+		err       error
 	)
 	if config.PayloadConfig != nil {
 		switch payload := config.PayloadConfig.Payload.(type) {
 		case *testpb.PayloadConfig_BytebufParams:
 			opts = append(opts, grpc.CustomCodec(byteBufCodec{}))
-			addr, close, err = benchmark.StartServer(benchmark.ServerInfo{
+			addr, closeFunc, err = benchmark.StartServer(benchmark.ServerInfo{
 				Addr:     ":" + strconv.Itoa(port),
 				Type:     "bytebuf",
 				Metadata: payload.BytebufParams.RespSize,
@@ -126,7 +124,7 @@ func startBenchmarkServer(config *testpb.ServerConfig, serverPort int) (*benchma
 				grpclog.Fatalf("failed to start server: %v", err)
 			}
 		case *testpb.PayloadConfig_SimpleParams:
-			addr, close, err = benchmark.StartServer(benchmark.ServerInfo{
+			addr, closeFunc, err = benchmark.StartServer(benchmark.ServerInfo{
 				Addr: ":" + strconv.Itoa(port),
 				Type: "protobuf",
 			}, opts...)
@@ -140,7 +138,7 @@ func startBenchmarkServer(config *testpb.ServerConfig, serverPort int) (*benchma
 		}
 	} else {
 		// Start protobuf server is payload config is nil.
-		addr, close, err = benchmark.StartServer(benchmark.ServerInfo{
+		addr, closeFunc, err = benchmark.StartServer(benchmark.ServerInfo{
 			Addr: ":" + strconv.Itoa(port),
 			Type: "protobuf",
 		}, opts...)
@@ -156,7 +154,7 @@ func startBenchmarkServer(config *testpb.ServerConfig, serverPort int) (*benchma
 		grpclog.Fatalf("failed to get port number from server address: %v", err)
 	}
 
-	return &benchmarkServer{port: p, cores: numOfCores, close: close, lastResetTime: time.Now()}, nil
+	return &benchmarkServer{port: p, cores: numOfCores, closeFunc: closeFunc, lastResetTime: time.Now()}, nil
 }
 
 func (bs *benchmarkServer) getStats() *testpb.ServerStats {

benchmark/worker/main.go

@@ -59,7 +59,7 @@ type byteBufCodec struct {
 func (byteBufCodec) Marshal(v interface{}) ([]byte, error) {
 	b, ok := v.(*[]byte)
 	if !ok {
-		return nil, fmt.Errorf("Failed to marshal: %v is not type of *[]byte")
+		return nil, fmt.Errorf("failed to marshal: %v is not type of *[]byte")
 	}
 	return *b, nil
 }
@@ -67,16 +67,18 @@ func (byteBufCodec) Marshal(v interface{}) ([]byte, error) {
 func (byteBufCodec) Unmarshal(data []byte, v interface{}) error {
 	b, ok := v.(*[]byte)
 	if !ok {
-		return fmt.Errorf("Failed to marshal: %v is not type of *[]byte")
+		return fmt.Errorf("failed to marshal: %v is not type of *[]byte")
 	}
 	*b = data
 	return nil
 }
 
 func (byteBufCodec) String() string {
-	return "byteBufCodec"
+	return "bytebuffer"
 }
 
+// workerServer implements WorkerService rpc handlers.
+// It can create benchmarkServer or benchmarkClient on demand.
 type workerServer struct {
 	stop       chan<- bool
 	serverPort int
@@ -88,7 +90,7 @@ func (s *workerServer) RunServer(stream testpb.WorkerService_RunServerServer) er
 		// Close benchmark server when stream ends.
 		grpclog.Printf("closing benchmark server")
 		if bs != nil {
-			bs.close()
+			bs.closeFunc()
 		}
 	}()
 	for {
@@ -110,7 +112,7 @@ func (s *workerServer) RunServer(stream testpb.WorkerService_RunServerServer) er
 			}
 			if bs != nil {
 				grpclog.Printf("server setup received when server already exists, closing the existing server")
-				bs.close()
+				bs.closeFunc()
 			}
 			bs = newbs
 			out = &testpb.ServerStatus{

benchmark/worker/util.go

@@ -38,10 +38,10 @@ import (
 	"path/filepath"
 )
 
-// Abs returns the absolute path the given relative file or directory path,
+// abs returns the absolute path the given relative file or directory path,
 // relative to the google.golang.org/grpc directory in the user's GOPATH.
 // If rel is already absolute, it is returned unmodified.
-func Abs(rel string) string {
+func abs(rel string) string {
 	if filepath.IsAbs(rel) {
 		return rel
 	}