feat: use BackgroundThreads in SessionPool (#1259)

* feat: use `BackgroundThreads` in `SessionPool`

Currently this just schedules a recurring 5s timer and cancels it in
the destructor.

* address review comments

* rewrite the comment in `~SessionPool`

* point to the proper background_threads

also change some ;'s to .'s per bww

* API/ABI update (internal namespace break only)

Author: mr-salty

ci/test-api/spanner_client.expected.abi.dump.gz

@@ -98,6 +98,7 @@ ConnectionImpl::ConnectionImpl(Database db,
       backoff_policy_prototype_(std::move(backoff_policy)),
       session_pool_(MakeSessionPool(db_, std::move(stubs),
                                     std::move(session_pool_options),
+                                    options.background_threads_factory()(),
                                     retry_policy_prototype_->clone(),
                                     backoff_policy_prototype_->clone())),
       rpc_stream_tracing_enabled_(options.tracing_enabled("rpc-streams")),

google/cloud/spanner/internal/connection_impl.cc

@@ -16,10 +16,15 @@
 #include "google/cloud/spanner/internal/connection_impl.h"
 #include "google/cloud/spanner/internal/retry_loop.h"
 #include "google/cloud/spanner/internal/session.h"
+#include "google/cloud/grpc_utils/completion_queue.h"
+#include "google/cloud/internal/background_threads_impl.h"
 #include "google/cloud/internal/make_unique.h"
+#include "google/cloud/log.h"
 #include "google/cloud/status.h"
 #include <algorithm>
+#include <chrono>
 #include <random>
+#include <thread>
 
 namespace google {
 namespace cloud {
@@ -31,23 +36,28 @@ namespace spanner_proto = ::google::spanner::v1;
 
 std::shared_ptr<SessionPool> MakeSessionPool(
     Database db, std::vector<std::shared_ptr<SpannerStub>> stubs,
-    SessionPoolOptions options, std::unique_ptr<RetryPolicy> retry_policy,
+    SessionPoolOptions options,
+    std::unique_ptr<BackgroundThreads> background_threads,
+    std::unique_ptr<RetryPolicy> retry_policy,
     std::unique_ptr<BackoffPolicy> backoff_policy) {
   auto pool = std::make_shared<SessionPool>(
       std::move(db), std::move(stubs), std::move(options),
-      std::move(retry_policy), std::move(backoff_policy));
+      std::move(background_threads), std::move(retry_policy),
+      std::move(backoff_policy));
   pool->Initialize();
   return pool;
 }
 
 SessionPool::SessionPool(Database db,
                          std::vector<std::shared_ptr<SpannerStub>> stubs,
                          SessionPoolOptions options,
+                         std::unique_ptr<BackgroundThreads> background_threads,
                          std::unique_ptr<RetryPolicy> retry_policy,
                          std::unique_ptr<BackoffPolicy> backoff_policy)
     : db_(std::move(db)),
       options_(std::move(
           options.EnforceConstraints(static_cast<int>(stubs.size())))),
+      background_threads_(std::move(background_threads)),
       retry_policy_prototype_(std::move(retry_policy)),
       backoff_policy_prototype_(std::move(backoff_policy)),
       max_pool_size_(options_.max_sessions_per_channel() *
@@ -89,6 +99,43 @@ void SessionPool::Initialize() {
       (void)CreateSessions(lk, channel, options_.labels(), num_sessions);
     }
   }
+  ScheduleBackgroundWork(std::chrono::seconds(5));
+}
+
+SessionPool::~SessionPool() {
+  // All references to this object are via `shared_ptr`; since we're in the
+  // destructor that implies there can be no concurrent accesses to any member
+  // variables, including `current_timer_`.
+  //
+  // Note that it *is* possible the timer lambda in `ScheduleBackgroundWork`
+  // is executing concurrently. However, since we are in the destructor we know
+  // that the lambda must not have yet successfully finished a call to `lock()`
+  // on the `weak_ptr` to `this` it holds. Any subsequent or in-progress calls
+  // must return `nullptr`, and the lambda will not do any work nor reschedule
+  // the timer.
+  current_timer_.cancel();
+}
+
+void SessionPool::ScheduleBackgroundWork(std::chrono::seconds relative_time) {
+  // See the comment in the destructor about the thread safety of this method.
+  std::weak_ptr<SessionPool> pool = shared_from_this();
+  current_timer_ =
+      background_threads_->cq()
+          .MakeRelativeTimer(relative_time)
+          .then([pool](future<StatusOr<std::chrono::system_clock::time_point>>
+                           result) {
+            if (result.get().ok()) {
+              if (auto shared_pool = pool.lock()) {
+                shared_pool->DoBackgroundWork();
+              }
+            }
+          });
+}
+
+void SessionPool::DoBackgroundWork() {
+  // TODO(#1171) Implement SessionPool session refresh
+  // TODO(#1172) maintain desired SessionPool size
+  ScheduleBackgroundWork(std::chrono::seconds(5));
 }
 
 StatusOr<SessionHolder> SessionPool::Allocate(bool dissociate_from_pool) {
@@ -219,6 +266,7 @@ Status SessionPool::CreateSessions(
 }
 
 void SessionPool::UpdateNextChannelForCreateSessions() {
+  // `mu_` must be held by the caller.
   next_channel_for_create_sessions_ = channels_.begin();
   for (auto it = channels_.begin(); it != channels_.end(); ++it) {
     if (it->session_count < next_channel_for_create_sessions_->session_count) {

google/cloud/spanner/internal/session_pool.cc

@@ -22,8 +22,11 @@
 #include "google/cloud/spanner/retry_policy.h"
 #include "google/cloud/spanner/session_pool_options.h"
 #include "google/cloud/spanner/version.h"
+#include "google/cloud/background_threads.h"
+#include "google/cloud/future.h"
 #include "google/cloud/status_or.h"
 #include <google/spanner/v1/spanner.pb.h>
+#include <chrono>
 #include <condition_variable>
 #include <cstddef>
 #include <map>
@@ -62,8 +65,12 @@ class SessionPool : public std::enable_shared_from_this<SessionPool> {
    */
   SessionPool(Database db, std::vector<std::shared_ptr<SpannerStub>> stubs,
               SessionPoolOptions options,
+              std::unique_ptr<BackgroundThreads> background_threads,
               std::unique_ptr<RetryPolicy> retry_policy,
               std::unique_ptr<BackoffPolicy> backoff_policy);
+
+  ~SessionPool();
+
   /**
    * If using the factory method is not possible, call `Initialize()` exactly
    * once, immediately after constructing the pool. This is necessary because
@@ -118,8 +125,12 @@ class SessionPool : public std::enable_shared_from_this<SessionPool> {
 
   void UpdateNextChannelForCreateSessions();  // EXCLUSIVE_LOCKS_REQUIRED(mu_)
 
+  void ScheduleBackgroundWork(std::chrono::seconds relative_time);
+  void DoBackgroundWork();
+
   Database const db_;
   SessionPoolOptions const options_;
+  std::unique_ptr<BackgroundThreads> const background_threads_;
   std::unique_ptr<RetryPolicy const> retry_policy_prototype_;
   std::unique_ptr<BackoffPolicy const> backoff_policy_prototype_;
   int const max_pool_size_;
@@ -130,6 +141,7 @@ class SessionPool : public std::enable_shared_from_this<SessionPool> {
   int total_sessions_ = 0;                          // GUARDED_BY(mu_)
   bool create_in_progress_ = false;                 // GUARDED_BY(mu_)
   int num_waiting_for_session_ = 0;                 // GUARDED_BY(mu_)
+  future<void> current_timer_;
 
   // `channels_` is guaranteed to be non-empty and will not be resized after
   // the constructor runs (so the iterators are guaranteed to always be valid).
@@ -150,7 +162,9 @@ class SessionPool : public std::enable_shared_from_this<SessionPool> {
  */
 std::shared_ptr<SessionPool> MakeSessionPool(
     Database db, std::vector<std::shared_ptr<SpannerStub>> stubs,
-    SessionPoolOptions options, std::unique_ptr<RetryPolicy> retry_policy,
+    SessionPoolOptions options,
+    std::unique_ptr<BackgroundThreads> background_threads,
+    std::unique_ptr<RetryPolicy> retry_policy,
     std::unique_ptr<BackoffPolicy> backoff_policy);
 
 }  // namespace internal

google/cloud/spanner/internal/session_pool.h

@@ -15,6 +15,7 @@
 #include "google/cloud/spanner/internal/session_pool.h"
 #include "google/cloud/spanner/internal/session.h"
 #include "google/cloud/spanner/testing/mock_spanner_stub.h"
+#include "google/cloud/internal/background_threads_impl.h"
 #include "google/cloud/internal/make_unique.h"
 #include "google/cloud/status.h"
 #include "google/cloud/testing_util/assert_ok.h"
@@ -67,6 +68,8 @@ std::shared_ptr<SessionPool> MakeSessionPool(
     SessionPoolOptions options) {
   return MakeSessionPool(
       std::move(db), std::move(stubs), std::move(options),
+      google::cloud::internal::make_unique<
+          google::cloud::internal::AutomaticallyCreatedBackgroundThreads>(),
       google::cloud::internal::make_unique<LimitedTimeRetryPolicy>(
           std::chrono::minutes(10)),
       google::cloud::internal::make_unique<ExponentialBackoffPolicy>(