SERVER-33563 add RAII-style class to opt out of replication's use of ParallelBatchWriterMode

6 files changed, 49 insertions, 40 deletions

diff --git a/src/mongo/db/concurrency/d_concurrency.cpp b/src/mongo/db/concurrency/d_concurrency.cpp
index 7981cfaa2a7..a47b155780b 100644
--- a/src/mongo/db/concurrency/d_concurrency.cpp
+++ b/src/mongo/db/concurrency/d_concurrency.cpp
@@ -306,14 +306,7 @@ void Lock::OplogIntentWriteLock::serializeIfNeeded() {
 
 Lock::ParallelBatchWriterMode::ParallelBatchWriterMode(Locker* lockState)
     : _pbwm(lockState, resourceIdParallelBatchWriterMode, MODE_X),
-      _lockState(lockState),
-      _orginalShouldConflict(_lockState->shouldConflictWithSecondaryBatchApplication()) {
-    _lockState->setShouldConflictWithSecondaryBatchApplication(false);
-}
-
-Lock::ParallelBatchWriterMode::~ParallelBatchWriterMode() {
-    _lockState->setShouldConflictWithSecondaryBatchApplication(_orginalShouldConflict);
-}
+      _shouldNotConflictBlock(lockState) {}
 
 void Lock::ResourceLock::lock(LockMode mode) {
     invariant(_result == LOCK_INVALID);
diff --git a/src/mongo/db/concurrency/d_concurrency.h b/src/mongo/db/concurrency/d_concurrency.h
index 9f6a1010efc..469dbb76769 100644
--- a/src/mongo/db/concurrency/d_concurrency.h
+++ b/src/mongo/db/concurrency/d_concurrency.h
@@ -388,12 +388,10 @@ public:
 
     public:
         explicit ParallelBatchWriterMode(Locker* lockState);
-        ~ParallelBatchWriterMode();
 
     private:
         ResourceLock _pbwm;
-        Locker* const _lockState;
-        const bool _orginalShouldConflict;
+        ShouldNotConflictWithSecondaryBatchApplicationBlock _shouldNotConflictBlock;
     };
 };
 
diff --git a/src/mongo/db/concurrency/locker.h b/src/mongo/db/concurrency/locker.h
index ea6a9f6f688..673bb165e2d 100644
--- a/src/mongo/db/concurrency/locker.h
+++ b/src/mongo/db/concurrency/locker.h
@@ -432,4 +432,26 @@ private:
     Locker* const _locker;
 };
 
+/**
+ * RAII-style class to opt out of replication's use of ParallelBatchWriterMode.
+ */
+class ShouldNotConflictWithSecondaryBatchApplicationBlock {
+    MONGO_DISALLOW_COPYING(ShouldNotConflictWithSecondaryBatchApplicationBlock);
+
+public:
+    explicit ShouldNotConflictWithSecondaryBatchApplicationBlock(Locker* lockState)
+        : _lockState(lockState),
+          _originalShouldConflict(_lockState->shouldConflictWithSecondaryBatchApplication()) {
+        _lockState->setShouldConflictWithSecondaryBatchApplication(false);
+    }
+
+    ~ShouldNotConflictWithSecondaryBatchApplicationBlock() {
+        _lockState->setShouldConflictWithSecondaryBatchApplication(_originalShouldConflict);
+    }
+
+private:
+    Locker* const _lockState;
+    const bool _originalShouldConflict;
+};
+
 }  // namespace mongo
