SERVER-20866: Fix race in rollback of oplog insertions

3 files changed, 48 insertions, 23 deletions

diff --git a/src/mongo/db/concurrency/d_concurrency.cpp b/src/mongo/db/concurrency/d_concurrency.cpp
index d6dffb01c9d..7acd55302ba 100644
--- a/src/mongo/db/concurrency/d_concurrency.cpp
+++ b/src/mongo/db/concurrency/d_concurrency.cpp
@@ -166,29 +166,14 @@ void Lock::CollectionLock::relockWithMode(LockMode mode, Lock::DBLock& dbLock) {
     }
 }
 
-namespace {
-boost::mutex oplogSerialization;  // for OplogIntentWriteLock
-}  // namespace
-
-Lock::OplogIntentWriteLock::OplogIntentWriteLock(Locker* lockState)
-    : _lockState(lockState), _serialized(false) {
+Lock::OplogIntentWriteLock::OplogIntentWriteLock(Locker* lockState) : _lockState(lockState) {
     _lockState->lock(resourceIdOplog, MODE_IX);
 }
 
 Lock::OplogIntentWriteLock::~OplogIntentWriteLock() {
-    if (_serialized) {
-        oplogSerialization.unlock();
-    }
     _lockState->unlock(resourceIdOplog);
 }
 
-void Lock::OplogIntentWriteLock::serializeIfNeeded() {
-    if (!supportsDocLocking() && !_serialized) {
-        oplogSerialization.lock();
-        _serialized = true;
-    }
-}
-
 Lock::ParallelBatchWriterMode::ParallelBatchWriterMode(Locker* lockState)
     : _pbwm(lockState, resourceIdParallelBatchWriterMode, MODE_X) {}
 
diff --git a/src/mongo/db/concurrency/d_concurrency.h b/src/mongo/db/concurrency/d_concurrency.h
index 5d8fb4f355a..eb1658bd793 100644
--- a/src/mongo/db/concurrency/d_concurrency.h
+++ b/src/mongo/db/concurrency/d_concurrency.h
@@ -238,10 +238,9 @@ public:
     };
 
     /**
-     * Like the CollectionLock, but optimized for the local oplog. Always locks in MODE_IX,
-     * must call serializeIfNeeded() before doing any concurrent operations in order to
-     * support storage engines without document level locking. It is an error, checked with a
-     * dassert(), to not have a suitable database lock when taking this lock.
+     * Like the CollectionLock, but optimized for the local oplog. Always locks in MODE_IX.
+     * This means that on storage engines without document level locking, an extra critical section
+     * is needed to ensure serialization until the unit of work is committed or rolled back.
      */
     class OplogIntentWriteLock {
         MONGO_DISALLOW_COPYING(OplogIntentWriteLock);
@@ -249,11 +248,9 @@ public:
     public:
         explicit OplogIntentWriteLock(Locker* lockState);
         ~OplogIntentWriteLock();
-        void serializeIfNeeded();
 
     private:
         Locker* const _lockState;
-        bool _serialized;
     };
 
 
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 5e6e78418f1..dbf1e83d24a 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -37,6 +37,8 @@
 #include <deque>
 #include <vector>
 
+#include <boost/thread/recursive_mutex.hpp>
+
 #include "mongo/db/auth/action_set.h"
 #include "mongo/db/auth/action_type.h"
 #include "mongo/db/auth/authorization_manager.h"
@@ -182,6 +184,42 @@ private:
     BSONObj _oField;
 };
 
+namespace {
+boost::recursive_mutex oplogSerialization;  // for OplogIntentWriteLock
+
+/**
+ *  This implements a critical section that extends until commit/rollback.
+ *  Used to avoid the condvar overhead of using a MODE_X lock on this critical path.
+ */
+class OplogSerialization : public RecoveryUnit::Change {
+public:
+    OplogSerialization() {
+        oplogSerialization.lock();
+        _locked = true;
+    }
+
+    ~OplogSerialization() {
+        invariant(!_locked);
+    }
+
+    virtual void commit() {
+        // use explicit unlock here rather than using a lock_guard, as we want to unlock ASAP
+        oplogSerialization.unlock();
+        _locked = false;
+    }
+
+    virtual void rollback() {
+        oplogSerialization.unlock();
+        _locked = false;
+        log() << "rolling back insert into oplog, as transaction aborted";
+    }
+
+private:
+    bool _locked;
+};
+}  // namespace
+
+
 /* we write to local.oplog.rs:
      { ts : ..., h: ..., v: ..., op: ..., etc }
    ts: an OpTime timestamp
@@ -234,7 +272,12 @@ void _logOpRS(OperationContext* txn,
         fassertFailed(17405);
     }
 
-    oplogLk.serializeIfNeeded();
+
+    // Need to be in a critical section until commit or rollback on non-doc-locking engines
+    if (!supportsDocLocking()) {
+        txn->recoveryUnit()->registerChange(new OplogSerialization);
+    }
+
     std::pair<OpTime, long long> slot =
         getNextOpTime(txn, localOplogRSCollection, ns, replCoord, opstr);