using completetion_queue to run background threads

Author: bajajneha27

google/cloud/storage/internal/tracing_connection.cc

@@ -17,7 +17,11 @@
 #include "google/cloud/storage/options.h"
 #include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/internal/opentelemetry.h"
+#include "google/cloud/internal/rest_pure_background_threads_impl.h"
 #include "google/cloud/options.h"
+#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
+#include "google/cloud/grpc_options.h"
+#endif
 #include <algorithm>
 #include <memory>
 #include <string>
@@ -29,15 +33,27 @@ namespace cloud {
 namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
+namespace {
+std::size_t DefaultThreadPoolSize(Options const& options) {
+#if GOOGLE_CLOUD_CPP_STORAGE_HAVE_GRPC
+  auto pool_size = options.get<GrpcBackgroundThreadPoolSizeOption>();
+  if (pool_size == 0) return 1U;
+  return pool_size;
+#else
+  (void)options;
+  return 1U;
+#endif
+}
+}  // namespace
+
 TracingConnection::TracingConnection(std::shared_ptr<StorageConnection> impl)
-    : impl_(std::move(impl)) {}
+    : impl_(std::move(impl)),
+      background_threads_(
+          std::make_unique<
+              rest_internal::AutomaticallyCreatedRestPureBackgroundThreads>(
+              DefaultThreadPoolSize(impl_->options()))) {}
 
-TracingConnection::~TracingConnection() {
-  std::lock_guard<std::mutex> lk(mu_);
-  for (auto& f : bg_tasks_) {
-    if (f.valid()) f.wait();
-  }
-}
+TracingConnection::~TracingConnection() = default;
 
 BucketMetadataCache& TracingConnection::cache() {
   static BucketMetadataCache instance(10000);
@@ -48,16 +64,6 @@ void TracingConnection::ResetCacheForTesting() { cache().Clear(); }
 
 Options TracingConnection::options() const { return impl_->options(); }
 
-void TracingConnection::CleanupCompletedTasks() {
-  std::unique_lock<std::mutex> lk(mu_);
-  bg_tasks_.erase(std::remove_if(bg_tasks_.begin(), bg_tasks_.end(),
-                                 [](std::future<void> const& f) {
-                                   return f.wait_for(std::chrono::seconds(0)) ==
-                                          std::future_status::ready;
-                                 }),
-                  bg_tasks_.end());
-}
-
 void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
                                    BucketCacheEntry const& entry) {
   span.SetAttribute("gcp.resource.destination.id", entry.id);
@@ -66,15 +72,12 @@ void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
 
 void TracingConnection::MaybeTriggerBackgroundFetch(
     std::string const& bucket_name) {
-  CleanupCompletedTasks();
-
   if (!cache().StartFetch(bucket_name)) {
     return;
   }
 
   auto current_options = google::cloud::internal::SaveCurrentOptions();
-  auto f = std::async(std::launch::async, [this, bucket_name,
-                                           current_options]() {
+  background_threads_->cq().RunAsync([this, bucket_name, current_options]() {
     google::cloud::internal::OptionsSpan span(current_options);
     storage::internal::GetBucketMetadataRequest request(bucket_name);
     auto result = impl_->GetBucketMetadata(request);
@@ -87,11 +90,6 @@ void TracingConnection::MaybeTriggerBackgroundFetch(
 
     cache().EndFetch(bucket_name);
   });
-
-  {
-    std::lock_guard<std::mutex> lk(mu_);
-    bg_tasks_.push_back(std::move(f));
-  }
 }
 
 void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
@@ -108,8 +106,9 @@ void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
   }
 }
 
-void TracingConnection::EnrichSpan(opentelemetry::trace::Span& span,
-                                   storage::BucketMetadata const& metadata) const {
+void TracingConnection::EnrichSpan(
+    opentelemetry::trace::Span& span,
+    storage::BucketMetadata const& metadata) const {
   auto const enabled =
       options().get<storage_experimental::OTelSpanEnrichmentOption>();
   if (!enabled) return;

google/cloud/storage/internal/tracing_connection.h

@@ -19,11 +19,9 @@
 #include "google/cloud/storage/internal/storage_connection.h"
 #include "google/cloud/storage/parallel_upload.h"
 #include "google/cloud/storage/version.h"
-#include <future>
+#include "google/cloud/internal/rest_pure_background_threads_impl.h"
 #include <memory>
-#include <mutex>
 #include <string>
-#include <vector>
 
 namespace google {
 namespace cloud {
@@ -190,7 +188,6 @@ class TracingConnection : public storage::internal::StorageConnection {
   static void EnrichSpan(opentelemetry::trace::Span& span,
                          BucketCacheEntry const& entry);
   void MaybeTriggerBackgroundFetch(std::string const& bucket_name);
-  void CleanupCompletedTasks();
 
   static void MaybeInvalidate(Status const& status,
                               std::string const& bucket_name) {
@@ -208,8 +205,8 @@ class TracingConnection : public storage::internal::StorageConnection {
   static BucketMetadataCache& cache();
 
   std::shared_ptr<StorageConnection> impl_;
-  std::mutex mu_;
-  std::vector<std::future<void>> bg_tasks_;
+  std::unique_ptr<google::cloud::rest_internal::RestPureBackgroundThreads>
+      background_threads_;
 };
 
 std::shared_ptr<storage::internal::StorageConnection> MakeTracingClient(

google/cloud/storage/internal/tracing_connection_test.cc

@@ -57,7 +57,8 @@ TEST(TracingClientTest, Options) {
   };
 
   auto mock = std::make_shared<MockClient>();
-  EXPECT_CALL(*mock, options).WillOnce(Return(Options{}.set<TestOption>(42)));
+  EXPECT_CALL(*mock, options)
+      .WillRepeatedly(Return(Options{}.set<TestOption>(42)));
   auto under_test = TracingConnection(mock);
   auto const options = under_test.options();
   EXPECT_EQ(42, options.get<TestOption>());