impl(bigtable): add option to select directpath initialization mode

google/cloud/bigtable/data_connection.cc

@@ -209,30 +209,34 @@ std::shared_ptr<bigtable::DataConnection> MakeInstanceAffinityDataConnection(
       std::move(components.limiter), std::move(components.options));
 }
 
-std::shared_ptr<bigtable::DataConnection> MakeDirectPathDataConnection(
-    DataConnectionComponents components,
-    std::vector<bigtable::InstanceResource> const& instances) {
-  // Step 1: Configure separate Options for DirectPath and CloudPath (fallback).
-  Options directpath_options = components.options;
-  directpath_options
-      .set<::google::cloud::bigtable_internal::DataEndpointOption>(
-          bigtable::internal::DefaultDirectPathDataEndpoint());
-  directpath_options.set<EndpointOption>(
+Options MakeDirectPathOptions(Options options) {
+  options.set<::google::cloud::bigtable_internal::DataEndpointOption>(
+      bigtable::internal::DefaultDirectPathDataEndpoint());
+  options.set<EndpointOption>(
       bigtable::internal::DefaultDirectPathDataEndpoint());
-  directpath_options.set<AuthorityOption>(
+  options.set<AuthorityOption>(
       bigtable::internal::DefaultDirectPathAuthority());
-  directpath_options.set<bigtable::experimental::DirectPathModeOption>(
+  options.set<bigtable::experimental::DirectPathModeOption>(
       bigtable::experimental::DirectPathMode::kEnabled);
+  return options;
+}
 
-  Options cloudpath_options = components.options;
-  cloudpath_options.set<::google::cloud::bigtable_internal::DataEndpointOption>(
+Options MakeCloudPathOptions(Options options) {
+  options.set<::google::cloud::bigtable_internal::DataEndpointOption>(
       "bigtable.googleapis.com");
-  cloudpath_options.set<EndpointOption>("bigtable.googleapis.com");
-  cloudpath_options.set<AuthorityOption>("bigtable.googleapis.com");
-  cloudpath_options.set<bigtable::experimental::DirectPathModeOption>(
+  options.set<EndpointOption>("bigtable.googleapis.com");
+  options.set<AuthorityOption>("bigtable.googleapis.com");
+  options.set<bigtable::experimental::DirectPathModeOption>(
       bigtable::experimental::DirectPathMode::kDisabled);
+  return options;
+}
 
-  // Step 2: Concurrently create StubManagers for both DirectPath and CloudPath
+std::shared_ptr<bigtable::DataConnection>
+MakeSpeculativeDirectPathDataConnection(
+    DataConnectionComponents components,
+    std::vector<bigtable::InstanceResource> const& instances,
+    Options directpath_options, Options cloudpath_options) {
+  // Step 1: Concurrently create StubManagers for both DirectPath and CloudPath
   // on background threads so that stubs are warming up while probing proceeds.
   promise<std::unique_ptr<bigtable_internal::StubManager>> directpath_promise;
   future<std::unique_ptr<bigtable_internal::StubManager>> directpath_future =
@@ -279,14 +283,14 @@ std::shared_ptr<bigtable::DataConnection> MakeDirectPathDataConnection(
             std::move(affinity_stubs), stub_creation_fn));
       });
 
-  // Step 3: Probe DirectPath connectivity and ALTS negotiation on the primary
+  // Step 2: Probe DirectPath connectivity and ALTS negotiation on the primary
   // instance.
   StatusOr<bigtable_internal::DirectPathProbeResult> const probe_result =
       bigtable_internal::DirectPathProber::Probe(
           components.auth, instances.front(), directpath_options,
           components.background->cq());
 
-  // Step 4a: If DirectPath probing succeeds, use DirectPath stubs, discard the
+  // Step 3a: If DirectPath probing succeeds, use DirectPath stubs, discard the
   // fallback CloudPath future asynchronously, and record successful metrics.
   if (probe_result.ok() && probe_result->success) {
     std::unique_ptr<bigtable_internal::StubManager> stub_manager =
@@ -309,7 +313,7 @@ std::shared_ptr<bigtable::DataConnection> MakeDirectPathDataConnection(
         std::move(components.limiter), std::move(directpath_options));
   }
 
-  // Step 4b: If DirectPath probing fails, fall back to the CloudPath
+  // Step 3b: If DirectPath probing fails, fall back to the CloudPath
   // StubManager, discard the unused DirectPath future asynchronously, and run
   // environmental diagnostics asynchronously to diagnose and record metrics for
   // the failure reason.
@@ -330,6 +334,88 @@ std::shared_ptr<bigtable::DataConnection> MakeDirectPathDataConnection(
       std::move(components.limiter), std::move(cloudpath_options));
 }
 
+std::shared_ptr<bigtable::DataConnection> MakeBlockingDirectPathDataConnection(
+    DataConnectionComponents components,
+    std::vector<bigtable::InstanceResource> const& instances,
+    Options directpath_options, Options cloudpath_options) {
+  // Step 1: Probe DirectPath connectivity and ALTS negotiation on the primary
+  // instance first.
+  StatusOr<bigtable_internal::DirectPathProbeResult> const probe_result =
+      bigtable_internal::DirectPathProber::Probe(
+          components.auth, instances.front(), directpath_options,
+          components.background->cq());
+
+  auto create_stub_manager = [&](Options const& options) {
+    auto stub_creation_fn =
+        [auth = components.auth, cq = components.background->cq(), options](
+            std::string_view instance_name,
+            bigtable_internal::StubManager::Priming priming) {
+          return bigtable_internal::CreateBigtableStub(auth, cq, instance_name,
+                                                       priming, options);
+        };
+    absl::flat_hash_map<std::string,
+                        std::shared_ptr<bigtable_internal::BigtableStub>>
+        affinity_stubs = bigtable_internal::CreateBigtableAffinityStubs(
+            instances, stub_creation_fn);
+    return std::make_unique<bigtable_internal::StubManager>(
+        std::move(affinity_stubs), std::move(stub_creation_fn));
+  };
+
+  // Step 2a: If DirectPath probing succeeds, create the DirectPath StubManager
+  // and record successful metrics.
+  if (probe_result.ok() && probe_result->success) {
+    auto stub_manager = create_stub_manager(directpath_options);
+
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+    if (components.direct_access_compatibility != nullptr) {
+      components.direct_access_compatibility->Record(
+          opentelemetry::context::RuntimeContext::GetCurrent(), 1,
+          bigtable_internal::DirectAccessCompatibilityLabels{
+              bigtable_internal::ToString(probe_result->ip_preference), ""});
+    }
+#endif
+    return std::make_shared<bigtable_internal::DataConnectionImpl>(
+        std::move(components.background), std::move(stub_manager),
+        std::move(components.operation_context_factory),
+        std::move(components.limiter), std::move(directpath_options));
+  }
+
+  // Step 2b: If DirectPath probing fails, create the CloudPath StubManager,
+  // and run environmental diagnostics asynchronously.
+  auto stub_manager = create_stub_manager(cloudpath_options);
+
+#ifdef GOOGLE_CLOUD_CPP_BIGTABLE_WITH_OTEL_METRICS
+  bigtable_internal::DirectPathDiagnostics::RunAsync(
+      components.background->cq(), directpath_options,
+      components.direct_access_compatibility);
+#endif
+  return std::make_shared<bigtable_internal::DataConnectionImpl>(
+      std::move(components.background), std::move(stub_manager),
+      std::move(components.operation_context_factory),
+      std::move(components.limiter), std::move(cloudpath_options));
+}
+
+std::shared_ptr<bigtable::DataConnection> MakeDirectPathDataConnection(
+    DataConnectionComponents components,
+    std::vector<bigtable::InstanceResource> const& instances) {
+  experimental::DirectPathInitializationMode const initialization_mode =
+      components.options
+          .get<bigtable::experimental::DirectPathInitializationModeOption>();
+  Options directpath_options = MakeDirectPathOptions(components.options);
+  Options cloudpath_options =
+      MakeCloudPathOptions(std::move(components.options));
+
+  if (initialization_mode ==
+      bigtable::experimental::DirectPathInitializationMode::kBlocking) {
+    return MakeBlockingDirectPathDataConnection(
+        std::move(components), instances, std::move(directpath_options),
+        std::move(cloudpath_options));
+  }
+  return MakeSpeculativeDirectPathDataConnection(
+      std::move(components), instances, std::move(directpath_options),
+      std::move(cloudpath_options));
+}
+
 std::shared_ptr<bigtable::DataConnection> WrapDataTracing(
     std::shared_ptr<bigtable::DataConnection> conn) {
   if (google::cloud::internal::TracingEnabled(conn->options())) {

google/cloud/bigtable/data_connection_test.cc

@@ -149,6 +149,30 @@ TEST(MakeDataConnection, DirectPathModeOptionDisabledNoGrpcMetrics) {
                             experimental::DirectPathMetricsMode::kEnabled));
   EXPECT_NE(conn, nullptr);
 }
+
+TEST(MakeDataConnection, DirectPathInitializationModeBlocking) {
+  InstanceResource instance_a{Project("my-project"), "instance-a"};
+  auto conn = MakeDataConnection(
+      {instance_a},
+      TestOptions()
+          .set<experimental::DirectPathModeOption>(
+              experimental::DirectPathMode::kEnabled)
+          .set<experimental::DirectPathInitializationModeOption>(
+              experimental::DirectPathInitializationMode::kBlocking));
+  EXPECT_NE(conn, nullptr);
+}
+
+TEST(MakeDataConnection, DirectPathInitializationModeSpeculative) {
+  InstanceResource instance_a{Project("my-project"), "instance-a"};
+  auto conn = MakeDataConnection(
+      {instance_a}, TestOptions()
+                        .set<experimental::DirectPathModeOption>(
+                            experimental::DirectPathMode::kEnabled)
+                        .set<experimental::DirectPathInitializationModeOption>(
+                            experimental::DirectPathInitializationMode::
+                                kAsynchronousSpeculative));
+  EXPECT_NE(conn, nullptr);
+}
 #endif
 
 }  // namespace

google/cloud/bigtable/internal/defaults.cc

@@ -260,6 +260,10 @@ Options DefaultOptions(Options opts) {
     opts.set<experimental::DirectPathDiagnosticsTimeoutOption>(
         DefaultDirectPathDiagnosticsTimeout());
   }
+  if (!opts.has<experimental::DirectPathInitializationModeOption>()) {
+    opts.set<experimental::DirectPathInitializationModeOption>(
+        DefaultDirectPathInitializationMode());
+  }
 
   return opts;
 }

google/cloud/bigtable/internal/defaults.h

@@ -52,6 +52,11 @@ constexpr std::chrono::milliseconds DefaultDirectPathDiagnosticsTimeout() {
   return std::chrono::seconds(60);
 }
 
+constexpr experimental::DirectPathInitializationMode
+DefaultDirectPathInitializationMode() {
+  return experimental::DirectPathInitializationMode::kBlocking;
+}
+
 /**
  * Returns true if Direct Path is enabled for Bigtable.
  */

google/cloud/bigtable/internal/defaults_test.cc

@@ -636,6 +636,8 @@ TEST(EndpointEnvTest, DirectPathTimeoutDefaults) {
               Eq(DefaultDirectPathProbeTimeout()));
   EXPECT_THAT(opts.get<experimental::DirectPathDiagnosticsTimeoutOption>(),
               Eq(DefaultDirectPathDiagnosticsTimeout()));
+  EXPECT_THAT(opts.get<experimental::DirectPathInitializationModeOption>(),
+              Eq(DefaultDirectPathInitializationMode()));
 }
 
 TEST(EndpointEnvTest, EmulatorOverridesCloudDirectPath) {

google/cloud/bigtable/options.h

@@ -265,6 +265,31 @@ struct DirectPathDiagnosticsTimeoutOption {
   using Type = std::chrono::milliseconds;
 };
 
+/**
+ * Option to configure the initialization behavior for DirectPath.
+ *
+ * - kBlocking - Channel Pool initialization begins only after
+ *   DirectPath or CloudPath is determined.
+ * - kAsynchronousSpeculative - Both DirectPath and CloudPath channel pools
+ *   are initialized on background threads while the probe runs. The unused
+ *   channel pool is discarded.
+ *
+ * @ingroup google-cloud-bigtable-options
+ */
+enum class DirectPathInitializationMode {
+  kBlocking,  // Default.
+  kAsynchronousSpeculative,
+};
+
+/**
+ * Option to configure the initialization behavior for DirectPath.
+ *
+ * @ingroup google-cloud-bigtable-options
+ */
+struct DirectPathInitializationModeOption {
+  using Type = DirectPathInitializationMode;
+};
+
 }  // namespace experimental
 
 /// The complete list of options accepted by `bigtable::*Client`
@@ -355,7 +380,8 @@ using DataPolicyOptionList =
     OptionList<DataRetryPolicyOption, DataBackoffPolicyOption, DeadlineOption,
                IdempotentMutationPolicyOption, EnableMetricsOption,
                MetricsPeriodOption,
-               experimental::DynamicChannelPoolSizingPolicyOption>;
+               experimental::DynamicChannelPoolSizingPolicyOption,
+               experimental::DirectPathInitializationModeOption>;
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace bigtable