diff options
-rw-r--r-- | src/mongo/db/SConscript | 1 | ||||
-rw-r--r-- | src/mongo/db/kill_sessions_local.cpp | 68 | ||||
-rw-r--r-- | src/mongo/db/op_observer_impl_test.cpp | 12 | ||||
-rw-r--r-- | src/mongo/db/pipeline/process_interface_standalone.cpp | 16 | ||||
-rw-r--r-- | src/mongo/db/s/session_catalog_migration_destination.cpp | 121 | ||||
-rw-r--r-- | src/mongo/db/s/session_catalog_migration_destination_test.cpp | 194 | ||||
-rw-r--r-- | src/mongo/db/session.cpp | 86 | ||||
-rw-r--r-- | src/mongo/db/session.h | 71 | ||||
-rw-r--r-- | src/mongo/db/session_catalog.cpp | 101 | ||||
-rw-r--r-- | src/mongo/db/session_catalog.h | 195 | ||||
-rw-r--r-- | src/mongo/db/session_catalog_mongod.cpp | 32 | ||||
-rw-r--r-- | src/mongo/db/session_catalog_test.cpp | 86 | ||||
-rw-r--r-- | src/mongo/db/transaction_participant.cpp | 2 | ||||
-rw-r--r-- | src/mongo/db/transaction_participant_test.cpp | 5 |
14 files changed, 515 insertions, 475 deletions
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript index 41bfd7cbaa8..5cd1763cf13 100644 --- a/src/mongo/db/SConscript +++ b/src/mongo/db/SConscript @@ -668,7 +668,6 @@ env.Library( target='session_catalog', source=[ 'session_catalog.cpp', - 'session.cpp', ], LIBDEPS=[ 'kill_sessions', diff --git a/src/mongo/db/kill_sessions_local.cpp b/src/mongo/db/kill_sessions_local.cpp index 56ffabf14cf..79b7283d4c3 100644 --- a/src/mongo/db/kill_sessions_local.cpp +++ b/src/mongo/db/kill_sessions_local.cpp @@ -55,17 +55,18 @@ namespace { * context with 'reason' as the code. * 3) Finish killing the selected and interrupted sessions through the 'killSessionFn'. */ -void killSessionsAction(OperationContext* opCtx, - const SessionKiller::Matcher& matcher, - const stdx::function<bool(Session*)>& filterFn, - const stdx::function<void(Session*)>& killSessionFn, - ErrorCodes::Error reason = ErrorCodes::Interrupted) { +void killSessionsAction( + OperationContext* opCtx, + const SessionKiller::Matcher& matcher, + const stdx::function<bool(const ObservableSession&)>& filterFn, + const stdx::function<void(OperationContext*, const SessionToKill&)>& killSessionFn, + ErrorCodes::Error reason = ErrorCodes::Interrupted) { const auto catalog = SessionCatalog::get(opCtx); - std::vector<Session::KillToken> sessionKillTokens; - catalog->scanSessions(matcher, [&](WithLock sessionCatalogLock, Session* session) { + std::vector<SessionCatalog::KillToken> sessionKillTokens; + catalog->scanSessions(matcher, [&](const ObservableSession& session) { if (filterFn(session)) - sessionKillTokens.emplace_back(session->kill(sessionCatalogLock, reason)); + sessionKillTokens.emplace_back(session.kill(reason)); }); for (auto& sessionKillToken : sessionKillTokens) { @@ -73,11 +74,11 @@ void killSessionsAction(OperationContext* opCtx, // TODO (SERVER-33850): Rename KillAllSessionsByPattern and // ScopedKillAllSessionsByPatternImpersonator to not refer to session kill - const KillAllSessionsByPattern* pattern = matcher.match(session->getSessionId()); + const KillAllSessionsByPattern* pattern = matcher.match(session.getSessionId()); invariant(pattern); ScopedKillAllSessionsByPatternImpersonator impersonator(opCtx, *pattern); - killSessionFn(session.get()); + killSessionFn(opCtx, session); } } @@ -86,12 +87,13 @@ void killSessionsAction(OperationContext* opCtx, void killSessionsLocalKillTransactions(OperationContext* opCtx, const SessionKiller::Matcher& matcher, ErrorCodes::Error reason) { - killSessionsAction( - opCtx, - matcher, - [](Session*) { return true; }, - [](Session* session) { TransactionParticipant::get(session)->abortArbitraryTransaction(); }, - reason); + killSessionsAction(opCtx, + matcher, + [](const ObservableSession&) { return true; }, + [](OperationContext* opCtx, const SessionToKill& session) { + TransactionParticipant::get(session.get())->abortArbitraryTransaction(); + }, + reason); } SessionKiller::Result killSessionsLocal(OperationContext* opCtx, @@ -112,21 +114,21 @@ void killAllExpiredTransactions(OperationContext* opCtx) { killSessionsAction( opCtx, matcherAllSessions, - [](Session* session) { - const auto txnParticipant = TransactionParticipant::get(session); + [when = opCtx->getServiceContext()->getPreciseClockSource()->now()]( + const ObservableSession& session) { - return txnParticipant->expired(); + return TransactionParticipant::get(session.get())->expired(); }, - [](Session* session) { - const auto txnParticipant = TransactionParticipant::get(session); + [](OperationContext* opCtx, const SessionToKill& session) { + auto txnParticipant = TransactionParticipant::get(session.get()); LOG(0) << "Aborting transaction with txnNumber " << txnParticipant->getActiveTxnNumber() - << " on session " << session->getSessionId().getId() + << " on session " << session.getSessionId().getId() << " because it has been running for longer than 'transactionLifetimeLimitSeconds'"; - // The try/catch block below is necessary because expired() in the filterFn above could - // return true for expired, but unprepared transaction, but by the time we get to + // The try/catch block below is necessary because expiredAsOf() in the filterFn above + // could return true for expired, but unprepared transaction, but by the time we get to // actually kill it, the participant could theoretically become prepared (being under // the SessionCatalog mutex doesn't prevent the concurrently running thread from doing // preparing the participant). @@ -136,8 +138,9 @@ void killAllExpiredTransactions(OperationContext* opCtx) { try { txnParticipant->abortArbitraryTransaction(); } catch (const DBException& ex) { + // TODO(schwerin): Can we catch a more specific exception? warning() << "May have failed to abort expired transaction on session " - << session->getSessionId().getId() << " due to " << redact(ex.toStatus()); + << session.getSessionId().getId() << " due to " << redact(ex.toStatus()); } }, ErrorCodes::ExceededTimeLimit); @@ -148,8 +151,10 @@ void killSessionsLocalShutdownAllTransactions(OperationContext* opCtx) { KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(opCtx)}); killSessionsAction(opCtx, matcherAllSessions, - [](Session*) { return true; }, - [](Session* session) { TransactionParticipant::get(session)->shutdown(); }, + [](const ObservableSession&) { return true; }, + [](OperationContext* opCtx, const SessionToKill& session) { + TransactionParticipant::get(session.get())->shutdown(); + }, ErrorCodes::InterruptedAtShutdown); } @@ -159,15 +164,14 @@ void killSessionsAbortAllPreparedTransactions(OperationContext* opCtx) { killSessionsAction( opCtx, matcherAllSessions, - [](Session* session) { + [](const ObservableSession& session) { // Filter for sessions that have a prepared transaction. - const auto txnParticipant = TransactionParticipant::get(session); - return txnParticipant->transactionIsPrepared(); + return TransactionParticipant::get(session.get())->transactionIsPrepared(); }, - [](Session* session) { + [](OperationContext* opCtx, const SessionToKill& session) { // Abort the prepared transaction and invalidate the session it is // associated with. - TransactionParticipant::get(session)->abortPreparedTransactionForRollback(); + TransactionParticipant::get(session.get())->abortPreparedTransactionForRollback(); }); } diff --git a/src/mongo/db/op_observer_impl_test.cpp b/src/mongo/db/op_observer_impl_test.cpp index ea3b80b605e..c8bd4509843 100644 --- a/src/mongo/db/op_observer_impl_test.cpp +++ b/src/mongo/db/op_observer_impl_test.cpp @@ -334,7 +334,6 @@ TEST_F(OpObserverSessionCatalogRollbackTest, OnRollbackDoesntInvalidateSessionCatalogIfNoSessionOpsRolledBack) { const NamespaceString nss("testDB", "testColl"); - auto sessionCatalog = SessionCatalog::get(getServiceContext()); auto sessionId = makeLogicalSessionIdForTest(); const TxnNumber txnNum = 0; @@ -343,10 +342,8 @@ TEST_F(OpObserverSessionCatalogRollbackTest, { auto opCtx = cc().makeOperationContext(); opCtx->setLogicalSessionId(sessionId); - - // Create a session and sync it from disk - auto session = sessionCatalog->checkOutSession(opCtx.get()); - const auto txnParticipant = TransactionParticipant::get(session.get()); + MongoDOperationContextSession ocs(opCtx.get()); + const auto txnParticipant = TransactionParticipant::get(opCtx.get()); txnParticipant->refreshFromStorageIfNeeded(); // Simulate a write occurring on that session @@ -369,9 +366,8 @@ TEST_F(OpObserverSessionCatalogRollbackTest, { auto opCtx = cc().makeOperationContext(); opCtx->setLogicalSessionId(sessionId); - - auto session = sessionCatalog->checkOutSession(opCtx.get()); - const auto txnParticipant = TransactionParticipant::get(session.get()); + MongoDOperationContextSession ocs(opCtx.get()); + const auto txnParticipant = TransactionParticipant::get(opCtx.get()); ASSERT(txnParticipant->checkStatementExecutedNoOplogEntryFetch(stmtId)); } } diff --git a/src/mongo/db/pipeline/process_interface_standalone.cpp b/src/mongo/db/pipeline/process_interface_standalone.cpp index 3ee21d3001b..aeef3174c5c 100644 --- a/src/mongo/db/pipeline/process_interface_standalone.cpp +++ b/src/mongo/db/pipeline/process_interface_standalone.cpp @@ -556,14 +556,14 @@ void MongoInterfaceStandalone::_reportCurrentOpsForIdleSessions(OperationContext ? makeSessionFilterForAuthenticatedUsers(opCtx) : KillAllSessionsByPatternSet{{}}); - sessionCatalog->scanSessions({std::move(sessionFilter)}, - [&](WithLock sessionCatalogLock, Session* session) { - auto op = - TransactionParticipant::get(session)->reportStashedState(); - if (!op.isEmpty()) { - ops->emplace_back(op); - } - }); + sessionCatalog->scanSessions( + {std::move(sessionFilter)}, + [&](const ObservableSession& session) { + auto op = TransactionParticipant::get(session.get())->reportStashedState(); + if (!op.isEmpty()) { + ops->emplace_back(op); + } + }); } std::unique_ptr<CollatorInterface> MongoInterfaceStandalone::_getCollectionDefaultCollator( diff --git a/src/mongo/db/s/session_catalog_migration_destination.cpp b/src/mongo/db/s/session_catalog_migration_destination.cpp index 0cbe33a8d74..28437321ef5 100644 --- a/src/mongo/db/s/session_catalog_migration_destination.cpp +++ b/src/mongo/db/s/session_catalog_migration_destination.cpp @@ -42,7 +42,7 @@ #include "mongo/db/repl/oplog.h" #include "mongo/db/repl/oplog_entry.h" #include "mongo/db/s/migration_session_id.h" -#include "mongo/db/session_catalog.h" +#include "mongo/db/session_catalog_mongod.h" #include "mongo/db/transaction_participant.h" #include "mongo/db/write_concern.h" #include "mongo/s/client/shard_registry.h" @@ -204,8 +204,7 @@ BSONObj getNextSessionOplogBatch(OperationContext* opCtx, * Insert a new oplog entry by converting the oplogBSON into type 'n' oplog with the session * information. The new oplogEntry will also link to prePostImageTs if not null. */ -ProcessOplogResult processSessionOplog(OperationContext* opCtx, - const BSONObj& oplogBSON, +ProcessOplogResult processSessionOplog(const BSONObj& oplogBSON, const ProcessOplogResult& lastResult) { auto oplogEntry = parseOplog(oplogBSON); const auto& sessionInfo = oplogEntry.getOperationSessionInfo(); @@ -247,9 +246,12 @@ ProcessOplogResult processSessionOplog(OperationContext* opCtx, const auto stmtId = *oplogEntry.getStatementId(); - auto scopedSession = SessionCatalog::get(opCtx)->checkOutSession(opCtx, result.sessionId); - auto const txnParticipant = TransactionParticipant::get(scopedSession.get()); - txnParticipant->refreshFromStorageIfNeeded(); + auto uniqueOpCtx = cc().makeOperationContext(); + auto opCtx = uniqueOpCtx.get(); + opCtx->setLogicalSessionId(result.sessionId); + opCtx->setTxnNumber(result.txnNum); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); txnParticipant->beginOrContinue(result.txnNum, boost::none, boost::none); try { @@ -396,10 +398,7 @@ void SessionCatalogMigrationDestination::join() { */ void SessionCatalogMigrationDestination::_retrieveSessionStateFromSource(ServiceContext* service) { Client::initThread( - "sessionCatalogMigration-" + _migrationSessionId.toString(), service, nullptr); - - auto uniqueCtx = cc().makeOperationContext(); - auto opCtx = uniqueCtx.get(); + "sessionCatalogMigrationProducer-" + _migrationSessionId.toString(), service, nullptr); bool oplogDrainedAfterCommiting = false; ProcessOplogResult lastResult; @@ -413,67 +412,75 @@ void SessionCatalogMigrationDestination::_retrieveSessionStateFromSource(Service } } - auto nextBatch = getNextSessionOplogBatch(opCtx, _fromShard, _migrationSessionId); - BSONArray oplogArray(nextBatch[kOplogField].Obj()); - BSONArrayIteratorSorted oplogIter(oplogArray); - - if (!oplogIter.more()) { - { - stdx::lock_guard<stdx::mutex> lk(_mutex); - if (_state == State::Committing) { - // The migration is considered done only when it gets an empty result from the - // source shard while this is in state committing. This is to make sure that it - // doesn't miss any new oplog created between the time window where this - // depleted the buffer from the source shard and receiving the commit command. - if (oplogDrainedAfterCommiting) { - break; + BSONObj nextBatch; + BSONArray oplogArray; + { + auto uniqueCtx = cc().makeOperationContext(); + auto opCtx = uniqueCtx.get(); + + nextBatch = getNextSessionOplogBatch(opCtx, _fromShard, _migrationSessionId); + oplogArray = BSONArray{nextBatch[kOplogField].Obj()}; + + if (oplogArray.isEmpty()) { + { + stdx::lock_guard<stdx::mutex> lk(_mutex); + if (_state == State::Committing) { + // The migration is considered done only when it gets an empty result from + // the source shard while this is in state committing. This is to make sure + // that it doesn't miss any new oplog created between the time window where + // this depleted the buffer from the source shard and receiving the commit + // command. + if (oplogDrainedAfterCommiting) { + break; + } + + oplogDrainedAfterCommiting = true; } + } - oplogDrainedAfterCommiting = true; + WriteConcernResult unusedWCResult; + uassertStatusOK( + waitForWriteConcern(opCtx, lastResult.oplogTime, kMajorityWC, &unusedWCResult)); + + // We depleted the buffer at least once, transition to ready for commit. + { + stdx::lock_guard<stdx::mutex> lk(_mutex); + // Note: only transition to "ready to commit" if state is not error/force stop. + if (_state == State::Migrating) { + _state = State::ReadyToCommit; + _isStateChanged.notify_all(); + } } - } - WriteConcernResult unusedWCResult; - uassertStatusOK( - waitForWriteConcern(opCtx, lastResult.oplogTime, kMajorityWC, &unusedWCResult)); - - // We depleted the buffer at least once, transition to ready for commit. - { - stdx::lock_guard<stdx::mutex> lk(_mutex); - // Note: only transition to "ready to commit" if state is not error/force stop. - if (_state == State::Migrating) { - _state = State::ReadyToCommit; - _isStateChanged.notify_all(); + if (lastOpTimeWaited == lastResult.oplogTime) { + // We got an empty result at least twice in a row from the source shard so space + // it + // out a little bit so we don't hammer the shard + opCtx->sleepFor(Milliseconds(200)); } - } - if (lastOpTimeWaited == lastResult.oplogTime) { - // We got an empty result at least twice in a row from the source shard so space it - // out a little bit so we don't hammer the shard - opCtx->sleepFor(Milliseconds(200)); + lastOpTimeWaited = lastResult.oplogTime; } - - lastOpTimeWaited = lastResult.oplogTime; } - - while (oplogIter.more()) { + for (BSONArrayIteratorSorted oplogIter(oplogArray); oplogIter.more();) { try { - lastResult = processSessionOplog(opCtx, oplogIter.next().Obj(), lastResult); - } catch (const DBException& ex) { - if (ex.code() == ErrorCodes::ConflictingOperationInProgress || - ex.code() == ErrorCodes::TransactionTooOld) { - // This means that the server has a newer txnNumber than the oplog being - // migrated, so just skip it - continue; - } - - throw; + lastResult = processSessionOplog(oplogIter.next().Obj(), lastResult); + } catch (const ExceptionFor<ErrorCodes::ConfigurationInProgress>&) { + // This means that the server has a newer txnNumber than the oplog being + // migrated, so just skip it + continue; + } catch (const ExceptionFor<ErrorCodes::TransactionTooOld>&) { + // This means that the server has a newer txnNumber than the oplog being + // migrated, so just skip it + continue; } } } WriteConcernResult unusedWCResult; - uassertStatusOK(waitForWriteConcern(opCtx, lastResult.oplogTime, kMajorityWC, &unusedWCResult)); + auto uniqueOpCtx = cc().makeOperationContext(); + uassertStatusOK( + waitForWriteConcern(uniqueOpCtx.get(), lastResult.oplogTime, kMajorityWC, &unusedWCResult)); { stdx::lock_guard<stdx::mutex> lk(_mutex); diff --git a/src/mongo/db/s/session_catalog_migration_destination_test.cpp b/src/mongo/db/s/session_catalog_migration_destination_test.cpp index 451ef2425cd..5cb0aa92c62 100644 --- a/src/mongo/db/s/session_catalog_migration_destination_test.cpp +++ b/src/mongo/db/s/session_catalog_migration_destination_test.cpp @@ -164,13 +164,14 @@ public: return _migrationId.value(); } - ScopedCheckedOutSession getSessionWithTxn(OperationContext* opCtx, - const LogicalSessionId& sessionId, - const TxnNumber& txnNum) { - auto scopedSession = SessionCatalog::get(opCtx)->checkOutSession(opCtx, sessionId); - const auto txnParticipant = TransactionParticipant::get(scopedSession.get()); + void setUpSessionWithTxn(OperationContext* opCtx, + const LogicalSessionId& sessionId, + const TxnNumber& txnNum) { + opCtx->setLogicalSessionId(sessionId); + opCtx->setTxnNumber(txnNum); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); txnParticipant->beginOrContinue(txnNum, boost::none, boost::none); - return scopedSession; } void checkOplog(const repl::OplogEntry& originalOplog, const repl::OplogEntry& oplogToCheck) { @@ -194,21 +195,17 @@ public: } } - void checkStatementExecuted(OperationContext* opCtx, - Session* session, - TxnNumber txnNumber, - StmtId stmtId) { - const auto txnParticipant = TransactionParticipant::get(session); + void checkStatementExecuted(OperationContext* opCtx, TxnNumber txnNumber, StmtId stmtId) { + auto txnParticipant = TransactionParticipant::get(opCtx); auto oplog = txnParticipant->checkStatementExecuted(stmtId); ASSERT_TRUE(oplog); } void checkStatementExecuted(OperationContext* opCtx, - Session* session, TxnNumber txnNumber, StmtId stmtId, const repl::OplogEntry& expectedOplog) { - const auto txnParticipant = TransactionParticipant::get(session); + const auto txnParticipant = TransactionParticipant::get(opCtx); auto oplog = txnParticipant->checkStatementExecuted(stmtId); ASSERT_TRUE(oplog); checkOplogWithNestedOplog(expectedOplog, *oplog); @@ -354,8 +351,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithSameTxn) { finishSessionExpectSuccess(&sessionMigration); auto opCtx = operationContext(); - auto session = getSessionWithTxn(opCtx, sessionId, 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -369,9 +368,9 @@ TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithSameTxn) { ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), 2, 23, oplog1); - checkStatementExecuted(opCtx, session.get(), 2, 45, oplog2); - checkStatementExecuted(opCtx, session.get(), 2, 5, oplog3); + checkStatementExecuted(opCtx, 2, 23, oplog1); + checkStatementExecuted(opCtx, 2, 45, oplog2); + checkStatementExecuted(opCtx, 2, 5, oplog3); } TEST_F(SessionCatalogMigrationDestinationTest, ShouldOnlyStoreHistoryOfLatestTxn) { @@ -416,15 +415,17 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldOnlyStoreHistoryOfLatestTxn finishSessionExpectSuccess(&sessionMigration); auto opCtx = operationContext(); - auto session = getSessionWithTxn(opCtx, sessionId, txnNum); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, txnNum); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); checkOplogWithNestedOplog(oplog3, historyIter.next(opCtx)); ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), txnNum, 5, oplog3); + checkStatementExecuted(opCtx, txnNum, 5, oplog3); } TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithSameTxnInSeparateBatches) { @@ -468,8 +469,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithSameTxnInSeparate finishSessionExpectSuccess(&sessionMigration); auto opCtx = operationContext(); - auto session = getSessionWithTxn(opCtx, sessionId, 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -483,9 +486,9 @@ TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithSameTxnInSeparate ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), 2, 23, oplog1); - checkStatementExecuted(opCtx, session.get(), 2, 45, oplog2); - checkStatementExecuted(opCtx, session.get(), 2, 5, oplog3); + checkStatementExecuted(opCtx, 2, 23, oplog1); + checkStatementExecuted(opCtx, 2, 45, oplog2); + checkStatementExecuted(opCtx, 2, 5, oplog3); } TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithDifferentSession) { @@ -536,32 +539,40 @@ TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithDifferentSession) auto opCtx = operationContext(); { - auto session = getSessionWithTxn(opCtx, sessionId1, 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId1, 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); checkOplogWithNestedOplog(oplog1, historyIter.next(opCtx)); ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), 2, 23, oplog1); + checkStatementExecuted(opCtx, 2, 23, oplog1); } { - auto session = getSessionWithTxn(opCtx, sessionId2, 42); - const auto txnParticipant = TransactionParticipant::get(session.get()); + // XXX TODO USE A DIFFERENT OPERATION CONTEXT! + auto client2 = getServiceContext()->makeClient("client2"); + AlternativeClientRegion acr(client2); + auto opCtx2 = cc().makeOperationContext(); + setUpSessionWithTxn(opCtx2.get(), sessionId2, 42); + MongoDOperationContextSession ocs(opCtx2.get()); + auto txnParticipant = TransactionParticipant::get(opCtx2.get()); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); - checkOplogWithNestedOplog(oplog3, historyIter.next(opCtx)); + checkOplogWithNestedOplog(oplog3, historyIter.next(opCtx2.get())); ASSERT_TRUE(historyIter.hasNext()); - checkOplogWithNestedOplog(oplog2, historyIter.next(opCtx)); + checkOplogWithNestedOplog(oplog2, historyIter.next(opCtx2.get())); ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), 42, 45, oplog2); - checkStatementExecuted(opCtx, session.get(), 42, 5, oplog3); + checkStatementExecuted(opCtx2.get(), 42, 45, oplog2); + checkStatementExecuted(opCtx2.get(), 42, 5, oplog3); } } @@ -612,8 +623,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldNotNestAlreadyNestedOplog) finishSessionExpectSuccess(&sessionMigration); auto opCtx = operationContext(); - auto session = getSessionWithTxn(opCtx, sessionId, 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); @@ -625,8 +638,8 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldNotNestAlreadyNestedOplog) ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), 2, 23); - checkStatementExecuted(opCtx, session.get(), 2, 45); + checkStatementExecuted(opCtx, 2, 23); + checkStatementExecuted(opCtx, 2, 45); } TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandlePreImageFindAndModify) { @@ -663,8 +676,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandlePreImageFindA finishSessionExpectSuccess(&sessionMigration); auto opCtx = operationContext(); - auto session = getSessionWithTxn(opCtx, sessionId, 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -716,7 +731,7 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandlePreImageFindA ASSERT_TRUE(newPreImageOplog.getObject2()); ASSERT_TRUE(newPreImageOplog.getObject2().value().isEmpty()); - checkStatementExecuted(opCtx, session.get(), 2, 45, updateOplog); + checkStatementExecuted(opCtx, 2, 45, updateOplog); } TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandlePostImageFindAndModify) { @@ -752,8 +767,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandlePostImageFind finishSessionExpectSuccess(&sessionMigration); auto opCtx = operationContext(); - auto session = getSessionWithTxn(opCtx, sessionId, 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -805,7 +822,7 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandlePostImageFind ASSERT_TRUE(newPostImageOplog.getObject2()); ASSERT_TRUE(newPostImageOplog.getObject2().value().isEmpty()); - checkStatementExecuted(opCtx, session.get(), 2, 45, updateOplog); + checkStatementExecuted(opCtx, 2, 45, updateOplog); } TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandleFindAndModifySplitIn2Batches) { @@ -844,8 +861,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandleFindAndModify ASSERT_TRUE(SessionCatalogMigrationDestination::State::Done == sessionMigration.getState()); auto opCtx = operationContext(); - auto session = getSessionWithTxn(opCtx, sessionId, 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -897,7 +916,7 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldBeAbleToHandleFindAndModify ASSERT_TRUE(newPreImageOplog.getObject2()); ASSERT_TRUE(newPreImageOplog.getObject2().value().isEmpty()); - checkStatementExecuted(opCtx, session.get(), 2, 45, updateOplog); + checkStatementExecuted(opCtx, 2, 45, updateOplog); } TEST_F(SessionCatalogMigrationDestinationTest, OlderTxnShouldBeIgnored) { @@ -944,8 +963,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, OlderTxnShouldBeIgnored) { ASSERT_TRUE(SessionCatalogMigrationDestination::State::Done == sessionMigration.getState()); - auto session = getSessionWithTxn(opCtx, sessionId, 20); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 20); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -956,7 +977,7 @@ TEST_F(SessionCatalogMigrationDestinationTest, OlderTxnShouldBeIgnored) { ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), 20, 0); + checkStatementExecuted(opCtx, 20, 0); } TEST_F(SessionCatalogMigrationDestinationTest, NewerTxnWriteShouldNotBeOverwrittenByOldMigrateTxn) { @@ -1006,8 +1027,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, NewerTxnWriteShouldNotBeOverwritt finishSessionExpectSuccess(&sessionMigration); - auto session = getSessionWithTxn(opCtx, sessionId, 20); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 20); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -1016,7 +1039,7 @@ TEST_F(SessionCatalogMigrationDestinationTest, NewerTxnWriteShouldNotBeOverwritt << "newerSess"), oplog.getObject()); - checkStatementExecuted(opCtx, session.get(), 20, 0); + checkStatementExecuted(opCtx, 20, 0); } TEST_F(SessionCatalogMigrationDestinationTest, ShouldJoinProperlyAfterNetworkError) { @@ -1186,8 +1209,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, finishSessionExpectSuccess(&sessionMigration); - auto session = getSessionWithTxn(opCtx, sessionId, 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -1203,9 +1228,9 @@ TEST_F(SessionCatalogMigrationDestinationTest, ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), 2, 0); - checkStatementExecuted(opCtx, session.get(), 2, 23, oplog1); - checkStatementExecuted(opCtx, session.get(), 2, 45, oplog2); + checkStatementExecuted(opCtx, 2, 0); + checkStatementExecuted(opCtx, 2, 23, oplog1); + checkStatementExecuted(opCtx, 2, 45, oplog2); } TEST_F(SessionCatalogMigrationDestinationTest, ShouldErrorForConsecutivePreImageOplog) { @@ -1490,8 +1515,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldIgnoreAlreadyExecutedStatem finishSessionExpectSuccess(&sessionMigration); - auto session = getSessionWithTxn(opCtx, sessionId, 19); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, sessionId, 19); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -1508,9 +1535,9 @@ TEST_F(SessionCatalogMigrationDestinationTest, ShouldIgnoreAlreadyExecutedStatem ASSERT_TRUE(firstInsertOplog.getStatementId()); ASSERT_EQ(30, *firstInsertOplog.getStatementId()); - checkStatementExecuted(opCtx, session.get(), 19, 23, oplog1); - checkStatementExecuted(opCtx, session.get(), 19, 30); - checkStatementExecuted(opCtx, session.get(), 19, 45, oplog3); + checkStatementExecuted(opCtx, 19, 23, oplog1); + checkStatementExecuted(opCtx, 19, 30); + checkStatementExecuted(opCtx, 19, 45, oplog3); } TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithIncompleteHistory) { @@ -1555,8 +1582,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithIncompleteHistory ASSERT_TRUE(SessionCatalogMigrationDestination::State::Done == sessionMigration.getState()); auto opCtx = operationContext(); - auto session = getSessionWithTxn(opCtx, *sessionInfo.getSessionId(), 2); - const auto txnParticipant = TransactionParticipant::get(session.get()); + setUpSessionWithTxn(opCtx, *sessionInfo.getSessionId(), 2); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + TransactionHistoryIterator historyIter(txnParticipant->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -1570,8 +1599,8 @@ TEST_F(SessionCatalogMigrationDestinationTest, OplogEntriesWithIncompleteHistory ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session.get(), 2, 23, oplogEntries[0]); - checkStatementExecuted(opCtx, session.get(), 2, 5, oplogEntries[2]); + checkStatementExecuted(opCtx, 2, 23, oplogEntries[0]); + checkStatementExecuted(opCtx, 2, 5, oplogEntries[2]); ASSERT_THROWS(txnParticipant->checkStatementExecuted(38), AssertionException); } @@ -1586,9 +1615,10 @@ TEST_F(SessionCatalogMigrationDestinationTest, // "Start" a new transaction on session 1, so that migrating the entries above will result // in TransactionTooOld. This should not preclude the entries for session 2 from getting // applied. - auto scopedSession = - SessionCatalog::get(opCtx)->checkOutSession(opCtx, *sessionInfo1.getSessionId()); - const auto txnParticipant = TransactionParticipant::get(scopedSession.get()); + setUpSessionWithTxn(opCtx, *sessionInfo1.getSessionId(), *sessionInfo1.getTxnNumber()); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant = TransactionParticipant::get(opCtx); + txnParticipant->refreshFromStorageIfNeeded(); txnParticipant->beginOrContinue(3, boost::none, boost::none); } @@ -1639,15 +1669,25 @@ TEST_F(SessionCatalogMigrationDestinationTest, // Check nothing was written for session 1 { - auto session1 = getSessionWithTxn(opCtx, *sessionInfo1.getSessionId(), 3); - const auto txnParticipant1 = TransactionParticipant::get(session1.get()); + auto c1 = getServiceContext()->makeClient("c1"); + AlternativeClientRegion acr(c1); + auto opCtx1 = cc().makeOperationContext(); + auto opCtx = opCtx1.get(); + setUpSessionWithTxn(opCtx, *sessionInfo1.getSessionId(), 3); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant1 = TransactionParticipant::get(opCtx); ASSERT(txnParticipant1->getLastWriteOpTime().isNull()); } // Check session 2 was correctly updated { - auto session2 = getSessionWithTxn(opCtx, *sessionInfo2.getSessionId(), 15); - const auto txnParticipant2 = TransactionParticipant::get(session2.get()); + auto c2 = getServiceContext()->makeClient("c2"); + AlternativeClientRegion acr(c2); + auto opCtx2 = cc().makeOperationContext(); + auto opCtx = opCtx2.get(); + setUpSessionWithTxn(opCtx, *sessionInfo2.getSessionId(), 15); + MongoDOperationContextSession ocs(opCtx); + auto txnParticipant2 = TransactionParticipant::get(opCtx); TransactionHistoryIterator historyIter(txnParticipant2->getLastWriteOpTime()); ASSERT_TRUE(historyIter.hasNext()); @@ -1658,8 +1698,8 @@ TEST_F(SessionCatalogMigrationDestinationTest, ASSERT_FALSE(historyIter.hasNext()); - checkStatementExecuted(opCtx, session2.get(), 15, 56, oplogEntries[1]); - checkStatementExecuted(opCtx, session2.get(), 15, 55, oplogEntries[2]); + checkStatementExecuted(opCtx, 15, 56, oplogEntries[1]); + checkStatementExecuted(opCtx, 15, 55, oplogEntries[2]); } } diff --git a/src/mongo/db/session.cpp b/src/mongo/db/session.cpp deleted file mode 100644 index f72457d7f94..00000000000 --- a/src/mongo/db/session.cpp +++ /dev/null @@ -1,86 +0,0 @@ - -/** - * Copyright (C) 2018-present MongoDB, Inc. - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the Server Side Public License, version 1, - * as published by MongoDB, Inc. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * Server Side Public License for more details. - * - * You should have received a copy of the Server Side Public License - * along with this program. If not, see - * <http://www.mongodb.com/licensing/server-side-public-license>. - * - * As a special exception, the copyright holders give permission to link the - * code of portions of this program with the OpenSSL library under certain - * conditions as described in each individual source file and distribute - * linked combinations including the program with the OpenSSL library. You - * must comply with the Server Side Public License in all respects for - * all of the code used other than as permitted herein. If you modify file(s) - * with this exception, you may extend this exception to your version of the - * file(s), but you are not obligated to do so. If you do not wish to do so, - * delete this exception statement from your version. If you delete this - * exception statement from all source files in the program, then also delete - * it in the license file. - */ - -#include "mongo/platform/basic.h" - -#include "mongo/db/session.h" - -#include "mongo/db/operation_context.h" - -namespace mongo { - -Session::Session(LogicalSessionId sessionId) : _sessionId(std::move(sessionId)) {} - -OperationContext* Session::currentOperation() const { - stdx::lock_guard<stdx::mutex> lk(_mutex); - return _checkoutOpCtx; -} - -Session::KillToken Session::kill(WithLock sessionCatalogLock, ErrorCodes::Error reason) { - stdx::lock_guard<stdx::mutex> lg(_mutex); - const bool firstKiller = (0 == _killsRequested); - ++_killsRequested; - - // For currently checked-out sessions, interrupt the operation context so that the current owner - // can release the session - if (firstKiller && _checkoutOpCtx) { - const auto serviceContext = _checkoutOpCtx->getServiceContext(); - - stdx::lock_guard<Client> clientLock(*_checkoutOpCtx->getClient()); - serviceContext->killOperation(_checkoutOpCtx, reason); - } - - return KillToken(getSessionId()); -} - -bool Session::killed() const { - stdx::lock_guard<stdx::mutex> lg(_mutex); - return _killsRequested > 0; -} - -void Session::_markCheckedOut(WithLock sessionCatalogLock, OperationContext* checkoutOpCtx) { - stdx::lock_guard<stdx::mutex> lk(_mutex); - invariant(!_checkoutOpCtx); - _checkoutOpCtx = checkoutOpCtx; -} - -void Session::_markCheckedIn(WithLock sessionCatalogLock) { - stdx::lock_guard<stdx::mutex> lk(_mutex); - invariant(_checkoutOpCtx); - _checkoutOpCtx = nullptr; -} - -void Session::_markNotKilled(WithLock sessionCatalogLock, KillToken killToken) { - stdx::lock_guard<stdx::mutex> lg(_mutex); - invariant(_killsRequested > 0); - --_killsRequested; -} - -} // namespace mongo diff --git a/src/mongo/db/session.h b/src/mongo/db/session.h index ba01d6458f5..ad5a9dfb82b 100644 --- a/src/mongo/db/session.h +++ b/src/mongo/db/session.h @@ -32,13 +32,12 @@ #include "mongo/base/disallow_copying.h" #include "mongo/db/logical_session_id.h" +#include "mongo/db/operation_context.h" #include "mongo/util/concurrency/with_lock.h" #include "mongo/util/decorable.h" namespace mongo { -class OperationContext; - /** * A decorable container for state associated with an active session running on a MongoD or MongoS * server. Refer to SessionCatalog for more information on the semantics of sessions. @@ -46,10 +45,11 @@ class OperationContext; class Session : public Decorable<Session> { MONGO_DISALLOW_COPYING(Session); + friend class ObservableSession; friend class SessionCatalog; public: - explicit Session(LogicalSessionId sessionId); + explicit Session(LogicalSessionId sessionId) : _sessionId(std::move(sessionId)) {} /** * The logical session id that this object represents. @@ -58,72 +58,23 @@ public: return _sessionId; } - /** - * Returns a pointer to the current operation running on this Session, or nullptr if there is no - * operation currently running on this Session. - */ - OperationContext* currentOperation() const; - - /** - * Increments the number of "killers" for this session and returns a 'kill token' to to be - * passed later on to 'checkOutSessionForKill' method of the SessionCatalog in order to permit - * the caller to execute any kill cleanup tasks. This token is later on passed to - * '_markNotKilled' in order to decrement the number of "killers". - * - * Marking session as killed is an internal property only that will cause any further calls to - * 'checkOutSession' to block until 'checkOutSessionForKill' is called the same number of times - * as 'kill' was called and the returned scoped object destroyed. - * - * If the first killer finds the session checked-out, this method will also interrupt the - * operation context which has it checked-out. - * - * Must be called under the owning SessionCatalog's lock. - */ - struct KillToken { - KillToken(LogicalSessionId lsid) : lsidToKill(std::move(lsid)) {} - KillToken(KillToken&&) = default; - KillToken& operator=(KillToken&&) = default; - - LogicalSessionId lsidToKill; - }; - KillToken kill(WithLock sessionCatalogLock, ErrorCodes::Error reason = ErrorCodes::Interrupted); - - /** - * Returns whether 'kill' has been called on this session. - */ - bool killed() const; + OperationContext* currentOperation_forTest() const { + return _checkoutOpCtx; + } private: - /** - * Set/clear the current check-out state of the session by storing the operation which has this - * session currently checked-out. - * - * Must be called under the SessionCatalog mutex and internally will acquire the Session mutex. - */ - void _markCheckedOut(WithLock sessionCatalogLock, OperationContext* checkoutOpCtx); - void _markCheckedIn(WithLock sessionCatalogLock); - - /** - * Used by the session catalog when checking a session back in after a call to 'kill'. See the - * comments for 'kill for more details. - */ - void _markNotKilled(WithLock sessionCatalogLock, KillToken killToken); - // The id of the session with which this object is associated const LogicalSessionId _sessionId; - // Protects the member variables below. The order of lock acquisition should always be: - // - // 1) SessionCatalog mutex (if applicable) - // 2) Session mutex - // 3) Any decoration mutexes and/or the currently running Client's lock - mutable stdx::mutex _mutex; - // A pointer back to the currently running operation on this Session, or nullptr if there // is no operation currently running for the Session. + // + // May be read by holders of the SessionCatalog mutex. May only be set when clear or cleared + // when set, and the opCtx being set or cleared must have its client locked at the time. OperationContext* _checkoutOpCtx{nullptr}; - // Incremented every time 'kill' is invoked and decremented by '_markNotKilled'. + // Counter indicating the number of times ObservableSession::kill has been called on this + // session, which have not yet had a corresponding call to checkOutSessionForKill. int _killsRequested{0}; }; diff --git a/src/mongo/db/session_catalog.cpp b/src/mongo/db/session_catalog.cpp index 4da7207b735..34bd12390e3 100644 --- a/src/mongo/db/session_catalog.cpp +++ b/src/mongo/db/session_catalog.cpp @@ -45,16 +45,16 @@ namespace { const auto sessionTransactionTableDecoration = ServiceContext::declareDecoration<SessionCatalog>(); const auto operationSessionDecoration = - OperationContext::declareDecoration<boost::optional<ScopedCheckedOutSession>>(); + OperationContext::declareDecoration<boost::optional<SessionCatalog::ScopedCheckedOutSession>>(); } // namespace SessionCatalog::~SessionCatalog() { stdx::lock_guard<stdx::mutex> lg(_mutex); for (const auto& entry : _sessions) { - auto& sri = entry.second; - invariant(!sri->session.currentOperation()); - invariant(!sri->session.killed()); + ObservableSession session(lg, entry.second->session); + invariant(!session.currentOperation()); + invariant(!session._killed()); } } @@ -72,56 +72,55 @@ SessionCatalog* SessionCatalog::get(ServiceContext* service) { return &sessionTransactionTable; } -ScopedCheckedOutSession SessionCatalog::checkOutSession(OperationContext* opCtx) { - invariant(opCtx->getLogicalSessionId()); - return checkOutSession(opCtx, *opCtx->getLogicalSessionId()); -} - -ScopedCheckedOutSession SessionCatalog::checkOutSession(OperationContext* opCtx, - const LogicalSessionId& lsid) { +SessionCatalog::ScopedCheckedOutSession SessionCatalog::_checkOutSession(OperationContext* opCtx) { // This method is not supposed to be called with an already checked-out session due to risk of // deadlock + invariant(opCtx->getLogicalSessionId()); invariant(!operationSessionDecoration(opCtx)); invariant(!opCtx->lockState()->inAWriteUnitOfWork()); invariant(!opCtx->lockState()->isLocked()); stdx::unique_lock<stdx::mutex> ul(_mutex); - auto sri = _getOrCreateSessionRuntimeInfo(ul, opCtx, lsid); + auto sri = _getOrCreateSessionRuntimeInfo(ul, opCtx, *opCtx->getLogicalSessionId()); // Wait until the session is no longer checked out and until the previously scheduled kill has // completed - opCtx->waitForConditionOrInterrupt(sri->availableCondVar, ul, [&sri]() { - return !sri->session.currentOperation() && !sri->session.killed(); + opCtx->waitForConditionOrInterrupt(sri->availableCondVar, ul, [&ul, &sri]() { + ObservableSession osession(ul, sri->session); + return !osession.currentOperation() && !osession._killed(); }); - sri->session._markCheckedOut(ul, opCtx); + { + stdx::lock_guard<Client> lockClient(*opCtx->getClient()); + sri->session._checkoutOpCtx = opCtx; + } return ScopedCheckedOutSession( *this, std::move(sri), boost::none /* Not checked out for kill */); } -ScopedCheckedOutSession SessionCatalog::checkOutSessionForKill(OperationContext* opCtx, - Session::KillToken killToken) { +SessionCatalog::SessionToKill SessionCatalog::checkOutSessionForKill(OperationContext* opCtx, + KillToken killToken) { // This method is not supposed to be called with an already checked-out session due to risk of // deadlock invariant(!operationSessionDecoration(opCtx)); invariant(!opCtx->getTxnNumber()); - const auto lsid = killToken.lsidToKill; - stdx::unique_lock<stdx::mutex> ul(_mutex); - auto sri = _getOrCreateSessionRuntimeInfo(ul, opCtx, lsid); - invariant(sri->session.killed()); + auto sri = _getOrCreateSessionRuntimeInfo(ul, opCtx, killToken.lsidToKill); + invariant(ObservableSession(ul, sri->session)._killed()); // Wait until the session is no longer checked out - opCtx->waitForConditionOrInterrupt( - sri->availableCondVar, ul, [&sri]() { return !sri->session.currentOperation(); }); + opCtx->waitForConditionOrInterrupt(sri->availableCondVar, ul, [&ul, &sri]() { + return !ObservableSession(ul, sri->session).currentOperation(); + }); - invariant(!sri->session.currentOperation()); - sri->session._markCheckedOut(ul, opCtx); + { + stdx::lock_guard<Client> lockClient(*opCtx->getClient()); + sri->session._checkoutOpCtx = opCtx; + } - return ScopedCheckedOutSession( - *this, std::move(sri), std::move(killToken) /* Checked out for kill */); + return SessionToKill(ScopedCheckedOutSession(*this, std::move(sri), std::move(killToken))); } void SessionCatalog::scanSessions(const SessionKiller::Matcher& matcher, @@ -132,18 +131,18 @@ void SessionCatalog::scanSessions(const SessionKiller::Matcher& matcher, for (auto& sessionEntry : _sessions) { if (matcher.match(sessionEntry.first)) { - workerFn(lg, &sessionEntry.second->session); + workerFn({lg, sessionEntry.second->session}); } } } -Session::KillToken SessionCatalog::killSession(const LogicalSessionId& lsid) { +SessionCatalog::KillToken SessionCatalog::killSession(const LogicalSessionId& lsid) { stdx::lock_guard<stdx::mutex> lg(_mutex); auto it = _sessions.find(lsid); uassert(ErrorCodes::NoSuchSession, "Session not found", it != _sessions.end()); auto& sri = it->second; - return sri->session.kill(lg); + return ObservableSession(lg, sri->session).kill(); } std::shared_ptr<SessionCatalog::SessionRuntimeInfo> SessionCatalog::_getOrCreateSessionRuntimeInfo( @@ -157,21 +156,46 @@ std::shared_ptr<SessionCatalog::SessionRuntimeInfo> SessionCatalog::_getOrCreate } void SessionCatalog::_releaseSession(std::shared_ptr<SessionCatalog::SessionRuntimeInfo> sri, - boost::optional<Session::KillToken> killToken) { + boost::optional<SessionCatalog::KillToken> killToken) { stdx::lock_guard<stdx::mutex> lg(_mutex); // Make sure we have exactly the same session on the map and that it is still associated with an // operation context (meaning checked-out) invariant(_sessions[sri->session.getSessionId()] == sri); - invariant(sri->session.currentOperation()); - - sri->session._markCheckedIn(lg); + invariant(sri->session._checkoutOpCtx); + { + stdx::lock_guard<Client> lockClient(*sri->session._checkoutOpCtx->getClient()); + sri->session._checkoutOpCtx = nullptr; + } sri->availableCondVar.notify_all(); if (killToken) { - invariant(sri->session.killed()); - sri->session._markNotKilled(lg, std::move(*killToken)); + invariant(sri->session._killsRequested > 0); + --sri->session._killsRequested; + } +} + +OperationContext* ObservableSession::currentOperation() const { + return _session->_checkoutOpCtx; +} + +SessionCatalog::KillToken ObservableSession::kill(ErrorCodes::Error reason) const { + const bool firstKiller = (0 == _session->_killsRequested); + ++_session->_killsRequested; + + // For currently checked-out sessions, interrupt the operation context so that the current owner + // can release the session + if (firstKiller && _session->_checkoutOpCtx) { + invariant(_clientLock); + const auto serviceContext = _session->_checkoutOpCtx->getServiceContext(); + serviceContext->killOperation(_session->_checkoutOpCtx, reason); } + + return SessionCatalog::KillToken(getSessionId()); +} + +bool ObservableSession::_killed() const { + return _session->_killsRequested > 0; } OperationContextSession::OperationContextSession(OperationContext* opCtx) : _opCtx(opCtx) { @@ -217,7 +241,8 @@ void OperationContextSession::checkIn(OperationContext* opCtx) { // but destruction of the checkedOutSession must not be, as it takes the SessionCatalog mutex, // and other code may take the Client lock while holding that mutex. stdx::unique_lock<Client> lk(*opCtx->getClient()); - ScopedCheckedOutSession sessionToReleaseOutOfLock(std::move(*checkedOutSession)); + SessionCatalog::ScopedCheckedOutSession sessionToReleaseOutOfLock( + std::move(*checkedOutSession)); // This destroys the moved-from ScopedCheckedOutSession, and must be done within the client lock checkedOutSession = boost::none; @@ -229,7 +254,7 @@ void OperationContextSession::checkOut(OperationContext* opCtx) { invariant(!checkedOutSession); const auto catalog = SessionCatalog::get(opCtx); - auto scopedCheckedOutSession = catalog->checkOutSession(opCtx); + auto scopedCheckedOutSession = catalog->_checkOutSession(opCtx); // We acquire a Client lock here to guard the construction of this session so that references to // this session are safe to use while the lock is held diff --git a/src/mongo/db/session_catalog.h b/src/mongo/db/session_catalog.h index 1ead5cad736..c22c96ff8ba 100644 --- a/src/mongo/db/session_catalog.h +++ b/src/mongo/db/session_catalog.h @@ -34,7 +34,9 @@ #include <vector> #include "mongo/base/disallow_copying.h" +#include "mongo/db/client.h" #include "mongo/db/logical_session_id.h" +#include "mongo/db/operation_context.h" #include "mongo/db/session.h" #include "mongo/db/session_killer.h" #include "mongo/stdx/condition_variable.h" @@ -44,10 +46,7 @@ namespace mongo { -class OperationContext; -class ScopedSession; -class ScopedCheckedOutSession; -class ServiceContext; +class ObservableSession; /** * Keeps track of the transaction runtime state for every active session on this instance. @@ -55,10 +54,21 @@ class ServiceContext; class SessionCatalog { MONGO_DISALLOW_COPYING(SessionCatalog); - friend class ScopedSession; - friend class ScopedCheckedOutSession; + friend class ObservableSession; + friend class OperationContextSession; public: + class ScopedCheckedOutSession; + class SessionToKill; + + struct KillToken { + KillToken(LogicalSessionId lsid) : lsidToKill(std::move(lsid)) {} + KillToken(KillToken&&) = default; + KillToken& operator=(KillToken&&) = default; + + LogicalSessionId lsidToKill; + }; + SessionCatalog() = default; ~SessionCatalog(); @@ -75,28 +85,10 @@ public: void reset_forTest(); /** - * Potentially blocking call, which either creates a brand new session object (if one doesn't - * exist) or "checks-out" the existing one (if it is not currently in use or marked for kill). - * - * The 'opCtx'-only variant uses the session information stored on the operation context and the - * variant, which has the 'lsid' parameter checks-out that session id. Neither of these methods - * can be called with an already checked-out session. - * - * Checking out a session puts it in the 'checked out' state and all subsequent calls to - * checkout will block until it is checked back in. This happens when the returned object goes - * out of scope. - * - * Throws exception on errors. - */ - ScopedCheckedOutSession checkOutSession(OperationContext* opCtx); - ScopedCheckedOutSession checkOutSession(OperationContext* opCtx, const LogicalSessionId& lsid); - - /** - * See the description of 'Session::kill' for more information on the session kill usage - * pattern. + * See the description of 'ObservableSession::kill' for more information on the session kill + * usage pattern. */ - ScopedCheckedOutSession checkOutSessionForKill(OperationContext* opCtx, - Session::KillToken killToken); + SessionToKill checkOutSessionForKill(OperationContext* opCtx, KillToken killToken); /** * Iterates through the SessionCatalog under the SessionCatalog mutex and applies 'workerFn' to @@ -109,7 +101,7 @@ public: * * TODO SERVER-33850: Take Matcher out of the SessionKiller namespace. */ - using ScanSessionsCallbackFn = stdx::function<void(WithLock, Session*)>; + using ScanSessionsCallbackFn = stdx::function<void(const ObservableSession&)>; void scanSessions(const SessionKiller::Matcher& matcher, const ScanSessionsCallbackFn& workerFn); @@ -117,7 +109,7 @@ public: * Shortcut to invoke 'kill' on the specified session under the SessionCatalog mutex. Throws a * NoSuchSession exception if the session doesn't exist. */ - Session::KillToken killSession(const LogicalSessionId& lsid); + KillToken killSession(const LogicalSessionId& lsid); private: struct SessionRuntimeInfo { @@ -132,6 +124,8 @@ private: stdx::condition_variable availableCondVar; }; + ScopedCheckedOutSession _checkOutSession(OperationContext* opCtx); + /** * May release and re-acquire it zero or more times before returning. The returned * 'SessionRuntimeInfo' is guaranteed to be linked on the catalog's _txnTable as long as the @@ -144,7 +138,7 @@ private: * Makes a session, previously checked out through 'checkoutSession', available again. */ void _releaseSession(std::shared_ptr<SessionRuntimeInfo> sri, - boost::optional<Session::KillToken> killToken); + boost::optional<KillToken> killToken); stdx::mutex _mutex; @@ -153,19 +147,20 @@ private: }; /** - * Scoped object representing a checked-out session. See comments for the 'checkoutSession' method - * for more information on its behaviour. + * Scoped object representing a checked-out session. This type is an implementation detail + * of the SessionCatalog. */ -class ScopedCheckedOutSession { - MONGO_DISALLOW_COPYING(ScopedCheckedOutSession); - - friend ScopedCheckedOutSession SessionCatalog::checkOutSession(OperationContext*, - const LogicalSessionId&); - friend ScopedCheckedOutSession SessionCatalog::checkOutSessionForKill(OperationContext*, - Session::KillToken); - +class SessionCatalog::ScopedCheckedOutSession { public: + ScopedCheckedOutSession(SessionCatalog& catalog, + std::shared_ptr<SessionCatalog::SessionRuntimeInfo> sri, + boost::optional<SessionCatalog::KillToken> killToken) + : _catalog(catalog), _sri(std::move(sri)), _killToken(std::move(killToken)) {} + ScopedCheckedOutSession(ScopedCheckedOutSession&&) = default; + ScopedCheckedOutSession& operator=(ScopedCheckedOutSession&&) = delete; + ScopedCheckedOutSession(const ScopedCheckedOutSession&) = delete; + ScopedCheckedOutSession& operator=(ScopedCheckedOutSession&) = delete; ~ScopedCheckedOutSession() { if (_sri) { @@ -190,19 +185,117 @@ public: } private: - ScopedCheckedOutSession(SessionCatalog& catalog, - std::shared_ptr<SessionCatalog::SessionRuntimeInfo> sri, - boost::optional<Session::KillToken> killToken) - : _catalog(catalog), _sri(std::move(sri)), _killToken(std::move(killToken)) {} - // The owning session catalog into which the session should be checked back SessionCatalog& _catalog; std::shared_ptr<SessionCatalog::SessionRuntimeInfo> _sri; + boost::optional<SessionCatalog::KillToken> _killToken; +}; + +/** + * RAII type returned by SessionCatalog::checkOutSessionForKill. + * + * After calling kill() on an ObservableSession, let that ObservableSession go out + * of scope and in a context outside of SessionCatalog::scanSessions, call checkOutSessionForKill + * to get an instance of this type. Then, while holding that instance, perform any cleanup + * you need to perform on a session as part of killing it. More details in the description of + * ObservableSession::kill, below. + */ +class SessionCatalog::SessionToKill { +public: + SessionToKill(ScopedCheckedOutSession&& scos) : _scos(std::move(scos)) {} + Session* get() const { + return _scos.get(); + } + + const LogicalSessionId& getSessionId() const { + return get()->getSessionId(); + } + OperationContext* currentOperation_forTest() const { + return get()->currentOperation_forTest(); + } - // Only set if the session was obtained though checkOutSessionForKill - boost::optional<Session::KillToken> _killToken; +private: + ScopedCheckedOutSession _scos; }; +using SessionToKill = SessionCatalog::SessionToKill; + +/** + * This type represents access to a session inside of a scanSessions loop. + * If you have one of these, you're in a scanSessions callback context, and so + * have locked the whole catalog and, if the observed session is bound to an operation context, + * you hold that operation context's client's mutex, as well. + */ +class ObservableSession { +public: + ObservableSession(const ObservableSession&) = delete; + ObservableSession(ObservableSession&&) = delete; + ObservableSession& operator=(const ObservableSession&) = delete; + ObservableSession& operator=(ObservableSession&&) = delete; + + /** + * The logical session id that this object represents. + */ + const LogicalSessionId& getSessionId() const { + return _session->_sessionId; + } + + /** + * Returns a pointer to the current operation running on this Session, or nullptr if there is no + * operation currently running on this Session. + */ + OperationContext* currentOperation() const; + + /** + * Increments the number of "killers" for this session and returns a 'kill token' to to be + * passed later on to 'checkOutSessionForKill' method of the SessionCatalog in order to permit + * the caller to execute any kill cleanup tasks. This token is later on passed to + * '_markNotKilled' in order to decrement the number of "killers". + * + * Marking session as killed is an internal property only that will cause any further calls to + * 'checkOutSession' to block until 'checkOutSessionForKill' is called the same number of times + * as 'kill' was called and the returned scoped object destroyed. + * + * If the first killer finds the session checked-out, this method will also interrupt the + * operation context which has it checked-out. + */ + SessionCatalog::KillToken kill(ErrorCodes::Error reason = ErrorCodes::Interrupted) const; + + /** + * Returns a pointer to the Session itself. + */ + Session* get() const { + return _session; + } + +private: + friend class SessionCatalog; + + static stdx::unique_lock<Client> _lockClientForSession(WithLock, Session* session) { + if (const auto opCtx = session->_checkoutOpCtx) { + return stdx::unique_lock<Client>{*opCtx->getClient()}; + } + return {}; + } + + ObservableSession(WithLock wl, Session& session) + : _session(&session), _clientLock(_lockClientForSession(std::move(wl), _session)) {} + + /** + * Returns whether 'kill' has been called on this session. + */ + bool _killed() const; + + /** + * Used by the session catalog when checking a session back in after a call to 'kill'. See the + * comments for 'kill for more details. + */ + void _markNotKilled(WithLock sessionCatalogLock, SessionCatalog::KillToken killToken); + + Session* _session; + stdx::unique_lock<Client> _clientLock; +}; + /** * Scoped object, which checks out the session specified in the passed operation context and stores @@ -213,11 +306,17 @@ class OperationContextSession { MONGO_DISALLOW_COPYING(OperationContextSession); public: + /** + * Acquires the session with id opCtx->getLogicalSessionId(). Because a session can only be + * checked out by one user at a time, construction of OperationContextSession can block waiting + * for the desired session to be checked in by another user. + */ OperationContextSession(OperationContext* opCtx); ~OperationContextSession(); /** - * Returns the session checked out in the constructor. + * Returns the session currently checked out by "opCtx", or nullptr if the opCtx has no + * checked out session. */ static Session* get(OperationContext* opCtx); diff --git a/src/mongo/db/session_catalog_mongod.cpp b/src/mongo/db/session_catalog_mongod.cpp index c5fae50fe9c..dd6072af24e 100644 --- a/src/mongo/db/session_catalog_mongod.cpp +++ b/src/mongo/db/session_catalog_mongod.cpp @@ -74,8 +74,9 @@ auto getThreadPool(OperationContext* opCtx) { return &sessionTasksExecutor(opCtx->getServiceContext()).threadPool; } -void killSessionTokensFunction(OperationContext* opCtx, - std::shared_ptr<std::vector<Session::KillToken>> sessionKillTokens) { +void killSessionTokensFunction( + OperationContext* opCtx, + std::shared_ptr<std::vector<SessionCatalog::KillToken>> sessionKillTokens) { if (sessionKillTokens->empty()) return; @@ -105,7 +106,7 @@ void MongoDSessionCatalog::onStepUp(OperationContext* opCtx) { const auto catalog = SessionCatalog::get(opCtx); // The use of shared_ptr here is in order to work around the limitation of stdx::function that // the functor must be copyable. - auto sessionKillTokens = std::make_shared<std::vector<Session::KillToken>>(); + auto sessionKillTokens = std::make_shared<std::vector<SessionCatalog::KillToken>>(); // Scan all sessions and reacquire locks for prepared transactions. // There may be sessions that are checked out during this scan, but none of them @@ -115,14 +116,14 @@ void MongoDSessionCatalog::onStepUp(OperationContext* opCtx) { SessionKiller::Matcher matcher( KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(opCtx)}); - catalog->scanSessions(matcher, [&](WithLock sessionCatalogLock, Session* session) { - const auto txnParticipant = TransactionParticipant::get(session); + catalog->scanSessions(matcher, [&](const ObservableSession& session) { + const auto txnParticipant = TransactionParticipant::get(session.get()); if (!txnParticipant->inMultiDocumentTransaction()) { - sessionKillTokens->emplace_back(session->kill(sessionCatalogLock)); + sessionKillTokens->emplace_back(session.kill()); } if (txnParticipant->transactionIsPrepared()) { - sessionIdToReacquireLocks.emplace_back(session->getSessionId()); + sessionIdToReacquireLocks.emplace_back(session.getSessionId()); } }); killSessionTokensFunction(opCtx, sessionKillTokens); @@ -131,10 +132,12 @@ void MongoDSessionCatalog::onStepUp(OperationContext* opCtx) { // Create a new opCtx because we need an empty locker to refresh the locks. auto newClient = opCtx->getServiceContext()->makeClient("restore-prepared-txn"); AlternativeClientRegion acr(newClient); - auto newOpCtx = cc().makeOperationContext(); for (const auto& sessionId : sessionIdToReacquireLocks) { - auto scopedSessionCheckOut = catalog->checkOutSession(newOpCtx.get(), sessionId); - auto txnParticipant = TransactionParticipant::get(scopedSessionCheckOut.get()); + auto newOpCtx = cc().makeOperationContext(); + newOpCtx->setLogicalSessionId(sessionId); + MongoDOperationContextSession ocs(newOpCtx.get()); + auto txnParticipant = + TransactionParticipant::get(OperationContextSession::get(newOpCtx.get())); txnParticipant->refreshLocksForPreparedTransaction(newOpCtx.get(), false); } } @@ -195,7 +198,7 @@ void MongoDSessionCatalog::invalidateSessions(OperationContext* opCtx, // The use of shared_ptr here is in order to work around the limitation of stdx::function that // the functor must be copyable. - auto sessionKillTokens = std::make_shared<std::vector<Session::KillToken>>(); + auto sessionKillTokens = std::make_shared<std::vector<SessionCatalog::KillToken>>(); if (singleSessionDoc) { sessionKillTokens->emplace_back(catalog->killSession(LogicalSessionId::parse( @@ -203,10 +206,9 @@ void MongoDSessionCatalog::invalidateSessions(OperationContext* opCtx, } else { SessionKiller::Matcher matcher( KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(opCtx)}); - catalog->scanSessions( - matcher, [&sessionKillTokens](WithLock sessionCatalogLock, Session* session) { - sessionKillTokens->emplace_back(session->kill(sessionCatalogLock)); - }); + catalog->scanSessions(matcher, [&sessionKillTokens](const ObservableSession& session) { + sessionKillTokens->emplace_back(session.kill()); + }); } killSessionTokensFunction(opCtx, sessionKillTokens); diff --git a/src/mongo/db/session_catalog_test.cpp b/src/mongo/db/session_catalog_test.cpp index 9d9e532abad..b65a933d6c7 100644 --- a/src/mongo/db/session_catalog_test.cpp +++ b/src/mongo/db/session_catalog_test.cpp @@ -74,11 +74,11 @@ private: TEST_F(SessionCatalogTestWithDefaultOpCtx, CheckoutAndReleaseSession) { _opCtx->setLogicalSessionId(makeLogicalSessionIdForTest()); + OperationContextSession ocs(_opCtx); - auto scopedSession = catalog()->checkOutSession(_opCtx); - - ASSERT(scopedSession.get()); - ASSERT_EQ(*_opCtx->getLogicalSessionId(), scopedSession->getSessionId()); + auto session = OperationContextSession::get(_opCtx); + ASSERT(session); + ASSERT_EQ(*_opCtx->getLogicalSessionId(), session->getSessionId()); } TEST_F(SessionCatalogTestWithDefaultOpCtx, OperationContextCheckedOutSession) { @@ -120,8 +120,8 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, NestedOperationContextSession) { TEST_F(SessionCatalogTestWithDefaultOpCtx, ScanSessions) { std::vector<LogicalSessionId> lsidsFound; - const auto workerFn = [&lsidsFound](WithLock, Session* session) { - lsidsFound.push_back(session->getSessionId()); + const auto workerFn = [&lsidsFound](const ObservableSession& session) { + lsidsFound.push_back(session.getSessionId()); }; // Scan over zero Sessions. @@ -139,7 +139,8 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, ScanSessions) { stdx::async(stdx::launch::async, [this, lsid] { ThreadClient tc(getServiceContext()); auto opCtx = makeOperationContext(); - const auto unusedSession(catalog()->checkOutSession(opCtx.get(), lsid)); + opCtx->setLogicalSessionId(lsid); + OperationContextSession ocs(opCtx.get()); }).get(); } @@ -163,7 +164,8 @@ TEST_F(SessionCatalogTest, KillSessionWhenSessionIsNotCheckedOut) { // Create the session so there is something to kill { auto opCtx = makeOperationContext(); - const auto unusedSession(catalog()->checkOutSession(opCtx.get(), lsid)); + opCtx->setLogicalSessionId(lsid); + OperationContextSession ocs(opCtx.get()); } auto killToken = catalog()->killSession(lsid); @@ -182,7 +184,8 @@ TEST_F(SessionCatalogTest, KillSessionWhenSessionIsNotCheckedOut) { auto future = stdx::async(stdx::launch::async, [this, lsid] { ThreadClient tc(getServiceContext()); auto sideOpCtx = Client::getCurrent()->makeOperationContext(); - const auto unusedSession(catalog()->checkOutSession(sideOpCtx.get(), lsid)); + sideOpCtx->setLogicalSessionId(lsid); + OperationContextSession ocs(sideOpCtx.get()); }); ASSERT(stdx::future_status::ready != future.wait_for(Milliseconds(10).toSystemDuration())); @@ -190,13 +193,14 @@ TEST_F(SessionCatalogTest, KillSessionWhenSessionIsNotCheckedOut) { { auto opCtx = makeOperationContext(); auto scopedSession = catalog()->checkOutSessionForKill(opCtx.get(), std::move(killToken)); - ASSERT_EQ(opCtx.get(), scopedSession->currentOperation()); + ASSERT_EQ(opCtx.get(), scopedSession.currentOperation_forTest()); } // Make sure that session check-out after kill succeeds again { auto opCtx = makeOperationContext(); - const auto unusedSession(catalog()->checkOutSession(opCtx.get(), lsid)); + opCtx->setLogicalSessionId(lsid); + OperationContextSession ocs(opCtx.get()); } // Make sure the "regular operation" eventually is able to proceed and use the just killed @@ -225,8 +229,7 @@ TEST_F(SessionCatalogTest, KillSessionWhenSessionIsCheckedOut) { auto sideOpCtx = Client::getCurrent()->makeOperationContext(); sideOpCtx->setLogicalSessionId(lsid); sideOpCtx->setDeadlineAfterNowBy(Milliseconds(10), ErrorCodes::MaxTimeMSExpired); - - const auto unusedSession(catalog()->checkOutSession(sideOpCtx.get(), lsid)); + OperationContextSession ocs(sideOpCtx.get()); }); ASSERT_THROWS_CODE(future.get(), AssertionException, ErrorCodes::MaxTimeMSExpired); @@ -248,7 +251,8 @@ TEST_F(SessionCatalogTest, KillSessionWhenSessionIsCheckedOut) { auto future = stdx::async(stdx::launch::async, [this, lsid] { ThreadClient tc(getServiceContext()); auto sideOpCtx = Client::getCurrent()->makeOperationContext(); - const auto unusedSession(catalog()->checkOutSession(sideOpCtx.get(), lsid)); + sideOpCtx->setLogicalSessionId(lsid); + OperationContextSession ocs(sideOpCtx.get()); }); ASSERT(stdx::future_status::ready != future.wait_for(Milliseconds(10).toSystemDuration())); @@ -256,13 +260,14 @@ TEST_F(SessionCatalogTest, KillSessionWhenSessionIsCheckedOut) { { auto opCtx = makeOperationContext(); auto scopedSession = catalog()->checkOutSessionForKill(opCtx.get(), std::move(killToken)); - ASSERT_EQ(opCtx.get(), scopedSession->currentOperation()); + ASSERT_EQ(opCtx.get(), scopedSession.currentOperation_forTest()); } // Make sure that session check-out after kill succeeds again { auto opCtx = makeOperationContext(); - const auto unusedSession(catalog()->checkOutSession(opCtx.get(), lsid)); + opCtx->setLogicalSessionId(lsid); + OperationContextSession ocs(opCtx.get()); } // Make sure the "regular operation" eventually is able to proceed and use the just killed @@ -276,7 +281,8 @@ TEST_F(SessionCatalogTest, MarkSessionAsKilledCanBeCalledMoreThanOnce) { // Create the session so there is something to kill { auto opCtx = makeOperationContext(); - const auto unusedSession(catalog()->checkOutSession(opCtx.get(), lsid)); + opCtx->setLogicalSessionId(lsid); + OperationContextSession ocs(opCtx.get()); } auto killToken1 = catalog()->killSession(lsid); @@ -292,13 +298,13 @@ TEST_F(SessionCatalogTest, MarkSessionAsKilledCanBeCalledMoreThanOnce) { OperationContextSession(opCtx.get()), AssertionException, ErrorCodes::MaxTimeMSExpired); } - boost::optional<Session::KillToken> killTokenWhileSessionIsCheckedOutForKill; + boost::optional<SessionCatalog::KillToken> killTokenWhileSessionIsCheckedOutForKill; // Finish the first killer of the session { auto opCtx = makeOperationContext(); auto scopedSession = catalog()->checkOutSessionForKill(opCtx.get(), std::move(killToken1)); - ASSERT_EQ(opCtx.get(), scopedSession->currentOperation()); + ASSERT_EQ(opCtx.get(), scopedSession.currentOperation_forTest()); // Killing a session while checked out for kill should not affect the killers killTokenWhileSessionIsCheckedOutForKill.emplace(catalog()->killSession(lsid)); @@ -316,13 +322,13 @@ TEST_F(SessionCatalogTest, MarkSessionAsKilledCanBeCalledMoreThanOnce) { { auto opCtx = makeOperationContext(); auto scopedSession = catalog()->checkOutSessionForKill(opCtx.get(), std::move(killToken2)); - ASSERT_EQ(opCtx.get(), scopedSession->currentOperation()); + ASSERT_EQ(opCtx.get(), scopedSession.currentOperation_forTest()); } { auto opCtx = makeOperationContext(); auto scopedSession = catalog()->checkOutSessionForKill( opCtx.get(), std::move(*killTokenWhileSessionIsCheckedOutForKill)); - ASSERT_EQ(opCtx.get(), scopedSession->currentOperation()); + ASSERT_EQ(opCtx.get(), scopedSession.currentOperation_forTest()); } } @@ -379,7 +385,8 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, KillSessionsThroughScanSessions) { { auto sideOpCtx = Client::getCurrent()->makeOperationContext(); - const auto unusedSession(catalog()->checkOutSession(sideOpCtx.get(), lsid)); + sideOpCtx->setLogicalSessionId(lsid); + OperationContextSession ocs(sideOpCtx.get()); firstUseOfTheSessionReachedBarrier.countDownAndWait(); @@ -391,8 +398,7 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, KillSessionsThroughScanSessions) { { auto sideOpCtx = Client::getCurrent()->makeOperationContext(); sideOpCtx->setLogicalSessionId(lsid); - - const auto unusedSession(catalog()->checkOutSession(sideOpCtx.get(), lsid)); + OperationContextSession ocs(sideOpCtx.get()); } })); } @@ -402,14 +408,13 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, KillSessionsThroughScanSessions) { // Kill the first and the third sessions { - std::vector<Session::KillToken> firstAndThirdTokens; + std::vector<SessionCatalog::KillToken> firstAndThirdTokens; catalog()->scanSessions( SessionKiller::Matcher( KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(_opCtx)}), - [&lsids, &firstAndThirdTokens](WithLock sessionCatalogLock, Session* session) { - if (session->getSessionId() == lsids[0] || session->getSessionId() == lsids[2]) - firstAndThirdTokens.emplace_back( - session->kill(sessionCatalogLock, ErrorCodes::ExceededTimeLimit)); + [&lsids, &firstAndThirdTokens](const ObservableSession& session) { + if (session.getSessionId() == lsids[0] || session.getSessionId() == lsids[2]) + firstAndThirdTokens.emplace_back(session.kill(ErrorCodes::ExceededTimeLimit)); }); ASSERT_EQ(2U, firstAndThirdTokens.size()); for (auto& killToken : firstAndThirdTokens) { @@ -422,14 +427,13 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, KillSessionsThroughScanSessions) { // Kill the second session { - std::vector<Session::KillToken> secondToken; + std::vector<SessionCatalog::KillToken> secondToken; catalog()->scanSessions( SessionKiller::Matcher( KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(_opCtx)}), - [&lsids, &secondToken](WithLock sessionCatalogLock, Session* session) { - if (session->getSessionId() == lsids[1]) - secondToken.emplace_back( - session->kill(sessionCatalogLock, ErrorCodes::ExceededTimeLimit)); + [&lsids, &secondToken](const ObservableSession& session) { + if (session.getSessionId() == lsids[1]) + secondToken.emplace_back(session.kill(ErrorCodes::ExceededTimeLimit)); }); ASSERT_EQ(1U, secondToken.size()); for (auto& killToken : secondToken) { @@ -472,13 +476,13 @@ TEST_F(SessionCatalogTestWithDefaultOpCtx, ConcurrentCheckOutAndKill) { sideOpCtx->setLogicalSessionId(lsid); // Kill the session - std::vector<Session::KillToken> killTokens; - catalog()->scanSessions(SessionKiller::Matcher(KillAllSessionsByPatternSet{ - makeKillAllSessionsByPattern(sideOpCtx.get())}), - [&killTokens](WithLock sessionCatalogLock, Session* session) { - killTokens.emplace_back(session->kill( - sessionCatalogLock, ErrorCodes::InternalError)); - }); + std::vector<SessionCatalog::KillToken> killTokens; + catalog()->scanSessions( + SessionKiller::Matcher( + KillAllSessionsByPatternSet{makeKillAllSessionsByPattern(sideOpCtx.get())}), + [&killTokens](const ObservableSession& session) { + killTokens.emplace_back(session.kill(ErrorCodes::InternalError)); + }); ASSERT_EQ(1U, killTokens.size()); auto checkOutSessionForKill( catalog()->checkOutSessionForKill(sideOpCtx.get(), std::move(killTokens[0]))); diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp index 066790d2236..83361ba3be8 100644 --- a/src/mongo/db/transaction_participant.cpp +++ b/src/mongo/db/transaction_participant.cpp @@ -307,7 +307,7 @@ const LogicalSessionId& TransactionParticipant::_sessionId() const { OperationContext* TransactionParticipant::_opCtx() const { const auto* owningSession = getTransactionParticipant.owner(this); - auto* opCtx = owningSession->currentOperation(); + auto* opCtx = owningSession->currentOperation_forTest(); invariant(opCtx); return opCtx; } diff --git a/src/mongo/db/transaction_participant_test.cpp b/src/mongo/db/transaction_participant_test.cpp index b20d33efa3a..16c2be1b588 100644 --- a/src/mongo/db/transaction_participant_test.cpp +++ b/src/mongo/db/transaction_participant_test.cpp @@ -1144,9 +1144,8 @@ TEST_F(TxnParticipantTest, CannotStartNewTransactionWhilePreparedTransactionInPr ](OperationContext * newOpCtx) { newOpCtx->setLogicalSessionId(lsid); newOpCtx->setTxnNumber(txnNumberToStart); - - auto session = SessionCatalog::get(newOpCtx)->checkOutSession(newOpCtx); - auto txnParticipant = TransactionParticipant::get(session.get()); + MongoDOperationContextSession ocs(newOpCtx); + auto txnParticipant = TransactionParticipant::get(newOpCtx); ASSERT_THROWS_CODE(txnParticipant->beginOrContinue(txnNumberToStart, false, true), AssertionException, |