Copy BackgroundThreads from google::cloud::spanner, planning to

later remove the class from `g-c-cpp-spanner`.

google/cloud/CMakeLists.txt

@@ -316,6 +316,7 @@ if (GOOGLE_CLOUD_CPP_ENABLE_GRPC_UTILS)
     add_library(
         google_cloud_cpp_grpc_utils
         async_operation.h
+        background_threads.h
         completion_queue.cc
         completion_queue.h
         grpc_error_delegate.cc
@@ -325,6 +326,8 @@ if (GOOGLE_CLOUD_CPP_ENABLE_GRPC_UTILS)
         grpc_utils/grpc_error_delegate.h
         grpc_utils/version.h
         internal/async_read_stream_impl.h
+        internal/background_threads_impl.cc
+        internal/background_threads_impl.h
         internal/completion_queue_impl.cc
         internal/completion_queue_impl.h)
     target_link_libraries(
@@ -352,7 +355,8 @@ if (GOOGLE_CLOUD_CPP_ENABLE_GRPC_UTILS)
         # List the unit tests, then setup the targets and dependencies.
         set(google_cloud_cpp_grpc_utils_unit_tests
             # cmake-format: sort
-            completion_queue_test.cc grpc_error_delegate_test.cc)
+            completion_queue_test.cc grpc_error_delegate_test.cc
+            internal/background_threads_impl_test.cc)
 
         # Export the list of unit tests so the Bazel BUILD file can pick it up.
         export_list_to_bazel("google_cloud_cpp_grpc_utils_unit_tests.bzl"

google/cloud/background_threads.h

@@ -0,0 +1,39 @@
+// Copyright 2020 Google LLC
+//
+// 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 GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BACKGROUND_THREADS_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BACKGROUND_THREADS_H
+
+#include "google/cloud/completion_queue.h"
+#include "google/cloud/version.h"
+
+namespace google {
+namespace cloud {
+inline namespace GOOGLE_CLOUD_CPP_NS {
+/**
+ * A object representing the background threads available to a Client.
+ */
+class BackgroundThreads {
+ public:
+  virtual ~BackgroundThreads() = default;
+
+  /// The completion queue used for the background operations.
+  virtual CompletionQueue cq() const = 0;
+};
+
+}  // namespace GOOGLE_CLOUD_CPP_NS
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_BACKGROUND_THREADS_H

google/cloud/google_cloud_cpp_grpc_utils.bzl

@@ -18,18 +18,21 @@
 
 google_cloud_cpp_grpc_utils_hdrs = [
     "async_operation.h",
+    "background_threads.h",
     "completion_queue.h",
     "grpc_error_delegate.h",
     "grpc_utils/async_operation.h",
     "grpc_utils/completion_queue.h",
     "grpc_utils/grpc_error_delegate.h",
     "grpc_utils/version.h",
     "internal/async_read_stream_impl.h",
+    "internal/background_threads_impl.h",
     "internal/completion_queue_impl.h",
 ]
 
 google_cloud_cpp_grpc_utils_srcs = [
     "completion_queue.cc",
     "grpc_error_delegate.cc",
+    "internal/background_threads_impl.cc",
     "internal/completion_queue_impl.cc",
 ]

google/cloud/google_cloud_cpp_grpc_utils_unit_tests.bzl

@@ -19,4 +19,5 @@
 google_cloud_cpp_grpc_utils_unit_tests = [
     "completion_queue_test.cc",
     "grpc_error_delegate_test.cc",
+    "internal/background_threads_impl_test.cc",
 ]

google/cloud/internal/background_threads_impl.cc

@@ -0,0 +1,38 @@
+// Copyright 2020 Google LLC
+//
+// 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 "google/cloud/internal/background_threads_impl.h"
+
+namespace google {
+namespace cloud {
+inline namespace GOOGLE_CLOUD_CPP_NS {
+namespace internal {
+
+AutomaticallyCreatedBackgroundThreads::AutomaticallyCreatedBackgroundThreads()
+    : runner_([](CompletionQueue cq) { cq.Run(); }, cq_) {}
+
+AutomaticallyCreatedBackgroundThreads::
+    ~AutomaticallyCreatedBackgroundThreads() {
+  Shutdown();
+}
+
+void AutomaticallyCreatedBackgroundThreads::Shutdown() {
+  cq_.Shutdown();
+  if (runner_.joinable()) runner_.join();
+}
+
+}  // namespace internal
+}  // namespace GOOGLE_CLOUD_CPP_NS
+}  // namespace cloud
+}  // namespace google

google/cloud/internal/background_threads_impl.h

@@ -0,0 +1,59 @@
+// Copyright 2020 Google LLC
+//
+// 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 GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_BACKGROUND_THREADS_IMPL_H
+#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_BACKGROUND_THREADS_IMPL_H
+
+#include "google/cloud/background_threads.h"
+#include "google/cloud/completion_queue.h"
+#include <thread>
+
+namespace google {
+namespace cloud {
+inline namespace GOOGLE_CLOUD_CPP_NS {
+namespace internal {
+
+/// Assume the user has provided the background threads and use them.
+class CustomerSuppliedBackgroundThreads : public BackgroundThreads {
+ public:
+  explicit CustomerSuppliedBackgroundThreads(CompletionQueue cq)
+      : cq_(std::move(cq)) {}
+  ~CustomerSuppliedBackgroundThreads() override = default;
+
+  CompletionQueue cq() const override { return cq_; }
+
+ private:
+  CompletionQueue cq_;
+};
+
+/// Create a background thread to perform background operations.
+class AutomaticallyCreatedBackgroundThreads : public BackgroundThreads {
+ public:
+  AutomaticallyCreatedBackgroundThreads();
+  ~AutomaticallyCreatedBackgroundThreads() override;
+
+  CompletionQueue cq() const override { return cq_; }
+  void Shutdown();
+
+ private:
+  CompletionQueue cq_;
+  std::thread runner_;
+};
+
+}  // namespace internal
+}  // namespace GOOGLE_CLOUD_CPP_NS
+}  // namespace cloud
+}  // namespace google
+
+#endif  // GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_BACKGROUND_THREADS_IMPL_H

google/cloud/internal/background_threads_impl_test.cc

@@ -0,0 +1,86 @@
+// Copyright 2019 Google LLC
+//
+// 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 "google/cloud/internal/background_threads_impl.h"
+#include <gmock/gmock.h>
+
+namespace google {
+namespace cloud {
+inline namespace GOOGLE_CLOUD_CPP_NS {
+namespace internal {
+namespace {
+
+/// @test Verify we can create and use a CustomerSuppliedBackgroundThreads
+/// without impacting the completion queue
+TEST(CustomerSuppliedBackgroundThreads, LifecycleNoShutdown) {
+  CompletionQueue cq;
+  promise<void> p;
+  std::thread t([&cq, &p] {
+    cq.Run();
+    p.set_value();
+  });
+
+  { CustomerSuppliedBackgroundThreads actual(cq); }
+
+  using ms = std::chrono::milliseconds;
+
+  auto has_shutdown = p.get_future();
+  EXPECT_NE(std::future_status::ready, has_shutdown.wait_for(ms(2)));
+
+  auto expired = cq.MakeRelativeTimer(ms(0));
+  EXPECT_EQ(std::future_status::ready, expired.wait_for(ms(100)));
+
+  cq.Shutdown();
+  EXPECT_EQ(std::future_status::ready, has_shutdown.wait_for(ms(100)));
+
+  t.join();
+}
+
+/// @test Verify that users can supply their own queue and threads.
+TEST(CustomerSuppliedBackgroundThreads, SharesCompletionQueue) {
+  CompletionQueue cq;
+
+  CustomerSuppliedBackgroundThreads actual(cq);
+
+  using ms = std::chrono::milliseconds;
+  // Verify the completion queue is shared. Scheduling work in actual.cq() works
+  // once a thread is blocked in cq.Run(). Start that thread after scheduling
+  // the work to avoid flaky failures where the timer expires immediately.
+  future<std::thread::id> id = actual.cq().MakeRelativeTimer(ms(1)).then(
+      [](future<StatusOr<std::chrono::system_clock::time_point>>) {
+        return std::this_thread::get_id();
+      });
+  std::thread t([&cq] { cq.Run(); });
+  EXPECT_EQ(std::future_status::ready, id.wait_for(ms(100)));
+  EXPECT_EQ(t.get_id(), id.get());
+
+  cq.Shutdown();
+  t.join();
+}
+
+/// @test Verify that automatically created completion queues are usable.
+TEST(AutomaticallyCreatedBackgroundThreads, IsActive) {
+  AutomaticallyCreatedBackgroundThreads actual;
+
+  using ms = std::chrono::milliseconds;
+
+  auto expired = actual.cq().MakeRelativeTimer(ms(0));
+  EXPECT_EQ(std::future_status::ready, expired.wait_for(ms(100)));
+}
+
+}  // namespace
+}  // namespace internal
+}  // namespace GOOGLE_CLOUD_CPP_NS
+}  // namespace cloud
+}  // namespace google