SERVER-13420 remove syncSourceFeedback connection management from the notifier thread

4 files changed, 4 insertions, 56 deletions

diff --git a/src/mongo/db/repl/bgsync.cpp b/src/mongo/db/repl/bgsync.cpp
index 4ac657ae71d..ab1921ec4dd 100644
--- a/src/mongo/db/repl/bgsync.cpp
+++ b/src/mongo/db/repl/bgsync.cpp
@@ -92,7 +92,6 @@ namespace replset {
                                        _appliedBuffer(true),
                                        _assumingPrimary(false),
                                        _currentSyncTarget(NULL),
-                                       _oplogMarkerTarget(NULL),
                                        _consumedOpTime(0, 0) {
     }
 
@@ -136,19 +135,9 @@ namespace replset {
         Client::initThread("rsSyncNotifier");
         replLocalAuth();
 
-        // This makes the initial connection to our sync source for oplog position notification.
-        // It also sets the supportsUpdater flag so we know which method to use.
-        // If this function fails, we ignore that situation because it will be taken care of
-        // the first time markOplog() is called in the loop below.
-        {
-            boost::unique_lock<boost::mutex> oplogLockSSF(theReplSet->syncSourceFeedback.oplock);
-            connectOplogNotifier();
-        }
         theReplSet->syncSourceFeedback.go();
 
         while (!inShutdown()) {
-            bool clearTarget = false;
-
             if (!theReplSet) {
                 sleepsecs(5);
                 continue;
@@ -171,20 +160,13 @@ namespace replset {
                 markOplog();
             }
             catch (DBException &e) {
-                clearTarget = true;
                 log() << "replset tracking exception: " << e.getInfo() << rsLog;
                 sleepsecs(1);
             }
             catch (std::exception &e2) {
-                clearTarget = true;
                 log() << "replset tracking error" << e2.what() << rsLog;
                 sleepsecs(1);
             }
-
-            if (clearTarget) {
-                boost::unique_lock<boost::mutex> lock(_mutex);
-                _oplogMarkerTarget = NULL;
-            }
         }
 
         cc().shutdown();
@@ -194,12 +176,13 @@ namespace replset {
         LOG(3) << "replset markOplog: " << _consumedOpTime << " "
                << theReplSet->lastOpTimeWritten << rsLog;
 
+        boost::unique_lock<boost::mutex> oplogLockSSF(theReplSet->syncSourceFeedback.oplock);
         if (theReplSet->syncSourceFeedback.supportsUpdater()) {
+            oplogLockSSF.unlock();
             _consumedOpTime = theReplSet->lastOpTimeWritten;
             theReplSet->syncSourceFeedback.updateSelfInMap(theReplSet->lastOpTimeWritten);
         }
         else {
-            boost::unique_lock<boost::mutex> oplogLockSSF(theReplSet->syncSourceFeedback.oplock);
             if (!hasCursor()) {
                 oplogLockSSF.unlock();
                 sleepmillis(500);
@@ -228,30 +211,7 @@ namespace replset {
         }
     }
 
-    bool BackgroundSync::connectOplogNotifier() {
-        boost::unique_lock<boost::mutex> lock(_mutex);
-
-        if (!_oplogMarkerTarget || _currentSyncTarget != _oplogMarkerTarget) {
-            if (!_currentSyncTarget) {
-                return false;
-            }
-
-            log() << "replset setting oplog notifier to "
-                  << _currentSyncTarget->fullName() << rsLog;
-            _oplogMarkerTarget = _currentSyncTarget;
-
-            if (!theReplSet->syncSourceFeedback.connect(_oplogMarkerTarget)) {
-                _oplogMarkerTarget = NULL;
-                return false;
-            }
-        }
-        return true;
-    }
-
     bool BackgroundSync::hasCursor() {
-        if (!connectOplogNotifier()) {
-            return false;
-        }
         if (!theReplSet->syncSourceFeedback.haveCursor()) {
             BSONObj fields = BSON("ts" << 1);
             theReplSet->syncSourceFeedback.tailingQueryGTE(rsoplog,
@@ -464,14 +424,6 @@ namespace replset {
 
 
     bool BackgroundSync::peek(BSONObj* op) {
-        {
-            boost::unique_lock<boost::mutex> lock(_mutex);
-
-            if (_currentSyncTarget != _oplogMarkerTarget &&
-                _currentSyncTarget != NULL) {
-                _oplogMarkerTarget = NULL;
-            }
-        }
         return _buffer.peek(*op);
     }
 
diff --git a/src/mongo/db/repl/bgsync.h b/src/mongo/db/repl/bgsync.h
index 2734890fe46..15288a54e30 100644
--- a/src/mongo/db/repl/bgsync.h
+++ b/src/mongo/db/repl/bgsync.h
@@ -97,7 +97,6 @@ namespace replset {
         boost::condition_variable _lastOpCond;
         boost::mutex _lastOpMutex;
 
-        const Member* _oplogMarkerTarget;
         OpTime _consumedOpTime; // not locked, only used by notifier thread
 
         BackgroundSync();
@@ -125,10 +124,6 @@ namespace replset {
         void markOplog();
         bool hasCursor();
 
-        // Sets _oplogMarkerTarget and calls connect();
-        // used for both the notifier command and the older OplogReader style notifier
-        bool connectOplogNotifier();
-
         bool isAssumingPrimary();
 
     public:
diff --git a/src/mongo/db/repl/sync_source_feedback.cpp b/src/mongo/db/repl/sync_source_feedback.cpp
index d2aebd020ee..04c98753e68 100644
--- a/src/mongo/db/repl/sync_source_feedback.cpp
+++ b/src/mongo/db/repl/sync_source_feedback.cpp
@@ -150,6 +150,7 @@ namespace mongo {
         if (hasConnection()) {
             return true;
         }
+        log() << "replset setting syncSourceFeedback to " << hostName << rsLog;
         _connection.reset(new DBClientConnection(false, 0, OplogReader::tcp_timeout));
         string errmsg;
         if (!_connection->connect(hostName.c_str(), errmsg) ||
diff --git a/src/mongo/db/repl/sync_source_feedback.h b/src/mongo/db/repl/sync_source_feedback.h
index 8238952e0b2..859e3efef66 100644
--- a/src/mongo/db/repl/sync_source_feedback.h
+++ b/src/mongo/db/repl/sync_source_feedback.h
@@ -42,7 +42,7 @@ namespace mongo {
         SyncSourceFeedback() : BackgroundJob(false /*don't selfdelete*/),
                               _syncTarget(NULL),
                               _oplogReader(new OplogReader()),
-                              _supportsUpdater(false),
+                              _supportsUpdater(true),
                               _positionChanged(false),
                               _handshakeNeeded(false) {}