void ChunkManager::loadExistingRanges(const ChunkManager* oldManager) {
        int tries = 3;

        while (tries--) {
            ChunkMap chunkMap;
            set<Shard> shards;
            ShardVersionMap shardVersions;

            Timer t;

            bool success = _load(chunkMap, shards, &shardVersions, oldManager);
            if (success) {
                log() << "ChunkManager: time to load chunks for " << _ns << ": "
                      << t.millis() << "ms"
                      << " sequenceNumber: " << _sequenceNumber
                      << " version: " << _version.toString()
                      << " based on: "
                            << (oldManager ? oldManager->getVersion().toString() : "(empty)");

                // TODO: Merge into diff code above, so we validate in one place
                if (isChunkMapValid(chunkMap)) {
                    _chunkMap.swap(chunkMap);
                    _shards.swap(shards);
                    _shardVersions.swap(shardVersions);
                    _chunkRanges.reloadAll(_chunkMap);

                    return;
                }
            }

            if (_chunkMap.size() < 10) {
                _printChunks();
            }

            warning() << "ChunkManager loaded an invalid config for " << _ns << ", trying again";

            sleepmillis(10 * (3 - tries));
        }

        // This will abort construction so we should never have a reference to an invalid config
        msgasserted(13282, str::stream() << "Couldn't load a valid config for " << _ns
                                         << " after 3 attempts. Please try again.");
    }

 bool waitForSyncToFinish(OperationContext* txn, string &errmsg) const {
     // Wait for slave thread to finish syncing, so sources will be be
     // reloaded with new saved state on next pass.
     Timer t;
     while ( 1 ) {
         if ( syncing == 0 || t.millis() > 30000 )
             break;
         {
             Lock::TempRelease t(txn->lockState());
             relinquishSyncingSome = 1;
             sleepmillis(1);
         }
     }
     if ( syncing ) {
         errmsg = "timeout waiting for sync() to finish";
         return false;
     }
     return true;
 }

 void MiniWebServer::run() {
     SockAddr from;
     while ( 1 ) {
         int s = accept(sock, from.getSockAddr(), &from.addressSize);
         if ( s < 0 ) {
             if ( errno == ECONNABORTED ) {
                 log() << "Listener on port " << port << " aborted." << endl;
                 return;
             }
             log() << "MiniWebServer: accept() returns " << s << " errno:" << errno << endl;
             sleepmillis(200);
             continue;
         }
         disableNagle(s);
         RARELY log() << "MiniWebServer: connection accepted from " << from.toString() << endl;
         accepted( s, from );
         closesocket(s);
     }
 }

void WiredTigerSessionCache::shuttingDown() {
    uint32_t actual = _shuttingDown.load();
    uint32_t expected;

    // Try to atomically set _shuttingDown flag, but just return if another thread was first.
    do {
        expected = actual;
        actual = _shuttingDown.compareAndSwap(expected, expected | kShuttingDownMask);
        if (actual & kShuttingDownMask)
            return;
    } while (actual != expected);

    // Spin as long as there are threads in releaseSession
    while (_shuttingDown.load() != kShuttingDownMask) {
        sleepmillis(1);
    }

    closeAll();
}

    /* static */
    BSONObj WriteBackListener::waitFor( const ConnectionIdent& ident, const OID& oid ) {

        Timer t;
        Timer lastMessageTimer;

        while ( t.minutes() < 60 ) {
            {
                scoped_lock lk( _seenWritebacksLock );
                WBStatus s = _seenWritebacks[ident];

                if ( oid < s.id ) {
                    // this means we're waiting for a GLE that already passed.
                    // it should be impossible because once we call GLE, no other
                    // writebacks should happen with that connection id

                    msgasserted( 14041 , str::stream() << "got writeback waitfor for older id " <<
                                 " oid: " << oid << " s.id: " << s.id << " ident: " << ident.toString() );
                }
                else if ( oid == s.id ) {
                    return s.gle;
                }

                // Stay in lock so we can use the status
                if( lastMessageTimer.seconds() > 10 ){

                    warning() << "waiting for writeback " << oid
                              << " from connection " << ident.toString()
                              << " for " << t.seconds() << " secs"
                              << ", currently at id " << s.id << endl;

                    lastMessageTimer.reset();
                }
            }

            sleepmillis( 10 );
        }

        uasserted( 13403 , str::stream() << "didn't get writeback for: " << oid
                                         << " after: " << t.millis() << " ms"
                                         << " from connection " << ident.toString() );

        throw 1; // never gets here
    }

