Revert "SERVER-6071 use command on local.slaves instead of cursor"

This reverts commit 2267744af0e972bceccb4ff4e9ed19a1ed639d2e.

Revert "SERVER-6071 correct assert code"

This reverts commit 85ca38c33e4aaebad539e78a05aed329b09c1208.

Conflicts:
	src/mongo/db/repl/sync_source_feedback.cpp

Revert "SERVER-6071 use command on local.slaves instead of cursor"

This reverts commit 83ecb9775b7997dd5115c53f6ea30d2e368a4244.

Conflicts:
	src/mongo/db/repl/replset_commands.cpp
	src/mongo/db/repl/sync_source_feedback.cpp

1 files changed, 30 insertions, 38 deletions

diff --git a/src/mongo/db/repl/bgsync.cpp b/src/mongo/db/repl/bgsync.cpp
index f453664262d..6e9e7538f5a 100644
--- a/src/mongo/db/repl/bgsync.cpp
+++ b/src/mongo/db/repl/bgsync.cpp
@@ -22,7 +22,6 @@
 #include "mongo/db/repl/bgsync.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/rs_sync.h"
-#include "mongo/db/repl/rs.h"
 #include "mongo/util/fail_point_service.h"
 #include "mongo/base/counter.h"
 #include "mongo/db/stats/timer_stats.h"
@@ -80,6 +79,7 @@ namespace replset {
                                        _assumingPrimary(false),
                                        _currentSyncTarget(NULL),
                                        _oplogMarkerTarget(NULL),
+                                       _oplogMarker(true /* doHandshake */),
                                        _consumedOpTime(0, 0) {
     }
 
@@ -122,7 +122,6 @@ namespace replset {
     void BackgroundSync::notifierThread() {
         Client::initThread("rsSyncNotifier");
         replLocalAuth();
-        theReplSet->syncSourceFeedback.go();
 
         while (!inShutdown()) {
             bool clearTarget = false;
@@ -169,44 +168,37 @@ namespace replset {
     }
 
     void BackgroundSync::markOplog() {
-        LOG(3) << "replset markOplog: " << _consumedOpTime << " "
-               << theReplSet->lastOpTimeWritten << rsLog;
+        LOG(3) << "replset markOplog: " << _consumedOpTime << " " << theReplSet->lastOpTimeWritten << rsLog;
 
-        if (theReplSet->syncSourceFeedback.supportsUpdater()) {
-            theReplSet->syncSourceFeedback.updateSelfInMap(theReplSet->lastOpTimeWritten);
-            _consumedOpTime = theReplSet->lastOpTimeWritten;
+        if (!hasCursor()) {
+            sleepsecs(1);
+            return;
         }
-        else {
-            if (!hasCursor()) {
-                return;
-            }
 
-            if (!theReplSet->syncSourceFeedback.moreInCurrentBatch()) {
-                theReplSet->syncSourceFeedback.more();
-            }
-
-            if (!theReplSet->syncSourceFeedback.more()) {
-                theReplSet->syncSourceFeedback.tailCheck();
-                return;
-            }
+        if (!_oplogMarker.moreInCurrentBatch()) {
+            _oplogMarker.more();
+        }
 
-            // if this member has written the op at optime T
-            // we want to nextSafe up to and including T
-            while (_consumedOpTime < theReplSet->lastOpTimeWritten
-                   && theReplSet->syncSourceFeedback.more()) {
-                BSONObj temp = theReplSet->syncSourceFeedback.nextSafe();
-                _consumedOpTime = temp["ts"]._opTime();
-            }
+        if (!_oplogMarker.more()) {
+            _oplogMarker.tailCheck();
+            sleepsecs(1);
+            return;
+        }
 
-            // call more() to signal the sync target that we've synced T
-            theReplSet->syncSourceFeedback.more();
+        // if this member has written the op at optime T, we want to nextSafe up to and including T
+        while (_consumedOpTime < theReplSet->lastOpTimeWritten && _oplogMarker.more()) {
+            BSONObj temp = _oplogMarker.nextSafe();
+            _consumedOpTime = temp["ts"]._opTime();
         }
+
+        // call more() to signal the sync target that we've synced T
+        _oplogMarker.more();
     }
 
     bool BackgroundSync::hasCursor() {
         {
             // prevent writers from blocking readers during fsync
-            SimpleMutex::scoped_lock fsynclk(filesLockedFsync);
+            SimpleMutex::scoped_lock fsynclk(filesLockedFsync); 
             // we don't need the local write lock yet, but it's needed by OplogReader::connect
             // so we take it preemptively to avoid deadlocking.
             Lock::DBWrite lk("local");
@@ -218,23 +210,25 @@ namespace replset {
                     return false;
                 }
 
-                log() << "replset setting oplog notifier to "
-                      << _currentSyncTarget->fullName() << rsLog;
+                log() << "replset setting oplog notifier to " << _currentSyncTarget->fullName() << rsLog;
                 _oplogMarkerTarget = _currentSyncTarget;
 
-                if (!theReplSet->syncSourceFeedback.connect(_oplogMarkerTarget)) {
+                _oplogMarker.resetConnection();
+
+                if (!_oplogMarker.connect(_oplogMarkerTarget->fullName())) {
+                    LOG(1) << "replset could not connect to " << _oplogMarkerTarget->fullName() << rsLog;
                     _oplogMarkerTarget = NULL;
                     return false;
                 }
             }
         }
-        if (!theReplSet->syncSourceFeedback.haveCursor()) {
+
+        if (!_oplogMarker.haveCursor()) {
             BSONObj fields = BSON("ts" << 1);
-            theReplSet->syncSourceFeedback.tailingQueryGTE(rsoplog,
-                                                theReplSet->lastOpTimeWritten, &fields);
+            _oplogMarker.tailingQueryGTE(rsoplog, theReplSet->lastOpTimeWritten, &fields);
         }
 
-        return theReplSet->syncSourceFeedback.haveCursor();
+        return _oplogMarker.haveCursor();
     }
 
     void BackgroundSync::producerThread() {
@@ -531,8 +525,6 @@ namespace replset {
                 _currentSyncTarget = target;
             }
 
-            theReplSet->syncSourceFeedback.connect(target);
-
             return;
         }