impl(bigtable): add DynamicChannelPool class

google/cloud/bigtable/CMakeLists.txt

@@ -184,6 +184,7 @@ add_library(
     internal/default_row_reader.h
     internal/defaults.cc
     internal/defaults.h
+    internal/dynamic_channel_pool.h
     internal/endpoint_options.h
     internal/google_bytes_traits.cc
     internal/google_bytes_traits.h
@@ -453,6 +454,7 @@ if (BUILD_TESTING)
         internal/data_tracing_connection_test.cc
         internal/default_row_reader_test.cc
         internal/defaults_test.cc
+        internal/dynamic_channel_pool_test.cc
         internal/google_bytes_traits_test.cc
         internal/logging_result_set_reader_test.cc
         internal/metrics_test.cc

google/cloud/bigtable/bigtable_client_unit_tests.bzl

@@ -52,6 +52,7 @@ bigtable_client_unit_tests = [
     "internal/data_tracing_connection_test.cc",
     "internal/default_row_reader_test.cc",
     "internal/defaults_test.cc",
+    "internal/dynamic_channel_pool_test.cc",
     "internal/google_bytes_traits_test.cc",
     "internal/logging_result_set_reader_test.cc",
     "internal/metrics_test.cc",

google/cloud/bigtable/google_cloud_cpp_bigtable.bzl

@@ -90,6 +90,7 @@ google_cloud_cpp_bigtable_hdrs = [
     "internal/data_tracing_connection.h",
     "internal/default_row_reader.h",
     "internal/defaults.h",
+    "internal/dynamic_channel_pool.h",
     "internal/endpoint_options.h",
     "internal/google_bytes_traits.h",
     "internal/logging_result_set_reader.h",

google/cloud/bigtable/internal/channel_usage.h

@@ -41,15 +41,27 @@ class ChannelUsage : public std::enable_shared_from_this<ChannelUsage<T>> {
                                                      std::make_shared<Clock>())
       : stub_(std::move(stub)), clock_(std::move(clock)) {}
 
+  // This constructor is only used in testing.
+  ChannelUsage(std::shared_ptr<T> stub, std::shared_ptr<Clock> clock,
+               int initial_outstanding_rpcs)
+      : stub_(std::move(stub)),
+        clock_(std::move(clock)),
+        outstanding_rpcs_(initial_outstanding_rpcs) {
+    std::cout << __func__
+              << ": initial_outstanding_rpcs=" << initial_outstanding_rpcs
+              << std::endl;
+  }
+
   // Computes the weighted average of outstanding RPCs on the channel over the
   // past 60 seconds.
   StatusOr<int> average_outstanding_rpcs() {
     auto constexpr kWindowSeconds = 60;
     auto constexpr kWindowDuration = std::chrono::seconds(kWindowSeconds);
     std::scoped_lock lk(mu_);
     if (!last_refresh_status_.ok()) return last_refresh_status_;
-    // If there are no measurements then the stub has never been used.
-    if (measurements_.empty()) return 0;
+    // If there are no measurements then the stub has never been used. In real
+    // use this will be 0. In testing we sometimes set an initial value.
+    if (measurements_.empty()) return outstanding_rpcs_;
     auto now = clock_->Now();
     auto last_time = now;
     auto window_start = now - kWindowDuration;
@@ -114,6 +126,8 @@ class ChannelUsage : public std::enable_shared_from_this<ChannelUsage<T>> {
 
   std::shared_ptr<T> AcquireStub() {
     std::scoped_lock lk(mu_);
+    std::cout << __func__ << ": outstanding_rpcs_=" << outstanding_rpcs_
+              << std::endl;
     ++outstanding_rpcs_;
     auto time = clock_->Now();
     measurements_.emplace_back(outstanding_rpcs_, time);