void BackgroundSync::_run() {
    Client::initThread("rsBackgroundSync");
    AuthorizationSession::get(cc())->grantInternalAuthorization();

    while (!inShutdown()) {
        try {
            _runProducer();
        } catch (const DBException& e) {
            std::string msg(str::stream() << "sync producer problem: " << redact(e));
            error() << msg;
            _replCoord->setMyHeartbeatMessage(msg);
            sleepmillis(100);  // sleep a bit to keep from hammering this thread with temp. errors.
        } catch (const std::exception& e2) {
            // redact(std::exception&) doesn't work
            severe() << "sync producer exception: " << redact(e2.what());
            fassertFailed(28546);
        }
    }
    stop();
}

 void t() {
     for( int i = 0; i < 20; i++ ) {
         sleepmillis(21);
         string fn = "/tmp/t1";
         MongoMMF f;
         unsigned long long len = 1 * 1024 * 1024;
         assert( f.create(fn, len, /*sequential*/rand()%2==0) );
         {
             char *p = (char *) f.getView();
             assert(p);
             // write something to the private view as a test
             strcpy(p, "hello");
         }
         if( cmdLine.dur ) {
             char *w = (char *) f.view_write();
             strcpy(w + 6, "world");
         }
         MongoFileFinder ff;
         ASSERT( ff.findByPath(fn) );
     }
 }

 static void durThread() { 
     Client::initThread("dur");
     const int HowOftenToGroupCommitMs = 100;
     while( 1 ) { 
         try {
             int millis = HowOftenToGroupCommitMs;
             {
                 Timer t;
                 journalRotate(); // note we do this part outside of mongomutex
                 millis -= t.millis();
                 if( millis < 5 || millis > HowOftenToGroupCommitMs )
                     millis = 5;
             }
             sleepmillis(millis);
             go();
         }
         catch(std::exception& e) { 
             log() << "exception in durThread " << e.what() << endl;
         }
     }
 }

        static void runThread() {
            while (keepGoing) {
                try {
                    if (current->lock_try( "test" )) {
                        int before = count.addAndFetch(1);
                        sleepmillis(3);
                        int after = count.loadRelaxed();

                        if (after != before) {
                            error() << " before: " << before << " after: " << after
                                    << endl;
                        }

                        current->unlock();
                    }
                }
                catch ( const DBException& ex ) {
                    log() << "*** !Could not try distributed lock." << causedBy( ex ) << endl;
                }
            }
        }

    void DataFileSync::run() {
        Client::initThread( name().c_str() );

        if (mmapv1GlobalOptions.syncdelay == 0) {
            log() << "warning: --syncdelay 0 is not recommended and can have strange performance" << endl;
        }
        else if (mmapv1GlobalOptions.syncdelay == 1) {
            log() << "--syncdelay 1" << endl;
        }
        else if (mmapv1GlobalOptions.syncdelay != 60) {
            LOG(1) << "--syncdelay " << mmapv1GlobalOptions.syncdelay << endl;
        }
        int time_flushing = 0;
        while ( ! inShutdown() ) {
            _diaglog.flush();
            if (mmapv1GlobalOptions.syncdelay == 0) {
                // in case at some point we add an option to change at runtime
                sleepsecs(5);
                continue;
            }

            sleepmillis((long long) std::max(0.0, (mmapv1GlobalOptions.syncdelay * 1000) - time_flushing));

            if ( inShutdown() ) {
                // occasional issue trying to flush during shutdown when sleep interrupted
                break;
            }

            Date_t start = jsTime();
            StorageEngine* storageEngine = getGlobalEnvironment()->getGlobalStorageEngine();
            int numFiles = storageEngine->flushAllFiles( true );
            time_flushing = (int) (jsTime() - start);

            _flushed(time_flushing);

            if( logger::globalLogDomain()->shouldLog(logger::LogSeverity::Debug(1)) || time_flushing >= 10000 ) {
                log() << "flushing mmaps took " << time_flushing << "ms " << " for " << numFiles << " files" << endl;
            }
        }
    }

 void BackgroundSync::handleSlaveDelay(uint64_t opTimestamp) {
     dassert(_opSyncRunning);
     uint64_t slaveDelayMillis = theReplSet->myConfig().slaveDelay * 1000;
     uint64_t currTime = curTimeMillis64();
     uint64_t timeOpShouldBeApplied = opTimestamp + slaveDelayMillis;
     while (currTime < timeOpShouldBeApplied) {
         uint64_t sleepTime = (timeOpShouldBeApplied - currTime);
         // let's sleep for at most one second
         sleepmillis((sleepTime < 1000) ? sleepTime : 1000);        
         // check if we should bail out, as we don't want to 
         // sleep the whole time possibly long delay time
         // if we see we should be stopping
         {
             boost::unique_lock<boost::mutex> lck(_mutex);
             if (!_opSyncShouldRun) {
                 break;
             }
         }
         // reset currTime
         currTime = curTimeMillis64();
     }
 }

        bool run(const string& , BSONObj& cmdObj, int, string& errmsg, BSONObjBuilder& result, bool) {

            cc().curop()->suppressFromCurop();
            cc().curop()->setExpectedLatencyMs( 30000 );

            BSONElement e = cmdObj.firstElement();
            if ( e.type() != jstOID ) {
                errmsg = "need oid as first value";
                return 0;
            }

            // get the command issuer's (a mongos) serverID
            const OID id = e.__oid();

            // the command issuer is blocked awaiting a response
            // we want to do return at least at every 5 minutes so sockets don't timeout
            BSONObj z;
            if ( writeBackManager.getWritebackQueue(id.str())->queue.blockingPop( z, 5 * 60 /* 5 minutes */ ) ) {
                LOG(1) << "WriteBackCommand got : " << z << endl;
                result.append( "data" , z );
            }
            else {
                result.appendBool( "noop" , true );
            }

#ifdef _DEBUG
            PseudoRandom r(static_cast<int64_t>(time(0)));
            // Sleep a short amount of time usually
            int sleepFor = r.nextInt32( 10 );
            sleepmillis( sleepFor );

            // Sleep a longer amount of time every once and awhile
            int sleepLong = r.nextInt32( 50 );
            if( sleepLong == 0 ) sleepsecs( 2 );
#endif

            return true;
        }

