vendor/google.golang.org/grpc/benchmark/worker/benchmark_server.go

   1 /*
   2  *
   3  * Copyright 2016 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 package main
  20
  21 import (
  22         "flag"
  23         "runtime"
  24         "strconv"
  25         "strings"
  26         "sync"
  27         "syscall"
  28         "time"
  29
  30         "google.golang.org/grpc"
  31         "google.golang.org/grpc/benchmark"
  32         testpb "google.golang.org/grpc/benchmark/grpc_testing"
  33         "google.golang.org/grpc/codes"
  34         "google.golang.org/grpc/credentials"
  35         "google.golang.org/grpc/grpclog"
  36         "google.golang.org/grpc/testdata"
  37 )
  38
  39 var (
  40         certFile = flag.String("tls_cert_file", "", "The TLS cert file")
  41         keyFile  = flag.String("tls_key_file", "", "The TLS key file")
  42 )
  43
  44 type benchmarkServer struct {
  45         port            int
  46         cores           int
  47         closeFunc       func()
  48         mu              sync.RWMutex
  49         lastResetTime   time.Time
  50         rusageLastReset *syscall.Rusage
  51 }
  52
  53 func printServerConfig(config *testpb.ServerConfig) {
  54         // Some config options are ignored:
  55         // - server type:
  56         //     will always start sync server
  57         // - async server threads
  58         // - core list
  59         grpclog.Printf(" * server type: %v (ignored, always starts sync server)", config.ServerType)
  60         grpclog.Printf(" * async server threads: %v (ignored)", config.AsyncServerThreads)
  61         // TODO: use cores specified by CoreList when setting list of cores is supported in go.
  62         grpclog.Printf(" * core list: %v (ignored)", config.CoreList)
  63
  64         grpclog.Printf(" - security params: %v", config.SecurityParams)
  65         grpclog.Printf(" - core limit: %v", config.CoreLimit)
  66         grpclog.Printf(" - port: %v", config.Port)
  67         grpclog.Printf(" - payload config: %v", config.PayloadConfig)
  68 }
  69
  70 func startBenchmarkServer(config *testpb.ServerConfig, serverPort int) (*benchmarkServer, error) {
  71         printServerConfig(config)
  72
  73         // Use all cpu cores available on machine by default.
  74         // TODO: Revisit this for the optimal default setup.
  75         numOfCores := runtime.NumCPU()
  76         if config.CoreLimit > 0 {
  77                 numOfCores = int(config.CoreLimit)
  78         }
  79         runtime.GOMAXPROCS(numOfCores)
  80
  81         var opts []grpc.ServerOption
  82
  83         // Sanity check for server type.
  84         switch config.ServerType {
  85         case testpb.ServerType_SYNC_SERVER:
  86         case testpb.ServerType_ASYNC_SERVER:
  87         case testpb.ServerType_ASYNC_GENERIC_SERVER:
  88         default:
  89                 return nil, grpc.Errorf(codes.InvalidArgument, "unknow server type: %v", config.ServerType)
  90         }
  91
  92         // Set security options.
  93         if config.SecurityParams != nil {
  94                 if *certFile == "" {
  95                         *certFile = testdata.Path("server1.pem")
  96                 }
  97                 if *keyFile == "" {
  98                         *keyFile = testdata.Path("server1.key")
  99                 }
 100                 creds, err := credentials.NewServerTLSFromFile(*certFile, *keyFile)
 101                 if err != nil {
 102                         grpclog.Fatalf("failed to generate credentials %v", err)
 103                 }
 104                 opts = append(opts, grpc.Creds(creds))
 105         }
 106
 107         // Priority: config.Port > serverPort > default (0).
 108         port := int(config.Port)
 109         if port == 0 {
 110                 port = serverPort
 111         }
 112
 113         // Create different benchmark server according to config.
 114         var (
 115                 addr      string
 116                 closeFunc func()
 117                 err       error
 118         )
 119         if config.PayloadConfig != nil {
 120                 switch payload := config.PayloadConfig.Payload.(type) {
 121                 case *testpb.PayloadConfig_BytebufParams:
 122                         opts = append(opts, grpc.CustomCodec(byteBufCodec{}))
 123                         addr, closeFunc = benchmark.StartServer(benchmark.ServerInfo{
 124                                 Addr:     ":" + strconv.Itoa(port),
 125                                 Type:     "bytebuf",
 126                                 Metadata: payload.BytebufParams.RespSize,
 127                         }, opts...)
 128                 case *testpb.PayloadConfig_SimpleParams:
 129                         addr, closeFunc = benchmark.StartServer(benchmark.ServerInfo{
 130                                 Addr: ":" + strconv.Itoa(port),
 131                                 Type: "protobuf",
 132                         }, opts...)
 133                 case *testpb.PayloadConfig_ComplexParams:
 134                         return nil, grpc.Errorf(codes.Unimplemented, "unsupported payload config: %v", config.PayloadConfig)
 135                 default:
 136                         return nil, grpc.Errorf(codes.InvalidArgument, "unknow payload config: %v", config.PayloadConfig)
 137                 }
 138         } else {
 139                 // Start protobuf server if payload config is nil.
 140                 addr, closeFunc = benchmark.StartServer(benchmark.ServerInfo{
 141                         Addr: ":" + strconv.Itoa(port),
 142                         Type: "protobuf",
 143                 }, opts...)
 144         }
 145
 146         grpclog.Printf("benchmark server listening at %v", addr)
 147         addrSplitted := strings.Split(addr, ":")
 148         p, err := strconv.Atoi(addrSplitted[len(addrSplitted)-1])
 149         if err != nil {
 150                 grpclog.Fatalf("failed to get port number from server address: %v", err)
 151         }
 152
 153         rusage := new(syscall.Rusage)
 154         syscall.Getrusage(syscall.RUSAGE_SELF, rusage)
 155
 156         return &benchmarkServer{
 157                 port:            p,
 158                 cores:           numOfCores,
 159                 closeFunc:       closeFunc,
 160                 lastResetTime:   time.Now(),
 161                 rusageLastReset: rusage,
 162         }, nil
 163 }
 164
 165 // getStats returns the stats for benchmark server.
 166 // It resets lastResetTime if argument reset is true.
 167 func (bs *benchmarkServer) getStats(reset bool) *testpb.ServerStats {
 168         bs.mu.RLock()
 169         defer bs.mu.RUnlock()
 170         wallTimeElapsed := time.Since(bs.lastResetTime).Seconds()
 171         rusageLatest := new(syscall.Rusage)
 172         syscall.Getrusage(syscall.RUSAGE_SELF, rusageLatest)
 173         uTimeElapsed, sTimeElapsed := cpuTimeDiff(bs.rusageLastReset, rusageLatest)
 174
 175         if reset {
 176                 bs.lastResetTime = time.Now()
 177                 bs.rusageLastReset = rusageLatest
 178         }
 179         return &testpb.ServerStats{
 180                 TimeElapsed: wallTimeElapsed,
 181                 TimeUser:    uTimeElapsed,
 182                 TimeSystem:  sTimeElapsed,
 183         }
 184 }