client_lb_end2end_test: refactor connection delay injection into its …

…own library (grpc#29320)

* refactor connection delay injection from client_lb_end2end_test

* fix build

* fix build on older compilers

* clang-format

* buildifier

src/core/lib/gprpp/time.h

@@ -120,7 +120,7 @@ class Timestamp {
 // Duration represents a span of time.
 class Duration {
  public:
-  constexpr Duration() : millis_(0) {}
+  constexpr Duration() noexcept : millis_(0) {}
 
   static Duration FromTimespec(gpr_timespec t);
   static Duration FromSecondsAndNanoseconds(int64_t seconds, int32_t nanos);

test/cpp/end2end/BUILD

@@ -56,6 +56,16 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "connection_delay_injector",
+    testonly = True,
+    srcs = ["connection_delay_injector.cc"],
+    hdrs = ["connection_delay_injector.h"],
+    deps = [
+        "//:grpc",
+    ],
+)
+
 grpc_cc_library(
     name = "interceptors_util",
     testonly = True,
@@ -478,6 +488,7 @@ grpc_cc_test(
     flaky = True,  # TODO(b/151315347)
     tags = ["no_windows"],  # TODO(jtattermusch): fix test on windows
     deps = [
+        ":connection_delay_injector",
         ":test_service_impl",
         "//:gpr",
         "//:grpc",

test/cpp/end2end/client_lb_end2end_test.cc

@@ -64,40 +64,18 @@
 #include "test/core/util/resolve_localhost_ip46.h"
 #include "test/core/util/test_config.h"
 #include "test/core/util/test_lb_policies.h"
+#include "test/cpp/end2end/connection_delay_injector.h"
 #include "test/cpp/end2end/test_service_impl.h"
 
 using grpc::testing::EchoRequest;
 using grpc::testing::EchoResponse;
 
-// defined in tcp_client.cc
-extern grpc_tcp_client_vtable* grpc_tcp_client_impl;
-
-static grpc_tcp_client_vtable* default_client_impl;
-
 namespace grpc {
 namespace testing {
 namespace {
 
 constexpr char kRequestMessage[] = "Live long and prosper.";
 
-gpr_atm g_connection_delay_ms;
-
-void tcp_client_connect_with_delay(grpc_closure* closure, grpc_endpoint** ep,
-                                   grpc_pollset_set* interested_parties,
-                                   const grpc_channel_args* channel_args,
-                                   const grpc_resolved_address* addr,
-                                   grpc_core::Timestamp deadline) {
-  const int delay_ms = gpr_atm_acq_load(&g_connection_delay_ms);
-  if (delay_ms > 0) {
-    gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(delay_ms));
-  }
-  default_client_impl->connect(
-      closure, ep, interested_parties, channel_args, addr,
-      deadline + grpc_core::Duration::Milliseconds(delay_ms));
-}
-
-grpc_tcp_client_vtable delayed_connect = {tcp_client_connect_with_delay};
-
 // Subclass of TestServiceImpl that increments a request counter for
 // every call to the Echo RPC.
 class MyTestServiceImpl : public TestServiceImpl {
@@ -627,9 +605,9 @@ TEST_F(ClientLbEnd2endTest, PickFirstBackOffMinReconnect) {
   response_generator.SetNextResolution(ports);
   // Make connection delay a 10% longer than it's willing to in order to make
   // sure we are hitting the codepath that waits for the min reconnect backoff.
-  gpr_atm_rel_store(&g_connection_delay_ms, kMinReconnectBackOffMs * 1.10);
-  default_client_impl = grpc_tcp_client_impl;
-  grpc_set_tcp_client_impl(&delayed_connect);
+  ConnectionDelayInjector delay_injector;
+  auto injected_delay = delay_injector.SetDelay(
+      grpc_core::Duration::Milliseconds(kMinReconnectBackOffMs * 1.10));
   const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
   channel->WaitForConnected(
       grpc_timeout_milliseconds_to_deadline(kMinReconnectBackOffMs * 2));
@@ -640,7 +618,6 @@ TEST_F(ClientLbEnd2endTest, PickFirstBackOffMinReconnect) {
   // We should have waited at least kMinReconnectBackOffMs. We substract one to
   // account for test and precision accuracy drift.
   EXPECT_GE(waited.millis(), kMinReconnectBackOffMs - 1);
-  gpr_atm_rel_store(&g_connection_delay_ms, 0);
 }
 
 TEST_F(ClientLbEnd2endTest, PickFirstResetConnectionBackoff) {

test/cpp/end2end/connection_delay_injector.cc

@@ -0,0 +1,116 @@
+// Copyright 2016 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "test/cpp/end2end/connection_delay_injector.h"
+
+#include <atomic>
+#include <memory>
+
+#include "absl/memory/memory.h"
+#include "absl/utility/utility.h"
+
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/gprpp/time.h"
+#include "src/core/lib/iomgr/tcp_client.h"
+#include "src/core/lib/iomgr/timer.h"
+
+// defined in tcp_client.cc
+extern grpc_tcp_client_vtable* grpc_tcp_client_impl;
+
+namespace grpc {
+namespace testing {
+
+namespace {
+
+grpc_tcp_client_vtable* g_original_vtable = nullptr;
+
+std::atomic<grpc_core::Duration> g_delay;
+
+class InjectedDelay {
+ public:
+  InjectedDelay(grpc_closure* closure, grpc_endpoint** ep,
+                grpc_pollset_set* interested_parties,
+                const grpc_channel_args* channel_args,
+                const grpc_resolved_address* addr,
+                grpc_core::Timestamp deadline)
+      : closure_(closure),
+        endpoint_(ep),
+        interested_parties_(interested_parties),
+        channel_args_(grpc_channel_args_copy(channel_args)),
+        deadline_(deadline) {
+    memcpy(&address_, addr, sizeof(grpc_resolved_address));
+    GRPC_CLOSURE_INIT(&timer_callback_, TimerCallback, this, nullptr);
+    auto duration = g_delay.load();
+    deadline_ += duration;
+    grpc_timer_init(&timer_, grpc_core::ExecCtx::Get()->Now() + duration,
+                    &timer_callback_);
+  }
+
+  ~InjectedDelay() { grpc_channel_args_destroy(channel_args_); }
+
+ private:
+  static void TimerCallback(void* arg, grpc_error_handle /*error*/) {
+    auto* self = static_cast<InjectedDelay*>(arg);
+    g_original_vtable->connect(self->closure_, self->endpoint_,
+                               self->interested_parties_, self->channel_args_,
+                               &self->address_, self->deadline_);
+    delete self;
+  }
+
+  grpc_timer timer_;
+  grpc_closure timer_callback_;
+
+  // Original args.
+  grpc_closure* closure_;
+  grpc_endpoint** endpoint_;
+  grpc_pollset_set* interested_parties_;
+  const grpc_channel_args* channel_args_;
+  grpc_resolved_address address_;
+  grpc_core::Timestamp deadline_;
+};
+
+void TcpConnectWithDelay(grpc_closure* closure, grpc_endpoint** ep,
+                         grpc_pollset_set* interested_parties,
+                         const grpc_channel_args* channel_args,
+                         const grpc_resolved_address* addr,
+                         grpc_core::Timestamp deadline) {
+  new InjectedDelay(closure, ep, interested_parties, channel_args, addr,
+                    deadline);
+}
+
+grpc_tcp_client_vtable kDelayedConnectVTable = {TcpConnectWithDelay};
+
+}  // namespace
+
+ConnectionDelayInjector::InjectedDelay::~InjectedDelay() {
+  g_delay.store(grpc_core::Duration());
+}
+
+ConnectionDelayInjector::ConnectionDelayInjector() {
+  g_original_vtable =
+      absl::exchange(grpc_tcp_client_impl, &kDelayedConnectVTable);
+}
+
+ConnectionDelayInjector::~ConnectionDelayInjector() {
+  grpc_tcp_client_impl = g_original_vtable;
+}
+
+std::unique_ptr<ConnectionDelayInjector::InjectedDelay>
+ConnectionDelayInjector::SetDelay(grpc_core::Duration duration) {
+  GPR_ASSERT(g_delay.exchange(duration) == grpc_core::Duration());
+  return absl::make_unique<ConnectionDelayInjector::InjectedDelay>();
+}
+
+}  // namespace testing
+}  // namespace grpc

test/cpp/end2end/connection_delay_injector.h

@@ -0,0 +1,58 @@
+// Copyright 2016 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GRPC_TEST_CPP_END2END_CONNECTION_DELAY_INJECTOR_H
+#define GRPC_TEST_CPP_END2END_CONNECTION_DELAY_INJECTOR_H
+
+#include <memory>
+
+#include "src/core/lib/gprpp/time.h"
+
+namespace grpc {
+namespace testing {
+
+// Allows injecting connection-establishment delays into C-core.
+// Typical usage:
+//
+//  ConnectionDelayInjector delay_injector;
+//  auto scoped_delay =
+//      delay_injector.SetDelay(grpc_core::Duration::Seconds(10));
+//
+// When ConnectionDelayInjector is instantiated, it replaces the iomgr
+// TCP client vtable, and it sets it back to the original value when it
+// is destroyed.
+//
+// When SetDelay() is called, it sets the global delay, which will
+// automatically be unset when the result goes out of scope.
+//
+// The injection is global, so there must be only one ConnectionDelayInjector
+// object at any one time, and there must be only one scoped delay in effect
+// at any one time.
+class ConnectionDelayInjector {
+ public:
+  class InjectedDelay {
+   public:
+    ~InjectedDelay();
+  };
+
+  ConnectionDelayInjector();
+  ~ConnectionDelayInjector();
+
+  std::unique_ptr<InjectedDelay> SetDelay(grpc_core::Duration duration);
+};
+
+}  // namespace testing
+}  // namespace grpc
+
+#endif  // GRPC_TEST_CPP_END2END_CONNECTION_DELAY_INJECTOR_H