summaryrefslogtreecommitdiff
path: root/src/mongo/db/vector_clock_mongod.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/vector_clock_mongod.cpp')
-rw-r--r--src/mongo/db/vector_clock_mongod.cpp124
1 files changed, 28 insertions, 96 deletions
diff --git a/src/mongo/db/vector_clock_mongod.cpp b/src/mongo/db/vector_clock_mongod.cpp
index c951daf3258..6ef97de2b27 100644
--- a/src/mongo/db/vector_clock_mongod.cpp
+++ b/src/mongo/db/vector_clock_mongod.cpp
@@ -31,7 +31,6 @@
#include "mongo/platform/basic.h"
-#include "mongo/db/concurrency/lock_state.h"
#include "mongo/db/logical_time_validator.h"
#include "mongo/db/persistent_task_store.h"
#include "mongo/db/repl/replica_set_aware_service.h"
@@ -49,15 +48,14 @@ namespace mongo {
namespace {
class VectorClockMongoD : public VectorClockMutable,
- public ReplicaSetAwareService<VectorClockMongoD>,
- public std::enable_shared_from_this<VectorClockMongoD> {
+ public ReplicaSetAwareService<VectorClockMongoD> {
VectorClockMongoD(const VectorClockMongoD&) = delete;
VectorClockMongoD& operator=(const VectorClockMongoD&) = delete;
public:
static VectorClockMongoD* get(ServiceContext* serviceContext);
- VectorClockMongoD(ServiceContext* ctx);
+ VectorClockMongoD();
virtual ~VectorClockMongoD();
private:
@@ -97,7 +95,7 @@ private:
void onStartup(OperationContext* opCtx) override {}
void onInitialDataAvailable(OperationContext* opCtx, bool isMajorityDataAvailable) override;
- void onShutdown() override;
+ void onShutdown() override {}
void onStepUpBegin(OperationContext* opCtx, long long term) override;
void onStepUpComplete(OperationContext* opCtx, long long term) override {}
void onStepDown() override;
@@ -141,7 +139,7 @@ private:
*/
SharedSemiFuture<void> _enqueueWaiterAndScheduleLoopIfNeeded(stdx::unique_lock<Mutex> ul,
VectorTime time);
- Future<void> _doWhileQueueNotEmptyOrError();
+ Future<void> _doWhileQueueNotEmptyOrError(ServiceContext* service);
// Protects the shared state below
Mutex _mutex = MONGO_MAKE_LATCH("VectorClockMongoD::_mutex");
@@ -162,14 +160,9 @@ private:
// Queue ordered in increasing order of the VectorTimes, which are waiting to be persisted
using Queue = std::map<ComparableVectorTime, std::unique_ptr<SharedPromise<void>>>;
Queue _queue;
-
- ServiceContext* _serviceContext;
-
- AtomicWord<bool> _shutdownInitiated{false};
};
-const auto vectorClockMongoDDecoration =
- ServiceContext::declareDecoration<std::shared_ptr<VectorClockMongoD>>();
+const auto vectorClockMongoDDecoration = ServiceContext::declareDecoration<VectorClockMongoD>();
const ReplicaSetAwareServiceRegistry::Registerer<VectorClockMongoD>
vectorClockMongoDServiceRegisterer("VectorClockMongoD-ReplicaSetAwareServiceRegistration");
@@ -178,20 +171,16 @@ const ServiceContext::ConstructorActionRegisterer vectorClockMongoDRegisterer(
"VectorClockMongoD-VectorClockRegistration",
{},
[](ServiceContext* service) {
- VectorClockMongoD::registerVectorClockOnServiceContext(service,
- VectorClockMongoD::get(service));
+ VectorClockMongoD::registerVectorClockOnServiceContext(
+ service, &vectorClockMongoDDecoration(service));
},
{});
VectorClockMongoD* VectorClockMongoD::get(ServiceContext* serviceContext) {
- auto& clock = vectorClockMongoDDecoration(serviceContext);
- if (!clock) {
- clock = std::make_shared<VectorClockMongoD>(serviceContext);
- }
- return clock.get();
+ return &vectorClockMongoDDecoration(serviceContext);
}
-VectorClockMongoD::VectorClockMongoD(ServiceContext* ctx) : _serviceContext(ctx) {}
+VectorClockMongoD::VectorClockMongoD() = default;
VectorClockMongoD::~VectorClockMongoD() = default;
@@ -243,16 +232,6 @@ void VectorClockMongoD::onInitialDataAvailable(OperationContext* opCtx,
}
}
-void VectorClockMongoD::onShutdown() {
- stdx::lock_guard lg(_mutex);
- _shutdownInitiated.store(true);
- for (auto& [_, promise] : _queue) {
- promise->setError(
- Status(ErrorCodes::ShutdownInProgress, "Not persisting vector clock due to shutdown"));
- }
- _queue.clear();
-}
-
void VectorClockMongoD::onBecomeArbiter() {
// The node has become an arbiter, hence will not need logical clock for external operations.
_disable();
@@ -318,60 +297,28 @@ VectorClock::VectorTime VectorClockMongoD::recoverDirect(OperationContext* opCtx
SharedSemiFuture<void> VectorClockMongoD::_enqueueWaiterAndScheduleLoopIfNeeded(
stdx::unique_lock<Mutex> ul, VectorTime time) {
- if (_shutdownInitiated.load()) {
- return Future<void>().makeReady(
- Status(ErrorCodes::ShutdownInProgress, "Not persisting vector clock due to shutdown"));
- }
-
auto [it, unusedEmplaced] =
_queue.try_emplace({std::move(time)}, std::make_unique<SharedPromise<void>>());
- auto future = it->second->getFuture();
-
if (!_loopScheduled) {
_loopScheduled = true;
auto joinPreviousLoop(_currentWhileLoop ? std::move(*_currentWhileLoop)
: Future<void>::makeReady());
- _currentWhileLoop.emplace(
- std::move(joinPreviousLoop).onCompletion([this, _ = shared_from_this()](auto) {
- return _doWhileQueueNotEmptyOrError();
- }));
+ _currentWhileLoop.emplace(std::move(joinPreviousLoop).onCompletion([this](auto) {
+ return _doWhileQueueNotEmptyOrError(vectorClockMongoDDecoration.owner(this));
+ }));
}
- return future;
+ return it->second->getFuture();
}
-Future<void> VectorClockMongoD::_doWhileQueueNotEmptyOrError() {
- if (_shutdownInitiated.load()) {
- return Future<void>::makeReady(
- Status(ErrorCodes::ShutdownInProgress, "Not persisting vector clock due to shutdown"));
- }
-
- // This lambda is only called from the .then and .onError functions.
- // The only possible way we can lock the mutex if the _shutdownInitiated is true when the
- // setting of _shutdownInitiated runs parallel with the .then or .onError.
- // That means the .then or .onError is already scheduled when the onShutdown happens.
- // Since the threadpool stops later then the onShutdown, the already scheduled tasks will run
- // on different thread, so it is safe to lock the mutex here.
- const auto checkShutdownStatusAndLockMutex =
- [this, _ = shared_from_this()]() -> boost::optional<stdx::unique_lock<Mutex>> {
- if (_shutdownInitiated.load()) {
- return boost::none;
- }
- return stdx::unique_lock(_mutex);
- };
-
+Future<void> VectorClockMongoD::_doWhileQueueNotEmptyOrError(ServiceContext* service) {
auto [p, f] = makePromiseFuture<VectorTime>();
auto future = std::move(f)
- .then([this, _ = shared_from_this(), checkShutdownStatusAndLockMutex](
- VectorTime newDurableTime) {
- auto ul = checkShutdownStatusAndLockMutex();
- if (!ul) {
- return;
- }
-
+ .then([this](VectorTime newDurableTime) {
+ stdx::unique_lock ul(_mutex);
_durableTime.emplace(newDurableTime);
ComparableVectorTime time{*_durableTime};
@@ -383,34 +330,24 @@ Future<void> VectorClockMongoD::_doWhileQueueNotEmptyOrError() {
promises.emplace_back(std::move(it->second));
it = _queue.erase(it);
}
- ul->unlock();
+ ul.unlock();
for (auto& p : promises)
p->emplaceValue();
})
- .onError([this, _ = shared_from_this(), checkShutdownStatusAndLockMutex](
- Status status) {
- auto ul = checkShutdownStatusAndLockMutex();
- if (!ul) {
- return;
- }
-
+ .onError([this](Status status) {
+ stdx::unique_lock ul(_mutex);
std::vector<Queue::value_type::second_type> promises;
for (auto it = _queue.begin(); it != _queue.end();) {
promises.emplace_back(std::move(it->second));
it = _queue.erase(it);
}
- ul->unlock();
+ ul.unlock();
for (auto& p : promises)
p->setError(status);
})
- .onCompletion([this, _ = shared_from_this()](auto) {
- if (_shutdownInitiated.load()) {
- return Future<void>::makeReady(
- Status(ErrorCodes::ShutdownInProgress,
- "Not persisting vector clock due to shutdown"));
- }
+ .onCompletion([this, service](auto) {
{
stdx::lock_guard lg(_mutex);
if (_queue.empty()) {
@@ -418,18 +355,18 @@ Future<void> VectorClockMongoD::_doWhileQueueNotEmptyOrError() {
return Future<void>::makeReady();
}
}
- return _doWhileQueueNotEmptyOrError();
+ return _doWhileQueueNotEmptyOrError(service);
});
// Blocking work to recover and/or persist the current vector time
- ExecutorFuture<void>(Grid::get(_serviceContext)->getExecutorPool()->getFixedExecutor())
- .then([this, _ = shared_from_this()] {
+ ExecutorFuture<void>(Grid::get(service)->getExecutorPool()->getFixedExecutor())
+ .then([this, service] {
auto mustRecoverDurableTime = [&] {
stdx::lock_guard lg(_mutex);
return !_durableTime;
}();
- ThreadClient tc("VectorClockStateOperation", _serviceContext);
+ ThreadClient tc("VectorClockStateOperation", service);
{
stdx::lock_guard<Client> lk(*tc.get());
@@ -439,11 +376,6 @@ Future<void> VectorClockMongoD::_doWhileQueueNotEmptyOrError() {
const auto opCtxHolder = tc->makeOperationContext();
auto* const opCtx = opCtxHolder.get();
- // This code is used by the TransactionCoordinator. As a result, we need to skip ticket
- // acquisition in order to prevent possible deadlock when participants are in the
- // prepared state. See SERVER-82883 and SERVER-60682.
- SkipTicketAcquisitionForLock skipTicketAcquisition(opCtx);
-
if (mustRecoverDurableTime) {
return recoverDirect(opCtx);
}
@@ -462,9 +394,9 @@ Future<void> VectorClockMongoD::_doWhileQueueNotEmptyOrError() {
return vectorTime;
})
- .getAsync(
- [this, _ = shared_from_this(), promise = std::move(p)](
- StatusWith<VectorTime> swResult) mutable { promise.setFrom(std::move(swResult)); });
+ .getAsync([this, promise = std::move(p)](StatusWith<VectorTime> swResult) mutable {
+ promise.setFrom(std::move(swResult));
+ });
return future;
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 18:09:26 +0000