SERVER-23163 Fix signaling of worker thread in replication executor.

Author: visualzhou

src/mongo/db/repl/replication_executor.cpp

@@ -274,6 +274,7 @@ StatusWith<ReplicationExecutor::CallbackHandle> ReplicationExecutor::onEvent(
         queue = &event->_waiters;
     } else {
         queue = &_readyQueue;
+        _networkInterface->signalWorkAvailable();
     }
     return enqueueWork_inlock(queue, work);
 }
@@ -382,6 +383,7 @@ StatusWith<ReplicationExecutor::CallbackHandle> ReplicationExecutor::scheduleWor
         ++insertBefore;
     _sleepersQueue.splice(insertBefore, temp, temp.begin());
     ++_counterScheduledWorkAts;
+    _networkInterface->signalWorkAvailable();
     return cbHandle;
 }
 

src/mongo/db/repl/replication_executor_test.cpp

@@ -52,6 +52,7 @@ namespace repl {
 namespace {
 
 using executor::NetworkInterfaceMock;
+using unittest::assertGet;
 
 const int64_t prngSeed = 1;
 
@@ -171,6 +172,85 @@ TEST_F(ReplicationExecutorTest, CancelBeforeRunningFutureWork) {
     ASSERT_EQUALS(1, executor.getDiagnosticBSON().getFieldDotted("queues.ready").Int());
 }
 
+// Equivalent to EventChainAndWaitingTest::onGo
+TEST_F(ReplicationExecutorTest, ScheduleCallbackOnFutureEvent) {
+    launchExecutorThread();
+    getNet()->exitNetwork();
+
+    ReplicationExecutor& executor = getReplExecutor();
+    // We signal this "ping" event and the executor will signal "pong" event in return.
+    auto ping = assertGet(executor.makeEvent());
+    auto pong = assertGet(executor.makeEvent());
+    auto fn = [&executor, pong](const ReplicationExecutor::CallbackArgs& cbData) {
+        ASSERT_OK(cbData.status);
+        executor.signalEvent(pong);
+    };
+
+    // Wait for a future event.
+    executor.onEvent(ping, fn);
+    ASSERT_EQUALS(0, executor.getDiagnosticBSON().getFieldDotted("queues.ready").Int());
+    executor.signalEvent(ping);
+    executor.waitForEvent(pong);
+}
+
+// Equivalent to EventChainAndWaitingTest::onGoAfterTriggered
+TEST_F(ReplicationExecutorTest, ScheduleCallbackOnSignaledEvent) {
+    launchExecutorThread();
+    getNet()->exitNetwork();
+
+    ReplicationExecutor& executor = getReplExecutor();
+    // We signal this "ping" event and the executor will signal "pong" event in return.
+    auto ping = assertGet(executor.makeEvent());
+    auto pong = assertGet(executor.makeEvent());
+    auto fn = [&executor, pong](const ReplicationExecutor::CallbackArgs& cbData) {
+        ASSERT_OK(cbData.status);
+        executor.signalEvent(pong);
+    };
+
+    // Wait for a signaled event.
+    executor.signalEvent(ping);
+    executor.onEvent(ping, fn);
+    executor.waitForEvent(pong);
+}
+
+TEST_F(ReplicationExecutorTest, ScheduleCallbackAtNow) {
+    launchExecutorThread();
+    getNet()->exitNetwork();
+
+    ReplicationExecutor& executor = getReplExecutor();
+    auto finishEvent = assertGet(executor.makeEvent());
+    auto fn = [&executor, finishEvent](const ReplicationExecutor::CallbackArgs& cbData) {
+        ASSERT_OK(cbData.status);
+        executor.signalEvent(finishEvent);
+    };
+
+    auto cb = executor.scheduleWorkAt(getNet()->now(), fn);
+    executor.waitForEvent(finishEvent);
+}
+
+TEST_F(ReplicationExecutorTest, ScheduleCallbackAtAFutureTime) {
+    launchExecutorThread();
+    getNet()->exitNetwork();
+
+    ReplicationExecutor& executor = getReplExecutor();
+    auto finishEvent = assertGet(executor.makeEvent());
+    auto fn = [&executor, finishEvent](const ReplicationExecutor::CallbackArgs& cbData) {
+        ASSERT_OK(cbData.status);
+        executor.signalEvent(finishEvent);
+    };
+
+    auto now = getNet()->now();
+    now += Milliseconds(1000);
+    auto cb = executor.scheduleWorkAt(now, fn);
+
+    getNet()->enterNetwork();
+    getNet()->runUntil(now);
+    getNet()->exitNetwork();
+
+    executor.waitForEvent(finishEvent);
+}
+
+
 }  // namespace
 }  // namespace repl
 }  // namespace mongo