test/cpp/interop/reconnect_interop_server.cc

   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 // Test description at doc/connection-backoff-interop-test-description.md
  20
  21 #include <signal.h>
  22
  23 #include <condition_variable>
  24 #include <memory>
  25 #include <mutex>
  26 #include <sstream>
  27
  28 #include "absl/flags/flag.h"
  29
  30 #include <grpc/grpc.h>
  31 #include <grpc/support/log.h>
  32 #include <grpcpp/server.h>
  33 #include <grpcpp/server_builder.h>
  34 #include <grpcpp/server_context.h>
  35
  36 #include "src/proto/grpc/testing/empty.pb.h"
  37 #include "src/proto/grpc/testing/messages.pb.h"
  38 #include "src/proto/grpc/testing/test.grpc.pb.h"
  39 #include "test/core/util/reconnect_server.h"
  40 #include "test/cpp/util/test_config.h"
  41
  42 ABSL_FLAG(int32_t, control_port, 0, "Server port for controlling the server.");
  43 ABSL_FLAG(int32_t, retry_port, 0,
  44           "Server port for raw tcp connections. All incoming "
  45           "connections will be closed immediately.");
  46
  47 using grpc::Server;
  48 using grpc::ServerBuilder;
  49 using grpc::ServerContext;
  50 using grpc::Status;
  51 using grpc::testing::Empty;
  52 using grpc::testing::ReconnectInfo;
  53 using grpc::testing::ReconnectParams;
  54 using grpc::testing::ReconnectService;
  55
  56 static bool got_sigint = false;
  57
  58 class ReconnectServiceImpl : public ReconnectService::Service {
  59  public:
  60   explicit ReconnectServiceImpl(int retry_port)
  61       : retry_port_(retry_port),
  62         serving_(false),
  63         server_started_(false),
  64         shutdown_(false) {
  65     reconnect_server_init(&tcp_server_);
  66   }
  67
  68   ~ReconnectServiceImpl() override {
  69     if (server_started_) {
  70       reconnect_server_destroy(&tcp_server_);
  71     }
  72   }
  73
  74   void Poll(int seconds) { reconnect_server_poll(&tcp_server_, seconds); }
  75
  76   Status Start(ServerContext* /*context*/, const ReconnectParams* request,
  77                Empty* /*response*/) override {
  78     bool start_server = true;
  79     std::unique_lock<std::mutex> lock(mu_);
  80     while (serving_ && !shutdown_) {
  81       cv_.wait(lock);
  82     }
  83     if (shutdown_) {
  84       return Status(grpc::StatusCode::UNAVAILABLE, "shutting down");
  85     }
  86     serving_ = true;
  87     if (server_started_) {
  88       start_server = false;
  89     } else {
  90       tcp_server_.max_reconnect_backoff_ms =
  91           request->max_reconnect_backoff_ms();
  92       server_started_ = true;
  93     }
  94     lock.unlock();
  95
  96     if (start_server) {
  97       reconnect_server_start(&tcp_server_, retry_port_);
  98     } else {
  99       reconnect_server_clear_timestamps(&tcp_server_);
 100     }
 101     return Status::OK;
 102   }
 103
 104   Status Stop(ServerContext* /*context*/, const Empty* /*request*/,
 105               ReconnectInfo* response) override {
 106     // extract timestamps and set response
 107     Verify(response);
 108     reconnect_server_clear_timestamps(&tcp_server_);
 109     std::lock_guard<std::mutex> lock(mu_);
 110     serving_ = false;
 111     cv_.notify_one();
 112     return Status::OK;
 113   }
 114
 115   void Verify(ReconnectInfo* response) {
 116     double expected_backoff = 1000.0;
 117     const double kTransmissionDelay = 100.0;
 118     const double kBackoffMultiplier = 1.6;
 119     const double kJitterFactor = 0.2;
 120     const int kMaxBackoffMs = tcp_server_.max_reconnect_backoff_ms
 121                                   ? tcp_server_.max_reconnect_backoff_ms
 122                                   : 120 * 1000;
 123     bool passed = true;
 124     for (timestamp_list* cur = tcp_server_.head; cur && cur->next;
 125          cur = cur->next) {
 126       double backoff = gpr_time_to_millis(
 127           gpr_time_sub(cur->next->timestamp, cur->timestamp));
 128       double min_backoff = expected_backoff * (1 - kJitterFactor);
 129       double max_backoff = expected_backoff * (1 + kJitterFactor);
 130       if (backoff < min_backoff - kTransmissionDelay ||
 131           backoff > max_backoff + kTransmissionDelay) {
 132         passed = false;
 133       }
 134       response->add_backoff_ms(static_cast<int32_t>(backoff));
 135       expected_backoff *= kBackoffMultiplier;
 136       expected_backoff =
 137           expected_backoff > kMaxBackoffMs ? kMaxBackoffMs : expected_backoff;
 138     }
 139     response->set_passed(passed);
 140   }
 141
 142   void Shutdown() {
 143     std::lock_guard<std::mutex> lock(mu_);
 144     shutdown_ = true;
 145     cv_.notify_all();
 146   }
 147
 148  private:
 149   int retry_port_;
 150   reconnect_server tcp_server_;
 151   bool serving_;
 152   bool server_started_;
 153   bool shutdown_;
 154   std::mutex mu_;
 155   std::condition_variable cv_;
 156 };
 157
 158 void RunServer() {
 159   std::ostringstream server_address;
 160   server_address << "0.0.0.0:" << absl::GetFlag(FLAGS_control_port);
 161   ReconnectServiceImpl service(absl::GetFlag(FLAGS_retry_port));
 162
 163   ServerBuilder builder;
 164   builder.RegisterService(&service);
 165   builder.AddListeningPort(server_address.str(),
 166                            grpc::InsecureServerCredentials());
 167   std::unique_ptr<Server> server(builder.BuildAndStart());
 168   gpr_log(GPR_INFO, "Server listening on %s", server_address.str().c_str());
 169   while (!got_sigint) {
 170     service.Poll(5);
 171   }
 172   service.Shutdown();
 173 }
 174
 175 static void sigint_handler(int /*x*/) { got_sigint = true; }
 176
 177 int main(int argc, char** argv) {
 178   grpc::testing::InitTest(&argc, &argv, true);
 179   signal(SIGINT, sigint_handler);
 180
 181   GPR_ASSERT(absl::GetFlag(FLAGS_control_port) != 0);
 182   GPR_ASSERT(absl::GetFlag(FLAGS_retry_port) != 0);
 183   RunServer();
 184
 185   return 0;
 186 }