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