void BackgroundSync::_rollback(OperationContext* txn,
                               const HostAndPort& source,
                               stdx::function<DBClientBase*()> getConnection) {
    // Abort only when syncRollback detects we are in a unrecoverable state.
    // In other cases, we log the message contained in the error status and retry later.
    auto status = syncRollback(txn,
                               OplogInterfaceLocal(txn, rsOplogName),
                               RollbackSourceImpl(getConnection, source, rsOplogName),
                               _replCoord);
    if (status.isOK()) {
        // When the syncTail thread sees there is no new data by adding something to the buffer.
        _signalNoNewDataForApplier();
        // Wait until the buffer is empty.
        // This is an indication that syncTail has removed the sentinal marker from the buffer
        // and reset its local lastAppliedOpTime via the replCoord.
        while (!_buffer.empty()) {
            sleepmillis(10);
            if (inShutdown()) {
                return;
            }
        }

        // It is now safe to clear the ROLLBACK state, which may result in the applier thread
        // transitioning to SECONDARY.  This is safe because the applier thread has now reloaded
        // the new rollback minValid from the database.
        if (!_replCoord->setFollowerMode(MemberState::RS_RECOVERING)) {
            warning() << "Failed to transition into " << MemberState(MemberState::RS_RECOVERING)
                      << "; expected to be in state " << MemberState(MemberState::RS_ROLLBACK)
                      << " but found self in " << _replCoord->getMemberState();
        }
        return;
    }
    if (ErrorCodes::UnrecoverableRollbackError == status.code()) {
        fassertNoTrace(28723, status);
    }
    warning() << "rollback cannot proceed at this time (retrying later): " << status;
}

    void BackgroundSync::markOplog() {
        LOG(3) << "replset markOplog: " << _consumedOpTime << " "
               << theReplSet->lastOpTimeWritten << rsLog;

        if (theReplSet->syncSourceFeedback.supportsUpdater()) {
            _consumedOpTime = theReplSet->lastOpTimeWritten;
            theReplSet->syncSourceFeedback.updateSelfInMap(theReplSet->lastOpTimeWritten);
        }
        else {
            boost::unique_lock<boost::mutex> oplogLockSSF(theReplSet->syncSourceFeedback.oplock);
            if (!hasCursor()) {
                oplogLockSSF.unlock();
                sleepmillis(500);
                return;
            }

            if (!theReplSet->syncSourceFeedback.moreInCurrentBatch()) {
                theReplSet->syncSourceFeedback.more();
            }

            if (!theReplSet->syncSourceFeedback.more()) {
                theReplSet->syncSourceFeedback.tailCheck();
                return;
            }

            // 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();
            }

            // call more() to signal the sync target that we've synced T
            theReplSet->syncSourceFeedback.more();
        }
    }

 void run() {
     Client::initThread("fsyncjob");
     Client& c = cc();
     {
         scoped_lock lk(lockedForWritingMutex);
         lockedForWriting++;
     }
     readlock lk("");
     MemoryMappedFile::flushAll(true);
     log() << "db is now locked for snapshotting, no writes allowed. use db.$cmd.sys.unlock.findOne() to unlock" << endl;
     _ready = true;
     while( 1 ) {
         if( unlockRequested ) {
             unlockRequested = false;
             break;
         }
         sleepmillis(20);
     }
     {
         scoped_lock lk(lockedForWritingMutex);
         lockedForWriting--;
     }
     c.shutdown();
 }

    void BackgroundSync::produce() {
        // this oplog reader does not do a handshake because we don't want the server it's syncing
        // from to track how far it has synced
        OplogReader r(false /* doHandshake */);

        // find a target to sync from the last op time written
        getOplogReader(r);

        // no server found
        {
            boost::unique_lock<boost::mutex> lock(_mutex);

            if (_currentSyncTarget == NULL) {
                lock.unlock();
                sleepsecs(1);
                // if there is no one to sync from
                return;
            }

            r.tailingQueryGTE(rsoplog, _lastOpTimeFetched);
        }

        // if target cut connections between connecting and querying (for
        // example, because it stepped down) we might not have a cursor
        if (!r.haveCursor()) {
            return;
        }

        while (MONGO_FAIL_POINT(rsBgSyncProduce)) {
            sleepmillis(0);
        }

        uassert(1000, "replSet source for syncing doesn't seem to be await capable -- is it an older version of mongodb?", r.awaitCapable() );

        if (isRollbackRequired(r)) {
            stop();
            return;
        }

        while (!inShutdown()) {
            while (!inShutdown()) {

                if (!r.moreInCurrentBatch()) {
                    int bs = r.currentBatchMessageSize();
                    if( bs > 0 && bs < BatchIsSmallish ) {
                        // on a very low latency network, if we don't wait a little, we'll be 
                        // getting ops to write almost one at a time.  this will both be expensive
                        // for the upstream server as well as postentiallyd efating our parallel 
                        // application of batches on the secondary.
                        //
                        // the inference here is basically if the batch is really small, we are 
                        // "caught up".
                        //
                        dassert( !Lock::isLocked() );
                        sleepmillis(SleepToAllowBatchingMillis);
                    }
  
                    if (theReplSet->gotForceSync()) {
                        return;
                    }

                    if (isAssumingPrimary() || theReplSet->isPrimary()) {
                        return;
                    }

                    // re-evaluate quality of sync target
                    if (shouldChangeSyncTarget()) {
                        return;
                    }
                    //record time for each getmore
                    {
                        TimerHolder batchTimer(&getmoreReplStats);
                        r.more();
                    }
                    //increment
                    networkByteStats.increment(r.currentBatchMessageSize());

                }

                if (!r.more())
                    break;

                BSONObj o = r.nextSafe().getOwned();
                opsReadStats.increment();

                {
                    boost::unique_lock<boost::mutex> lock(_mutex);
                    _appliedBuffer = false;
                }

                OCCASIONALLY {
                    LOG(2) << "bgsync buffer has " << _buffer.size() << " bytes" << rsLog;
                }
                // the blocking queue will wait (forever) until there's room for us to push
                _buffer.push(o);
                bufferCountGauge.increment();
                bufferSizeGauge.increment(getSize(o));

                {
                    boost::unique_lock<boost::mutex> lock(_mutex);
//.........这里部分代码省略.........

        bool run(const char *ns, BSONObj& cmdObj, string& errmsg, BSONObjBuilder& result, bool fromRepl) {
            log(1) << " authenticate: " << cmdObj << endl;

            string user = cmdObj.getStringField("user");
            string key = cmdObj.getStringField("key");
            string received_nonce = cmdObj.getStringField("nonce");
            
            if( user.empty() || key.empty() || received_nonce.empty() ) { 
                log() << "field missing/wrong type in received authenticate command " 
                    << cc().database()->name
                    << endl;               
                errmsg = "auth fails";
                sleepmillis(10);
                return false;
            }
            
            stringstream digestBuilder;

            {
                bool reject = false;
                nonce *ln = lastNonce.release();
                if ( ln == 0 ) {
                    reject = true;
                } else {
                    digestBuilder << hex << *ln;
                    reject = digestBuilder.str() != received_nonce;
                }
                    
                if ( reject ) {
                    log() << "auth: bad nonce received or getnonce not called. could be a driver bug or a security attack. db:" << cc().database()->name << endl;
                    errmsg = "auth fails";
                    sleepmillis(30);
                    return false;
                }
            }

            static BSONObj userPattern = fromjson("{\"user\":1}");
            string systemUsers = cc().database()->name + ".system.users";
            OCCASIONALLY Helpers::ensureIndex(systemUsers.c_str(), userPattern, false, "user_1");

            BSONObj userObj;
            {
                BSONObjBuilder b;
                b << "user" << user;
                BSONObj query = b.done();
                if( !Helpers::findOne(systemUsers.c_str(), query, userObj) ) { 
                    log() << "auth: couldn't find user " << user << ", " << systemUsers << endl;
                    errmsg = "auth fails";
                    return false;
                }
            }
            
            md5digest d;
            {
                
                string pwd = userObj.getStringField("pwd");
                digestBuilder << user << pwd;
                string done = digestBuilder.str();
                
                md5_state_t st;
                md5_init(&st);
                md5_append(&st, (const md5_byte_t *) done.c_str(), done.size());
                md5_finish(&st, d);
            }
            
            string computed = digestToString( d );
            
            if ( key != computed ){
                log() << "auth: key mismatch " << user << ", ns:" << ns << endl;
                errmsg = "auth fails";
                return false;
            }

            AuthenticationInfo *ai = cc().getAuthenticationInfo();
            
            if ( userObj[ "readOnly" ].isBoolean() && userObj[ "readOnly" ].boolean() ) {
                if ( readLockSupported() ){
                    ai->authorizeReadOnly( cc().database()->name.c_str() );
                }
                else {
                    log() << "warning: old version of boost, read-only users not supported" << endl;
                    ai->authorize( cc().database()->name.c_str() );
                }
            } else {
                ai->authorize( cc().database()->name.c_str() );
            }
            return true;
        }

    virtual void subthread(int tnumber) {
        Client::initThread("mongomutextest");

        const ServiceContext::UniqueOperationContext txnPtr = cc().makeOperationContext();
        OperationContext& txn = *txnPtr;

        sleepmillis(0);
        for (int i = 0; i < N; i++) {
            int x = std::rand();
            bool sometimes = (x % 15 == 0);
            if (i % 7 == 0) {
                Lock::GlobalRead r(txn.lockState());  // nested test
                Lock::GlobalRead r2(txn.lockState());
            } else if (i % 7 == 1) {
                Lock::GlobalRead r(txn.lockState());
                ASSERT(txn.lockState()->isReadLocked());
            } else if (i % 7 == 4 && tnumber == 1 /*only one upgrader legal*/) {
                Lock::GlobalWrite w(txn.lockState());
                ASSERT(txn.lockState()->isW());
                if (i % 7 == 2) {
                    Lock::TempRelease t(txn.lockState());
                }
            } else if (i % 7 == 2) {
                Lock::GlobalWrite w(txn.lockState());
                ASSERT(txn.lockState()->isW());
                if (sometimes) {
                    Lock::TempRelease t(txn.lockState());
                }
            } else if (i % 7 == 3) {
                Lock::GlobalWrite w(txn.lockState());
                { Lock::TempRelease t(txn.lockState()); }
                Lock::GlobalRead r(txn.lockState());
                ASSERT(txn.lockState()->isW());
                if (sometimes) {
                    Lock::TempRelease t(txn.lockState());
                }
            } else if (i % 7 == 5) {
                {
                    ScopedTransaction scopedXact(&txn, MODE_IS);
                    Lock::DBLock r(txn.lockState(), "foo", MODE_S);
                }
                {
                    ScopedTransaction scopedXact(&txn, MODE_IS);
                    Lock::DBLock r(txn.lockState(), "bar", MODE_S);
                }
            } else if (i % 7 == 6) {
                if (i > N / 2) {
                    int q = i % 11;
                    if (q == 0) {
                        ScopedTransaction scopedXact(&txn, MODE_IS);

                        Lock::DBLock r(txn.lockState(), "foo", MODE_S);
                        ASSERT(txn.lockState()->isDbLockedForMode("foo", MODE_S));

                        Lock::DBLock r2(txn.lockState(), "foo", MODE_S);
                        ASSERT(txn.lockState()->isDbLockedForMode("foo", MODE_S));

                        Lock::DBLock r3(txn.lockState(), "local", MODE_S);
                        ASSERT(txn.lockState()->isDbLockedForMode("foo", MODE_S));
                        ASSERT(txn.lockState()->isDbLockedForMode("local", MODE_S));
                    } else if (q == 1) {
                        // test locking local only -- with no preceding lock
                        {
                            ScopedTransaction scopedXact(&txn, MODE_IS);
                            Lock::DBLock x(txn.lockState(), "local", MODE_S);
                        }
                        {
                            ScopedTransaction scopedXact(&txn, MODE_IX);
                            Lock::DBLock x(txn.lockState(), "local", MODE_X);

                            //  No actual writing here, so no WriteUnitOfWork
                            if (sometimes) {
                                Lock::TempRelease t(txn.lockState());
                            }
                        }
                    } else if (q == 1) {
                        {
                            ScopedTransaction scopedXact(&txn, MODE_IS);
                            Lock::DBLock x(txn.lockState(), "admin", MODE_S);
                        }

                        {
                            ScopedTransaction scopedXact(&txn, MODE_IX);
                            Lock::DBLock x(txn.lockState(), "admin", MODE_X);
                        }
                    } else if (q == 3) {
                        ScopedTransaction scopedXact(&txn, MODE_IX);

                        Lock::DBLock x(txn.lockState(), "foo", MODE_X);
                        Lock::DBLock y(txn.lockState(), "admin", MODE_S);
                    } else if (q == 4) {
                        ScopedTransaction scopedXact(&txn, MODE_IS);

                        Lock::DBLock x(txn.lockState(), "foo2", MODE_S);
                        Lock::DBLock y(txn.lockState(), "admin", MODE_S);
                    } else {
                        ScopedTransaction scopedXact(&txn, MODE_IX);

                        Lock::DBLock w(txn.lockState(), "foo", MODE_X);

//.........这里部分代码省略.........

    void BackgroundSync::getOplogReader(OplogReader& r) {
        const Member *target = NULL, *stale = NULL;
        BSONObj oldest;

        {
            boost::unique_lock<boost::mutex> lock(_mutex);
            if (_lastOpTimeFetched.isNull()) {
                // then we're initial syncing and we're still waiting for this to be set
                _currentSyncTarget = NULL;
                return;
            }

            // Wait until we've applied the ops we have before we choose a sync target
            while (!_appliedBuffer) {
                _condvar.wait(lock);
            }
        }

        while (MONGO_FAIL_POINT(rsBgSyncProduce)) {
            sleepmillis(0);
        }

        verify(r.conn() == NULL);

        while ((target = theReplSet->getMemberToSyncTo()) != NULL) {
            string current = target->fullName();

            if (!r.connect(current)) {
                LOG(2) << "replSet can't connect to " << current << " to read operations" << rsLog;
                r.resetConnection();
                theReplSet->veto(current);
                sleepsecs(1);
                continue;
            }

            if (isStale(r, oldest)) {
                r.resetConnection();
                theReplSet->veto(current, 600);
                stale = target;
                continue;
            }

            // if we made it here, the target is up and not stale
            {
                boost::unique_lock<boost::mutex> lock(_mutex);
                _currentSyncTarget = target;
            }

            boost::unique_lock<boost::mutex> oplogLockSSF(theReplSet->syncSourceFeedback.oplock);
            theReplSet->syncSourceFeedback.connect(target);

            return;
        }

        // the only viable sync target was stale
        if (stale) {
            theReplSet->goStale(stale, oldest);
            sleepsecs(120);
        }

        {
            boost::unique_lock<boost::mutex> lock(_mutex);
            _currentSyncTarget = NULL;
        }
    }

    void BackgroundSync::produce() {
        // this oplog reader does not do a handshake because we don't want the server it's syncing
        // from to track how far it has synced
        OplogReader r;
        OpTime lastOpTimeFetched;
        // find a target to sync from the last op time written
        getOplogReader(r);

        // no server found
        {
            boost::unique_lock<boost::mutex> lock(_mutex);

            if (_currentSyncTarget == NULL) {
                lock.unlock();
                sleepsecs(1);
                // if there is no one to sync from
                return;
            }
            lastOpTimeFetched = _lastOpTimeFetched;
        }

        r.tailingQueryGTE(rsoplog, lastOpTimeFetched);

        // if target cut connections between connecting and querying (for
        // example, because it stepped down) we might not have a cursor
        if (!r.haveCursor()) {
            return;
        }

        uassert(1000, "replSet source for syncing doesn't seem to be await capable -- is it an older version of mongodb?", r.awaitCapable() );

        if (isRollbackRequired(r)) {
            stop();
            return;
        }

        while (!inShutdown()) {
            if (!r.moreInCurrentBatch()) {
                // Check some things periodically
                // (whenever we run out of items in the
                // current cursor batch)

                int bs = r.currentBatchMessageSize();
                if( bs > 0 && bs < BatchIsSmallish ) {
                    // on a very low latency network, if we don't wait a little, we'll be 
                    // getting ops to write almost one at a time.  this will both be expensive
                    // for the upstream server as well as potentially defeating our parallel 
                    // application of batches on the secondary.
                    //
                    // the inference here is basically if the batch is really small, we are 
                    // "caught up".
                    //
                    dassert( !Lock::isLocked() );
                    sleepmillis(SleepToAllowBatchingMillis);
                }
  
                if (theReplSet->gotForceSync()) {
                    return;
                }
                // If we are transitioning to primary state, we need to leave
                // this loop in order to go into bgsync-pause mode.
                if (isAssumingPrimary() || theReplSet->isPrimary()) {
                    return;
                }

                // re-evaluate quality of sync target
                if (shouldChangeSyncTarget()) {
                    return;
                }


                {
                    //record time for each getmore
                    TimerHolder batchTimer(&getmoreReplStats);
                    
                    // This calls receiveMore() on the oplogreader cursor.
                    // It can wait up to five seconds for more data.
                    r.more();
                }
                networkByteStats.increment(r.currentBatchMessageSize());

                if (!r.moreInCurrentBatch()) {
                    // If there is still no data from upstream, check a few more things
                    // and then loop back for another pass at getting more data
                    {
                        boost::unique_lock<boost::mutex> lock(_mutex);
                        if (_pause || 
                            !_currentSyncTarget || 
                            !_currentSyncTarget->hbinfo().hbstate.readable()) {
                            return;
                        }
                    }

                    r.tailCheck();
                    if( !r.haveCursor() ) {
                        LOG(1) << "replSet end syncTail pass" << rsLog;
                        return;
                    }

                    continue;
//.........这里部分代码省略.........