SERVER-18590 Implement lock & unlock for replSets distlock manager

20 files changed, 1262 insertions, 128 deletions

diff --git a/src/mongo/s/catalog/SConscript b/src/mongo/s/catalog/SConscript
index cb074a225e7..d950e70365f 100644
--- a/src/mongo/s/catalog/SConscript
+++ b/src/mongo/s/catalog/SConscript
@@ -5,6 +5,7 @@ Import("env")
 env.SConscript(
     dirs=[
         'legacy',
+        'replset',
     ],
 )
 
@@ -111,3 +112,13 @@ env.CppUnitTest(
     ]
 )
 
+env.Library(
+    target='dist_lock_catalog_mock',
+    source=[
+        'dist_lock_catalog_mock.cpp',
+    ],
+    LIBDEPS=[
+        'dist_lock_catalog',
+    ]
+)
+
diff --git a/src/mongo/s/catalog/dist_lock_catalog_mock.cpp b/src/mongo/s/catalog/dist_lock_catalog_mock.cpp
new file mode 100644
index 00000000000..7525098cd82
--- /dev/null
+++ b/src/mongo/s/catalog/dist_lock_catalog_mock.cpp
@@ -0,0 +1,161 @@
+/**
+ *    Copyright (C) 2015 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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/s/catalog/dist_lock_catalog_mock.h"
+
+#include "mongo/base/status.h"
+#include "mongo/base/status_with.h"
+#include "mongo/s/type_lockpings.h"
+#include "mongo/s/type_locks.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/mongoutils/str.h"
+
+namespace mongo {
+
+namespace {
+    Status kBadRetValue(ErrorCodes::InternalError, "no return value");
+    StatusWith<LocksType> kLocksTypeBadRetValue(kBadRetValue);
+
+    void noGrabLockFuncSet(StringData lockID,
+                           const OID& lockSessionID,
+                           StringData who,
+                           StringData processId,
+                           Date_t time,
+                           StringData why) {
+        FAIL(str::stream() << "grabLock not expected to be called. "
+                           << "lockID: " << lockID
+                           << ", who: " << who
+                           << ", processId: " << processId
+                           << ", why: " << why);
+    };
+
+    void noUnLockFuncSet(const OID& lockSessionID) {
+        FAIL(str::stream() << "unlock not expected to be called. "
+                           << "lockSessionID: " << lockSessionID);
+    };
+} // unnamed namespace
+
+    DistLockCatalogMock::DistLockCatalogMock():
+            _grabLockChecker(noGrabLockFuncSet),
+            _grabLockReturnValue(kLocksTypeBadRetValue),
+            _unlockChecker(noUnLockFuncSet),
+            _unlockReturnValue(kBadRetValue) {
+    }
+
+    DistLockCatalogMock::~DistLockCatalogMock() {
+    }
+
+    StatusWith<LockpingsType> DistLockCatalogMock::getPing(StringData processID) {
+        invariant(false);
+        return {ErrorCodes::InternalError, "not yet implemented"};
+    }
+
+    Status DistLockCatalogMock::ping(StringData processID, Date_t ping) {
+        invariant(false);
+        return {ErrorCodes::InternalError, "not yet implemented"};
+    }
+
+    StatusWith<LocksType> DistLockCatalogMock::grabLock(StringData lockID,
+                                                        const OID& lockSessionID,
+                                                        StringData who,
+                                                        StringData processId,
+                                                        Date_t time,
+                                                        StringData why) {
+        auto ret = kLocksTypeBadRetValue;
+        GrabLockFunc checkerFunc = noGrabLockFuncSet;
+
+        {
+            stdx::lock_guard<stdx::mutex> ul(_mutex);
+            ret = _grabLockReturnValue;
+            checkerFunc = _grabLockChecker;
+        }
+
+        checkerFunc(lockID, lockSessionID, who, processId, time, why);
+        return ret;
+    }
+
+    StatusWith<LocksType> DistLockCatalogMock::overtakeLock(StringData lockID,
+                                                            const OID& lockSessionID,
+                                                            const OID& lockTS,
+                                                            StringData who,
+                                                            StringData processId,
+                                                            Date_t time,
+                                                            StringData why) {
+        invariant(false);
+        return {ErrorCodes::InternalError, "not yet implemented"};
+    }
+
+    Status DistLockCatalogMock::unlock(const OID& lockSessionID) {
+        auto ret = kBadRetValue;
+        UnlockFunc checkerFunc = noUnLockFuncSet;
+
+        {
+            stdx::lock_guard<stdx::mutex> ul(_mutex);
+            ret = _unlockReturnValue;
+            checkerFunc = _unlockChecker;
+        }
+
+        checkerFunc(lockSessionID);
+        return ret;
+    }
+
+    StatusWith<DistLockCatalog::ServerInfo> DistLockCatalogMock::getServerInfo() {
+        invariant(false);
+        return {ErrorCodes::InternalError, "not yet implemented"};
+    }
+
+    StatusWith<LocksType> DistLockCatalogMock::getLockByTS(const OID& ts) {
+        invariant(false);
+        return {ErrorCodes::InternalError, "not yet implemented"};
+    }
+
+    void DistLockCatalogMock::setSucceedingExpectedGrabLock(
+            DistLockCatalogMock::GrabLockFunc checkerFunc,
+            StatusWith<LocksType> returnThis) {
+        stdx::lock_guard<stdx::mutex> ul(_mutex);
+        _grabLockChecker = checkerFunc;
+        _grabLockReturnValue = returnThis;
+    }
+
+    void DistLockCatalogMock::expectNoGrabLock() {
+        stdx::lock_guard<stdx::mutex> ul(_mutex);
+        _grabLockChecker = noGrabLockFuncSet;
+        _grabLockReturnValue = kLocksTypeBadRetValue;
+    }
+
+    void DistLockCatalogMock::setSucceedingExpectedUnLock(
+            DistLockCatalogMock::UnlockFunc checkerFunc,
+            Status returnThis) {
+        stdx::lock_guard<stdx::mutex> ul(_mutex);
+        _unlockChecker = checkerFunc;
+        _unlockReturnValue = returnThis;
+    }
+
+}
diff --git a/src/mongo/s/catalog/dist_lock_catalog_mock.h b/src/mongo/s/catalog/dist_lock_catalog_mock.h
new file mode 100644
index 00000000000..9642b27f012
--- /dev/null
+++ b/src/mongo/s/catalog/dist_lock_catalog_mock.h
@@ -0,0 +1,132 @@
+/**
+ *    Copyright (C) 2015 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#pragma once
+
+#include "mongo/base/status_with.h"
+#include "mongo/s/catalog/dist_lock_catalog.h"
+#include "mongo/s/type_lockpings.h"
+#include "mongo/s/type_locks.h"
+#include "mongo/stdx/functional.h"
+#include "mongo/stdx/mutex.h"
+
+namespace mongo {
+
+    /**
+     * Mock implementation of DistLockCatalog for testing.
+     *
+     * Example usage:
+     *
+     * DistLockCatalogMock mock;
+     * LocksType badLock;
+     * mock.setSucceedingExpectedGrabLock([](StringData lockID,
+     *                                       const OID& lockSessionID,
+     *                                       StringData who,
+     *                                       StringData processId,
+     *                                       Date_t time,
+     *                                       StringData why) {
+     *   ASSERT_EQUALS("test", lockID);
+     * }, badLock);
+     *
+     * mock.grabLock("test", OID(), "me", "x", Date_t::now(), "end");
+     *
+     * It is also possible to chain the callbacks. For example, if we want to set the test
+     * such that grabLock can only be called once, you can do this:
+     *
+     * DistLockCatalogMock mock;
+     * mock.setSucceedingExpectedGrabLock([&mock](...) {
+     *   mock.expectNoGrabLock();
+     * }, Status::OK());
+     */
+    class DistLockCatalogMock : public DistLockCatalog {
+    public:
+        DistLockCatalogMock();
+        virtual ~DistLockCatalogMock();
+
+        using GrabLockFunc = stdx::function<void (StringData lockID,
+                                                  const OID& lockSessionID,
+                                                  StringData who,
+                                                  StringData processId,
+                                                  Date_t time,
+                                                  StringData why)>;
+        using UnlockFunc = stdx::function<void (const OID& lockSessionID)>;
+
+        virtual StatusWith<LockpingsType> getPing(StringData processID) override;
+
+        virtual Status ping(StringData processID, Date_t ping) override;
+
+        virtual StatusWith<LocksType> grabLock(StringData lockID,
+                                               const OID& lockSessionID,
+                                               StringData who,
+                                               StringData processId,
+                                               Date_t time,
+                                               StringData why) override;
+
+        virtual StatusWith<LocksType> overtakeLock(StringData lockID,
+                                                   const OID& lockSessionID,
+                                                   const OID& lockTS,
+                                                   StringData who,
+                                                   StringData processId,
+                                                   Date_t time,
+                                                   StringData why) override;
+
+        virtual Status unlock(const OID& lockSessionID) override;
+
+        virtual StatusWith<ServerInfo> getServerInfo() override;
+
+        virtual StatusWith<LocksType> getLockByTS(const OID& ts) override;
+
+        /**
+         * Sets the checker method to use and the return value for grabLock to return every
+         * time it is called.
+         */
+        void setSucceedingExpectedGrabLock(GrabLockFunc checkerFunc,
+                                           StatusWith<LocksType> returnThis);
+
+        /**
+         * Expect grabLock to never be called after this is called.
+         */
+        void expectNoGrabLock();
+
+        /**
+         * Sets the checker method to use and the return value for unlock to return every
+         * time it is called.
+         */
+        void setSucceedingExpectedUnLock(UnlockFunc checkerFunc, Status returnThis);
+
+    private:
+        // Protects all the member variables.
+        stdx::mutex _mutex;
+
+        GrabLockFunc _grabLockChecker;
+        StatusWith<LocksType> _grabLockReturnValue;
+
+        UnlockFunc _unlockChecker;
+        Status _unlockReturnValue;
+    };
+}
diff --git a/src/mongo/s/catalog/dist_lock_manager.cpp b/src/mongo/s/catalog/dist_lock_manager.cpp
index 7803feb7423..e40504e2ff9 100644
--- a/src/mongo/s/catalog/dist_lock_manager.cpp
+++ b/src/mongo/s/catalog/dist_lock_manager.cpp
@@ -36,6 +36,9 @@
 
 namespace mongo {
 
+    const stdx::chrono::milliseconds DistLockManager::kDefaultSingleLockAttemptTimeout(0);
+    const stdx::chrono::milliseconds DistLockManager::kDefaultLockRetryInterval(1000);
+
     DistLockManager::ScopedDistLock::ScopedDistLock(): _lockManager(nullptr) {
     }
 
diff --git a/src/mongo/s/catalog/dist_lock_manager.h b/src/mongo/s/catalog/dist_lock_manager.h
index 8235496d521..17fc7ca0ad3 100644
--- a/src/mongo/s/catalog/dist_lock_manager.h
+++ b/src/mongo/s/catalog/dist_lock_manager.h
@@ -28,7 +28,6 @@
 
 #pragma once
 
-#include "mongo/base/status_with.h"
 #include "mongo/base/string_data.h"
 #include "mongo/bson/oid.h"
 #include "mongo/stdx/chrono.h"
@@ -36,6 +35,8 @@
 namespace mongo {
 
     using DistLockHandle = OID;
+    class Status;
+    template <typename T> class StatusWith;
 
     /**
      * Interface for handling distributed locks.
@@ -59,6 +60,9 @@ namespace mongo {
     class DistLockManager {
     public:
 
+        static const stdx::chrono::milliseconds kDefaultSingleLockAttemptTimeout;
+        static const stdx::chrono::milliseconds kDefaultLockRetryInterval;
+
         /**
          * RAII type for distributed lock. Not meant to be shared across multiple threads.
          */
@@ -104,8 +108,8 @@ namespace mongo {
         virtual StatusWith<ScopedDistLock> lock(
                 StringData name,
                 StringData whyMessage,
-                stdx::chrono::milliseconds waitFor = stdx::chrono::milliseconds(0),
-                stdx::chrono::milliseconds lockTryInterval = stdx::chrono::milliseconds(1000)) = 0;
+                stdx::chrono::milliseconds waitFor = kDefaultSingleLockAttemptTimeout,
+                stdx::chrono::milliseconds lockTryInterval = kDefaultLockRetryInterval) = 0;
 
     protected:
 
@@ -113,7 +117,7 @@ namespace mongo {
          * Unlocks the given lockHandle. Will attempt to retry again later if the config
          * server is not reachable.
          */
-        virtual void unlock(const DistLockHandle& lockHandle) BOOST_NOEXCEPT = 0;
+        virtual void unlock(const DistLockHandle& lockHandle) = 0;
 
         /**
          * Checks if the lockHandle still exists in the config server.
diff --git a/src/mongo/s/catalog/dist_lock_manager_mock.cpp b/src/mongo/s/catalog/dist_lock_manager_mock.cpp
index 4b105261e2d..38fe8b9584a 100644
--- a/src/mongo/s/catalog/dist_lock_manager_mock.cpp
+++ b/src/mongo/s/catalog/dist_lock_manager_mock.cpp
@@ -76,7 +76,7 @@ namespace mongo {
         return DistLockManager::ScopedDistLock(info.lockID, this);
     }
 
-    void DistLockManagerMock::unlock(const DistLockHandle& lockHandle) BOOST_NOEXCEPT {
+    void DistLockManagerMock::unlock(const DistLockHandle& lockHandle) {
         std::vector<LockInfo>::iterator it =
                 std::find_if(
                         _locks.begin(),
diff --git a/src/mongo/s/catalog/dist_lock_manager_mock.h b/src/mongo/s/catalog/dist_lock_manager_mock.h
index 45898e8b265..9fb13ee1786 100644
--- a/src/mongo/s/catalog/dist_lock_manager_mock.h
+++ b/src/mongo/s/catalog/dist_lock_manager_mock.h
@@ -54,7 +54,7 @@ namespace mongo {
 
     protected:
 
-        virtual void unlock(const DistLockHandle& lockHandle) BOOST_NOEXCEPT override;
+        virtual void unlock(const DistLockHandle& lockHandle) override;
 
         virtual Status checkStatus(const DistLockHandle& lockHandle) override;
 
diff --git a/src/mongo/s/catalog/legacy/legacy_dist_lock_manager.cpp b/src/mongo/s/catalog/legacy/legacy_dist_lock_manager.cpp
index 0a7709a3f01..3fffd1595b5 100644
--- a/src/mongo/s/catalog/legacy/legacy_dist_lock_manager.cpp
+++ b/src/mongo/s/catalog/legacy/legacy_dist_lock_manager.cpp
@@ -102,7 +102,7 @@ namespace {
 
         Timer timer;
         Timer msgTimer;
-        while (waitFor <= milliseconds::zero() || timer.millis() < waitFor.count()) {
+        while (waitFor <= milliseconds::zero() || milliseconds(timer.millis()) < waitFor) {
             bool acquired = false;
             BSONObj lockDoc;
             try {
@@ -165,7 +165,7 @@ namespace {
 
             milliseconds timeRemaining =
                     std::max(milliseconds::zero(), waitFor - milliseconds(timer.millis()));
-            sleepmillis(std::min(lockTryInterval, timeRemaining).count());
+            sleepFor(std::min(lockTryInterval, timeRemaining));
         }
 
         return lastStatus;
diff --git a/src/mongo/s/catalog/replset/SConscript b/src/mongo/s/catalog/replset/SConscript
new file mode 100644
index 00000000000..1ba9bc533b6
--- /dev/null
+++ b/src/mongo/s/catalog/replset/SConscript
@@ -0,0 +1,25 @@
+# -*- mode: python -*-
+
+Import("env")
+
+env.Library(
+    target='replset_dist_lock_manager',
+    source=[
+        'replset_dist_lock_manager.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/s/catalog/dist_lock_catalog',
+        '$BUILD_DIR/mongo/s/catalog/dist_lock_manager'
+    ],
+)
+
+env.CppUnitTest(
+    target='replset_dist_lock_manager_test',
+    source=[
+        'replset_dist_lock_manager_test.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/s/catalog/dist_lock_catalog_mock',
+        'replset_dist_lock_manager'
+    ]
+)
diff --git a/src/mongo/s/catalog/replset/replset_dist_lock_manager.cpp b/src/mongo/s/catalog/replset/replset_dist_lock_manager.cpp
new file mode 100644
index 00000000000..80ae92d7fe9
--- /dev/null
+++ b/src/mongo/s/catalog/replset/replset_dist_lock_manager.cpp
@@ -0,0 +1,138 @@
+/**
+ *    Copyright (C) 2015 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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.
+ */
+
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/s/catalog/replset/replset_dist_lock_manager.h"
+
+#include "mongo/base/status.h"
+#include "mongo/base/status_with.h"
+#include "mongo/s/catalog/dist_lock_catalog.h"
+#include "mongo/s/type_locks.h"
+#include "mongo/util/concurrency/thread_name.h"
+#include "mongo/util/log.h"
+#include "mongo/util/mongoutils/str.h"
+#include "mongo/util/timer.h"
+
+namespace mongo {
+
+    using std::string;
+    using std::unique_ptr;
+    using stdx::chrono::milliseconds;
+
+    ReplSetDistLockManager::ReplSetDistLockManager(StringData processID,
+                                                   unique_ptr<DistLockCatalog> catalog):
+        _processID(processID.toString()),
+        _catalog(std::move(catalog)) {
+    }
+
+    ReplSetDistLockManager::~ReplSetDistLockManager() = default;
+
+    void ReplSetDistLockManager::startUp() {
+        // TODO
+    }
+
+    void ReplSetDistLockManager::shutDown() {
+        // TODO
+    }
+
+    StatusWith<DistLockManager::ScopedDistLock> ReplSetDistLockManager::lock(
+            StringData name,
+            StringData whyMessage,
+            milliseconds waitFor,
+            milliseconds lockTryInterval) {
+        Timer timer;
+        Timer msgTimer;
+
+        while (waitFor <= milliseconds::zero() || milliseconds(timer.millis()) < waitFor) {
+            OID lockSessionID = OID::gen();
+            string who = str::stream() << _processID << ":" << getThreadName();
+            auto lockResult = _catalog->grabLock(name,
+                                                 lockSessionID,
+                                                 who,
+                                                 _processID,
+                                                 Date_t::now(),
+                                                 whyMessage);
+
+            auto status = lockResult.getStatus();
+
+            if (!status.isOK()) {
+                // An error occurred but the write might have actually been applied on the
+                // other side. Schedule an unlock to clean it up just in case.
+                queueUnlock(lockSessionID);
+                return status;
+            }
+
+            const auto& lockDoc = lockResult.getValue();
+            if (lockDoc.isValid(nullptr)) {
+                // Lock is acquired since findAndModify was able to successfully modify
+                // the lock document.
+                return ScopedDistLock(lockSessionID, this);
+            }
+
+            // TODO: implement lock overtaking here.
+
+            if (waitFor == milliseconds::zero()) {
+                break;
+            }
+
+            // Periodically message for debugging reasons
+            if (msgTimer.seconds() > 10) {
+                LOG(0) << "waited " << timer.seconds() << "s for distributed lock " << name
+                       << " for " << whyMessage;
+
+                msgTimer.reset();
+            }
+
+            milliseconds timeRemaining =
+                    std::max(milliseconds::zero(), waitFor - milliseconds(timer.millis()));
+            sleepFor(std::min(lockTryInterval, timeRemaining));
+        }
+
+        return {ErrorCodes::LockBusy, str::stream() << "timed out waiting for " << name};
+    }
+
+    void ReplSetDistLockManager::unlock(const DistLockHandle& lockSessionID) {
+        auto unlockStatus = _catalog->unlock(lockSessionID);
+
+        if (!unlockStatus.isOK()) {
+            queueUnlock(lockSessionID);
+        }
+    }
+
+    Status ReplSetDistLockManager::checkStatus(const DistLockHandle& lockHandle) {
+        invariant(false);
+    }
+
+    void ReplSetDistLockManager::queueUnlock(const OID& lockSessionID) {
+        // TODO: implement
+        invariant(false);
+    }
+}
diff --git a/src/mongo/s/repl_set_dist_lock_manager.h b/src/mongo/s/catalog/replset/replset_dist_lock_manager.h
index d40d89ebe92..d4f9b2ce467 100644
--- a/src/mongo/s/repl_set_dist_lock_manager.h
+++ b/src/mongo/s/catalog/replset/replset_dist_lock_manager.h
@@ -28,36 +28,44 @@
 
 #pragma once
 
-#include "mongo/client/dbclientinterface.h"
-#include "mongo/s/dist_lock_manager.h"
+#include <memory>
+#include <string>
+
+#include "mongo/base/string_data.h"
+#include "mongo/s/catalog/dist_lock_manager.h"
 
 namespace mongo {
 
-    class CatalogManager;
+    class DistLockCatalog;
 
     class ReplSetDistLockManager: public DistLockManager {
     public:
-        ReplSetDistLockManager(CatalogManager* lockCatalogue);
+        ReplSetDistLockManager(StringData processID,
+                               std::unique_ptr<DistLockCatalog> catalog);
 
-        virtual ~ReplSetDistLockManager() = default;
+        virtual ~ReplSetDistLockManager();
 
-        virtual void startUp();
-        virtual void shutDown();
+        virtual void startUp() override;
+        virtual void shutDown() override;
 
         virtual StatusWith<DistLockManager::ScopedDistLock> lock(
                 StringData name,
                 StringData whyMessage,
                 stdx::chrono::milliseconds waitFor,
-                stdx::chrono::milliseconds lockTryInterval);
+                stdx::chrono::milliseconds lockTryInterval) override;
 
     protected:
 
-        virtual void unlock(const DistLockHandle& lockHandle) BOOST_NOEXCEPT override;
+        virtual void unlock(const DistLockHandle& lockSessionID) override;
 
-        virtual Status checkStatus(const DistLockHandle& lockHandle) override;
+        virtual Status checkStatus(const DistLockHandle& lockSessionID) override;
 
     private:
-        CatalogManager* _lockCatalogue; // Not owned here.
-    };
 
+        void queueUnlock(const OID& lockSessionID);
+
+        std::string _processID;
+        std::unique_ptr<DistLockCatalog> _catalog;
+
+    };
 }
diff --git a/src/mongo/s/catalog/replset/replset_dist_lock_manager_test.cpp b/src/mongo/s/catalog/replset/replset_dist_lock_manager_test.cpp
new file mode 100644
index 00000000000..de43d58349c
--- /dev/null
+++ b/src/mongo/s/catalog/replset/replset_dist_lock_manager_test.cpp
@@ -0,0 +1,303 @@
+/**
+ *    Copyright (C) 2015 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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/s/catalog/replset/replset_dist_lock_manager.h"
+
+#include "mongo/platform/basic.h"
+
+#include <string>
+
+#include "mongo/base/status.h"
+#include "mongo/base/status_with.h"
+#include "mongo/bson/json.h"
+#include "mongo/db/jsobj.h"
+#include "mongo/s/catalog/dist_lock_catalog_mock.h"
+#include "mongo/s/type_lockpings.h"
+#include "mongo/s/type_locks.h"
+#include "mongo/stdx/condition_variable.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/stdx/mutex.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/time_support.h"
+
+namespace mongo {
+namespace {
+
+    using std::string;
+
+    const Milliseconds kUnlockTimeout(100);
+
+    /**
+     * Test scenario:
+     * 1. Grab lock.
+     * 2. Unlock (on destructor of ScopedDistLock).
+     * 3. Check lock id used in lock and unlock are the same.
+     */
+    TEST(ReplSetDistLockManager, BasicLockLifeCycle) {
+        string lockName("test");
+        string processID("abcd");
+        Date_t now(Date_t::now());
+        string whyMsg("because");
+
+        LocksType retLockDoc;
+        retLockDoc.setName(lockName);
+        retLockDoc.setState(LocksType::LOCKED);
+        retLockDoc.setProcess(processID);
+        retLockDoc.setWho("me");
+        retLockDoc.setWhy(whyMsg);
+        // Will be different from the actual lock session id. For testing only.
+        retLockDoc.setLockID(OID::gen());
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        auto rawMock = mock.get();
+
+        OID lockSessionIDPassed;
+
+        mock->setSucceedingExpectedGrabLock(
+                [&lockName, &processID, &now, &whyMsg, &lockSessionIDPassed, &rawMock](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+            ASSERT_EQUALS(lockName, lockID);
+            ASSERT_TRUE(lockSessionID.isSet());
+            ASSERT_EQUALS(processID, processId);
+            ASSERT_GREATER_THAN_OR_EQUALS(time, now);
+            ASSERT_EQUALS(whyMsg, why);
+
+            lockSessionIDPassed = lockSessionID;
+            rawMock->expectNoGrabLock(); // Call only once.
+        }, retLockDoc);
+
+        ReplSetDistLockManager mgr(processID, std::move(mock));
+
+        int unlockCallCount = 0;
+        OID unlockSessionIDPassed;
+
+        {
+            auto lockStatus = mgr.lock(lockName,
+                                       whyMsg,
+                                       DistLockManager::kDefaultSingleLockAttemptTimeout,
+                                       DistLockManager::kDefaultLockRetryInterval);
+            ASSERT_OK(lockStatus.getStatus());
+
+            rawMock->expectNoGrabLock();
+            rawMock->setSucceedingExpectedUnLock([&unlockCallCount, &unlockSessionIDPassed](
+                            const OID& lockSessionID) {
+                unlockCallCount++;
+                unlockSessionIDPassed = lockSessionID;
+            }, Status::OK());
+        }
+
+        ASSERT_EQUALS(1, unlockCallCount);
+        ASSERT_EQUALS(lockSessionIDPassed, unlockSessionIDPassed);
+    }
+
+    /**
+     * Test scenario:
+     * 1. Grab lock fails up to 3 times.
+     * 2. Check that each attempt uses a unique lock session id.
+     * 3. Unlock (on destructor of ScopedDistLock).
+     * 4. Check lock id used in lock and unlock are the same.
+     */
+    TEST(ReplSetDistLockManager, LockSuccessAfterRetry) {
+        string lockName("test");
+        string me("me");
+        string processID("abcd");
+        OID lastTS;
+        Date_t lastTime(Date_t::now());
+        string whyMsg("because");
+
+        int retryAttempt = 0;
+        const int kMaxRetryAttempt = 3;
+        LocksType invalidLockDoc;
+
+        LocksType goodLockDoc;
+        goodLockDoc.setName(lockName);
+        goodLockDoc.setState(LocksType::LOCKED);
+        goodLockDoc.setProcess(processID);
+        goodLockDoc.setWho("me");
+        goodLockDoc.setWhy(whyMsg);
+        goodLockDoc.setLockID(OID::gen());
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        auto rawMock = mock.get();
+        mock->setSucceedingExpectedGrabLock(
+                [&lockName,
+                 &lastTS,
+                 &me,
+                 &processID,
+                 &lastTime,
+                 &whyMsg,
+                 &retryAttempt,
+                 &kMaxRetryAttempt,
+                 &rawMock,
+                 &goodLockDoc](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+            ASSERT_EQUALS(lockName, lockID);
+            // Every attempt should have a unique sesssion ID.
+            ASSERT_NOT_EQUALS(lastTS, lockSessionID);
+            ASSERT_EQUALS(processID, processId);
+            ASSERT_GREATER_THAN_OR_EQUALS(time, lastTime);
+            ASSERT_EQUALS(whyMsg, why);
+
+            lastTS = lockSessionID;
+            lastTime = time;
+
+            if (++retryAttempt >= kMaxRetryAttempt) {
+                rawMock->setSucceedingExpectedGrabLock([&lockName,
+                                                        &lastTS,
+                                                        &me,
+                                                        &processID,
+                                                        &lastTime,
+                                                        &whyMsg,
+                                                        &retryAttempt,
+                                                        &rawMock](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+                    ASSERT_EQUALS(lockName, lockID);
+                    // Every attempt should have a unique sesssion ID.
+                    ASSERT_NOT_EQUALS(lastTS, lockSessionID);
+                    lastTS = lockSessionID;
+                    ASSERT_TRUE(lockSessionID.isSet());
+                    ASSERT_EQUALS(processID, processId);
+                    ASSERT_GREATER_THAN_OR_EQUALS(time, lastTime);
+                    ASSERT_EQUALS(whyMsg, why);
+                }, goodLockDoc);
+            }
+        }, invalidLockDoc);
+
+        ReplSetDistLockManager mgr(processID, std::move(mock));
+
+        int unlockCallCount = 0;
+        OID unlockSessionIDPassed;
+
+        {
+            auto lockStatus = mgr.lock(lockName, whyMsg, Milliseconds(10), Milliseconds(1));
+            ASSERT_OK(lockStatus.getStatus());
+
+            rawMock->expectNoGrabLock();
+            rawMock->setSucceedingExpectedUnLock([&unlockCallCount, &unlockSessionIDPassed](
+                            const OID& lockSessionID) {
+                unlockCallCount++;
+                unlockSessionIDPassed = lockSessionID;
+            }, Status::OK());
+        }
+
+        ASSERT_EQUALS(1, unlockCallCount);
+        ASSERT_EQUALS(lastTS, unlockSessionIDPassed);
+    }
+
+    TEST(ReplSetDistLockManager, LockBusyNoRetry) {
+        int retryAttempt = 0;
+        LocksType invalidLockDoc;
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        mock->setSucceedingExpectedGrabLock([&retryAttempt](StringData lockID,
+                                                            const OID& lockSessionID,
+                                                            StringData who,
+                                                            StringData processId,
+                                                            Date_t time,
+                                                            StringData why) {
+            retryAttempt++;
+        }, invalidLockDoc);
+
+        ReplSetDistLockManager mgr("", std::move(mock));
+        auto status = mgr.lock("", "", Milliseconds(0), Milliseconds(0)).getStatus();
+        ASSERT_NOT_OK(status);
+        ASSERT_EQUALS(ErrorCodes::LockBusy, status.code());
+        ASSERT_EQUALS(1, retryAttempt);
+    }
+
+    /**
+     * Test scenario:
+     * 1. Attempt to grab lock.
+     * 2. Check that each attempt uses a unique lock session id.
+     * 3. Times out trying.
+     * 4. Checks result is error.
+     * 5. Implicitly check that unlock is not called (default setting of mock catalog).
+     */
+    TEST(ReplSetDistLockManager, LockRetryTimeout) {
+        string lockName("test");
+        string me("me");
+        string processID("abcd");
+        OID lastTS;
+        Date_t lastTime(Date_t::now());
+        string whyMsg("because");
+
+        int retryAttempt = 0;
+        LocksType invalidLockDoc;
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        mock->setSucceedingExpectedGrabLock(
+                [&lockName,
+                 &lastTS,
+                 &me,
+                 &processID,
+                 &lastTime,
+                 &whyMsg,
+                 &retryAttempt](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+            ASSERT_EQUALS(lockName, lockID);
+            // Every attempt should have a unique sesssion ID.
+            ASSERT_NOT_EQUALS(lastTS, lockSessionID);
+            ASSERT_EQUALS(processID, processId);
+            ASSERT_GREATER_THAN_OR_EQUALS(time, lastTime);
+            ASSERT_EQUALS(whyMsg, why);
+
+            lastTS = lockSessionID;
+            lastTime = time;
+            retryAttempt++;
+        }, invalidLockDoc);
+
+        ReplSetDistLockManager mgr(processID, std::move(mock));
+        auto lockStatus = mgr.lock(lockName, whyMsg, Milliseconds(5), Milliseconds(1)).getStatus();
+        ASSERT_NOT_OK(lockStatus);
+
+        ASSERT_EQUALS(ErrorCodes::LockBusy, lockStatus.code());
+        ASSERT_GREATER_THAN(retryAttempt, 1);
+    }
+
+} // unnamed namespace
+} // namespace mongo
diff --git a/src/mongo/s/repl_set_dist_lock_manager.cpp b/src/mongo/s/catalog/replset_dist_lock_manager.cpp
index e6d1e42cbd5..475d56983c2 100644
--- a/src/mongo/s/repl_set_dist_lock_manager.cpp
+++ b/src/mongo/s/catalog/replset_dist_lock_manager.cpp
@@ -30,23 +30,33 @@
 
 #include "mongo/platform/basic.h"
 
