SERVER-20928 Synchronize stepdown process with events

Author: visualzhou

src/mongo/db/repl/replication_coordinator_impl.cpp

@@ -405,6 +405,7 @@ void ReplicationCoordinatorImpl::_finishLoadLocalConfig(
         _setCurrentRSConfig_inlock(cbData, localConfig, myIndex.getValue());
     _setMyLastOptimeAndReport_inlock(&lk, lastOpTime, false);
     _externalState->setGlobalTimestamp(lastOpTime.getTimestamp());
+    // Step down is impossible, so we don't need to wait for the returned event.
     _updateTerm_incallback(term);
     LOG(1) << "Current term is now " << term;
     if (lk.owns_lock()) {
@@ -1724,22 +1725,27 @@ void ReplicationCoordinatorImpl::processReplSetGetConfig(BSONObjBuilder* result)
 }
 
 void ReplicationCoordinatorImpl::processReplSetMetadata(const rpc::ReplSetMetadata& replMetadata) {
-    _scheduleWorkAndWaitForCompletion(stdx::bind(
-        &ReplicationCoordinatorImpl::_processReplSetMetadata_incallback, this, replMetadata));
+    EventHandle evh;
+    _scheduleWorkAndWaitForCompletion([this, &evh, &replMetadata](const CallbackArgs& args) {
+        evh = _processReplSetMetadata_incallback(replMetadata);
+    });
+    if (evh.isValid()) {
+        _replExecutor.waitForEvent(evh);
+    }
 }
 
 void ReplicationCoordinatorImpl::cancelAndRescheduleElectionTimeout() {
     stdx::lock_guard<stdx::mutex> lock(_mutex);
     _cancelAndRescheduleElectionTimeout_inlock();
 }
 
-void ReplicationCoordinatorImpl::_processReplSetMetadata_incallback(
+EventHandle ReplicationCoordinatorImpl::_processReplSetMetadata_incallback(
     const rpc::ReplSetMetadata& replMetadata) {
     if (replMetadata.getConfigVersion() != _rsConfig.getConfigVersion()) {
-        return;
+        return EventHandle();
     }
     _setLastCommittedOpTime(replMetadata.getLastOpCommitted());
-    _updateTerm_incallback(replMetadata.getTerm());
+    return _updateTerm_incallback(replMetadata.getTerm());
 }
 
 bool ReplicationCoordinatorImpl::getMaintenanceMode() {
@@ -1908,11 +1914,6 @@ Status ReplicationCoordinatorImpl::processHeartbeat(const ReplSetHeartbeatArgs&
     }
     fassert(18508, cbh.getStatus());
     _replExecutor.wait(cbh.getValue());
-
-    // Wait if heartbeat causes stepdown.
-    if (_stepDownFinishedEvent.isValid()) {
-        _replExecutor.waitForEvent(_stepDownFinishedEvent);
-    }
     return result;
 }
 
@@ -2849,12 +2850,6 @@ Status ReplicationCoordinatorImpl::processReplSetRequestVotes(
     if (!termStatus.isOK() && termStatus.code() != ErrorCodes::StaleTerm)
         return termStatus;
 
-    // Term update may cause current primary step down, we need to wait until it
-    // finishes so that it won't close our connection.
-    if (_stepDownFinishedEvent.isValid()) {
-        _replExecutor.waitForEvent(_stepDownFinishedEvent);
-    }
-
     Status result{ErrorCodes::InternalError, "didn't set status in processReplSetRequestVotes"};
     CBHStatus cbh = _replExecutor.scheduleWork(
         stdx::bind(&ReplicationCoordinatorImpl::_processReplSetRequestVotes_finish,
@@ -2993,10 +2988,6 @@ Status ReplicationCoordinatorImpl::processHeartbeatV1(const ReplSetHeartbeatArgs
     fassert(28645, cbh.getStatus());
     _replExecutor.wait(cbh.getValue());
 
-    // Wait if heartbeat causes stepdown.
-    if (_stepDownFinishedEvent.isValid()) {
-        _replExecutor.waitForEvent(_stepDownFinishedEvent);
-    }
     return result;
 }
 
@@ -3084,65 +3075,80 @@ void ReplicationCoordinatorImpl::_getTerm_helper(const ReplicationExecutor::Call
     *term = _topCoord->getTerm();
 }
 
-StatusWith<ReplicationExecutor::CallbackHandle> ReplicationCoordinatorImpl::updateTerm_nonBlocking(
-    long long term, bool* updated) {
+EventHandle ReplicationCoordinatorImpl::updateTerm_forTest(long long term, bool* updated) {
+    auto finishEvhStatus = _replExecutor.makeEvent();
+    invariantOK(finishEvhStatus.getStatus());
+    EventHandle finishEvh = finishEvhStatus.getValue();
+    auto signalFinishEvent =
+        [this, finishEvh](const CallbackArgs&) { this->_replExecutor.signalEvent(finishEvh); };
+    auto work = [this, term, updated, signalFinishEvent](const CallbackArgs& args) {
+        auto evh = _updateTerm_incallback(term, updated);
+        if (evh.isValid()) {
+            _replExecutor.onEvent(evh, signalFinishEvent);
+        } else {
+            signalFinishEvent(args);
+        }
+    };
+    _scheduleWork(work);
+    return finishEvh;
+}
+
+Status ReplicationCoordinatorImpl::updateTerm(long long term) {
     // Term is only valid if we are replicating.
     if (getReplicationMode() != modeReplSet) {
         return {ErrorCodes::BadValue, "cannot supply 'term' without active replication"};
     }
 
     if (!isV1ElectionProtocol()) {
         // Do not update if not in V1 protocol.
-        return ReplicationExecutor::CallbackHandle();
+        return Status::OK();
     }
 
-    auto work =
-        [this, term, updated](const CallbackArgs&) { *updated = _updateTerm_incallback(term); };
-    return _scheduleWork(work);
-}
-
-Status ReplicationCoordinatorImpl::updateTerm(long long term) {
     bool updated = false;
-    auto result = updateTerm_nonBlocking(term, &updated);
-    if (!result.isOK()) {
-        return result.getStatus();
-    }
-    auto handle = result.getValue();
-    if (handle.isValid()) {
-        _replExecutor.wait(handle);
+    EventHandle finishEvh;
+    auto work = [this, term, &updated, &finishEvh](const CallbackArgs&) {
+        finishEvh = _updateTerm_incallback(term, &updated);
+    };
+    _scheduleWorkAndWaitForCompletion(work);
+    // Wait for potential stepdown to finish.
+    if (finishEvh.isValid()) {
+        _replExecutor.waitForEvent(finishEvh);
     }
-
     if (updated) {
         return {ErrorCodes::StaleTerm, "Replication term of this node was stale; retry query"};
     }
 
     return Status::OK();
 }
 
-bool ReplicationCoordinatorImpl::_updateTerm_incallback(long long term) {
+EventHandle ReplicationCoordinatorImpl::_updateTerm_incallback(long long term, bool* updated) {
     if (!isV1ElectionProtocol()) {
         LOG(3) << "Cannot update term in election protocol version 0";
-        return false;
+        return EventHandle();
     }
 
     auto now = _replExecutor.now();
-    bool updated = _topCoord->updateTerm(term, now);
+    bool termUpdated = _topCoord->updateTerm(term, now);
     {
         stdx::lock_guard<stdx::mutex> lock(_mutex);
         _cachedTerm = _topCoord->getTerm();
 
-        if (updated) {
+        if (termUpdated) {
             _cancelPriorityTakeover_inlock();
             _cancelAndRescheduleElectionTimeout_inlock();
         }
     }
 
-    if (updated && getMemberState().primary()) {
+    if (updated) {
+        *updated = termUpdated;
+    }
+
+    if (termUpdated && getMemberState().primary()) {
         log() << "stepping down from primary, because a new term has begun: " << term;
         _topCoord->prepareForStepDown();
-        _stepDownStart();
+        return _stepDownStart();
     }
-    return updated;
+    return EventHandle();
 }
 
 SnapshotName ReplicationCoordinatorImpl::reserveSnapshotName(OperationContext* txn) {
@@ -3246,12 +3252,6 @@ void ReplicationCoordinatorImpl::waitForElectionDryRunFinish_forTest() {
     }
 }
 
-void ReplicationCoordinatorImpl::waitForStepDownFinish_forTest() {
-    if (_stepDownFinishedEvent.isValid()) {
-        _replExecutor.waitForEvent(_stepDownFinishedEvent);
-    }
-}
-
 void ReplicationCoordinatorImpl::_resetElectionInfoOnProtocolVersionUpgrade(
     const ReplicaSetConfig& newConfig) {
     // On protocol version upgrade, reset last vote as if I just learned the term 0 from other

src/mongo/db/repl/replication_coordinator_impl.h

@@ -363,12 +363,11 @@ class ReplicationCoordinatorImpl : public ReplicationCoordinator, public KillOpL
 
     /**
      * Non-blocking version of updateTerm.
-     * Returns callback handle that we can use to wait for the operation to complete.
-     * When the operation is complete (wait() returns), 'updated' will be set to true
-     * if the term increased.
+     * Returns event handle that we can use to wait for the operation to complete.
+     * When the operation is complete (waitForEvent() returns), 'updated' will be set to true
+     * if the term increased and potential stepdown has finished.
      */
-    StatusWith<ReplicationExecutor::CallbackHandle> updateTerm_nonBlocking(long long term,
-                                                                           bool* updated);
+    ReplicationExecutor::EventHandle updateTerm_forTest(long long term, bool* updated);
 
     /**
      * If called after _startElectSelfV1(), blocks until all asynchronous
@@ -383,12 +382,6 @@ class ReplicationCoordinatorImpl : public ReplicationCoordinator, public KillOpL
      */
     void waitForElectionDryRunFinish_forTest();
 
-    /**
-     * If called after a stepdown starts, blocks until all asynchronous activities associated with
-     * stepdown complete.
-     */
-    void waitForStepDownFinish_forTest();
-
 private:
     using CallbackFn = executor::TaskExecutor::CallbackFn;
 
@@ -968,7 +961,7 @@ class ReplicationCoordinatorImpl : public ReplicationCoordinator, public KillOpL
      */
     void _requestRemotePrimaryStepdown(const HostAndPort& target);
 
-    void _stepDownStart();
+    ReplicationExecutor::EventHandle _stepDownStart();
 
     /**
      * Completes a step-down of the current node.  Must be run with a global
@@ -1050,16 +1043,19 @@ class ReplicationCoordinatorImpl : public ReplicationCoordinator, public KillOpL
     /**
      * Callback that attempts to set the current term in topology coordinator and
      * relinquishes primary if the term actually changes and we are primary.
-     * Returns true if the term increased.
+     * *updated will be true if the term increased.
+     * Returns the finish event if it does not finish in this function, for example,
+     * due to stepdown, otherwise the returned EventHandle is invalid.
      */
-    bool _updateTerm_incallback(long long term);
+    EventHandle _updateTerm_incallback(long long term, bool* updated = nullptr);
 
     /**
      * Callback that processes the ReplSetMetadata returned from a command run against another
      * replica set member and updates protocol version 1 information (most recent optime that is
      * committed, member id of the current PRIMARY, the current config version and the current term)
+     * Returns the finish event which is invalid if the process has already finished.
      */
-    void _processReplSetMetadata_incallback(const rpc::ReplSetMetadata& replMetadata);
+    EventHandle _processReplSetMetadata_incallback(const rpc::ReplSetMetadata& replMetadata);
 
     /**
      * Blesses a snapshot to be used for new committed reads.

src/mongo/db/repl/replication_coordinator_impl_elect_v1.cpp

@@ -180,7 +180,10 @@ void ReplicationCoordinatorImpl::_onDryRunComplete(long long originalTerm) {
     }
 
     log() << "dry election run succeeded, running for election";
-    _updateTerm_incallback(originalTerm + 1);
+    // Stepdown is impossible from this term update.
+    bool updated = false;
+    _updateTerm_incallback(originalTerm + 1, &updated);
+    invariant(updated);
     // Secure our vote for ourself first
     _topCoord->voteForMyselfV1();
 

src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp

@@ -132,6 +132,8 @@ void ReplicationCoordinatorImpl::_handleHeartbeatResponse(
         StatusWith<rpc::ReplSetMetadata> replMetadata =
             rpc::ReplSetMetadata::readFromMetadata(cbData.response.getValue().metadata);
         if (replMetadata.isOK()) {
+            // Asynchronous stepdown could happen, but it will be queued in executor after
+            // this function, so we cannot and don't need to wait for it to finish.
             _processReplSetMetadata_incallback(replMetadata.getValue());
         }
     }
@@ -142,6 +144,8 @@ void ReplicationCoordinatorImpl::_handleHeartbeatResponse(
 
     if (responseStatus.isOK()) {
         networkTime = cbData.response.getValue().elapsedMillis;
+        // TODO(sz) Because the term is duplicated in ReplSetMetaData, we can get rid of this
+        // and update tests.
         _updateTerm_incallback(hbStatusResponse.getValue().getTerm());
         // Postpone election timeout if we have a successful heartbeat response from the primary.
         const auto& hbResponse = hbStatusResponse.getValue();
@@ -225,6 +229,7 @@ void ReplicationCoordinatorImpl::_handleHeartbeatResponseAction(
             invariant(action.getPrimaryConfigIndex() == _selfIndex);
             log() << "Stepping down from primary in response to heartbeat";
             _topCoord->prepareForStepDown();
+            // Don't need to wait for stepdown to finish.
             _stepDownStart();
             break;
         case HeartbeatResponseAction::StepDownRemotePrimary: {
@@ -281,17 +286,17 @@ void ReplicationCoordinatorImpl::_requestRemotePrimaryStepdown(const HostAndPort
     }
 }
 
-void ReplicationCoordinatorImpl::_stepDownStart() {
-    auto event = _makeEvent();
-    if (!event) {
-        return;
+ReplicationExecutor::EventHandle ReplicationCoordinatorImpl::_stepDownStart() {
+    auto finishEvent = _makeEvent();
+    if (!finishEvent) {
+        return finishEvent;
     }
-    _stepDownFinishedEvent = event;
     _replExecutor.scheduleWorkWithGlobalExclusiveLock(
         stdx::bind(&ReplicationCoordinatorImpl::_stepDownFinish,
                    this,
                    stdx::placeholders::_1,
-                   _stepDownFinishedEvent));
+                   finishEvent));
+    return finishEvent;
 }
 
 void ReplicationCoordinatorImpl::_stepDownFinish(
@@ -588,6 +593,8 @@ void ReplicationCoordinatorImpl::_handleLivenessTimeout(
                 // downstream.
                 HeartbeatResponseAction action =
                     _topCoord->setMemberAsDown(now, memberIndex, _getMyLastOptime_inlock());
+                // Don't mind potential asynchronous stepdown as this is the last step of
+                // liveness check.
                 _handleHeartbeatResponseAction(action, makeStatusWith<ReplSetHeartbeatResponse>());
             }
         }

src/mongo/db/repl/replication_coordinator_impl_test.cpp

@@ -1284,7 +1284,6 @@ TEST_F(ReplCoordTest, UpdateTerm) {
     Handle cbHandle;
     ASSERT_EQUALS(ErrorCodes::StaleTerm, getReplCoord()->updateTerm(2).code());
     ASSERT_EQUALS(2, getReplCoord()->getTerm());
-    getReplCoord()->waitForStepDownFinish_forTest();
     ASSERT_TRUE(getReplCoord()->getMemberState().secondary());
 }
 
@@ -1322,33 +1321,25 @@ TEST_F(ReplCoordTest, ConcurrentStepDownShouldNotSignalTheSameFinishEventMoreTha
     replExec->scheduleWorkWithGlobalExclusiveLock(stepDownFinishBlocker);
 
     bool termUpdated2 = false;
-    auto updateTermResult2 = getReplCoord()->updateTerm_nonBlocking(2, &termUpdated2);
-    ASSERT_OK(updateTermResult2.getStatus());
+    auto updateTermEvh2 = getReplCoord()->updateTerm_forTest(2, &termUpdated2);
+    ASSERT(updateTermEvh2.isValid());
 
     bool termUpdated3 = false;
-    auto updateTermResult3 = getReplCoord()->updateTerm_nonBlocking(3, &termUpdated3);
-    ASSERT_OK(updateTermResult3.getStatus());
+    auto updateTermEvh3 = getReplCoord()->updateTerm_forTest(3, &termUpdated3);
+    ASSERT(updateTermEvh3.isValid());
 
     // Unblock 'stepDownFinishBlocker'. Tasks for updateTerm and _stepDownFinish should proceed.
     barrier.countDownAndWait();
 
     // Both _updateTerm_incallback tasks should be scheduled.
-    auto handle2 = updateTermResult2.getValue();
-    ASSERT_TRUE(handle2.isValid());
-    replExec->wait(handle2);
+    replExec->waitForEvent(updateTermEvh2);
     ASSERT_TRUE(termUpdated2);
-
-    auto handle3 = updateTermResult3.getValue();
-    ASSERT_TRUE(handle3.isValid());
-    replExec->wait(handle3);
+    replExec->waitForEvent(updateTermEvh3);
     ASSERT_TRUE(termUpdated3);
 
     ASSERT_EQUALS(3, getReplCoord()->getTerm());
 
-    // Ensure all global exclusive lock tasks (eg. _stepDownFinish) run to completion.
-    auto work = [](const executor::TaskExecutor::CallbackArgs&) {};
-    replExec->wait(unittest::assertGet(replExec->scheduleWorkWithGlobalExclusiveLock(work)));
-    getReplCoord()->waitForStepDownFinish_forTest();
+    // Update term event handles will wait for potential stepdown.
     ASSERT_TRUE(getReplCoord()->getMemberState().secondary());
 }
 
@@ -3310,7 +3301,11 @@ TEST_F(ReplCoordTest, LivenessElectionTimeout) {
         }
     }
     getNet()->exitNetwork();
-    getReplCoord()->waitForStepDownFinish_forTest();
+
+    // Ensure all global exclusive lock tasks (eg. _stepDownFinish) run to completion.
+    auto exec = getReplExec();
+    auto work = [](const executor::TaskExecutor::CallbackArgs&) {};
+    exec->wait(unittest::assertGet(exec->scheduleWorkWithGlobalExclusiveLock(work)));
     ASSERT_EQUALS(MemberState::RS_SECONDARY, getReplCoord()->getMemberState().s);
 }