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