diff --git a/src/mongo/db/ftdc/collector.cpp b/src/mongo/db/ftdc/collector.cpp
index c1e0c47158d..4f564c22192 100644
--- a/src/mongo/db/ftdc/collector.cpp
+++ b/src/mongo/db/ftdc/collector.cpp
@@ -66,7 +66,7 @@ std::tuple<BSONObj, Date_t> FTDCCollectorCollection::collect(Client* client) {
     // batches. This is desirable because we want to be able to collect data in the middle of
     // batches that are taking a long time.
     auto opCtx = client->makeOperationContext();
-    opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);
+    ShouldNotConflictWithSecondaryBatchApplicationBlock shouldNotConflictBlock(opCtx->lockState());
     opCtx->lockState()->setShouldAcquireTicket(false);
 
     for (auto& collector : _collectors) {
diff --git a/src/mongo/db/index_builder.cpp b/src/mongo/db/index_builder.cpp
index bde3e35dd1d..0a6249ea830 100644
--- a/src/mongo/db/index_builder.cpp
+++ b/src/mongo/db/index_builder.cpp
@@ -82,24 +82,23 @@ void IndexBuilder::run() {
     Client::initThread(name().c_str());
     LOG(2) << "IndexBuilder building index " << _index;
 
-    const ServiceContext::UniqueOperationContext opCtxPtr = cc().makeOperationContext();
-    OperationContext& opCtx = *opCtxPtr;
-    opCtx.lockState()->setShouldConflictWithSecondaryBatchApplication(false);
+    auto opCtx = cc().makeOperationContext();
+    ShouldNotConflictWithSecondaryBatchApplicationBlock shouldNotConflictBlock(opCtx->lockState());
 
-    AuthorizationSession::get(opCtx.getClient())->grantInternalAuthorization();
+    AuthorizationSession::get(opCtx->getClient())->grantInternalAuthorization();
 
     {
-        stdx::lock_guard<Client> lk(*opCtx.getClient());
-        CurOp::get(opCtx)->setNetworkOp_inlock(dbInsert);
+        stdx::lock_guard<Client> lk(*(opCtx->getClient()));
+        CurOp::get(opCtx.get())->setNetworkOp_inlock(dbInsert);
     }
     NamespaceString ns(_index["ns"].String());
 
-    Lock::DBLock dlk(&opCtx, ns.db(), MODE_X);
-    OldClientContext ctx(&opCtx, ns.getSystemIndexesCollection());
+    Lock::DBLock dlk(opCtx.get(), ns.db(), MODE_X);
+    OldClientContext ctx(opCtx.get(), ns.getSystemIndexesCollection());
 
-    Database* db = dbHolder().get(&opCtx, ns.db().toString());
+    Database* db = dbHolder().get(opCtx.get(), ns.db().toString());
 
-    Status status = _build(&opCtx, db, true, &dlk);
+    Status status = _build(opCtx.get(), db, true, &dlk);
     if (!status.isOK()) {
         error() << "IndexBuilder could not build index: " << redact(status);
         fassert(28555, ErrorCodes::isInterruption(status.code()));
diff --git a/src/mongo/db/repl/sync_tail.cpp b/src/mongo/db/repl/sync_tail.cpp
index a871cfb79e8..339e40df721 100644
--- a/src/mongo/db/repl/sync_tail.cpp
+++ b/src/mongo/db/repl/sync_tail.cpp
@@ -480,10 +480,10 @@ void scheduleWritesToOplog(OperationContext* opCtx,
         // guarantees that 'ops' will stay in scope until the spawned threads complete.
         return [&ops, begin, end] {
             initializeWriterThread();
-            const auto opCtxHolder = cc().makeOperationContext();
-            const auto opCtx = opCtxHolder.get();
-            opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);
-            UnreplicatedWritesBlock uwb(opCtx);
+            auto opCtx = cc().makeOperationContext();
+            UnreplicatedWritesBlock uwb(opCtx.get());
+            ShouldNotConflictWithSecondaryBatchApplicationBlock shouldNotConflictBlock(
+                opCtx->lockState());
 
             std::vector<InsertStatement> docs;
             docs.reserve(end - begin);
@@ -494,9 +494,10 @@ void scheduleWritesToOplog(OperationContext* opCtx,
                     ops[i].raw, ops[i].getOpTime().getTimestamp(), ops[i].getOpTime().getTerm()});
             }
 
-            fassertStatusOK(40141,
-                            StorageInterface::get(opCtx)->insertDocuments(
-                                opCtx, NamespaceString::kRsOplogNamespace, docs));
+            fassertStatusOK(
+                40141,
+                StorageInterface::get(opCtx.get())
+                    ->insertDocuments(opCtx.get(), NamespaceString::kRsOplogNamespace, docs));
         };
     };
 
@@ -539,11 +540,10 @@ void scheduleTxnTableUpdates(OperationContext* opCtx,
 
         threadPool->schedule([&record]() {
             initializeWriterThread();
-            const auto opCtxHolder = cc().makeOperationContext();
-            const auto opCtx = opCtxHolder.get();
-            opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);
-
-            Session::updateSessionRecordOnSecondary(opCtx, record);
+            auto opCtx = cc().makeOperationContext();
+            ShouldNotConflictWithSecondaryBatchApplicationBlock shouldNotConflictBlock(
+                opCtx->lockState());
+            Session::updateSessionRecordOnSecondary(opCtx.get(), record);
         });
     }
 }
@@ -1278,9 +1278,7 @@ Status multiSyncApply_noAbort(OperationContext* opCtx,
                               SyncApplyFn syncApply) {
     UnreplicatedWritesBlock uwb(opCtx);
     DisableDocumentValidation validationDisabler(opCtx);
-
-    // Allow us to get through the magic barrier.
-    opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);
+    ShouldNotConflictWithSecondaryBatchApplicationBlock shouldNotConflictBlock(opCtx->lockState());
 
     // Sort the oplog entries by namespace, so that entries from the same namespace will be next to
     // each other in the list.
@@ -1466,13 +1464,12 @@ Status multiInitialSyncApply_noAbort(OperationContext* opCtx,
                                      WorkerMultikeyPathInfo* workerMultikeyPathInfo) {
     UnreplicatedWritesBlock uwb(opCtx);
     DisableDocumentValidation validationDisabler(opCtx);
+    ShouldNotConflictWithSecondaryBatchApplicationBlock shouldNotConflictBlock(opCtx->lockState());
+
     {  // Ensure that the MultikeyPathTracker stops tracking paths.
         ON_BLOCK_EXIT([opCtx] { MultikeyPathTracker::get(opCtx).stopTrackingMultikeyPathInfo(); });
         MultikeyPathTracker::get(opCtx).startTrackingMultikeyPathInfo();
 
-        // allow us to get through the magic barrier
-        opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);
-
         for (auto it = ops->begin(); it != ops->end(); ++it) {
             auto& entry = **it;
             try {