test/cpp/qps/server.h

   1 /*
   2  *
   3  * Copyright 2015 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 #ifndef TEST_QPS_SERVER_H
  20 #define TEST_QPS_SERVER_H
  21
  22 #include <vector>
  23
  24 #include <grpc/support/cpu.h>
  25 #include <grpc/support/log.h>
  26 #include <grpcpp/channel.h>
  27 #include <grpcpp/resource_quota.h>
  28 #include <grpcpp/security/server_credentials.h>
  29 #include <grpcpp/server_builder.h>
  30
  31 #include "src/cpp/util/core_stats.h"
  32 #include "src/proto/grpc/testing/control.pb.h"
  33 #include "src/proto/grpc/testing/messages.pb.h"
  34 #include "test/core/end2end/data/ssl_test_data.h"
  35 #include "test/core/util/port.h"
  36 #include "test/cpp/qps/usage_timer.h"
  37 #include "test/cpp/util/test_credentials_provider.h"
  38
  39 namespace grpc {
  40 namespace testing {
  41
  42 class Server {
  43  public:
  44   explicit Server(const ServerConfig& config)
  45       : timer_(new UsageTimer), last_reset_poll_count_(0) {
  46     cores_ = gpr_cpu_num_cores();
  47     if (config.port()) {  // positive for a fixed port, negative for inproc
  48       port_ = config.port();
  49     } else {  // zero for dynamic port
  50       port_ = grpc_pick_unused_port_or_die();
  51     }
  52   }
  53   virtual ~Server() {}
  54
  55   ServerStats Mark(bool reset) {
  56     UsageTimer::Result timer_result;
  57     int cur_poll_count = GetPollCount();
  58     int poll_count = cur_poll_count - last_reset_poll_count_;
  59     if (reset) {
  60       std::unique_ptr<UsageTimer> timer(new UsageTimer);
  61       timer.swap(timer_);
  62       timer_result = timer->Mark();
  63       last_reset_poll_count_ = cur_poll_count;
  64     } else {
  65       timer_result = timer_->Mark();
  66     }
  67
  68     grpc_stats_data core_stats;
  69     grpc_stats_collect(&core_stats);
  70
  71     ServerStats stats;
  72     stats.set_time_elapsed(timer_result.wall);
  73     stats.set_time_system(timer_result.system);
  74     stats.set_time_user(timer_result.user);
  75     stats.set_total_cpu_time(timer_result.total_cpu_time);
  76     stats.set_idle_cpu_time(timer_result.idle_cpu_time);
  77     stats.set_cq_poll_count(poll_count);
  78     CoreStatsToProto(core_stats, stats.mutable_core_stats());
  79     return stats;
  80   }
  81
  82   static bool SetPayload(PayloadType type, int size, Payload* payload) {
  83     // TODO(yangg): Support UNCOMPRESSABLE payload.
  84     if (type != PayloadType::COMPRESSABLE) {
  85       return false;
  86     }
  87     payload->set_type(type);
  88     // Don't waste time creating a new payload of identical size.
  89     if (payload->body().length() != static_cast<size_t>(size)) {
  90       std::unique_ptr<char[]> body(new char[size]());
  91       payload->set_body(body.get(), size);
  92     }
  93     return true;
  94   }
  95
  96   int port() const { return port_; }
  97   int cores() const { return cores_; }
  98   static std::shared_ptr<ServerCredentials> CreateServerCredentials(
  99       const ServerConfig& config) {
 100     if (config.has_security_params()) {
 101       std::string type;
 102       if (config.security_params().cred_type().empty()) {
 103         type = kTlsCredentialsType;
 104       } else {
 105         type = config.security_params().cred_type();
 106       }
 107
 108       return GetCredentialsProvider()->GetServerCredentials(type);
 109     } else {
 110       return InsecureServerCredentials();
 111     }
 112   }
 113
 114   virtual int GetPollCount() {
 115     // For sync server.
 116     return 0;
 117   }
 118
 119   virtual std::shared_ptr<Channel> InProcessChannel(
 120       const ChannelArguments& args) = 0;
 121
 122  protected:
 123   static void ApplyConfigToBuilder(const ServerConfig& config,
 124                                    ServerBuilder* builder) {
 125     if (config.resource_quota_size() > 0) {
 126       builder->SetResourceQuota(ResourceQuota("AsyncQpsServerTest")
 127                                     .Resize(config.resource_quota_size()));
 128     }
 129     for (const auto& channel_arg : config.channel_args()) {
 130       switch (channel_arg.value_case()) {
 131         case ChannelArg::kStrValue:
 132           builder->AddChannelArgument(channel_arg.name(),
 133                                       channel_arg.str_value());
 134           break;
 135         case ChannelArg::kIntValue:
 136           builder->AddChannelArgument(channel_arg.name(),
 137                                       channel_arg.int_value());
 138           break;
 139         case ChannelArg::VALUE_NOT_SET:
 140           gpr_log(GPR_ERROR, "Channel arg '%s' does not have a value",
 141                   channel_arg.name().c_str());
 142           break;
 143       }
 144     }
 145   }
 146
 147  private:
 148   int port_;
 149   int cores_;
 150   std::unique_ptr<UsageTimer> timer_;
 151   int last_reset_poll_count_;
 152 };
 153
 154 std::unique_ptr<Server> CreateSynchronousServer(const ServerConfig& config);
 155 std::unique_ptr<Server> CreateAsyncServer(const ServerConfig& config);
 156 std::unique_ptr<Server> CreateAsyncGenericServer(const ServerConfig& config);
 157 std::unique_ptr<Server> CreateCallbackServer(const ServerConfig& config);
 158
 159 }  // namespace testing
 160 }  // namespace grpc
 161
 162 #endif