feat(storage): separate read and hedging thread pools

- Extract lazy, dynamically scaling ThreadPool primitive from HedgingThreadPool.
- Separate StorageConnectionImpl thread pool into a dedicated ReadThreadPool (for primary stream opens) and a HedgingThreadPool (for speculative secondary hedges).
- Add ReadThreadPoolSizeOption and HedgingThreadPoolSizeOption with auto-scaling defaults to prevent read bottlenecking under high concurrency.
- Extract DefaultReadThreadPoolSize() and DefaultHedgingThreadPoolSize() helpers to share sizing logic between DefaultOptions() and connection initialization.
- Enqueue primary read attempt to ReadThreadPool and speculative hedge attempts to HedgingThreadPool, ensuring complete fault and stall isolation.
- Clamp ThreadPool capacity to at least 1 to prevent deadlock on zero sizing.
- Add unit tests verifying thread pool execution, default sizes, zero-size handling, lazy spawning, and pool isolation under saturation.

Author: ajayky-os

google/cloud/storage/client.cc

@@ -17,6 +17,7 @@
 #include "google/cloud/storage/idempotency_policy.h"
 #include "google/cloud/storage/internal/base64.h"
 #include "google/cloud/storage/internal/connection_factory.h"
+#include "google/cloud/storage/internal/hedging_thread_pool.h"
 #include "google/cloud/storage/options.h"
 #include "google/cloud/internal/curl_handle.h"
 #include "google/cloud/internal/curl_options.h"
