3 * Copyright 2016 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
26 #include <grpc/grpc.h>
27 #include <grpc/support/alloc.h>
28 #include <grpc/support/atm.h>
29 #include <grpc/support/log.h>
30 #include <grpc/support/string_util.h>
31 #include <grpc/support/time.h>
32 #include <grpcpp/channel.h>
33 #include <grpcpp/client_context.h>
34 #include <grpcpp/create_channel.h>
35 #include <grpcpp/health_check_service_interface.h>
36 #include <grpcpp/impl/codegen/sync.h>
37 #include <grpcpp/server.h>
38 #include <grpcpp/server_builder.h>
40 #include "src/core/ext/filters/client_channel/backup_poller.h"
41 #include "src/core/ext/filters/client_channel/global_subchannel_pool.h"
42 #include "src/core/ext/filters/client_channel/parse_address.h"
43 #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
44 #include "src/core/ext/filters/client_channel/server_address.h"
45 #include "src/core/lib/backoff/backoff.h"
46 #include "src/core/lib/channel/channel_args.h"
47 #include "src/core/lib/gprpp/debug_location.h"
48 #include "src/core/lib/gprpp/ref_counted_ptr.h"
49 #include "src/core/lib/iomgr/tcp_client.h"
50 #include "src/core/lib/security/credentials/fake/fake_credentials.h"
51 #include "src/cpp/client/secure_credentials.h"
52 #include "src/cpp/server/secure_server_credentials.h"
54 #include "src/proto/grpc/testing/echo.grpc.pb.h"
55 #include "test/core/util/port.h"
56 #include "test/core/util/test_config.h"
57 #include "test/core/util/test_lb_policies.h"
58 #include "test/cpp/end2end/test_service_impl.h"
60 #include <gmock/gmock.h>
61 #include <gtest/gtest.h>
63 using grpc::testing::EchoRequest;
64 using grpc::testing::EchoResponse;
65 using std::chrono::system_clock;
67 // defined in tcp_client.cc
68 extern grpc_tcp_client_vtable* grpc_tcp_client_impl;
70 static grpc_tcp_client_vtable* default_client_impl;
76 gpr_atm g_connection_delay_ms;
78 void tcp_client_connect_with_delay(grpc_closure* closure, grpc_endpoint** ep,
79 grpc_pollset_set* interested_parties,
80 const grpc_channel_args* channel_args,
81 const grpc_resolved_address* addr,
82 grpc_millis deadline) {
83 const int delay_ms = gpr_atm_acq_load(&g_connection_delay_ms);
85 gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(delay_ms));
87 default_client_impl->connect(closure, ep, interested_parties, channel_args,
88 addr, deadline + delay_ms);
91 grpc_tcp_client_vtable delayed_connect = {tcp_client_connect_with_delay};
93 // Subclass of TestServiceImpl that increments a request counter for
94 // every call to the Echo RPC.
95 class MyTestServiceImpl : public TestServiceImpl {
97 MyTestServiceImpl() : request_count_(0) {}
99 Status Echo(ServerContext* context, const EchoRequest* request,
100 EchoResponse* response) override {
102 grpc::internal::MutexLock lock(&mu_);
105 AddClient(context->peer());
106 return TestServiceImpl::Echo(context, request, response);
109 int request_count() {
110 grpc::internal::MutexLock lock(&mu_);
111 return request_count_;
114 void ResetCounters() {
115 grpc::internal::MutexLock lock(&mu_);
119 std::set<grpc::string> clients() {
120 grpc::internal::MutexLock lock(&clients_mu_);
125 void AddClient(const grpc::string& client) {
126 grpc::internal::MutexLock lock(&clients_mu_);
127 clients_.insert(client);
130 grpc::internal::Mutex mu_;
132 grpc::internal::Mutex clients_mu_;
133 std::set<grpc::string> clients_;
136 class ClientLbEnd2endTest : public ::testing::Test {
138 ClientLbEnd2endTest()
139 : server_host_("localhost"),
140 kRequestMessage_("Live long and prosper."),
141 creds_(new SecureChannelCredentials(
142 grpc_fake_transport_security_credentials_create())) {}
144 void SetUp() override {
146 response_generator_ =
147 grpc_core::MakeRefCounted<grpc_core::FakeResolverResponseGenerator>();
150 void TearDown() override {
151 for (size_t i = 0; i < servers_.size(); ++i) {
152 servers_[i]->Shutdown();
154 // Explicitly destroy all the members so that we can make sure grpc_shutdown
155 // has finished by the end of this function, and thus all the registered
156 // LB policy factories are removed.
160 grpc_shutdown_blocking();
163 void CreateServers(size_t num_servers,
164 std::vector<int> ports = std::vector<int>()) {
166 for (size_t i = 0; i < num_servers; ++i) {
168 if (ports.size() == num_servers) port = ports[i];
169 servers_.emplace_back(new ServerData(port));
173 void StartServer(size_t index) { servers_[index]->Start(server_host_); }
175 void StartServers(size_t num_servers,
176 std::vector<int> ports = std::vector<int>()) {
177 CreateServers(num_servers, std::move(ports));
178 for (size_t i = 0; i < num_servers; ++i) {
183 grpc_core::Resolver::Result BuildFakeResults(const std::vector<int>& ports) {
184 grpc_core::Resolver::Result result;
185 for (const int& port : ports) {
187 gpr_asprintf(&lb_uri_str, "ipv4:127.0.0.1:%d", port);
188 grpc_uri* lb_uri = grpc_uri_parse(lb_uri_str, true);
189 GPR_ASSERT(lb_uri != nullptr);
190 grpc_resolved_address address;
191 GPR_ASSERT(grpc_parse_uri(lb_uri, &address));
192 result.addresses.emplace_back(address.addr, address.len,
194 grpc_uri_destroy(lb_uri);
195 gpr_free(lb_uri_str);
200 void SetNextResolution(const std::vector<int>& ports) {
201 grpc_core::ExecCtx exec_ctx;
202 response_generator_->SetResponse(BuildFakeResults(ports));
205 void SetNextResolutionUponError(const std::vector<int>& ports) {
206 grpc_core::ExecCtx exec_ctx;
207 response_generator_->SetReresolutionResponse(BuildFakeResults(ports));
210 void SetFailureOnReresolution() {
211 grpc_core::ExecCtx exec_ctx;
212 response_generator_->SetFailureOnReresolution();
215 std::vector<int> GetServersPorts(size_t start_index = 0) {
216 std::vector<int> ports;
217 for (size_t i = start_index; i < servers_.size(); ++i) {
218 ports.push_back(servers_[i]->port_);
223 std::unique_ptr<grpc::testing::EchoTestService::Stub> BuildStub(
224 const std::shared_ptr<Channel>& channel) {
225 return grpc::testing::EchoTestService::NewStub(channel);
228 std::shared_ptr<Channel> BuildChannel(
229 const grpc::string& lb_policy_name,
230 ChannelArguments args = ChannelArguments()) {
231 if (lb_policy_name.size() > 0) {
232 args.SetLoadBalancingPolicyName(lb_policy_name);
233 } // else, default to pick first
234 args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
235 response_generator_.get());
236 return ::grpc::CreateCustomChannel("fake:///", creds_, args);
240 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
241 EchoResponse* response = nullptr, int timeout_ms = 1000,
242 Status* result = nullptr, bool wait_for_ready = false) {
243 const bool local_response = (response == nullptr);
244 if (local_response) response = new EchoResponse;
246 request.set_message(kRequestMessage_);
247 ClientContext context;
248 context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms));
249 if (wait_for_ready) context.set_wait_for_ready(true);
250 Status status = stub->Echo(&context, request, response);
251 if (result != nullptr) *result = status;
252 if (local_response) delete response;
257 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
258 const grpc_core::DebugLocation& location, bool wait_for_ready = false) {
259 EchoResponse response;
262 SendRpc(stub, &response, 2000, &status, wait_for_ready);
263 ASSERT_TRUE(success) << "From " << location.file() << ":" << location.line()
265 << "Error: " << status.error_message() << " "
266 << status.error_details();
267 ASSERT_EQ(response.message(), kRequestMessage_)
268 << "From " << location.file() << ":" << location.line();
269 if (!success) abort();
272 void CheckRpcSendFailure(
273 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub) {
274 const bool success = SendRpc(stub);
275 EXPECT_FALSE(success);
280 std::unique_ptr<Server> server_;
281 MyTestServiceImpl service_;
282 std::unique_ptr<std::thread> thread_;
283 bool server_ready_ = false;
284 bool started_ = false;
286 explicit ServerData(int port = 0) {
287 port_ = port > 0 ? port : grpc_pick_unused_port_or_die();
290 void Start(const grpc::string& server_host) {
291 gpr_log(GPR_INFO, "starting server on port %d", port_);
293 grpc::internal::Mutex mu;
294 grpc::internal::MutexLock lock(&mu);
295 grpc::internal::CondVar cond;
296 thread_.reset(new std::thread(
297 std::bind(&ServerData::Serve, this, server_host, &mu, &cond)));
298 cond.WaitUntil(&mu, [this] { return server_ready_; });
299 server_ready_ = false;
300 gpr_log(GPR_INFO, "server startup complete");
303 void Serve(const grpc::string& server_host, grpc::internal::Mutex* mu,
304 grpc::internal::CondVar* cond) {
305 std::ostringstream server_address;
306 server_address << server_host << ":" << port_;
307 ServerBuilder builder;
308 std::shared_ptr<ServerCredentials> creds(new SecureServerCredentials(
309 grpc_fake_transport_security_server_credentials_create()));
310 builder.AddListeningPort(server_address.str(), std::move(creds));
311 builder.RegisterService(&service_);
312 server_ = builder.BuildAndStart();
313 grpc::internal::MutexLock lock(mu);
314 server_ready_ = true;
319 if (!started_) return;
320 server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0));
325 void SetServingStatus(const grpc::string& service, bool serving) {
326 server_->GetHealthCheckService()->SetServingStatus(service, serving);
330 void ResetCounters() {
331 for (const auto& server : servers_) server->service_.ResetCounters();
335 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
336 size_t server_idx, const grpc_core::DebugLocation& location,
337 bool ignore_failure = false) {
339 if (ignore_failure) {
342 CheckRpcSendOk(stub, location, true);
344 } while (servers_[server_idx]->service_.request_count() == 0);
348 bool WaitForChannelNotReady(Channel* channel, int timeout_seconds = 5) {
349 const gpr_timespec deadline =
350 grpc_timeout_seconds_to_deadline(timeout_seconds);
351 grpc_connectivity_state state;
352 while ((state = channel->GetState(false /* try_to_connect */)) ==
353 GRPC_CHANNEL_READY) {
354 if (!channel->WaitForStateChange(state, deadline)) return false;
359 bool WaitForChannelReady(Channel* channel, int timeout_seconds = 5) {
360 const gpr_timespec deadline =
361 grpc_timeout_seconds_to_deadline(timeout_seconds);
362 grpc_connectivity_state state;
363 while ((state = channel->GetState(true /* try_to_connect */)) !=
364 GRPC_CHANNEL_READY) {
365 if (!channel->WaitForStateChange(state, deadline)) return false;
370 bool SeenAllServers() {
371 for (const auto& server : servers_) {
372 if (server->service_.request_count() == 0) return false;
377 // Updates \a connection_order by appending to it the index of the newly
378 // connected server. Must be called after every single RPC.
379 void UpdateConnectionOrder(
380 const std::vector<std::unique_ptr<ServerData>>& servers,
381 std::vector<int>* connection_order) {
382 for (size_t i = 0; i < servers.size(); ++i) {
383 if (servers[i]->service_.request_count() == 1) {
384 // Was the server index known? If not, update connection_order.
386 std::find(connection_order->begin(), connection_order->end(), i);
387 if (it == connection_order->end()) {
388 connection_order->push_back(i);
395 const grpc::string server_host_;
396 std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
397 std::vector<std::unique_ptr<ServerData>> servers_;
398 grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
400 const grpc::string kRequestMessage_;
401 std::shared_ptr<ChannelCredentials> creds_;
404 TEST_F(ClientLbEnd2endTest, ChannelStateConnectingWhenResolving) {
405 const int kNumServers = 3;
406 StartServers(kNumServers);
407 auto channel = BuildChannel("");
408 auto stub = BuildStub(channel);
409 // Initial state should be IDLE.
410 EXPECT_EQ(channel->GetState(false /* try_to_connect */), GRPC_CHANNEL_IDLE);
411 // Tell the channel to try to connect.
412 // Note that this call also returns IDLE, since the state change has
413 // not yet occurred; it just gets triggered by this call.
414 EXPECT_EQ(channel->GetState(true /* try_to_connect */), GRPC_CHANNEL_IDLE);
415 // Now that the channel is trying to connect, we should be in state
417 EXPECT_EQ(channel->GetState(false /* try_to_connect */),
418 GRPC_CHANNEL_CONNECTING);
419 // Return a resolver result, which allows the connection attempt to proceed.
420 SetNextResolution(GetServersPorts());
421 // We should eventually transition into state READY.
422 EXPECT_TRUE(WaitForChannelReady(channel.get()));
425 TEST_F(ClientLbEnd2endTest, PickFirst) {
426 // Start servers and send one RPC per server.
427 const int kNumServers = 3;
428 StartServers(kNumServers);
429 auto channel = BuildChannel(""); // test that pick first is the default.
430 auto stub = BuildStub(channel);
431 SetNextResolution(GetServersPorts());
432 for (size_t i = 0; i < servers_.size(); ++i) {
433 CheckRpcSendOk(stub, DEBUG_LOCATION);
435 // All requests should have gone to a single server.
437 for (size_t i = 0; i < servers_.size(); ++i) {
438 const int request_count = servers_[i]->service_.request_count();
439 if (request_count == kNumServers) {
442 EXPECT_EQ(0, request_count);
446 // Check LB policy name for the channel.
447 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
450 TEST_F(ClientLbEnd2endTest, PickFirstProcessPending) {
451 StartServers(1); // Single server
452 auto channel = BuildChannel(""); // test that pick first is the default.
453 auto stub = BuildStub(channel);
454 SetNextResolution({servers_[0]->port_});
455 WaitForServer(stub, 0, DEBUG_LOCATION);
456 // Create a new channel and its corresponding PF LB policy, which will pick
457 // the subchannels in READY state from the previous RPC against the same
458 // target (even if it happened over a different channel, because subchannels
459 // are globally reused). Progress should happen without any transition from
461 auto second_channel = BuildChannel("");
462 auto second_stub = BuildStub(second_channel);
463 SetNextResolution({servers_[0]->port_});
464 CheckRpcSendOk(second_stub, DEBUG_LOCATION);
467 TEST_F(ClientLbEnd2endTest, PickFirstSelectsReadyAtStartup) {
468 ChannelArguments args;
469 constexpr int kInitialBackOffMs = 5000;
470 args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
471 // Create 2 servers, but start only the second one.
472 std::vector<int> ports = {grpc_pick_unused_port_or_die(),
473 grpc_pick_unused_port_or_die()};
474 CreateServers(2, ports);
476 auto channel1 = BuildChannel("pick_first", args);
477 auto stub1 = BuildStub(channel1);
478 SetNextResolution(ports);
479 // Wait for second server to be ready.
480 WaitForServer(stub1, 1, DEBUG_LOCATION);
481 // Create a second channel with the same addresses. Its PF instance
482 // should immediately pick the second subchannel, since it's already
484 auto channel2 = BuildChannel("pick_first", args);
485 SetNextResolution(ports);
486 // Check that the channel reports READY without waiting for the
488 EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1 /* timeout_seconds */));
491 TEST_F(ClientLbEnd2endTest, PickFirstBackOffInitialReconnect) {
492 ChannelArguments args;
493 constexpr int kInitialBackOffMs = 100;
494 args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
495 const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
496 const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
497 auto channel = BuildChannel("pick_first", args);
498 auto stub = BuildStub(channel);
499 SetNextResolution(ports);
500 // The channel won't become connected (there's no server).
501 ASSERT_FALSE(channel->WaitForConnected(
502 grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 2)));
503 // Bring up a server on the chosen port.
504 StartServers(1, ports);
506 ASSERT_TRUE(channel->WaitForConnected(
507 grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 2)));
508 const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
509 const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
510 gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
511 // We should have waited at least kInitialBackOffMs. We substract one to
512 // account for test and precision accuracy drift.
513 EXPECT_GE(waited_ms, kInitialBackOffMs - 1);
514 // But not much more.
517 grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 1.10), t1),
521 TEST_F(ClientLbEnd2endTest, PickFirstBackOffMinReconnect) {
522 ChannelArguments args;
523 constexpr int kMinReconnectBackOffMs = 1000;
524 args.SetInt(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, kMinReconnectBackOffMs);
525 const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
526 auto channel = BuildChannel("pick_first", args);
527 auto stub = BuildStub(channel);
528 SetNextResolution(ports);
529 // Make connection delay a 10% longer than it's willing to in order to make
530 // sure we are hitting the codepath that waits for the min reconnect backoff.
531 gpr_atm_rel_store(&g_connection_delay_ms, kMinReconnectBackOffMs * 1.10);
532 default_client_impl = grpc_tcp_client_impl;
533 grpc_set_tcp_client_impl(&delayed_connect);
534 const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
535 channel->WaitForConnected(
536 grpc_timeout_milliseconds_to_deadline(kMinReconnectBackOffMs * 2));
537 const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
538 const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
539 gpr_log(GPR_DEBUG, "Waited %" PRId64 " ms", waited_ms);
540 // We should have waited at least kMinReconnectBackOffMs. We substract one to
541 // account for test and precision accuracy drift.
542 EXPECT_GE(waited_ms, kMinReconnectBackOffMs - 1);
543 gpr_atm_rel_store(&g_connection_delay_ms, 0);
546 TEST_F(ClientLbEnd2endTest, PickFirstResetConnectionBackoff) {
547 ChannelArguments args;
548 constexpr int kInitialBackOffMs = 1000;
549 args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
550 const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
551 auto channel = BuildChannel("pick_first", args);
552 auto stub = BuildStub(channel);
553 SetNextResolution(ports);
554 // The channel won't become connected (there's no server).
556 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
557 // Bring up a server on the chosen port.
558 StartServers(1, ports);
559 const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
560 // Wait for connect, but not long enough. This proves that we're
561 // being throttled by initial backoff.
563 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
564 // Reset connection backoff.
565 experimental::ChannelResetConnectionBackoff(channel.get());
566 // Wait for connect. Should happen ~immediately.
568 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
569 const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
570 const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
571 gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
572 // We should have waited less than kInitialBackOffMs.
573 EXPECT_LT(waited_ms, kInitialBackOffMs);
576 TEST_F(ClientLbEnd2endTest,
577 PickFirstResetConnectionBackoffNextAttemptStartsImmediately) {
578 ChannelArguments args;
579 constexpr int kInitialBackOffMs = 1000;
580 args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
581 const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
582 auto channel = BuildChannel("pick_first", args);
583 auto stub = BuildStub(channel);
584 SetNextResolution(ports);
585 // Wait for connect, which should fail ~immediately, because the server
587 gpr_log(GPR_INFO, "=== INITIAL CONNECTION ATTEMPT");
589 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
590 // Reset connection backoff.
591 // Note that the time at which the third attempt will be started is
592 // actually computed at this point, so we record the start time here.
593 gpr_log(GPR_INFO, "=== RESETTING BACKOFF");
594 const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
595 experimental::ChannelResetConnectionBackoff(channel.get());
596 // Trigger a second connection attempt. This should also fail
597 // ~immediately, but the retry should be scheduled for
598 // kInitialBackOffMs instead of applying the multiplier.
599 gpr_log(GPR_INFO, "=== POLLING FOR SECOND CONNECTION ATTEMPT");
601 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
602 // Bring up a server on the chosen port.
603 gpr_log(GPR_INFO, "=== STARTING BACKEND");
604 StartServers(1, ports);
605 // Wait for connect. Should happen within kInitialBackOffMs.
606 // Give an extra 100ms to account for the time spent in the second and
607 // third connection attempts themselves (since what we really want to
608 // measure is the time between the two). As long as this is less than
609 // the 1.6x increase we would see if the backoff state was not reset
610 // properly, the test is still proving that the backoff was reset.
611 constexpr int kWaitMs = kInitialBackOffMs + 100;
612 gpr_log(GPR_INFO, "=== POLLING FOR THIRD CONNECTION ATTEMPT");
613 EXPECT_TRUE(channel->WaitForConnected(
614 grpc_timeout_milliseconds_to_deadline(kWaitMs)));
615 const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
616 const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
617 gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
618 EXPECT_LT(waited_ms, kWaitMs);
621 TEST_F(ClientLbEnd2endTest, PickFirstUpdates) {
622 // Start servers and send one RPC per server.
623 const int kNumServers = 3;
624 StartServers(kNumServers);
625 auto channel = BuildChannel("pick_first");
626 auto stub = BuildStub(channel);
628 std::vector<int> ports;
630 // Perform one RPC against the first server.
631 ports.emplace_back(servers_[0]->port_);
632 SetNextResolution(ports);
633 gpr_log(GPR_INFO, "****** SET [0] *******");
634 CheckRpcSendOk(stub, DEBUG_LOCATION);
635 EXPECT_EQ(servers_[0]->service_.request_count(), 1);
637 // An empty update will result in the channel going into TRANSIENT_FAILURE.
639 SetNextResolution(ports);
640 gpr_log(GPR_INFO, "****** SET none *******");
641 grpc_connectivity_state channel_state;
643 channel_state = channel->GetState(true /* try to connect */);
644 } while (channel_state == GRPC_CHANNEL_READY);
645 ASSERT_NE(channel_state, GRPC_CHANNEL_READY);
646 servers_[0]->service_.ResetCounters();
648 // Next update introduces servers_[1], making the channel recover.
650 ports.emplace_back(servers_[1]->port_);
651 SetNextResolution(ports);
652 gpr_log(GPR_INFO, "****** SET [1] *******");
653 WaitForServer(stub, 1, DEBUG_LOCATION);
654 EXPECT_EQ(servers_[0]->service_.request_count(), 0);
656 // And again for servers_[2]
658 ports.emplace_back(servers_[2]->port_);
659 SetNextResolution(ports);
660 gpr_log(GPR_INFO, "****** SET [2] *******");
661 WaitForServer(stub, 2, DEBUG_LOCATION);
662 EXPECT_EQ(servers_[0]->service_.request_count(), 0);
663 EXPECT_EQ(servers_[1]->service_.request_count(), 0);
665 // Check LB policy name for the channel.
666 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
669 TEST_F(ClientLbEnd2endTest, PickFirstUpdateSuperset) {
670 // Start servers and send one RPC per server.
671 const int kNumServers = 3;
672 StartServers(kNumServers);
673 auto channel = BuildChannel("pick_first");
674 auto stub = BuildStub(channel);
676 std::vector<int> ports;
678 // Perform one RPC against the first server.
679 ports.emplace_back(servers_[0]->port_);
680 SetNextResolution(ports);
681 gpr_log(GPR_INFO, "****** SET [0] *******");
682 CheckRpcSendOk(stub, DEBUG_LOCATION);
683 EXPECT_EQ(servers_[0]->service_.request_count(), 1);
684 servers_[0]->service_.ResetCounters();
686 // Send and superset update
688 ports.emplace_back(servers_[1]->port_);
689 ports.emplace_back(servers_[0]->port_);
690 SetNextResolution(ports);
691 gpr_log(GPR_INFO, "****** SET superset *******");
692 CheckRpcSendOk(stub, DEBUG_LOCATION);
693 // We stick to the previously connected server.
694 WaitForServer(stub, 0, DEBUG_LOCATION);
695 EXPECT_EQ(0, servers_[1]->service_.request_count());
697 // Check LB policy name for the channel.
698 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
701 TEST_F(ClientLbEnd2endTest, PickFirstGlobalSubchannelPool) {
703 const int kNumServers = 1;
704 StartServers(kNumServers);
705 std::vector<int> ports = GetServersPorts();
706 // Create two channels that (by default) use the global subchannel pool.
707 auto channel1 = BuildChannel("pick_first");
708 auto stub1 = BuildStub(channel1);
709 SetNextResolution(ports);
710 auto channel2 = BuildChannel("pick_first");
711 auto stub2 = BuildStub(channel2);
712 SetNextResolution(ports);
713 WaitForServer(stub1, 0, DEBUG_LOCATION);
714 // Send one RPC on each channel.
715 CheckRpcSendOk(stub1, DEBUG_LOCATION);
716 CheckRpcSendOk(stub2, DEBUG_LOCATION);
717 // The server receives two requests.
718 EXPECT_EQ(2, servers_[0]->service_.request_count());
719 // The two requests are from the same client port, because the two channels
720 // share subchannels via the global subchannel pool.
721 EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
724 TEST_F(ClientLbEnd2endTest, PickFirstLocalSubchannelPool) {
726 const int kNumServers = 1;
727 StartServers(kNumServers);
728 std::vector<int> ports = GetServersPorts();
729 // Create two channels that use local subchannel pool.
730 ChannelArguments args;
731 args.SetInt(GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL, 1);
732 auto channel1 = BuildChannel("pick_first", args);
733 auto stub1 = BuildStub(channel1);
734 SetNextResolution(ports);
735 auto channel2 = BuildChannel("pick_first", args);
736 auto stub2 = BuildStub(channel2);
737 SetNextResolution(ports);
738 WaitForServer(stub1, 0, DEBUG_LOCATION);
739 // Send one RPC on each channel.
740 CheckRpcSendOk(stub1, DEBUG_LOCATION);
741 CheckRpcSendOk(stub2, DEBUG_LOCATION);
742 // The server receives two requests.
743 EXPECT_EQ(2, servers_[0]->service_.request_count());
744 // The two requests are from two client ports, because the two channels didn't
745 // share subchannels with each other.
746 EXPECT_EQ(2UL, servers_[0]->service_.clients().size());
749 TEST_F(ClientLbEnd2endTest, PickFirstManyUpdates) {
750 const int kNumUpdates = 1000;
751 const int kNumServers = 3;
752 StartServers(kNumServers);
753 auto channel = BuildChannel("pick_first");
754 auto stub = BuildStub(channel);
755 std::vector<int> ports = GetServersPorts();
756 for (size_t i = 0; i < kNumUpdates; ++i) {
757 std::shuffle(ports.begin(), ports.end(),
758 std::mt19937(std::random_device()()));
759 SetNextResolution(ports);
760 // We should re-enter core at the end of the loop to give the resolution
761 // setting closure a chance to run.
762 if ((i + 1) % 10 == 0) CheckRpcSendOk(stub, DEBUG_LOCATION);
764 // Check LB policy name for the channel.
765 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
768 TEST_F(ClientLbEnd2endTest, PickFirstReresolutionNoSelected) {
769 // Prepare the ports for up servers and down servers.
770 const int kNumServers = 3;
771 const int kNumAliveServers = 1;
772 StartServers(kNumAliveServers);
773 std::vector<int> alive_ports, dead_ports;
774 for (size_t i = 0; i < kNumServers; ++i) {
775 if (i < kNumAliveServers) {
776 alive_ports.emplace_back(servers_[i]->port_);
778 dead_ports.emplace_back(grpc_pick_unused_port_or_die());
781 auto channel = BuildChannel("pick_first");
782 auto stub = BuildStub(channel);
783 // The initial resolution only contains dead ports. There won't be any
784 // selected subchannel. Re-resolution will return the same result.
785 SetNextResolution(dead_ports);
786 gpr_log(GPR_INFO, "****** INITIAL RESOLUTION SET *******");
787 for (size_t i = 0; i < 10; ++i) CheckRpcSendFailure(stub);
788 // Set a re-resolution result that contains reachable ports, so that the
789 // pick_first LB policy can recover soon.
790 SetNextResolutionUponError(alive_ports);
791 gpr_log(GPR_INFO, "****** RE-RESOLUTION SET *******");
792 WaitForServer(stub, 0, DEBUG_LOCATION, true /* ignore_failure */);
793 CheckRpcSendOk(stub, DEBUG_LOCATION);
794 EXPECT_EQ(servers_[0]->service_.request_count(), 1);
795 // Check LB policy name for the channel.
796 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
799 TEST_F(ClientLbEnd2endTest, PickFirstReconnectWithoutNewResolverResult) {
800 std::vector<int> ports = {grpc_pick_unused_port_or_die()};
801 StartServers(1, ports);
802 auto channel = BuildChannel("pick_first");
803 auto stub = BuildStub(channel);
804 SetNextResolution(ports);
805 gpr_log(GPR_INFO, "****** INITIAL CONNECTION *******");
806 WaitForServer(stub, 0, DEBUG_LOCATION);
807 gpr_log(GPR_INFO, "****** STOPPING SERVER ******");
808 servers_[0]->Shutdown();
809 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
810 gpr_log(GPR_INFO, "****** RESTARTING SERVER ******");
811 StartServers(1, ports);
812 WaitForServer(stub, 0, DEBUG_LOCATION);
815 TEST_F(ClientLbEnd2endTest,
816 PickFirstReconnectWithoutNewResolverResultStartsFromTopOfList) {
817 std::vector<int> ports = {grpc_pick_unused_port_or_die(),
818 grpc_pick_unused_port_or_die()};
819 CreateServers(2, ports);
821 auto channel = BuildChannel("pick_first");
822 auto stub = BuildStub(channel);
823 SetNextResolution(ports);
824 gpr_log(GPR_INFO, "****** INITIAL CONNECTION *******");
825 WaitForServer(stub, 1, DEBUG_LOCATION);
826 gpr_log(GPR_INFO, "****** STOPPING SERVER ******");
827 servers_[1]->Shutdown();
828 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
829 gpr_log(GPR_INFO, "****** STARTING BOTH SERVERS ******");
830 StartServers(2, ports);
831 WaitForServer(stub, 0, DEBUG_LOCATION);
834 TEST_F(ClientLbEnd2endTest, PickFirstCheckStateBeforeStartWatch) {
835 std::vector<int> ports = {grpc_pick_unused_port_or_die()};
836 StartServers(1, ports);
837 auto channel_1 = BuildChannel("pick_first");
838 auto stub_1 = BuildStub(channel_1);
839 SetNextResolution(ports);
840 gpr_log(GPR_INFO, "****** RESOLUTION SET FOR CHANNEL 1 *******");
841 WaitForServer(stub_1, 0, DEBUG_LOCATION);
842 gpr_log(GPR_INFO, "****** CHANNEL 1 CONNECTED *******");
843 servers_[0]->Shutdown();
844 // Channel 1 will receive a re-resolution containing the same server. It will
845 // create a new subchannel and hold a ref to it.
846 StartServers(1, ports);
847 gpr_log(GPR_INFO, "****** SERVER RESTARTED *******");
848 auto channel_2 = BuildChannel("pick_first");
849 auto stub_2 = BuildStub(channel_2);
850 // TODO(juanlishen): This resolution result will only be visible to channel 2
851 // since the response generator is only associated with channel 2 now. We
852 // should change the response generator to be able to deliver updates to
853 // multiple channels at once.
854 SetNextResolution(ports);
855 gpr_log(GPR_INFO, "****** RESOLUTION SET FOR CHANNEL 2 *******");
856 WaitForServer(stub_2, 0, DEBUG_LOCATION, true);
857 gpr_log(GPR_INFO, "****** CHANNEL 2 CONNECTED *******");
858 servers_[0]->Shutdown();
859 // Wait until the disconnection has triggered the connectivity notification.
860 // Otherwise, the subchannel may be picked for next call but will fail soon.
861 EXPECT_TRUE(WaitForChannelNotReady(channel_2.get()));
862 // Channel 2 will also receive a re-resolution containing the same server.
863 // Both channels will ref the same subchannel that failed.
864 StartServers(1, ports);
865 gpr_log(GPR_INFO, "****** SERVER RESTARTED AGAIN *******");
866 gpr_log(GPR_INFO, "****** CHANNEL 2 STARTING A CALL *******");
867 // The first call after the server restart will succeed.
868 CheckRpcSendOk(stub_2, DEBUG_LOCATION);
869 gpr_log(GPR_INFO, "****** CHANNEL 2 FINISHED A CALL *******");
870 // Check LB policy name for the channel.
871 EXPECT_EQ("pick_first", channel_1->GetLoadBalancingPolicyName());
872 // Check LB policy name for the channel.
873 EXPECT_EQ("pick_first", channel_2->GetLoadBalancingPolicyName());
876 TEST_F(ClientLbEnd2endTest, PickFirstIdleOnDisconnect) {
877 // Start server, send RPC, and make sure channel is READY.
878 const int kNumServers = 1;
879 StartServers(kNumServers);
880 auto channel = BuildChannel(""); // pick_first is the default.
881 auto stub = BuildStub(channel);
882 SetNextResolution(GetServersPorts());
883 CheckRpcSendOk(stub, DEBUG_LOCATION);
884 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
885 // Stop server. Channel should go into state IDLE.
886 SetFailureOnReresolution();
887 servers_[0]->Shutdown();
888 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
889 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
893 TEST_F(ClientLbEnd2endTest, PickFirstPendingUpdateAndSelectedSubchannelFails) {
894 auto channel = BuildChannel(""); // pick_first is the default.
895 auto stub = BuildStub(channel);
896 // Create a number of servers, but only start 1 of them.
899 // Initially resolve to first server and make sure it connects.
900 gpr_log(GPR_INFO, "Phase 1: Connect to first server.");
901 SetNextResolution({servers_[0]->port_});
902 CheckRpcSendOk(stub, DEBUG_LOCATION, true /* wait_for_ready */);
903 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
904 // Send a resolution update with the remaining servers, none of which are
905 // running yet, so the update will stay pending. Note that it's important
906 // to have multiple servers here, or else the test will be flaky; with only
907 // one server, the pending subchannel list has already gone into
908 // TRANSIENT_FAILURE due to hitting the end of the list by the time we
911 "Phase 2: Resolver update pointing to remaining "
912 "(not started) servers.");
913 SetNextResolution(GetServersPorts(1 /* start_index */));
914 // RPCs will continue to be sent to the first server.
915 CheckRpcSendOk(stub, DEBUG_LOCATION);
916 // Now stop the first server, so that the current subchannel list
917 // fails. This should cause us to immediately swap over to the
918 // pending list, even though it's not yet connected. The state should
919 // be set to CONNECTING, since that's what the pending subchannel list
920 // was doing when we swapped over.
921 gpr_log(GPR_INFO, "Phase 3: Stopping first server.");
922 servers_[0]->Shutdown();
923 WaitForChannelNotReady(channel.get());
924 // TODO(roth): This should always return CONNECTING, but it's flaky
925 // between that and TRANSIENT_FAILURE. I suspect that this problem
926 // will go away once we move the backoff code out of the subchannel
927 // and into the LB policies.
928 EXPECT_THAT(channel->GetState(false),
929 ::testing::AnyOf(GRPC_CHANNEL_CONNECTING,
930 GRPC_CHANNEL_TRANSIENT_FAILURE));
931 // Now start the second server.
932 gpr_log(GPR_INFO, "Phase 4: Starting second server.");
934 // The channel should go to READY state and RPCs should go to the
936 WaitForChannelReady(channel.get());
937 WaitForServer(stub, 1, DEBUG_LOCATION, true /* ignore_failure */);
940 TEST_F(ClientLbEnd2endTest, PickFirstStaysIdleUponEmptyUpdate) {
941 // Start server, send RPC, and make sure channel is READY.
942 const int kNumServers = 1;
943 StartServers(kNumServers);
944 auto channel = BuildChannel(""); // pick_first is the default.
945 auto stub = BuildStub(channel);
946 SetNextResolution(GetServersPorts());
947 CheckRpcSendOk(stub, DEBUG_LOCATION);
948 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
949 // Stop server. Channel should go into state IDLE.
950 servers_[0]->Shutdown();
951 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
952 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
953 // Now send resolver update that includes no addresses. Channel
954 // should stay in state IDLE.
955 SetNextResolution({});
956 EXPECT_FALSE(channel->WaitForStateChange(
957 GRPC_CHANNEL_IDLE, grpc_timeout_seconds_to_deadline(3)));
958 // Now bring the backend back up and send a non-empty resolver update,
959 // and then try to send an RPC. Channel should go back into state READY.
961 SetNextResolution(GetServersPorts());
962 CheckRpcSendOk(stub, DEBUG_LOCATION);
963 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
966 TEST_F(ClientLbEnd2endTest, RoundRobin) {
967 // Start servers and send one RPC per server.
968 const int kNumServers = 3;
969 StartServers(kNumServers);
970 auto channel = BuildChannel("round_robin");
971 auto stub = BuildStub(channel);
972 SetNextResolution(GetServersPorts());
973 // Wait until all backends are ready.
975 CheckRpcSendOk(stub, DEBUG_LOCATION);
976 } while (!SeenAllServers());
978 // "Sync" to the end of the list. Next sequence of picks will start at the
979 // first server (index 0).
980 WaitForServer(stub, servers_.size() - 1, DEBUG_LOCATION);
981 std::vector<int> connection_order;
982 for (size_t i = 0; i < servers_.size(); ++i) {
983 CheckRpcSendOk(stub, DEBUG_LOCATION);
984 UpdateConnectionOrder(servers_, &connection_order);
986 // Backends should be iterated over in the order in which the addresses were
988 const auto expected = std::vector<int>{0, 1, 2};
989 EXPECT_EQ(expected, connection_order);
990 // Check LB policy name for the channel.
991 EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
994 TEST_F(ClientLbEnd2endTest, RoundRobinProcessPending) {
995 StartServers(1); // Single server
996 auto channel = BuildChannel("round_robin");
997 auto stub = BuildStub(channel);
998 SetNextResolution({servers_[0]->port_});
999 WaitForServer(stub, 0, DEBUG_LOCATION);
1000 // Create a new channel and its corresponding RR LB policy, which will pick
1001 // the subchannels in READY state from the previous RPC against the same
1002 // target (even if it happened over a different channel, because subchannels
1003 // are globally reused). Progress should happen without any transition from
1004 // this READY state.
1005 auto second_channel = BuildChannel("round_robin");
1006 auto second_stub = BuildStub(second_channel);
1007 SetNextResolution({servers_[0]->port_});
1008 CheckRpcSendOk(second_stub, DEBUG_LOCATION);
1011 TEST_F(ClientLbEnd2endTest, RoundRobinUpdates) {
1012 // Start servers and send one RPC per server.
1013 const int kNumServers = 3;
1014 StartServers(kNumServers);
1015 auto channel = BuildChannel("round_robin");
1016 auto stub = BuildStub(channel);
1017 std::vector<int> ports;
1018 // Start with a single server.
1019 gpr_log(GPR_INFO, "*** FIRST BACKEND ***");
1020 ports.emplace_back(servers_[0]->port_);
1021 SetNextResolution(ports);
1022 WaitForServer(stub, 0, DEBUG_LOCATION);
1023 // Send RPCs. They should all go servers_[0]
1024 for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1025 EXPECT_EQ(10, servers_[0]->service_.request_count());
1026 EXPECT_EQ(0, servers_[1]->service_.request_count());
1027 EXPECT_EQ(0, servers_[2]->service_.request_count());
1028 servers_[0]->service_.ResetCounters();
1029 // And now for the second server.
1030 gpr_log(GPR_INFO, "*** SECOND BACKEND ***");
1032 ports.emplace_back(servers_[1]->port_);
1033 SetNextResolution(ports);
1034 // Wait until update has been processed, as signaled by the second backend
1035 // receiving a request.
1036 EXPECT_EQ(0, servers_[1]->service_.request_count());
1037 WaitForServer(stub, 1, DEBUG_LOCATION);
1038 for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1039 EXPECT_EQ(0, servers_[0]->service_.request_count());
1040 EXPECT_EQ(10, servers_[1]->service_.request_count());
1041 EXPECT_EQ(0, servers_[2]->service_.request_count());
1042 servers_[1]->service_.ResetCounters();
1043 // ... and for the last server.
1044 gpr_log(GPR_INFO, "*** THIRD BACKEND ***");
1046 ports.emplace_back(servers_[2]->port_);
1047 SetNextResolution(ports);
1048 WaitForServer(stub, 2, DEBUG_LOCATION);
1049 for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1050 EXPECT_EQ(0, servers_[0]->service_.request_count());
1051 EXPECT_EQ(0, servers_[1]->service_.request_count());
1052 EXPECT_EQ(10, servers_[2]->service_.request_count());
1053 servers_[2]->service_.ResetCounters();
1054 // Back to all servers.
1055 gpr_log(GPR_INFO, "*** ALL BACKENDS ***");
1057 ports.emplace_back(servers_[0]->port_);
1058 ports.emplace_back(servers_[1]->port_);
1059 ports.emplace_back(servers_[2]->port_);
1060 SetNextResolution(ports);
1061 WaitForServer(stub, 0, DEBUG_LOCATION);
1062 WaitForServer(stub, 1, DEBUG_LOCATION);
1063 WaitForServer(stub, 2, DEBUG_LOCATION);
1064 // Send three RPCs, one per server.
1065 for (size_t i = 0; i < 3; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1066 EXPECT_EQ(1, servers_[0]->service_.request_count());
1067 EXPECT_EQ(1, servers_[1]->service_.request_count());
1068 EXPECT_EQ(1, servers_[2]->service_.request_count());
1069 // An empty update will result in the channel going into TRANSIENT_FAILURE.
1070 gpr_log(GPR_INFO, "*** NO BACKENDS ***");
1072 SetNextResolution(ports);
1073 grpc_connectivity_state channel_state;
1075 channel_state = channel->GetState(true /* try to connect */);
1076 } while (channel_state == GRPC_CHANNEL_READY);
1077 ASSERT_NE(channel_state, GRPC_CHANNEL_READY);
1078 servers_[0]->service_.ResetCounters();
1079 // Next update introduces servers_[1], making the channel recover.
1080 gpr_log(GPR_INFO, "*** BACK TO SECOND BACKEND ***");
1082 ports.emplace_back(servers_[1]->port_);
1083 SetNextResolution(ports);
1084 WaitForServer(stub, 1, DEBUG_LOCATION);
1085 channel_state = channel->GetState(false /* try to connect */);
1086 ASSERT_EQ(channel_state, GRPC_CHANNEL_READY);
1087 // Check LB policy name for the channel.
1088 EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
1091 TEST_F(ClientLbEnd2endTest, RoundRobinUpdateInError) {
1092 const int kNumServers = 3;
1093 StartServers(kNumServers);
1094 auto channel = BuildChannel("round_robin");
1095 auto stub = BuildStub(channel);
1096 std::vector<int> ports;
1098 // Start with a single server.
1099 ports.emplace_back(servers_[0]->port_);
1100 SetNextResolution(ports);
1101 WaitForServer(stub, 0, DEBUG_LOCATION);
1102 // Send RPCs. They should all go to servers_[0]
1103 for (size_t i = 0; i < 10; ++i) SendRpc(stub);
1104 EXPECT_EQ(10, servers_[0]->service_.request_count());
1105 EXPECT_EQ(0, servers_[1]->service_.request_count());
1106 EXPECT_EQ(0, servers_[2]->service_.request_count());
1107 servers_[0]->service_.ResetCounters();
1109 // Shutdown one of the servers to be sent in the update.
1110 servers_[1]->Shutdown();
1111 ports.emplace_back(servers_[1]->port_);
1112 ports.emplace_back(servers_[2]->port_);
1113 SetNextResolution(ports);
1114 WaitForServer(stub, 0, DEBUG_LOCATION);
1115 WaitForServer(stub, 2, DEBUG_LOCATION);
1117 // Send three RPCs, one per server.
1118 for (size_t i = 0; i < kNumServers; ++i) SendRpc(stub);
1119 // The server in shutdown shouldn't receive any.
1120 EXPECT_EQ(0, servers_[1]->service_.request_count());
1123 TEST_F(ClientLbEnd2endTest, RoundRobinManyUpdates) {
1124 // Start servers and send one RPC per server.
1125 const int kNumServers = 3;
1126 StartServers(kNumServers);
1127 auto channel = BuildChannel("round_robin");
1128 auto stub = BuildStub(channel);
1129 std::vector<int> ports = GetServersPorts();
1130 for (size_t i = 0; i < 1000; ++i) {
1131 std::shuffle(ports.begin(), ports.end(),
1132 std::mt19937(std::random_device()()));
1133 SetNextResolution(ports);
1134 if (i % 10 == 0) CheckRpcSendOk(stub, DEBUG_LOCATION);
1136 // Check LB policy name for the channel.
1137 EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
1140 TEST_F(ClientLbEnd2endTest, RoundRobinConcurrentUpdates) {
1141 // TODO(dgq): replicate the way internal testing exercises the concurrent
1142 // update provisions of RR.
1145 TEST_F(ClientLbEnd2endTest, RoundRobinReresolve) {
1146 // Start servers and send one RPC per server.
1147 const int kNumServers = 3;
1148 std::vector<int> first_ports;
1149 std::vector<int> second_ports;
1150 first_ports.reserve(kNumServers);
1151 for (int i = 0; i < kNumServers; ++i) {
1152 first_ports.push_back(grpc_pick_unused_port_or_die());
1154 second_ports.reserve(kNumServers);
1155 for (int i = 0; i < kNumServers; ++i) {
1156 second_ports.push_back(grpc_pick_unused_port_or_die());
1158 StartServers(kNumServers, first_ports);
1159 auto channel = BuildChannel("round_robin");
1160 auto stub = BuildStub(channel);
1161 SetNextResolution(first_ports);
1162 // Send a number of RPCs, which succeed.
1163 for (size_t i = 0; i < 100; ++i) {
1164 CheckRpcSendOk(stub, DEBUG_LOCATION);
1167 gpr_log(GPR_INFO, "****** ABOUT TO KILL SERVERS *******");
1168 for (size_t i = 0; i < servers_.size(); ++i) {
1169 servers_[i]->Shutdown();
1171 gpr_log(GPR_INFO, "****** SERVERS KILLED *******");
1172 gpr_log(GPR_INFO, "****** SENDING DOOMED REQUESTS *******");
1173 // Client requests should fail. Send enough to tickle all subchannels.
1174 for (size_t i = 0; i < servers_.size(); ++i) CheckRpcSendFailure(stub);
1175 gpr_log(GPR_INFO, "****** DOOMED REQUESTS SENT *******");
1176 // Bring servers back up on a different set of ports. We need to do this to be
1177 // sure that the eventual success is *not* due to subchannel reconnection
1178 // attempts and that an actual re-resolution has happened as a result of the
1179 // RR policy going into transient failure when all its subchannels become
1180 // unavailable (in transient failure as well).
1181 gpr_log(GPR_INFO, "****** RESTARTING SERVERS *******");
1182 StartServers(kNumServers, second_ports);
1183 // Don't notify of the update. Wait for the LB policy's re-resolution to
1184 // "pull" the new ports.
1185 SetNextResolutionUponError(second_ports);
1186 gpr_log(GPR_INFO, "****** SERVERS RESTARTED *******");
1187 gpr_log(GPR_INFO, "****** SENDING REQUEST TO SUCCEED *******");
1188 // Client request should eventually (but still fairly soon) succeed.
1189 const gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
1190 gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
1191 while (gpr_time_cmp(deadline, now) > 0) {
1192 if (SendRpc(stub)) break;
1193 now = gpr_now(GPR_CLOCK_MONOTONIC);
1195 ASSERT_GT(gpr_time_cmp(deadline, now), 0);
1198 TEST_F(ClientLbEnd2endTest, RoundRobinSingleReconnect) {
1199 const int kNumServers = 3;
1200 StartServers(kNumServers);
1201 const auto ports = GetServersPorts();
1202 auto channel = BuildChannel("round_robin");
1203 auto stub = BuildStub(channel);
1204 SetNextResolution(ports);
1205 for (size_t i = 0; i < kNumServers; ++i) {
1206 WaitForServer(stub, i, DEBUG_LOCATION);
1208 for (size_t i = 0; i < servers_.size(); ++i) {
1209 CheckRpcSendOk(stub, DEBUG_LOCATION);
1210 EXPECT_EQ(1, servers_[i]->service_.request_count()) << "for backend #" << i;
1212 // One request should have gone to each server.
1213 for (size_t i = 0; i < servers_.size(); ++i) {
1214 EXPECT_EQ(1, servers_[i]->service_.request_count());
1216 const auto pre_death = servers_[0]->service_.request_count();
1217 // Kill the first server.
1218 servers_[0]->Shutdown();
1219 // Client request still succeed. May need retrying if RR had returned a pick
1220 // before noticing the change in the server's connectivity.
1221 while (!SendRpc(stub)) {
1222 } // Retry until success.
1223 // Send a bunch of RPCs that should succeed.
1224 for (int i = 0; i < 10 * kNumServers; ++i) {
1225 CheckRpcSendOk(stub, DEBUG_LOCATION);
1227 const auto post_death = servers_[0]->service_.request_count();
1228 // No requests have gone to the deceased server.
1229 EXPECT_EQ(pre_death, post_death);
1230 // Bring the first server back up.
1232 // Requests should start arriving at the first server either right away (if
1233 // the server managed to start before the RR policy retried the subchannel) or
1234 // after the subchannel retry delay otherwise (RR's subchannel retried before
1235 // the server was fully back up).
1236 WaitForServer(stub, 0, DEBUG_LOCATION);
1239 // If health checking is required by client but health checking service
1240 // is not running on the server, the channel should be treated as healthy.
1241 TEST_F(ClientLbEnd2endTest,
1242 RoundRobinServersHealthCheckingUnimplementedTreatedAsHealthy) {
1243 StartServers(1); // Single server
1244 ChannelArguments args;
1245 args.SetServiceConfigJSON(
1246 "{\"healthCheckConfig\": "
1247 "{\"serviceName\": \"health_check_service_name\"}}");
1248 auto channel = BuildChannel("round_robin", args);
1249 auto stub = BuildStub(channel);
1250 SetNextResolution({servers_[0]->port_});
1251 EXPECT_TRUE(WaitForChannelReady(channel.get()));
1252 CheckRpcSendOk(stub, DEBUG_LOCATION);
1255 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthChecking) {
1256 EnableDefaultHealthCheckService(true);
1258 const int kNumServers = 3;
1259 StartServers(kNumServers);
1260 ChannelArguments args;
1261 args.SetServiceConfigJSON(
1262 "{\"healthCheckConfig\": "
1263 "{\"serviceName\": \"health_check_service_name\"}}");
1264 auto channel = BuildChannel("round_robin", args);
1265 auto stub = BuildStub(channel);
1266 SetNextResolution(GetServersPorts());
1267 // Channel should not become READY, because health checks should be failing.
1269 "*** initial state: unknown health check service name for "
1271 EXPECT_FALSE(WaitForChannelReady(channel.get(), 1));
1272 // Now set one of the servers to be healthy.
1273 // The channel should become healthy and all requests should go to
1274 // the healthy server.
1275 gpr_log(GPR_INFO, "*** server 0 healthy");
1276 servers_[0]->SetServingStatus("health_check_service_name", true);
1277 EXPECT_TRUE(WaitForChannelReady(channel.get()));
1278 for (int i = 0; i < 10; ++i) {
1279 CheckRpcSendOk(stub, DEBUG_LOCATION);
1281 EXPECT_EQ(10, servers_[0]->service_.request_count());
1282 EXPECT_EQ(0, servers_[1]->service_.request_count());
1283 EXPECT_EQ(0, servers_[2]->service_.request_count());
1284 // Now set a second server to be healthy.
1285 gpr_log(GPR_INFO, "*** server 2 healthy");
1286 servers_[2]->SetServingStatus("health_check_service_name", true);
1287 WaitForServer(stub, 2, DEBUG_LOCATION);
1288 for (int i = 0; i < 10; ++i) {
1289 CheckRpcSendOk(stub, DEBUG_LOCATION);
1291 EXPECT_EQ(5, servers_[0]->service_.request_count());
1292 EXPECT_EQ(0, servers_[1]->service_.request_count());
1293 EXPECT_EQ(5, servers_[2]->service_.request_count());
1294 // Now set the remaining server to be healthy.
1295 gpr_log(GPR_INFO, "*** server 1 healthy");
1296 servers_[1]->SetServingStatus("health_check_service_name", true);
1297 WaitForServer(stub, 1, DEBUG_LOCATION);
1298 for (int i = 0; i < 9; ++i) {
1299 CheckRpcSendOk(stub, DEBUG_LOCATION);
1301 EXPECT_EQ(3, servers_[0]->service_.request_count());
1302 EXPECT_EQ(3, servers_[1]->service_.request_count());
1303 EXPECT_EQ(3, servers_[2]->service_.request_count());
1304 // Now set one server to be unhealthy again. Then wait until the
1305 // unhealthiness has hit the client. We know that the client will see
1306 // this when we send kNumServers requests and one of the remaining servers
1307 // sees two of the requests.
1308 gpr_log(GPR_INFO, "*** server 0 unhealthy");
1309 servers_[0]->SetServingStatus("health_check_service_name", false);
1312 for (int i = 0; i < kNumServers; ++i) {
1313 CheckRpcSendOk(stub, DEBUG_LOCATION);
1315 } while (servers_[1]->service_.request_count() != 2 &&
1316 servers_[2]->service_.request_count() != 2);
1317 // Now set the remaining two servers to be unhealthy. Make sure the
1318 // channel leaves READY state and that RPCs fail.
1319 gpr_log(GPR_INFO, "*** all servers unhealthy");
1320 servers_[1]->SetServingStatus("health_check_service_name", false);
1321 servers_[2]->SetServingStatus("health_check_service_name", false);
1322 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
1323 CheckRpcSendFailure(stub);
1325 EnableDefaultHealthCheckService(false);
1328 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthCheckingInhibitPerChannel) {
1329 EnableDefaultHealthCheckService(true);
1331 const int kNumServers = 1;
1332 StartServers(kNumServers);
1333 // Create a channel with health-checking enabled.
1334 ChannelArguments args;
1335 args.SetServiceConfigJSON(
1336 "{\"healthCheckConfig\": "
1337 "{\"serviceName\": \"health_check_service_name\"}}");
1338 auto channel1 = BuildChannel("round_robin", args);
1339 auto stub1 = BuildStub(channel1);
1340 std::vector<int> ports = GetServersPorts();
1341 SetNextResolution(ports);
1342 // Create a channel with health checking enabled but inhibited.
1343 args.SetInt(GRPC_ARG_INHIBIT_HEALTH_CHECKING, 1);
1344 auto channel2 = BuildChannel("round_robin", args);
1345 auto stub2 = BuildStub(channel2);
1346 SetNextResolution(ports);
1347 // First channel should not become READY, because health checks should be
1349 EXPECT_FALSE(WaitForChannelReady(channel1.get(), 1));
1350 CheckRpcSendFailure(stub1);
1351 // Second channel should be READY.
1352 EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1));
1353 CheckRpcSendOk(stub2, DEBUG_LOCATION);
1354 // Enable health checks on the backend and wait for channel 1 to succeed.
1355 servers_[0]->SetServingStatus("health_check_service_name", true);
1356 CheckRpcSendOk(stub1, DEBUG_LOCATION, true /* wait_for_ready */);
1357 // Check that we created only one subchannel to the backend.
1358 EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
1360 EnableDefaultHealthCheckService(false);
1363 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthCheckingServiceNamePerChannel) {
1364 EnableDefaultHealthCheckService(true);
1366 const int kNumServers = 1;
1367 StartServers(kNumServers);
1368 // Create a channel with health-checking enabled.
1369 ChannelArguments args;
1370 args.SetServiceConfigJSON(
1371 "{\"healthCheckConfig\": "
1372 "{\"serviceName\": \"health_check_service_name\"}}");
1373 auto channel1 = BuildChannel("round_robin", args);
1374 auto stub1 = BuildStub(channel1);
1375 std::vector<int> ports = GetServersPorts();
1376 SetNextResolution(ports);
1377 // Create a channel with health-checking enabled with a different
1379 ChannelArguments args2;
1380 args2.SetServiceConfigJSON(
1381 "{\"healthCheckConfig\": "
1382 "{\"serviceName\": \"health_check_service_name2\"}}");
1383 auto channel2 = BuildChannel("round_robin", args2);
1384 auto stub2 = BuildStub(channel2);
1385 SetNextResolution(ports);
1386 // Allow health checks from channel 2 to succeed.
1387 servers_[0]->SetServingStatus("health_check_service_name2", true);
1388 // First channel should not become READY, because health checks should be
1390 EXPECT_FALSE(WaitForChannelReady(channel1.get(), 1));
1391 CheckRpcSendFailure(stub1);
1392 // Second channel should be READY.
1393 EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1));
1394 CheckRpcSendOk(stub2, DEBUG_LOCATION);
1395 // Enable health checks for channel 1 and wait for it to succeed.
1396 servers_[0]->SetServingStatus("health_check_service_name", true);
1397 CheckRpcSendOk(stub1, DEBUG_LOCATION, true /* wait_for_ready */);
1398 // Check that we created only one subchannel to the backend.
1399 EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
1401 EnableDefaultHealthCheckService(false);
1404 class ClientLbInterceptTrailingMetadataTest : public ClientLbEnd2endTest {
1406 void SetUp() override {
1407 ClientLbEnd2endTest::SetUp();
1408 grpc_core::RegisterInterceptRecvTrailingMetadataLoadBalancingPolicy(
1409 ReportTrailerIntercepted, this);
1412 void TearDown() override { ClientLbEnd2endTest::TearDown(); }
1414 int trailers_intercepted() {
1415 grpc::internal::MutexLock lock(&mu_);
1416 return trailers_intercepted_;
1420 static void ReportTrailerIntercepted(void* arg) {
1421 ClientLbInterceptTrailingMetadataTest* self =
1422 static_cast<ClientLbInterceptTrailingMetadataTest*>(arg);
1423 grpc::internal::MutexLock lock(&self->mu_);
1424 self->trailers_intercepted_++;
1427 grpc::internal::Mutex mu_;
1428 int trailers_intercepted_ = 0;
1431 TEST_F(ClientLbInterceptTrailingMetadataTest, InterceptsRetriesDisabled) {
1432 const int kNumServers = 1;
1433 const int kNumRpcs = 10;
1434 StartServers(kNumServers);
1435 auto channel = BuildChannel("intercept_trailing_metadata_lb");
1436 auto stub = BuildStub(channel);
1437 SetNextResolution(GetServersPorts());
1438 for (size_t i = 0; i < kNumRpcs; ++i) {
1439 CheckRpcSendOk(stub, DEBUG_LOCATION);
1441 // Check LB policy name for the channel.
1442 EXPECT_EQ("intercept_trailing_metadata_lb",
1443 channel->GetLoadBalancingPolicyName());
1444 EXPECT_EQ(kNumRpcs, trailers_intercepted());
1447 TEST_F(ClientLbInterceptTrailingMetadataTest, InterceptsRetriesEnabled) {
1448 const int kNumServers = 1;
1449 const int kNumRpcs = 10;
1450 StartServers(kNumServers);
1451 ChannelArguments args;
1452 args.SetServiceConfigJSON(
1454 " \"methodConfig\": [ {\n"
1456 " { \"service\": \"grpc.testing.EchoTestService\" }\n"
1458 " \"retryPolicy\": {\n"
1459 " \"maxAttempts\": 3,\n"
1460 " \"initialBackoff\": \"1s\",\n"
1461 " \"maxBackoff\": \"120s\",\n"
1462 " \"backoffMultiplier\": 1.6,\n"
1463 " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
1467 auto channel = BuildChannel("intercept_trailing_metadata_lb", args);
1468 auto stub = BuildStub(channel);
1469 SetNextResolution(GetServersPorts());
1470 for (size_t i = 0; i < kNumRpcs; ++i) {
1471 CheckRpcSendOk(stub, DEBUG_LOCATION);
1473 // Check LB policy name for the channel.
1474 EXPECT_EQ("intercept_trailing_metadata_lb",
1475 channel->GetLoadBalancingPolicyName());
1476 EXPECT_EQ(kNumRpcs, trailers_intercepted());
1480 } // namespace testing
1483 int main(int argc, char** argv) {
1484 // Make the backup poller poll very frequently in order to pick up
1485 // updates from all the subchannels's FDs.
1486 GPR_GLOBAL_CONFIG_SET(grpc_client_channel_backup_poll_interval_ms, 1);
1487 ::testing::InitGoogleTest(&argc, argv);
1488 grpc::testing::TestEnvironment env(argc, argv);
1489 const auto result = RUN_ALL_TESTS();