SERVER-37837 Examine and reap sessions from the SessionsCatalog

This change makes the logical sessions cache query and reap sessions,
which are possibly only in-memory on the SessionsCatalog.

1 files changed, 102 insertions, 0 deletions

diff --git a/src/mongo/db/session_catalog_test.cpp b/src/mongo/db/session_catalog_test.cpp
index 3bd63277365..da7712f89d4 100644
--- a/src/mongo/db/session_catalog_test.cpp
+++ b/src/mongo/db/session_catalog_test.cpp
@@ -117,6 +117,67 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, NestedOperationContextSession) {
     ASSERT(!OperationContextSession::get(_opCtx));
 }
 
+TEST_F(SessionCatalogTest, ScanSession) {
+    // Create three sessions in the catalog.
+    const std::vector<LogicalSessionId> lsids{makeLogicalSessionIdForTest(),
+                                              makeLogicalSessionIdForTest(),
+                                              makeLogicalSessionIdForTest()};
+    for (const auto& lsid : lsids) {
+        stdx::async(stdx::launch::async, [this, lsid] {
+            ThreadClient tc(getServiceContext());
+            auto opCtx = makeOperationContext();
+            opCtx->setLogicalSessionId(lsid);
+            OperationContextSession ocs(opCtx.get());
+        }).get();
+    }
+
+    catalog()->scanSession(lsids[0], [&lsids](const ObservableSession& session) {
+        ASSERT_EQ(lsids[0], session.get()->getSessionId());
+    });
+
+    catalog()->scanSession(lsids[1], [&lsids](const ObservableSession& session) {
+        ASSERT_EQ(lsids[1], session.get()->getSessionId());
+    });
+
+    catalog()->scanSession(lsids[2], [&lsids](const ObservableSession& session) {
+        ASSERT_EQ(lsids[2], session.get()->getSessionId());
+    });
+
+    catalog()->scanSession(makeLogicalSessionIdForTest(), [](const ObservableSession&) {
+        FAIL("The callback was called for non-existent session");
+    });
+}
+
+TEST_F(SessionCatalogTest, ScanSessionMarkForReapWhenSessionIsIdle) {
+    // Create three sessions in the catalog.
+    const std::vector<LogicalSessionId> lsids{makeLogicalSessionIdForTest(),
+                                              makeLogicalSessionIdForTest(),
+                                              makeLogicalSessionIdForTest()};
+    for (const auto& lsid : lsids) {
+        stdx::async(stdx::launch::async, [this, lsid] {
+            ThreadClient tc(getServiceContext());
+            auto opCtx = makeOperationContext();
+            opCtx->setLogicalSessionId(lsid);
+            OperationContextSession ocs(opCtx.get());
+        }).get();
+    }
+
+    catalog()->scanSession(lsids[0],
+                           [&lsids](ObservableSession& session) { session.markForReap(); });
+
+    catalog()->scanSession(lsids[0], [](const ObservableSession&) {
+        FAIL("The callback was called for non-existent session");
+    });
+
+    catalog()->scanSession(lsids[1], [&lsids](const ObservableSession& session) {
+        ASSERT_EQ(lsids[1], session.get()->getSessionId());
+    });
+
+    catalog()->scanSession(lsids[2], [&lsids](const ObservableSession& session) {
+        ASSERT_EQ(lsids[2], session.get()->getSessionId());
+    });
+}
+
 TEST_F(SessionCatalogTestWithDefaultOpCtx, ScanSessions) {
     std::vector<LogicalSessionId> lsidsFound;
     const auto workerFn = [&lsidsFound](const ObservableSession& session) {
@@ -157,6 +218,47 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, ScanSessions) {
     lsidsFound.clear();
 }
 
+TEST_F(SessionCatalogTestWithDefaultOpCtx, ScanSessionsMarkForReap) {
+    // Create three sessions in the catalog.
+    const std::vector<LogicalSessionId> lsids{makeLogicalSessionIdForTest(),
+                                              makeLogicalSessionIdForTest(),
+                                              makeLogicalSessionIdForTest()};
+
+    unittest::Barrier sessionsCheckedOut(2);
+    unittest::Barrier sessionsCheckedIn(2);
+
+    auto f = stdx::async(stdx::launch::async, [&] {
+        ThreadClient tc(getServiceContext());
+        auto opCtx = makeOperationContext();
+        opCtx->setLogicalSessionId(lsids[1]);
+        OperationContextSession ocs(opCtx.get());
+        sessionsCheckedOut.countDownAndWait();
+        sessionsCheckedIn.countDownAndWait();
+    });
+
+    // After this wait, session 1 is checked-out and waiting on the barrier, because of which only
+    // sessions 0 and 2 will be reaped
+    sessionsCheckedOut.countDownAndWait();
+
+    SessionKiller::Matcher matcherAllSessions(
+        KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(_opCtx)});
+
+    catalog()->scanSessions(matcherAllSessions,
+                            [&](ObservableSession& session) { session.markForReap(); });
+
+    catalog()->scanSessions(matcherAllSessions, [&](const ObservableSession& session) {
+        ASSERT_EQ(lsids[1], session.get()->getSessionId());
+    });
+
+    // After this point, session 1 is checked back in
+    sessionsCheckedIn.countDownAndWait();
+    f.get();
+
+    catalog()->scanSessions(matcherAllSessions, [&](const ObservableSession& session) {
+        ASSERT_EQ(lsids[1], session.get()->getSessionId());
+    });
+}
+
 TEST_F(SessionCatalogTest, KillSessionWhenSessionIsNotCheckedOut) {
     const auto lsid = makeLogicalSessionIdForTest();