@@ -600,6 +601,15 @@ Options DefaultOptions(Options opts) {
   if (!o.has<storage_experimental::MaxReadHedgesOption>()) {
     o.set<storage_experimental::MaxReadHedgesOption>(2);
   }
+  if (!o.has<storage_experimental::ReadThreadPoolSizeOption>()) {
+    o.set<storage_experimental::ReadThreadPoolSizeOption>(
+        internal::DefaultReadThreadPoolSize());
+  }
+  if (!o.has<storage_experimental::HedgingThreadPoolSizeOption>()) {
+    o.set<storage_experimental::HedgingThreadPoolSizeOption>(
+        internal::DefaultHedgingThreadPoolSize(
+            o.get<storage_experimental::MaxConcurrentHedgesOption>()));
+  }
 
   auto logging = GetEnv("CLOUD_STORAGE_ENABLE_TRACING");
   if (logging) {

google/cloud/storage/internal/connection_impl.cc

@@ -159,23 +159,29 @@ StorageConnectionImpl::StorageConnectionImpl(
     : stub_(std::move(stub)),
       options_(MergeOptions(std::move(options), stub_->options())) {
   if (options_.get<storage_experimental::EnableReadHedgingOption>()) {
-    // The pool only runs stream-open attempts: one primary and (at most) a few
-    // hedges per stream being opened. Size it to the number of connections the
-    // REST layer can use, falling back to the hardware concurrency when the
-    // connection pool is unbounded (`ConnectionPoolSizeOption == 0`).
-    auto pool_size = options_.get<ConnectionPoolSizeOption>();
-    if (pool_size == 0) {
-      pool_size =
-          (std::max<std::size_t>)(4, std::thread::hardware_concurrency());
+    // `DefaultOptions()` normally resolves these, but a connection can be
+    // built without it, in which case the option is left at 0 ("automatic").
+    // A pool sized 0 would accept reads it never runs, hanging the caller.
+    std::size_t read_threads =
+        options_.get<storage_experimental::ReadThreadPoolSizeOption>();
+    if (read_threads == 0) read_threads = DefaultReadThreadPoolSize();
+    // The read pool only ever has one thread per in-flight application read,
+    // and each one blocks inside a synchronous read. Once it saturates, new
+    // primaries queue behind blocked ones and reads degrade to hedge-only.
+    read_pool_ = std::make_shared<ThreadPool>(read_threads);
+
+    std::int64_t const max_concurrent =
+        options_.get<storage_experimental::MaxConcurrentHedgesOption>();
+    std::size_t hedge_threads =
+        options_.get<storage_experimental::HedgingThreadPoolSizeOption>();
+    if (hedge_threads == 0) {
+      hedge_threads = DefaultHedgingThreadPoolSize(max_concurrent);
     }
-    auto const max_threads = 2 * pool_size;
-    auto const rate_limit =
+    double const rate_limit =
         options_.get<storage_experimental::ReadHedgeRateLimitOption>();
-    auto const max_concurrent =
-        options_.get<storage_experimental::MaxConcurrentHedgesOption>();
     // Allow bursts of up to one second worth of hedges.
     hedge_pool_ = std::make_shared<HedgingThreadPool>(
-        max_threads, rate_limit, rate_limit, max_concurrent);
+        hedge_threads, rate_limit, rate_limit, max_concurrent);
   }
 }
 
@@ -435,14 +441,14 @@ StatusOr<std::unique_ptr<ObjectReadSource>> StorageConnectionImpl::ReadObject(
   auto const max_buffer =
       current->get<storage_experimental::MaximumHedgeBufferOption>();
 
-  if (!enable_hedging || max_hedges <= 0 || !hedge_pool_) {
+  if (!enable_hedging || max_hedges <= 0 || !hedge_pool_ || !read_pool_) {
     return retry_source_factory();
   }
 
   // `max_buffer` bounds the size of an individual read, which is only known
   // when the application calls `Read()`; the source applies it there.
   return std::unique_ptr<ObjectReadSource>(
-      std::make_unique<HedgedObjectReadSource>(hedge_pool_,
+      std::make_unique<HedgedObjectReadSource>(read_pool_, hedge_pool_,
                                                std::move(retry_source_factory),
                                                delay, max_hedges, max_buffer));
 }

google/cloud/storage/internal/connection_impl.h

@@ -188,6 +188,7 @@ class StorageConnectionImpl
 
   std::unique_ptr<storage_internal::GenericStub> stub_;
   Options options_;
+  std::shared_ptr<ThreadPool> read_pool_;
   std::shared_ptr<HedgingThreadPool> hedge_pool_;
   google::cloud::internal::InvocationIdGenerator invocation_id_generator_;
 };

google/cloud/storage/internal/hedged_object_read_source.cc

@@ -55,7 +55,7 @@ void RunAttempt(std::shared_ptr<RaceState> const& state,
   auto source = factory();
   if (!source) {
     if (!resolve_on_open_error) return;
-    auto expected = false;
+    bool expected = false;
     if (state->resolved.compare_exchange_strong(expected, true)) {
       state->promise.set_value(
           RaceResult{std::move(source).status(), nullptr, {}});
@@ -65,7 +65,7 @@ void RunAttempt(std::shared_ptr<RaceState> const& state,
   std::unique_ptr<char[]> buffer(new (std::nothrow) char[n]);
   if (!buffer) {
     if (!resolve_on_open_error) return;
-    auto expected = false;
+    bool expected = false;
     if (state->resolved.compare_exchange_strong(expected, true)) {
       state->promise.set_value(RaceResult{
           google::cloud::internal::ResourceExhaustedError(
@@ -76,7 +76,7 @@ void RunAttempt(std::shared_ptr<RaceState> const& state,
     return;
   }
   auto result = (*source)->Read(buffer.get(), n);
-  auto expected = false;
+  bool expected = false;
   if (state->resolved.compare_exchange_strong(expected, true)) {
     state->promise.set_value(
         RaceResult{std::move(result), *std::move(source), std::move(buffer)});
@@ -88,9 +88,11 @@ void RunAttempt(std::shared_ptr<RaceState> const& state,
 }  // namespace
 
 HedgedObjectReadSource::HedgedObjectReadSource(
+    std::shared_ptr<ThreadPool> read_pool,
     std::shared_ptr<HedgingThreadPool> hedge_pool, ChildFactory child_factory,
     std::chrono::milliseconds delay, int max_hedges, std::size_t max_buffer)
-    : hedge_pool_(std::move(hedge_pool)),
+    : read_pool_(std::move(read_pool)),
+      hedge_pool_(std::move(hedge_pool)),
       child_factory_(std::move(child_factory)),
       delay_(delay),
       max_hedges_(max_hedges),
@@ -135,9 +137,10 @@ StatusOr<ReadSourceResult> HedgedObjectReadSource::Read(char* buf,
   auto primary = [state, factory = child_factory_, n] {
     RunAttempt(state, factory, n, /*resolve_on_open_error=*/true, nullptr);
   };
+  // The primary attempt is scheduled on the dedicated read pool.
   // If the pool is shutting down run the attempt inline, the read must
   // complete either way.
-  if (!hedge_pool_->Enqueue(primary)) primary();
+  if (!read_pool_->Enqueue(primary)) primary();
 
   for (int i = 0; i != max_hedges_; ++i) {
     if (future.wait_for(delay_) != std::future_status::timeout) break;

google/cloud/storage/internal/hedged_object_read_source.h

@@ -54,7 +54,8 @@ class HedgedObjectReadSource : public ObjectReadSource {
   using ChildFactory =
       std::function<StatusOr<std::unique_ptr<ObjectReadSource>>()>;
 
-  HedgedObjectReadSource(std::shared_ptr<HedgingThreadPool> hedge_pool,
+  HedgedObjectReadSource(std::shared_ptr<ThreadPool> read_pool,
+                         std::shared_ptr<HedgingThreadPool> hedge_pool,
                          ChildFactory child_factory,
                          std::chrono::milliseconds delay, int max_hedges,
                          std::size_t max_buffer);
@@ -66,6 +67,7 @@ class HedgedObjectReadSource : public ObjectReadSource {
   StatusOr<ReadSourceResult> Read(char* buf, std::size_t n) override;
 
  private:
+  std::shared_ptr<ThreadPool> read_pool_;
   std::shared_ptr<HedgingThreadPool> hedge_pool_;
   ChildFactory child_factory_;
   std::chrono::milliseconds delay_;