-#include "mongo/s/repl_set_dist_lock_manager.h"
+#include "mongo/s/catalog/replset_dist_lock_manager.h"
 
-#include "mongo/s/dist_lock_logic.h"
+#include "mongo/base/status.h"
+#include "mongo/base/status_with.h"
+#include "mongo/s/catalog/dist_lock_catalog.h"
+#include "mongo/s/type_locks.h"
+#include "mongo/util/concurrency/thread_name.h"
 #include "mongo/util/log.h"
+#include "mongo/util/mongoutils/str.h"
 #include "mongo/util/timer.h"
 
 namespace mongo {
 
     using std::string;
+    using std::unique_ptr;
     using stdx::chrono::milliseconds;
 
-    ReplSetDistLockManager::ReplSetDistLockManager(CatalogManager* lockCatalogue):
-        _lockCatalogue(lockCatalogue) {
+    ReplSetDistLockManager::ReplSetDistLockManager(StringData processID,
+                                                   unique_ptr<DistLockCatalog> catalog):
+        _processID(processID.toString()),
+        _catalog(std::move(catalog)) {
     }
 
+    ReplSetDistLockManager::~ReplSetDistLockManager() = default;
+
     void ReplSetDistLockManager::startUp() {
-        // TODO: start background task.
+        // TODO
     }
 
     void ReplSetDistLockManager::shutDown() {
@@ -58,31 +68,45 @@ namespace mongo {
             StringData whyMessage,
             milliseconds waitFor,
             milliseconds lockTryInterval) {
-        auto lastStatus = StatusWith<DistLockHandle>(
-                ErrorCodes::LockBusy, str::stream() << "timed out waiting for " << name);
 
         Timer timer;
         Timer msgTimer;
+
         while (waitFor <= milliseconds::zero() || timer.millis() < waitFor.count()) {
-            lastStatus = dist_lock_logic::lock(_lockCatalogue,
-                                               name.toString(),
-                                               whyMessage.toString());
+            OID lockSessionID = OID::gen();
+            string who = str::stream() << _processID << ":" << getThreadName();
+            auto lockResult = _catalog->grabLock(name,
+                                                 lockSessionID,
+                                                 who,
+                                                 _processID,
+                                                 Date_t::now(),
+                                                 whyMessage);
+
+            auto status = lockResult.getStatus();
+            const auto& lockDoc = lockResult.getValue();
+
+            if (!status.isOK()) {
+                // An error occurred but the write might have actually been applied on the
+                // other side. Schedule an unlock to clean it up just in case.
+                queueUnlock(lockSessionID);
+                return status;
+            }
 
-            if (lastStatus.isOK()) {
-                // TODO: add to pinger.
-                return StatusWith<ScopedDistLock>(ScopedDistLock(lastStatus.getValue(), this));
+            string errmsg;
+            if (lockDoc.isValid(&errmsg)) {
+                // Lock is acquired since findAndModify was able to successfully modify
+                // the lock document.
+                return ScopedDistLock(lockSessionID, this);
             }
 
-            if (waitFor.count() == 0) break;
+            // TODO: implement lock overtaking here.
 
-            if (lastStatus.getStatus() != ErrorCodes::LockBusy) {
-                return StatusWith<ScopedDistLock>(lastStatus.getStatus());
-            }
+            if (waitFor == milliseconds::zero()) break;
 
             // Periodically message for debugging reasons
             if (msgTimer.seconds() > 10) {
                 LOG(0) << "waited " << timer.seconds() << "s for distributed lock " << name
-                       << " for " << whyMessage << ": " << causedBy(lastStatus.getStatus());
+                       << " for " << whyMessage;
 
                 msgTimer.reset();
             }
@@ -92,19 +116,20 @@ namespace mongo {
             sleepmillis(std::min(lockTryInterval, timeRemaining).count());
         }
 
-        return StatusWith<ScopedDistLock>(lastStatus.getStatus());
+        return {ErrorCodes::LockBusy, str::stream() << "timed out waiting for " << name};
     }
 
-    void ReplSetDistLockManager::unlock(const DistLockHandle& lockHandle) BOOST_NOEXCEPT {
-        // TODO: stop pinging.
-
-        if (!dist_lock_logic::unlock(_lockCatalogue, lockHandle)) {
-            // TODO: deferred unlocking
-        }
+    void ReplSetDistLockManager::unlock(const DistLockHandle& lockHandle) {
+        queueUnlock(lockHandle);
     }
 
     Status ReplSetDistLockManager::checkStatus(const DistLockHandle& lockHandle) {
-        // TODO: use catalogue to check.
-        return Status::OK();
+        invariant(false);
+        return {ErrorCodes::InternalError, "not yet implemented"};
+    }
+
+    void ReplSetDistLockManager::queueUnlock(const OID& lockSessionID) {
+        // TODO: make this asynchronous
+        auto unlockStatus = _catalog->unlock(lockSessionID);
     }
 }
diff --git a/src/mongo/s/dist_lock_logic.h b/src/mongo/s/catalog/replset_dist_lock_manager.h
index 9084a70881d..4650c8af105 100644
--- a/src/mongo/s/dist_lock_logic.h
+++ b/src/mongo/s/catalog/replset_dist_lock_manager.h
@@ -28,32 +28,45 @@
 
 #pragma once
 
-#include "mongo/base/status_with.h"
-#include "mongo/s/dist_lock_manager.h"
+#include <memory>
+#include <string>
 
-/**
- * Helper methods for handling distributed locks that are backed by replica set config servers.
- */
+#include "mongo/base/string_data.h"
+#include "mongo/s/catalog/dist_lock_manager.h"
 
 namespace mongo {
 
-    class CatalogManager;
+    class DistLockCatalog;
+
+    class ReplSetDistLockManager: public DistLockManager {
+    public:
+        ReplSetDistLockManager(StringData processID,
+                               std::unique_ptr<DistLockCatalog> catalog);
+
+        virtual ~ReplSetDistLockManager();
+
+        virtual void startUp() override;
+        virtual void shutDown() override;
+
+        virtual StatusWith<DistLockManager::ScopedDistLock> lock(
+                StringData name,
+                StringData whyMessage,
+                stdx::chrono::milliseconds waitFor = stdx::chrono::milliseconds(0),
+                stdx::chrono::milliseconds lockTryInterval =
+                        stdx::chrono::milliseconds(1000)) override;
+
+    protected:
+
+        virtual void unlock(const DistLockHandle& lockHandle) override;
+
+        virtual Status checkStatus(const DistLockHandle& lockHandle) override;
 
-namespace dist_lock_logic {
+    private:
 
-    /**
-     * Tries to acquire the distributed lock with the given name and returns the
-     * handle of the newly acquired lock on success.
-     */
-    StatusWith<DistLockHandle> lock(CatalogManager* lockCatalogue,
-                                    const std::string& name,
-                                    const std::string& whyMessage) BOOST_NOEXCEPT;
+        void queueUnlock(const OID& lockSessionID);
 
-    /**
-     * Unlocks the distributed lock with the given lock handle. Returns true on success.
-     */
-    bool unlock(CatalogManager* lockCatalogue,
-                const DistLockHandle& lockHandle) BOOST_NOEXCEPT;
+        std::string _processID;
+        std::unique_ptr<DistLockCatalog> _catalog;
 
-} // namespace dist_lock_logic
-} // namespace mongo
+    };
+}
diff --git a/src/mongo/s/catalog/replset_dist_lock_manager_test.cpp b/src/mongo/s/catalog/replset_dist_lock_manager_test.cpp
new file mode 100644
index 00000000000..e0a9fc8016b
--- /dev/null
+++ b/src/mongo/s/catalog/replset_dist_lock_manager_test.cpp
@@ -0,0 +1,362 @@
+/**
+ *    Copyright (C) 2015 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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 <string>
+
+#include "mongo/base/status.h"
+#include "mongo/base/status_with.h"
+#include "mongo/bson/json.h"
+#include "mongo/db/jsobj.h"
+#include "mongo/s/catalog/dist_lock_catalog_mock.h"
+#include "mongo/s/catalog/replset_dist_lock_manager.h"
+#include "mongo/s/type_lockpings.h"
+#include "mongo/s/type_locks.h"
+#include "mongo/stdx/condition_variable.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/stdx/mutex.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/time_support.h"
+
+namespace mongo {
+namespace {
+
+    using std::string;
+
+    const Milliseconds kUnlockTimeout(100);
+
+    TEST(ReplSetDistLockManager, BasicLockLifeCycle) {
+        string lockName("test");
+        string processID("abcd");
+        Date_t now(Date_t::now());
+        string whyMsg("because");
+
+        LocksType retLockDoc;
+        retLockDoc.setName(lockName);
+        retLockDoc.setState(LocksType::LOCKED);
+        retLockDoc.setProcess(processID);
+        retLockDoc.setWho("me");
+        retLockDoc.setWhy(whyMsg);
+        // Will be different from the actual lock session id. For testing only.
+        retLockDoc.setLockID(OID::gen());
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        auto rawMock = mock.get();
+
+        OID lockSessionIDPassed;
+
+        mock->setSucceedingExpectedGrabLock(
+                [&lockName, &processID, &now, &whyMsg, &lockSessionIDPassed, &rawMock](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+            ASSERT_EQUALS(lockName, lockID);
+            ASSERT_TRUE(lockSessionID.isSet());
+            ASSERT_EQUALS(processID, processId);
+            ASSERT_GREATER_THAN_OR_EQUALS(time, now);
+            ASSERT_EQUALS(whyMsg, why);
+
+            lockSessionIDPassed = lockSessionID;
+            rawMock->expectNoGrabLock(); // Call only once.
+        }, retLockDoc);
+
+        ReplSetDistLockManager mgr(processID, std::move(mock));
+
+        stdx::mutex unlockMutex;
+        stdx::condition_variable unlockCV;
+        int unlockCallCount = 0;
+        OID unlockSessionIDPassed;
+
+        {
+            auto lockStatus = mgr.lock(lockName, whyMsg);
+            ASSERT_OK(lockStatus.getStatus());
+
+            rawMock->expectNoGrabLock();
+            rawMock->setSucceedingExpectedUnLock(
+                    [&unlockMutex, &unlockCV, &unlockCallCount, &unlockSessionIDPassed](
+                            const OID& lockSessionID) {
+                stdx::unique_lock<stdx::mutex> ul(unlockMutex);
+                unlockCallCount++;
+                unlockSessionIDPassed = lockSessionID;
+                unlockCV.notify_all();
+            }, Status::OK());
+        }
+
+        stdx::unique_lock<stdx::mutex> ul(unlockMutex);
+        if (unlockCallCount == 0) {
+            ASSERT(unlockCV.wait_for(ul, kUnlockTimeout) == stdx::cv_status::no_timeout);
+        }
+
+        ASSERT_EQUALS(1, unlockCallCount);
+        ASSERT_EQUALS(lockSessionIDPassed, unlockSessionIDPassed);
+    }
+
+    TEST(ReplSetDistLockManager, LockSuccessAfterRetry) {
+        string lockName("test");
+        string me("me");
+        string processID("abcd");
+        OID lastTS;
+        Date_t lastTime(Date_t::now());
+        string whyMsg("because");
+
+        int retryAttempt = 0;
+        const int kMaxRetryAttempt = 3;
+        LocksType invalidLockDoc;
+
+        LocksType goodLockDoc;
+        goodLockDoc.setName(lockName);
+        goodLockDoc.setState(LocksType::LOCKED);
+        goodLockDoc.setProcess(processID);
+        goodLockDoc.setWho("me");
+        goodLockDoc.setWhy(whyMsg);
+        goodLockDoc.setLockID(OID::gen());
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        auto rawMock = mock.get();
+        mock->setSucceedingExpectedGrabLock(
+                [&lockName,
+                 &lastTS,
+                 &me,
+                 &processID,
+                 &lastTime,
+                 &whyMsg,
+                 &retryAttempt,
+                 &kMaxRetryAttempt,
+                 &rawMock,
+                 &goodLockDoc](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+            ASSERT_EQUALS(lockName, lockID);
+            // Every attempt should have a unique sesssion ID.
+            ASSERT_NOT_EQUALS(lastTS, lockSessionID);
+            ASSERT_EQUALS(processID, processId);
+            ASSERT_GREATER_THAN_OR_EQUALS(time, lastTime);
+            ASSERT_EQUALS(whyMsg, why);
+
+            lastTS = lockSessionID;
+            lastTime = time;
+
+            if (++retryAttempt >= kMaxRetryAttempt) {
+                rawMock->setSucceedingExpectedGrabLock([&lockName,
+                                                        &lastTS,
+                                                        &me,
+                                                        &processID,
+                                                        &lastTime,
+                                                        &whyMsg,
+                                                        &retryAttempt,
+                                                        &rawMock](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+                    ASSERT_EQUALS(lockName, lockID);
+                    // Every attempt should have a unique sesssion ID.
+                    ASSERT_NOT_EQUALS(lastTS, lockSessionID);
+                    lastTS = lockSessionID;
+                    ASSERT_TRUE(lockSessionID.isSet());
+                    ASSERT_EQUALS(processID, processId);
+                    ASSERT_GREATER_THAN_OR_EQUALS(time, lastTime);
+                    ASSERT_EQUALS(whyMsg, why);
+                }, goodLockDoc);
+            }
+        }, invalidLockDoc);
+
+        ReplSetDistLockManager mgr(processID, std::move(mock));
+
+        stdx::mutex unlockMutex;
+        stdx::condition_variable unlockCV;
+        int unlockCallCount = 0;
+        OID unlockSessionIDPassed;
+
+        {
+            auto lockStatus = mgr.lock(lockName, whyMsg, Milliseconds(10), Milliseconds(1));
+            ASSERT_OK(lockStatus.getStatus());
+
+            rawMock->expectNoGrabLock();
+            rawMock->setSucceedingExpectedUnLock(
+                    [&unlockMutex, &unlockCV, &unlockCallCount, &unlockSessionIDPassed](
+                            const OID& lockSessionID) {
+                stdx::unique_lock<stdx::mutex> ul(unlockMutex);
+                unlockCallCount++;
+                unlockSessionIDPassed = lockSessionID;
+                unlockCV.notify_all();
+            }, Status::OK());
+        }
+
+        stdx::unique_lock<stdx::mutex> ul(unlockMutex);
+        if (unlockCallCount == 0) {
+            ASSERT(unlockCV.wait_for(ul, kUnlockTimeout) == stdx::cv_status::no_timeout);
+        }
+
+        ASSERT_EQUALS(1, unlockCallCount);
+        ASSERT_EQUALS(lastTS, unlockSessionIDPassed);
+    }
+
+    TEST(ReplSetDistLockManager, LockBusyNoRetry) {
+        int retryAttempt = 0;
+        LocksType invalidLockDoc;
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        mock->setSucceedingExpectedGrabLock([&retryAttempt](StringData lockID,
+                                                            const OID& lockSessionID,
+                                                            StringData who,
+                                                            StringData processId,
+                                                            Date_t time,
+                                                            StringData why) {
+            retryAttempt++;
+        }, invalidLockDoc);
+
+        ReplSetDistLockManager mgr("", std::move(mock));
+        auto status = mgr.lock("", "").getStatus();
+        ASSERT_NOT_OK(status);
+        ASSERT_EQUALS(ErrorCodes::LockBusy, status.code());
+        ASSERT_EQUALS(1, retryAttempt);
+    }
+
+    TEST(ReplSetDistLockManager, LockRetryTimeout) {
+        string lockName("test");
+        string me("me");
+        string processID("abcd");
+        OID lastTS;
+        Date_t lastTime(Date_t::now());
+        string whyMsg("because");
+
+        int retryAttempt = 0;
+        LocksType invalidLockDoc;
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        mock->setSucceedingExpectedGrabLock(
+                [&lockName,
+                 &lastTS,
+                 &me,
+                 &processID,
+                 &lastTime,
+                 &whyMsg,
+                 &retryAttempt](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+            ASSERT_EQUALS(lockName, lockID);
+            // Every attempt should have a unique sesssion ID.
+            ASSERT_NOT_EQUALS(lastTS, lockSessionID);
+            ASSERT_EQUALS(processID, processId);
+            ASSERT_GREATER_THAN_OR_EQUALS(time, lastTime);
+            ASSERT_EQUALS(whyMsg, why);
+
+            lastTS = lockSessionID;
+            lastTime = time;
+            retryAttempt++;
+        }, invalidLockDoc);
+
+        ReplSetDistLockManager mgr(processID, std::move(mock));
+        auto lockStatus = mgr.lock(lockName, whyMsg, Milliseconds(5), Milliseconds(1)).getStatus();
+        ASSERT_NOT_OK(lockStatus);
+
+        ASSERT_EQUALS(ErrorCodes::LockBusy, lockStatus.code());
+        ASSERT_GREATER_THAN(retryAttempt, 1);
+    }
+
+    TEST(ReplSetDistLockManager, MustUnlockOnLockError) {
+        string lockName("test");
+        string me("me");
+        string processID("abcd");
+        OID lastTS;
+        string whyMsg("because");
+
+        auto mock = stdx::make_unique<DistLockCatalogMock>();
+        auto rawMock = mock.get();
+        mock->setSucceedingExpectedGrabLock(
+                [&lockName,
+                 &lastTS,
+                 &me,
+                 &processID,
+                 &whyMsg,
+                 &rawMock](
+                        StringData lockID,
+                        const OID& lockSessionID,
+                        StringData who,
+                        StringData processId,
+                        Date_t time,
+                        StringData why) {
+            ASSERT_EQUALS(lockName, lockID);
+            // Every attempt should have a unique sesssion ID.
+            ASSERT_TRUE(lockSessionID.isSet());
+            ASSERT_EQUALS(processID, processId);
+            ASSERT_EQUALS(whyMsg, why);
+
+            lastTS = lockSessionID;
+            rawMock->expectNoGrabLock();
+        }, {ErrorCodes::NetworkTimeout, "bad test network"});
+
+        ReplSetDistLockManager mgr(processID, std::move(mock));
+
+        stdx::mutex unlockMutex;
+        stdx::condition_variable unlockCV;
+        int unlockCallCount = 0;
+        OID unlockSessionIDPassed;
+
+        rawMock->setSucceedingExpectedUnLock(
+                [&unlockMutex, &unlockCV, &unlockCallCount, &unlockSessionIDPassed](
+                        const OID& lockSessionID) {
+            stdx::unique_lock<stdx::mutex> ul(unlockMutex);
+            unlockCallCount++;
+            unlockSessionIDPassed = lockSessionID;
+            unlockCV.notify_all();
+        }, Status::OK());
+
+        auto lockStatus = mgr.lock(lockName,
+                                   whyMsg,
+                                   Milliseconds(10),
+                                   Milliseconds(1)).getStatus();
+        ASSERT_NOT_OK(lockStatus);
+        ASSERT_EQUALS(ErrorCodes::NetworkTimeout, lockStatus.code());
+
+        stdx::unique_lock<stdx::mutex> ul(unlockMutex);
+        if (unlockCallCount == 0) {
+            ASSERT(unlockCV.wait_for(ul, kUnlockTimeout) == stdx::cv_status::no_timeout);
+        }
+
+        ASSERT_EQUALS(1, unlockCallCount);
+        ASSERT_EQUALS(lastTS, unlockSessionIDPassed);
+    }
+
+} // unnamed namespace
+} // namespace mongo
diff --git a/src/mongo/s/dist_lock_logic.cpp b/src/mongo/s/dist_lock_logic.cpp
deleted file mode 100644
index ca45c577779..00000000000
--- a/src/mongo/s/dist_lock_logic.cpp
+++ /dev/null
@@ -1,57 +0,0 @@
-/**
- *    Copyright (C) 2015 MongoDB Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    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
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    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 GNU Affero General 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.
- */
-
-#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/s/dist_lock_logic.h"
-
-#include "mongo/s/distlock.h"
-#include "mongo/s/type_locks.h"
-#include "mongo/stdx/memory.h"
-
-namespace mongo {
-
-namespace dist_lock_logic {
-    StatusWith<DistLockHandle> lock(CatalogManager* lockCatalogue,
-                                    const std::string& name,
-                                    const std::string& whyMessage) BOOST_NOEXCEPT {
-        // TODO
-        DistLockHandle dummy;
-        return StatusWith<DistLockHandle>(dummy);
-    }
-
-     bool unlock(CatalogManager* lockCatalogue,
-                 const DistLockHandle& lockHandle) BOOST_NOEXCEPT {
-        // TODO
-        return true;
-    }
-} // namespace dist_lock_logic
-
-} // namespace mongo
diff --git a/src/mongo/s/type_locks.h b/src/mongo/s/type_locks.h
index 20241f237ad..3eb6c3f86dd 100644
--- a/src/mongo/s/type_locks.h
+++ b/src/mongo/s/type_locks.h
@@ -156,7 +156,7 @@ namespace mongo {
         }
 
         // Optional Fields
-        void setProcess(StringData& process) {
+        void setProcess(StringData process) {
             _process = process.toString();
             _isProcessSet = true;
         }
@@ -198,7 +198,7 @@ namespace mongo {
                 return lockID.getDefault();
             }
         }
-        void setWho(StringData& who) {
+        void setWho(StringData who) {
             _who = who.toString();
             _isWhoSet = true;
         }
@@ -219,7 +219,7 @@ namespace mongo {
                 return who.getDefault();
             }
         }
-        void setWhy(StringData& why) {
+        void setWhy(StringData why) {
             _why = why.toString();
             _isWhySet = true;
         }
diff --git a/src/mongo/stdx/mutex.h b/src/mongo/stdx/mutex.h
index b495943b8b1..071fa64bc0d 100644
--- a/src/mongo/stdx/mutex.h
+++ b/src/mongo/stdx/mutex.h
@@ -28,6 +28,7 @@
 
 #pragma once
 
+#include <boost/thread/lock_guard.hpp>
 #include <boost/thread/mutex.hpp>
 
 namespace mongo {
diff --git a/src/mongo/util/time_support.cpp b/src/mongo/util/time_support.cpp
index 57eefa80775..1ac4b81427a 100644
--- a/src/mongo/util/time_support.cpp
+++ b/src/mongo/util/time_support.cpp
@@ -837,6 +837,10 @@ namespace {
     }
 #endif
 
+    void sleepFor(const Milliseconds& time) {
+        sleepmillis(time.count());
+    }
+
     void Backoff::nextSleepMillis(){
 
         // Get the current time
diff --git a/src/mongo/util/time_support.h b/src/mongo/util/time_support.h
index 3d5128f43e9..1d1e0cb29ed 100644
--- a/src/mongo/util/time_support.h
+++ b/src/mongo/util/time_support.h
@@ -304,6 +304,7 @@ namespace mongo {
     void sleepsecs(int s);
     void sleepmillis(long long ms);
     void sleepmicros(long long micros);
+    void sleepFor(const Milliseconds& time);
 
     class Backoff {
     public: