test/cpp/client/client_channel_stress_test.cc

   1 /*
   2  *
   3  * Copyright 2017 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 #include <atomic>
  20 #include <memory>
  21 #include <mutex>
  22 #include <random>
  23 #include <sstream>
  24 #include <thread>
  25
  26 #include <grpc/grpc.h>
  27 #include <grpc/support/alloc.h>
  28 #include <grpc/support/log.h>
  29 #include <grpc/support/string_util.h>
  30 #include <grpc/support/time.h>
  31 #include <grpcpp/channel.h>
  32 #include <grpcpp/client_context.h>
  33 #include <grpcpp/create_channel.h>
  34 #include <grpcpp/impl/codegen/sync.h>
  35 #include <grpcpp/server.h>
  36 #include <grpcpp/server_builder.h>
  37
  38 #include "src/core/ext/filters/client_channel/parse_address.h"
  39 #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
  40 #include "src/core/ext/filters/client_channel/server_address.h"
  41 #include "src/core/lib/gprpp/ref_counted_ptr.h"
  42 #include "src/core/lib/gprpp/thd.h"
  43 #include "src/core/lib/iomgr/sockaddr.h"
  44
  45 #include "test/core/util/port.h"
  46 #include "test/core/util/test_config.h"
  47 #include "test/cpp/end2end/test_service_impl.h"
  48
  49 #include "src/proto/grpc/lb/v1/load_balancer.grpc.pb.h"
  50 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  51
  52 using grpc::lb::v1::LoadBalanceRequest;
  53 using grpc::lb::v1::LoadBalanceResponse;
  54 using grpc::lb::v1::LoadBalancer;
  55
  56 namespace grpc {
  57 namespace testing {
  58 namespace {
  59
  60 const size_t kNumBackends = 10;
  61 const size_t kNumBalancers = 5;
  62 const size_t kNumClientThreads = 100;
  63 const int kResolutionUpdateIntervalMs = 50;
  64 const int kServerlistUpdateIntervalMs = 10;
  65 const int kTestDurationSec = 30;
  66
  67 using BackendServiceImpl = TestServiceImpl;
  68
  69 class BalancerServiceImpl : public LoadBalancer::Service {
  70  public:
  71   using Stream = ServerReaderWriter<LoadBalanceResponse, LoadBalanceRequest>;
  72
  73   explicit BalancerServiceImpl(const std::vector<int>& all_backend_ports)
  74       : all_backend_ports_(all_backend_ports) {}
  75
  76   Status BalanceLoad(ServerContext* /*context*/, Stream* stream) override {
  77     gpr_log(GPR_INFO, "LB[%p]: Start BalanceLoad.", this);
  78     LoadBalanceRequest request;
  79     stream->Read(&request);
  80     while (!shutdown_) {
  81       stream->Write(BuildRandomResponseForBackends());
  82       std::this_thread::sleep_for(
  83           std::chrono::milliseconds(kServerlistUpdateIntervalMs));
  84     }
  85     gpr_log(GPR_INFO, "LB[%p]: Finish BalanceLoad.", this);
  86     return Status::OK;
  87   }
  88
  89   void Shutdown() { shutdown_ = true; }
  90
  91  private:
  92   grpc::string Ip4ToPackedString(const char* ip_str) {
  93     struct in_addr ip4;
  94     GPR_ASSERT(inet_pton(AF_INET, ip_str, &ip4) == 1);
  95     return grpc::string(reinterpret_cast<const char*>(&ip4), sizeof(ip4));
  96   }
  97
  98   LoadBalanceResponse BuildRandomResponseForBackends() {
  99     // Generate a random serverlist with varying size (if N =
 100     // all_backend_ports_.size(), num_non_drop_entry is in [0, 2N],
 101     // num_drop_entry is in [0, N]), order, duplicate, and drop rate.
 102     size_t num_non_drop_entry =
 103         std::rand() % (all_backend_ports_.size() * 2 + 1);
 104     size_t num_drop_entry = std::rand() % (all_backend_ports_.size() + 1);
 105     std::vector<int> random_backend_indices;
 106     for (size_t i = 0; i < num_non_drop_entry; ++i) {
 107       random_backend_indices.push_back(std::rand() % all_backend_ports_.size());
 108     }
 109     for (size_t i = 0; i < num_drop_entry; ++i) {
 110       random_backend_indices.push_back(-1);
 111     }
 112     std::shuffle(random_backend_indices.begin(), random_backend_indices.end(),
 113                  std::mt19937(std::random_device()()));
 114     // Build the response according to the random list generated above.
 115     LoadBalanceResponse response;
 116     for (int index : random_backend_indices) {
 117       auto* server = response.mutable_server_list()->add_servers();
 118       if (index < 0) {
 119         server->set_drop(true);
 120         server->set_load_balance_token("load_balancing");
 121       } else {
 122         server->set_ip_address(Ip4ToPackedString("127.0.0.1"));
 123         server->set_port(all_backend_ports_[index]);
 124       }
 125     }
 126     return response;
 127   }
 128
 129   std::atomic_bool shutdown_{false};
 130   const std::vector<int> all_backend_ports_;
 131 };
 132
 133 class ClientChannelStressTest {
 134  public:
 135   void Run() {
 136     Start();
 137     // Keep updating resolution for the test duration.
 138     gpr_log(GPR_INFO, "Start updating resolution.");
 139     const auto wait_duration =
 140         std::chrono::milliseconds(kResolutionUpdateIntervalMs);
 141     std::vector<AddressData> addresses;
 142     auto start_time = std::chrono::steady_clock::now();
 143     while (true) {
 144       if (std::chrono::duration_cast<std::chrono::seconds>(
 145               std::chrono::steady_clock::now() - start_time)
 146               .count() > kTestDurationSec) {
 147         break;
 148       }
 149       // Generate a random subset of balancers.
 150       addresses.clear();
 151       for (const auto& balancer_server : balancer_servers_) {
 152         // Select each address with probability of 0.8.
 153         if (std::rand() % 10 < 8) {
 154           addresses.emplace_back(AddressData{balancer_server.port_, true, ""});
 155         }
 156       }
 157       std::shuffle(addresses.begin(), addresses.end(),
 158                    std::mt19937(std::random_device()()));
 159       SetNextResolution(addresses);
 160       std::this_thread::sleep_for(wait_duration);
 161     }
 162     gpr_log(GPR_INFO, "Finish updating resolution.");
 163     Shutdown();
 164   }
 165
 166  private:
 167   template <typename T>
 168   struct ServerThread {
 169     explicit ServerThread(const grpc::string& type,
 170                           const grpc::string& server_host, T* service)
 171         : type_(type), service_(service) {
 172       grpc::internal::Mutex mu;
 173       // We need to acquire the lock here in order to prevent the notify_one
 174       // by ServerThread::Start from firing before the wait below is hit.
 175       grpc::internal::MutexLock lock(&mu);
 176       port_ = grpc_pick_unused_port_or_die();
 177       gpr_log(GPR_INFO, "starting %s server on port %d", type_.c_str(), port_);
 178       grpc::internal::CondVar cond;
 179       thread_.reset(new std::thread(
 180           std::bind(&ServerThread::Start, this, server_host, &mu, &cond)));
 181       cond.Wait(&mu);
 182       gpr_log(GPR_INFO, "%s server startup complete", type_.c_str());
 183     }
 184
 185     void Start(const grpc::string& server_host, grpc::internal::Mutex* mu,
 186                grpc::internal::CondVar* cond) {
 187       // We need to acquire the lock here in order to prevent the notify_one
 188       // below from firing before its corresponding wait is executed.
 189       grpc::internal::MutexLock lock(mu);
 190       std::ostringstream server_address;
 191       server_address << server_host << ":" << port_;
 192       ServerBuilder builder;
 193       builder.AddListeningPort(server_address.str(),
 194                                InsecureServerCredentials());
 195       builder.RegisterService(service_);
 196       server_ = builder.BuildAndStart();
 197       cond->Signal();
 198     }
 199
 200     void Shutdown() {
 201       gpr_log(GPR_INFO, "%s about to shutdown", type_.c_str());
 202       server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0));
 203       thread_->join();
 204       gpr_log(GPR_INFO, "%s shutdown completed", type_.c_str());
 205     }
 206
 207     int port_;
 208     grpc::string type_;
 209     std::unique_ptr<Server> server_;
 210     T* service_;
 211     std::unique_ptr<std::thread> thread_;
 212   };
 213
 214   struct AddressData {
 215     int port;
 216     bool is_balancer;
 217     grpc::string balancer_name;
 218   };
 219
 220   void SetNextResolution(const std::vector<AddressData>& address_data) {
 221     grpc_core::ExecCtx exec_ctx;
 222     grpc_core::Resolver::Result result;
 223     for (const auto& addr : address_data) {
 224       char* lb_uri_str;
 225       gpr_asprintf(&lb_uri_str, "ipv4:127.0.0.1:%d", addr.port);
 226       grpc_uri* lb_uri = grpc_uri_parse(lb_uri_str, true);
 227       GPR_ASSERT(lb_uri != nullptr);
 228       grpc_resolved_address address;
 229       GPR_ASSERT(grpc_parse_uri(lb_uri, &address));
 230       std::vector<grpc_arg> args_to_add;
 231       if (addr.is_balancer) {
 232         args_to_add.emplace_back(grpc_channel_arg_integer_create(
 233             const_cast<char*>(GRPC_ARG_ADDRESS_IS_BALANCER), 1));
 234         args_to_add.emplace_back(grpc_channel_arg_string_create(
 235             const_cast<char*>(GRPC_ARG_ADDRESS_BALANCER_NAME),
 236             const_cast<char*>(addr.balancer_name.c_str())));
 237       }
 238       grpc_channel_args* args = grpc_channel_args_copy_and_add(
 239           nullptr, args_to_add.data(), args_to_add.size());
 240       result.addresses.emplace_back(address.addr, address.len, args);
 241       grpc_uri_destroy(lb_uri);
 242       gpr_free(lb_uri_str);
 243     }
 244     response_generator_->SetResponse(std::move(result));
 245   }
 246
 247   void KeepSendingRequests() {
 248     gpr_log(GPR_INFO, "Start sending requests.");
 249     while (!shutdown_) {
 250       ClientContext context;
 251       context.set_deadline(grpc_timeout_milliseconds_to_deadline(1000));
 252       EchoRequest request;
 253       request.set_message("test");
 254       EchoResponse response;
 255       {
 256         std::lock_guard<std::mutex> lock(stub_mutex_);
 257         Status status = stub_->Echo(&context, request, &response);
 258       }
 259     }
 260     gpr_log(GPR_INFO, "Finish sending requests.");
 261   }
 262
 263   void CreateStub() {
 264     ChannelArguments args;
 265     response_generator_ =
 266         grpc_core::MakeRefCounted<grpc_core::FakeResolverResponseGenerator>();
 267     args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
 268                     response_generator_.get());
 269     std::ostringstream uri;
 270     uri << "fake:///servername_not_used";
 271     channel_ = ::grpc::CreateCustomChannel(uri.str(),
 272                                            InsecureChannelCredentials(), args);
 273     stub_ = grpc::testing::EchoTestService::NewStub(channel_);
 274   }
 275
 276   void Start() {
 277     // Start the backends.
 278     std::vector<int> backend_ports;
 279     for (size_t i = 0; i < kNumBackends; ++i) {
 280       backends_.emplace_back(new BackendServiceImpl());
 281       backend_servers_.emplace_back(ServerThread<BackendServiceImpl>(
 282           "backend", server_host_, backends_.back().get()));
 283       backend_ports.push_back(backend_servers_.back().port_);
 284     }
 285     // Start the load balancers.
 286     for (size_t i = 0; i < kNumBalancers; ++i) {
 287       balancers_.emplace_back(new BalancerServiceImpl(backend_ports));
 288       balancer_servers_.emplace_back(ServerThread<BalancerServiceImpl>(
 289           "balancer", server_host_, balancers_.back().get()));
 290     }
 291     // Start sending RPCs in multiple threads.
 292     CreateStub();
 293     for (size_t i = 0; i < kNumClientThreads; ++i) {
 294       client_threads_.emplace_back(
 295           std::thread(&ClientChannelStressTest::KeepSendingRequests, this));
 296     }
 297   }
 298
 299   void Shutdown() {
 300     shutdown_ = true;
 301     for (size_t i = 0; i < client_threads_.size(); ++i) {
 302       client_threads_[i].join();
 303     }
 304     for (size_t i = 0; i < balancers_.size(); ++i) {
 305       balancers_[i]->Shutdown();
 306       balancer_servers_[i].Shutdown();
 307     }
 308     for (size_t i = 0; i < backends_.size(); ++i) {
 309       backend_servers_[i].Shutdown();
 310     }
 311   }
 312
 313   std::atomic_bool shutdown_{false};
 314   const grpc::string server_host_ = "localhost";
 315   std::shared_ptr<Channel> channel_;
 316   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
 317   std::mutex stub_mutex_;
 318   std::vector<std::unique_ptr<BackendServiceImpl>> backends_;
 319   std::vector<std::unique_ptr<BalancerServiceImpl>> balancers_;
 320   std::vector<ServerThread<BackendServiceImpl>> backend_servers_;
 321   std::vector<ServerThread<BalancerServiceImpl>> balancer_servers_;
 322   grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
 323       response_generator_;
 324   std::vector<std::thread> client_threads_;
 325 };
 326
 327 }  // namespace
 328 }  // namespace testing
 329 }  // namespace grpc
 330
 331 int main(int argc, char** argv) {
 332   grpc::testing::TestEnvironment env(argc, argv);
 333   grpc::testing::ClientChannelStressTest test;
 334   grpc_init();
 335   test.Run();
 336   grpc_shutdown();
 337   return 0;
 338 }