MB-40267: ActiveStreamCheckpointProcessorTask should be a NonIO task

The ActiveStreamCheckpointProcessorTask is incorrectly assigned to the
AuxIO thread pool (which is intended for IO-bound tasks not directly
related to reading / flushing data).

This bug traces back to when ActiveStreamCheckpointProcessorTask was
created on the 3.x branch - from the review comments of the original
patch
(http://review.couchbase.org/c/ep-engine/+/57148/5/src/dcp-producer.cc#152):

    Jim Walker
    2015-11-24
    The initial reason is to try and distribute the load. NONIO is
    already running the DcpConsumer Processor(s), itempager,
    expirypager, checkpoint remover, hashtable resizer and many more
    (http://src.couchbase.org/source/search?q=&defs=&refs=NONIO_TASK_IDX&path=&hist=&type=&project=3.1.1)
    whereas AUXIO is running DcpBackfill and access scanner (other
    tasks appear to be TAP related). As stated in the comments, for
    future (master branch) this should be revisited, maybe creating a
    DCP task type or increasing NONIO threads?

There is arguably two problems with it running AuxIO:

1. AuxIO tasks can often have long runtimes (latencies) as they are
   performing synchronous disk IO - and hence could impact the
   scheduling latency of the ActiveStreamCheckpointProcessorTask which
   should ideally have low latency (given it directly affects overall
   replication latency).

2. The AuxIO thread pool is small (compared to the NonIO thread pool)
   - thread counts for a few different machine sizes:

    #CPUs  #AuxIO thread  #NonIO threads
    8	   1	          2
    16     2	          4
    32	   4	          8

As such, DCP streaming (for example for in-memory phase of rebalance)
can be limited in how much CPU it can use and be prematurely
CPU-bound.

Change-Id: I9e4549428dc3bd08dd33abe05da8dac521de9ef2
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/140915
Reviewed-by: Ben Huddleston <[email protected]>
Tested-by: Dave Rigby <[email protected]>

Author: daverigby

engines/ep/src/tasks.def.h

@@ -50,7 +50,6 @@ TASK(VBucketMemoryAndDiskDeletionTask, AUXIO_TASK_IDX, 0)
 TASK(PersistCollectionsManifest, AUXIO_TASK_IDX, 1)
 TASK(AccessScanner, AUXIO_TASK_IDX, 2)
 TASK(AccessScannerVisitor, AUXIO_TASK_IDX, 2)
-TASK(ActiveStreamCheckpointProcessorTask, AUXIO_TASK_IDX, 3)
 TASK(BackfillManagerTask, AUXIO_TASK_IDX, 4)
 
 // Read/Write IO tasks
@@ -71,6 +70,7 @@ TASK(DcpConsumerTask, NONIO_TASK_IDX, 2)
 TASK(DurabilityCompletionTask, NONIO_TASK_IDX, 1)
 TASK(DurabilityTimeoutTask, NONIO_TASK_IDX, 1)
 TASK(DurabilityTimeoutVisitor, NONIO_TASK_IDX, 1)
+TASK(ActiveStreamCheckpointProcessorTask, NONIO_TASK_IDX, 3)
 TASK(ConnNotifierCallback, NONIO_TASK_IDX, 5)
 TASK(ClosedUnrefCheckpointRemoverTask, NONIO_TASK_IDX, 6)
 TASK(ClosedUnrefCheckpointRemoverVisitorTask, NONIO_TASK_IDX, 6)

engines/ep/src/vbucket_notify_context.h

@@ -44,9 +44,9 @@ struct VBNotifyCtx {
      *
      * 4) When there are no items in an ActiveStream's ready queue the front end
      *    worker stepping will schedule the
-     *    ActiveStreamCheckpointProcessorTask. This will run on an AuxIO thread
+     *    ActiveStreamCheckpointProcessorTask. This will run on an NonIO thread
      *    and enqueue nothing into the ActiveStream's ready queue if the only
-     *    item is a prepare. This will slow down other SyncWrites if AuxIO
+     *    item is a prepare. This will slow down other SyncWrites if NonIO
      *    threads are a bottleneck.
      *
      * 5) The ActiveStreamCheckpointProcessorTask would then notify the front

engines/ep/tests/module_tests/dcp_reflection_test.cc

@@ -53,6 +53,12 @@ static TaskQueue* getLpAuxQ() {
     return task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
 }
 
+static TaskQueue* getLpNonIoQ() {
+    auto* task_executor =
+            reinterpret_cast<SingleThreadedExecutorPool*>(ExecutorPool::get());
+    return task_executor->getLpTaskQ()[NONIO_TASK_IDX];
+}
+
 /**
  * Test fixture which creates two ep-engine (bucket) instances, using one
  * as a source for DCP replication and the second as the destination.
@@ -385,17 +391,30 @@ std::unique_ptr<DcpResponse>
 DCPLoopbackStreamTest::DcpRoute::getNextProducerMsg(ActiveStream* stream) {
     std::unique_ptr<DcpResponse> producerMsg(stream->next());
     if (!producerMsg) {
-        auto queueSize = getLpAuxQ()->getReadyQueueSize() +
-                         getLpAuxQ()->getFutureQueueSize();
-        EXPECT_GE(queueSize, 1)
-                << "Expected to have at least "
-                   "ActiveStreamCheckpointProcessorTask "
-                   "in ready/future queue after null producerMsg";
-
-        // Run the next waiting task to populate the streams' items.
-        CheckedExecutor executor(ExecutorPool::get(), *getLpAuxQ());
-        executor.runCurrentTask();
-        executor.completeCurrentTask();
+        // Run the next ready task to populate the streams' items. This could
+        // either be a NonIO task (ActiveStreamCheckpointProcessorTask) or
+        // AuxIO task (
+
+        // Note that the actual count of ready tasks isn't just the reaadyQueue
+        // - tasks in the futureQ whose waketime is less than or equal to now
+        // can also be run.
+        const auto auxIoQueueSize = getLpAuxQ()->getReadyQueueSize() +
+                                    getLpAuxQ()->getFutureQueueSize();
+        const auto nonIoQueueSize = getLpNonIoQ()->getReadyQueueSize() +
+                                    getLpNonIoQ()->getFutureQueueSize();
+        if (auxIoQueueSize > 0) {
+            CheckedExecutor executor(ExecutorPool::get(), *getLpAuxQ());
+            executor.runCurrentTask();
+            executor.completeCurrentTask();
+        } else if (nonIoQueueSize > 0) {
+            CheckedExecutor executor(ExecutorPool::get(), *getLpNonIoQ());
+            executor.runCurrentTask();
+            executor.completeCurrentTask();
+        } else {
+            ADD_FAILURE() << "Expected to have at least one task in AuxIO / "
+                             "NonIO ready/future queues after null "
+                             "producerMsg, but both are zero";
+        }
         if (!stream->getItemsRemaining()) {
             return {};
         }
@@ -559,7 +578,7 @@ DCPLoopbackStreamTest::DcpRoute::doStreamRequest(int flags) {
                   2)
                 << "Should have both persistence and DCP producer cursor on "
                    "producer VB";
-        EXPECT_GE(getLpAuxQ()->getFutureQueueSize(), 1);
+        EXPECT_GE(getLpNonIoQ()->getFutureQueueSize(), 1);
         // Finally the stream-request response sends the failover table back
         // to the consumer... simulate that
         auto failoverLog = producerVb->failovers->getFailoverLog();

engines/ep/tests/module_tests/ephemeral_bucket_test.cc

@@ -207,12 +207,12 @@ TEST_F(SingleThreadedEphemeralTest, RangeIteratorVBDeleteRaceTest) {
     EXPECT_FALSE(
             task_executor->isTaskScheduled(NONIO_TASK_IDX, vbDeleteTaskName));
 
-    auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
     // Now bin the producer
     producer->cancelCheckpointCreatorTask();
     /* Checkpoint processor task finishes up and releases its producer
        reference */
-    runNextTask(lpAuxioQ, "Process checkpoint(s) for DCP producer " + testName);
+    auto& lpNonIoQ = *task_executor->getLpTaskQ()[NONIO_TASK_IDX];
+    runNextTask(lpNonIoQ, "Process checkpoint(s) for DCP producer " + testName);
 
     engine->getDcpConnMap().shutdownAllConnections();
     mock_stream.reset();
@@ -221,6 +221,7 @@ TEST_F(SingleThreadedEphemeralTest, RangeIteratorVBDeleteRaceTest) {
     // run the backfill task so the backfill can reach state
     // backfill_finished and be destroyed destroying the range iterator
     // in the process
+    auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
     runNextTask(lpAuxioQ, "Backfilling items for a DCP Connection");
 
     // Now the backfill is gone, the evb can be deleted
@@ -429,13 +430,13 @@ TEST_F(SingleThreadedEphemeralPurgerTest, PurgeAcrossAllVbuckets) {
     bucket->enableTombstonePurgerTask();
     bucket->attemptToFreeMemory(); // this wakes up the HTCleaner task
 
-    auto& lpAuxioQ = *task_executor->getLpTaskQ()[NONIO_TASK_IDX];
+    auto& lpNonIoQ = *task_executor->getLpTaskQ()[NONIO_TASK_IDX];
     /* Run the HTCleaner and EphTombstoneStaleItemDeleter tasks. We expect
        pause and resume of EphTombstoneStaleItemDeleter atleast once and we run
        until all the deleted items across all the vbuckets are purged */
     int numPaused = 0;
     while (!checkAllPurged(expPurgeUpto)) {
-        runNextTask(lpAuxioQ);
+        runNextTask(lpNonIoQ);
         ++numPaused;
     }
     EXPECT_GT(numPaused, 2 /* 1 run of 'HTCleaner' and more than 1 run of
@@ -519,4 +520,4 @@ TEST_F(SingleThreadedEphemeralPurgerTest, HTCleanerSkipsPrepares) {
         ASSERT_FALSE(res.pending);
         ASSERT_FALSE(res.committed);
     }
-}
+}

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -980,10 +980,10 @@ TEST_F(MB29369_SingleThreadedEPBucketTest,
     auto producer = createDcpProducer(cookie, IncludeDeleteTime::Yes);
     producer->scheduleCheckpointProcessorTask();
 
-    auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
-    EXPECT_EQ(1, lpAuxioQ.getFutureQueueSize()) << "Expected to have "
-                                                   "ActiveStreamCheckpointProce"
-                                                   "ssorTask in AuxIO Queue";
+    auto& lpNonIoQ = *task_executor->getLpTaskQ()[NONIO_TASK_IDX];
+    EXPECT_EQ(1, lpNonIoQ.getFutureQueueSize())
+            << "Expected to have ActiveStreamCheckpointProcessorTask in NonIO "
+               "Queue";
 
     // Create dcp_producer_snapshot_marker_yield_limit + 1 streams -
     // this means that we don't process all pending vBuckets on a single
@@ -1075,11 +1075,13 @@ TEST_F(MB29369_SingleThreadedEPBucketTest,
     // If the Backfilling task then runs, which returns a disk snapshot and
     // re-registers the cursor; we still have an
     // ActiveStreamCheckpointProcessorTask outstanding with the vb in the queue.
-    EXPECT_EQ(2, lpAuxioQ.getFutureQueueSize());
+    EXPECT_EQ(1, lpNonIoQ.getFutureQueueSize());
+    auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
+    EXPECT_EQ(1, lpAuxioQ.getFutureQueueSize());
 
     // Run the ActiveStreamCheckpointProcessorTask; which should re-schedule
     // due to having items outstanding.
-    runNextTask(lpAuxioQ,
+    runNextTask(lpNonIoQ,
                 "Process checkpoint(s) for DCP producer test_producer");
 
     // Now run backfilling task.
@@ -1101,7 +1103,7 @@ TEST_F(MB29369_SingleThreadedEPBucketTest,
 
     // Now run chkptProcessorTask to complete it's queue. With the bug, this
     // results in us discarding the last item we just added to vBucket.
-    runNextTask(lpAuxioQ,
+    runNextTask(lpNonIoQ,
                 "Process checkpoint(s) for DCP producer test_producer");
 
     // Let the backfill task complete running (it requires multiple steps to
@@ -1127,7 +1129,7 @@ TEST_F(MB29369_SingleThreadedEPBucketTest,
 
     // Now run chkptProcessorTask to complete it's queue, this will now be able
     // to access the checkpoint and get key2
-    runNextTask(lpAuxioQ,
+    runNextTask(lpNonIoQ,
                 "Process checkpoint(s) for DCP producer test_producer");
 
     result = stream->next();
@@ -1160,9 +1162,10 @@ TEST_P(STParamPersistentBucketTest, MB29585_backfilling_whilst_snapshot_runs) {
     setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
 
     auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
-    EXPECT_EQ(1, lpAuxioQ.getFutureQueueSize()) << "Expected to have "
-                                                   "ActiveStreamCheckpointProce"
-                                                   "ssorTask in AuxIO Queue";
+    auto& lpNonIoQ = *task_executor->getLpTaskQ()[NONIO_TASK_IDX];
+    EXPECT_EQ(1, lpNonIoQ.getFutureQueueSize())
+            << "Expected to have ActiveStreamCheckpointProcessorTask in NonIO "
+               "Queue";
 
     // Create first stream
     auto vb = store->getVBucket(vbid);
@@ -1224,7 +1227,7 @@ TEST_P(STParamPersistentBucketTest, MB29585_backfilling_whilst_snapshot_runs) {
 
     // Next we must deque, but not run the snapshot task, we will interleave it
     // with backfill later
-    CheckedExecutor checkpointTask(task_executor, lpAuxioQ);
+    CheckedExecutor checkpointTask(task_executor, lpNonIoQ);
     EXPECT_STREQ("Process checkpoint(s) for DCP producer test_producer",
                  checkpointTask.getTaskName().data());
 
@@ -1276,7 +1279,7 @@ TEST_P(STParamPersistentBucketTest, MB29585_backfilling_whilst_snapshot_runs) {
         FAIL() << "Expected Event::Mutation named 'key2'";
     }
 
-    runNextTask(lpAuxioQ,
+    runNextTask(lpNonIoQ,
                 "Process checkpoint(s) for DCP producer test_producer");
 
     result = stream->next();
@@ -1539,7 +1542,9 @@ TEST_P(STParamPersistentBucketTest, MB22960_cursor_dropping_data_loss) {
     ckpt_mgr.removeCursor(mock_stream->getCursor().lock().get());
 
     auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
-    EXPECT_EQ(2, lpAuxioQ.getFutureQueueSize());
+    EXPECT_EQ(1, lpAuxioQ.getFutureQueueSize());
+    auto& lpNonIoQ = *task_executor->getLpTaskQ()[NONIO_TASK_IDX];
+    EXPECT_EQ(1, lpNonIoQ.getFutureQueueSize());
     // backfill:create()
     runNextTask(lpAuxioQ);
     // backfill:scan()
@@ -1743,7 +1748,9 @@ TEST_P(STParamPersistentBucketTest,
     mock_stream->next();
 
     auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
-    EXPECT_EQ(2, lpAuxioQ.getFutureQueueSize());
+    EXPECT_EQ(1, lpAuxioQ.getFutureQueueSize());
+    auto& lpNonIoQ = *task_executor->getLpTaskQ()[NONIO_TASK_IDX];
+    EXPECT_EQ(1, lpNonIoQ.getFutureQueueSize());
     // backfill:create()
     runNextTask(lpAuxioQ, "Backfilling items for a DCP Connection");
     // backfill:scan()
@@ -1783,7 +1790,7 @@ TEST_P(STParamPersistentBucketTest,
     ASSERT_EQ(1, registerCursorCount);
 
     // ActiveStreamCheckpointProcessorTask
-    runNextTask(lpAuxioQ, "Process checkpoint(s) for DCP producer " + testName);
+    runNextTask(lpNonIoQ, "Process checkpoint(s) for DCP producer " + testName);
     // BackfillManagerTask
     runNextTask(lpAuxioQ, "Backfilling items for a DCP Connection");
 
@@ -1917,7 +1924,7 @@ TEST_P(STParamPersistentBucketTest, MB19428_no_streams_against_dead_vbucket) {
               getEPBucket().flushVBucket(vbid));
 
     setVBucketStateAndRunPersistTask(vbid, vbucket_state_dead);
-    auto& lpAuxioQ = *task_executor->getLpTaskQ()[AUXIO_TASK_IDX];
+    auto& lpNonIoQ = *task_executor->getLpTaskQ()[NONIO_TASK_IDX];
 
     {
         // Create a Mock Dcp producer
@@ -1928,7 +1935,7 @@ TEST_P(STParamPersistentBucketTest, MB19428_no_streams_against_dead_vbucket) {
 
         // Creating a producer will not create an
         // ActiveStreamCheckpointProcessorTask until a stream is created.
-        EXPECT_EQ(0, lpAuxioQ.getFutureQueueSize());
+        EXPECT_EQ(0, lpNonIoQ.getFutureQueueSize());
 
         uint64_t rollbackSeqno;
         auto err = producer->streamRequest(
@@ -1948,7 +1955,7 @@ TEST_P(STParamPersistentBucketTest, MB19428_no_streams_against_dead_vbucket) {
 
         // The streamRequest failed and should not of created anymore tasks than
         // ActiveStreamCheckpointProcessorTask.
-        EXPECT_EQ(1, lpAuxioQ.getFutureQueueSize());
+        EXPECT_EQ(1, lpNonIoQ.getFutureQueueSize());
 
         // Stop Producer checkpoint processor task
         producer->cancelCheckpointCreatorTask();
@@ -1999,8 +2006,9 @@ TEST_F(SingleThreadedEPBucketTest, ReadyQueueMaintainsWakeTimeOrder) {
 TEST_F(SingleThreadedEPBucketTest, MB20235_wake_and_work_count) {
     class TestTask : public GlobalTask {
     public:
-        TestTask(EventuallyPersistentEngine *e, double s) :
-                 GlobalTask(e, TaskId::ActiveStreamCheckpointProcessorTask, s) {}
+        TestTask(EventuallyPersistentEngine* e, double s)
+            : GlobalTask(e, TaskId::AccessScanner, s) {
+        }
         bool run() override {
             return false;
         }
@@ -2504,6 +2512,12 @@ TEST_P(MB20054_SingleThreadedEPStoreTest,
     EXPECT_EQ(0, lpAuxioQ->getFutureQueueSize());
     EXPECT_EQ(0, lpAuxioQ->getReadyQueueSize());
 
+    auto lpNonIoQ = task_executor->getLpTaskQ()[NONIO_TASK_IDX];
+    // Initially have ConnNotifierCallback and ConnManagerTasks on NonIO queue.
+    const size_t numInitialNonIoTasks = 2;
+    EXPECT_EQ(numInitialNonIoTasks, lpNonIoQ->getFutureQueueSize());
+    EXPECT_EQ(0, lpNonIoQ->getReadyQueueSize());
+
     // Directly flush the vbucket, ensuring data is on disk.
     //  (This would normally also wake up the checkpoint remover task, but
     //   as that task was never registered with the ExecutorPool in this test
@@ -2550,9 +2564,12 @@ TEST_P(MB20054_SingleThreadedEPStoreTest,
                                  dummy_dcp_add_failover_cb,
                                  {}));
 
-    // FutureQ should now have an additional DCPBackfill task.
-    EXPECT_EQ(2, lpAuxioQ->getFutureQueueSize());
+    // FutureQ should now have an additional DCPBackfill task /
+    // ActiveStreamCheckpointProcessorTask.
+    EXPECT_EQ(1, lpAuxioQ->getFutureQueueSize());
     EXPECT_EQ(0, lpAuxioQ->getReadyQueueSize());
+    EXPECT_EQ(numInitialNonIoTasks + 1, lpNonIoQ->getFutureQueueSize());
+    EXPECT_EQ(0, lpNonIoQ->getReadyQueueSize());
 
     // Create an executor 'thread' to obtain shared ownership of the next
     // AuxIO task (which should be BackfillManagerTask). As long as this