feat!: implement SessionPool session refresh (#1425)

Call `GetSession()` during `SessionPool::DoBackgroundWork()` for any
session whose last-use time is older than the keep-alive interval.

Also change the type of `SessionPoolOptions::keep_alive_interval_`
from `std::chrono::minutes` to `std::chrono::seconds`.  This is API
compatible for callers of `SessionPoolOptions::set_keep_alive_interval()`,
but is a breakage if the result of `SessionPoolOptions::keep_alive_interval()`
is assigned to a `std::chrono::minutes`.  We believe this is unlikely to
impact many/any users.

Fixes #1171.

Author: devbww

google/cloud/spanner/internal/session.h

@@ -40,7 +40,7 @@ class Session {
       : session_name_(std::move(session_name)),
         channel_(std::move(channel)),
         is_bad_(false),
-        last_use_time_(std::chrono::steady_clock::now()) {}
+        last_use_time_(Clock::now()) {}
 
   // Not copyable or moveable.
   Session(Session const&) = delete;
@@ -55,23 +55,20 @@ class Session {
   bool is_bad() const { return is_bad_.load(std::memory_order_relaxed); }
 
  private:
-  // Give `SessionPool` access to the private methods below.
+  // Give `SessionPool` access to the private types/methods below.
   friend class SessionPool;
+  using Clock = std::chrono::steady_clock;
   std::shared_ptr<Channel> const& channel() const { return channel_; }
 
   // The caller is responsible for ensuring these methods are used in a
   // thread-safe manner (i.e. using external locking).
-  std::chrono::steady_clock::time_point last_use_time() const {
-    return last_use_time_;
-  }
-  void update_last_use_time() {
-    last_use_time_ = std::chrono::steady_clock::now();
-  }
+  Clock::time_point last_use_time() const { return last_use_time_; }
+  void update_last_use_time() { last_use_time_ = Clock::now(); }
 
   std::string const session_name_;
   std::shared_ptr<Channel> const channel_;
   std::atomic<bool> is_bad_;
-  std::chrono::steady_clock::time_point last_use_time_;
+  Clock::time_point last_use_time_;
 };
 
 /**

google/cloud/spanner/internal/session_pool.cc

@@ -25,6 +25,7 @@
 #include <chrono>
 #include <random>
 #include <thread>
+#include <utility>
 #include <vector>
 
 namespace google {
@@ -114,7 +115,7 @@ void SessionPool::ScheduleBackgroundWork(std::chrono::seconds relative_time) {
 
 void SessionPool::DoBackgroundWork() {
   MaintainPoolSize();
-  // TODO(#1171) Implement SessionPool session refresh
+  RefreshExpiringSessions();
   ScheduleBackgroundWork(std::chrono::seconds(5));
 }
 
@@ -129,6 +130,43 @@ void SessionPool::MaintainPoolSize() {
   }
 }
 
+// Initiate an async GetSession() call on any session whose last-use time is
+// older than the keep-alive interval.
+void SessionPool::RefreshExpiringSessions() {
+  std::vector<std::pair<std::shared_ptr<SpannerStub>, std::string>>
+      sessions_to_refresh;
+  Session::Clock::time_point now = Session::Clock::now();
+  Session::Clock::time_point refresh_limit =
+      now - options_.keep_alive_interval();
+  {
+    std::unique_lock<std::mutex> lk(mu_);
+    if (last_use_time_lower_bound_ <= refresh_limit) {
+      last_use_time_lower_bound_ = now;
+      for (auto const& session : sessions_) {
+        Session::Clock::time_point last_use_time = session->last_use_time();
+        if (last_use_time <= refresh_limit) {
+          sessions_to_refresh.emplace_back(session->channel()->stub,
+                                           session->session_name());
+          session->update_last_use_time();
+        } else if (last_use_time < last_use_time_lower_bound_) {
+          last_use_time_lower_bound_ = last_use_time;
+        }
+      }
+    }
+  }
+  for (auto& refresh : sessions_to_refresh) {
+    AsyncGetSession(cq_, std::move(refresh.first), std::move(refresh.second))
+        .then([](future<StatusOr<spanner_proto::Session>> result) {
+          // We simply discard the response as handling IsSessionNotFound()
+          // by removing the session from the pool is problematic (and would
+          // not eliminate the possibility of IsSessionNotFound() elsewhere).
+          // The last-use time has already been updated to throttle attempts.
+          // TODO(#1430): Re-evaluate these decisions.
+          (void)result.get();
+        });
+  }
+}
+
 /**
  * Grow the session pool by creating up to `sessions_to_create` sessions and
  * adding them to the pool.  Note that `lk` may be released and reacquired in

google/cloud/spanner/internal/session_pool.h

@@ -156,6 +156,7 @@ class SessionPool : public std::enable_shared_from_this<SessionPool> {
   void ScheduleBackgroundWork(std::chrono::seconds relative_time);
   void DoBackgroundWork();
   void MaintainPoolSize();
+  void RefreshExpiringSessions();
 
   Database const db_;
   SessionPoolOptions const options_;
@@ -171,6 +172,11 @@ class SessionPool : public std::enable_shared_from_this<SessionPool> {
   int total_sessions_ = 0;                          // GUARDED_BY(mu_)
   int create_calls_in_progress_ = 0;                // GUARDED_BY(mu_)
   int num_waiting_for_session_ = 0;                 // GUARDED_BY(mu_)
+
+  // Lower bound on all `sessions_[i]->last_use_time()` values.
+  Session::Clock::time_point last_use_time_lower_bound_ =
+      Session::Clock::now();  // GUARDED_BY(mu_)
+
   future<void> current_timer_;
 
   // `channels_` is guaranteed to be non-empty and will not be resized after

google/cloud/spanner/internal/session_pool_test.cc

@@ -14,6 +14,8 @@
 
 #include "google/cloud/spanner/internal/session_pool.h"
 #include "google/cloud/spanner/internal/session.h"
+#include "google/cloud/spanner/testing/mock_completion_queue.h"
+#include "google/cloud/spanner/testing/mock_response_reader.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"
@@ -36,7 +38,9 @@ namespace {
 using ::testing::_;
 using ::testing::ByMove;
 using ::testing::HasSubstr;
+using ::testing::Invoke;
 using ::testing::Return;
+using ::testing::StrictMock;
 using ::testing::UnorderedElementsAre;
 
 namespace spanner_proto = ::google::spanner::v1;
@@ -382,6 +386,69 @@ TEST(SessionPool, GetStubForStublessSession) {
   EXPECT_EQ(pool->GetStub(*session), mock);
 }
 
+// TODO(#1428): This test runs in real time. SessionPool does not currently
+// provide any mechanism to inject a clock source, or to control its
+// background-work scheduling. This makes the test slower and more fragile
+// than desired.
+TEST(SessionPool, SessionRefresh) {
+  auto mock = std::make_shared<StrictMock<spanner_testing::MockSpannerStub>>();
+  EXPECT_CALL(*mock, BatchCreateSessions(_, _))
+      .WillOnce(Return(ByMove(MakeSessionsResponse({"s1"}))))
+      .WillOnce(Return(ByMove(MakeSessionsResponse({"s2"}))));
+
+  auto reader = google::cloud::internal::make_unique<
+      StrictMock<google::cloud::spanner::testing::MockAsyncResponseReader<
+          spanner_proto::Session>>>();
+  EXPECT_CALL(*mock, AsyncGetSession(_, _, _))
+      .WillOnce(Invoke([&reader](
+                           grpc::ClientContext&,
+                           spanner_proto::GetSessionRequest const& request,
+                           grpc::CompletionQueue*) {
+        EXPECT_EQ("s2", request.name());
+        // This is safe. See comments in MockAsyncResponseReader.
+        return std::unique_ptr<
+            grpc::ClientAsyncResponseReaderInterface<spanner_proto::Session>>(
+            reader.get());
+      }));
+  EXPECT_CALL(*reader, Finish(_, _, _))
+      .WillOnce(Invoke(
+          [](spanner_proto::Session* session, grpc::Status* status, void*) {
+            session->set_name("s2");
+            *status = grpc::Status::OK;
+          }));
+
+  auto db = Database("project", "instance", "database");
+  SessionPoolOptions options;
+  options.set_keep_alive_interval(std::chrono::seconds(10));
+  auto impl = std::make_shared<testing::MockCompletionQueue>();
+  auto pool = MakeSessionPool(db, {mock}, options, CompletionQueue(impl));
+
+  // now == t0: Allocate and release two sessions such that "s1" will expire
+  // at t0 + 18s, and "s2" will expire at t0 + 10s, and after which both the
+  // BatchCreateSessions() expectations have been satisfied.
+  {
+    auto s1 = pool->Allocate();
+    ASSERT_STATUS_OK(s1);
+    EXPECT_EQ("s1", (*s1)->session_name());
+    {
+      auto s2 = pool->Allocate();
+      ASSERT_STATUS_OK(s2);
+      EXPECT_EQ("s2", (*s2)->session_name());
+    }
+    std::this_thread::sleep_for(std::chrono::seconds(8));
+  }
+  std::this_thread::sleep_for(std::chrono::seconds(8));
+
+  // now == t0 + 16s: Session "s2" has expired, but "s1" has not, and
+  // RefreshExpiringSessions() has run exactly once since "s2" expired.
+  impl->SimulateCompletion(true);  // make the async GetSession() RPC
+
+  // The AsyncGetSession() and Finish() expectations should now have been
+  // satisfied. If anything goes wrong we'll get unsatisfied/uninteresting
+  // gmock errors.
+  impl->SimulateCompletion(true);  // run the completion callback
+}
+
 }  // namespace
 }  // namespace internal
 }  // namespace SPANNER_CLIENT_NS

google/cloud/spanner/session_pool_options.h

@@ -122,13 +122,13 @@ class SessionPoolOptions {
    * minutes, so any duration below that (less some slack to allow the calls
    * to be made to refresh the sessions) should suffice.
    */
-  SessionPoolOptions& set_keep_alive_interval(std::chrono::minutes minutes) {
-    keep_alive_interval_ = minutes;
+  SessionPoolOptions& set_keep_alive_interval(std::chrono::seconds interval) {
+    keep_alive_interval_ = interval;
     return *this;
   }
 
   /// Return the interval at which we refresh sessions to prevent GC.
-  std::chrono::minutes keep_alive_interval() const {
+  std::chrono::seconds keep_alive_interval() const {
     return keep_alive_interval_;
   }
 
@@ -151,7 +151,7 @@ class SessionPoolOptions {
   int max_sessions_per_channel_ = 100;
   int max_idle_sessions_ = 0;
   ActionOnExhaustion action_on_exhaustion_ = ActionOnExhaustion::kBlock;
-  std::chrono::minutes keep_alive_interval_ = std::chrono::minutes(55);
+  std::chrono::seconds keep_alive_interval_ = std::chrono::minutes(55);
   std::map<std::string, std::string> labels_;
 };