MB-23987: Ignore repeatedly scheduled tasks

A timeout was observed in a test when running against an ephemeral
bucket.

This arose during tear down from a deadlock between
ExecutorPool::_unregisterTaskable and ExecutorPool::_cancel.

Normally, this should not be possible, as _unregisterTaskable waits for
the task locator to be empty after cancelling each of the tasks,
indicating that all tasks have been taken up by a thead and cleaned up.

The problem developed from the ItemPager as used in ephemeral buckets.
KVBucket::enableItemPager cancels and schedules the same task object.
Cancelling a task simply marks it as dead; it must later be cleaned up
by a ExecutorThread.

ExecutorPool::_schedule was altered to reset dead tasks to running to
facilitate enableItemPager. Therefore, if a thread did not clean up the
task between the cancel and schedule, enableItemPager reduced down to
just a schedule. If the task was already scheduled, this would be a
duplicate schedule.

Duplicate scheduling does not adversely affect the taskLocator as it is
a map. However, the taskQueues /will/ have a second copy of the task
added.

Now, when unregisterTaskable finds the taskLocator to be empty, it is
still possible for a thread to take up the duplicate copy of the task.
The task is already marked dead, and so the thread immediately calls
_cancel to clean it up.

if _unregisterTaskable already holds the mutex, _cancel will block.
_unregisterTaskable will then join all the ExecutorThreads, deadlocking
when it reaches the thread blocked on _cancel.

Change-Id: Ia0dc7e8e8c3423e1098c71cf376653b23b4393e6
Reviewed-on: http://review.couchbase.org/77035
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Dave Rigby <daver@couchbase.com>

diff --git a/src/executorpool.cc b/src/executorpool.cc
index fd6eefa..5206a2e 100644 (file)
--- a/src/executorpool.cc
+++ b/src/executorpool.cc
@@ -432,14 +432,21 @@ TaskQueue* ExecutorPool::_getTaskQueue(const Taskable& t,
 
 size_t ExecutorPool::_schedule(ExTask task) {
     LockHolder lh(tMutex);
+    const size_t taskId = task->getId();
+
     TaskQueue* q = _getTaskQueue(task->getTaskable(),
                                  GlobalTask::getTaskType(task->getTypeId()));
     TaskQpair tqp(task, q);
-    taskLocator[task->getId()] = tqp;
 
-    q->schedule(task);
+    auto result = taskLocator.insert(std::make_pair(taskId, tqp));
+
+    if (result.second) {
+        // tqp was inserted; it was not already present. Prevents multiple
+        // copies of a task being present in the task queues.
+        q->schedule(task);
+    }
 
-    return task->getId();
+    return taskId;
 }
 
 size_t ExecutorPool::schedule(ExTask task) {

diff --git a/src/fakes/fake_executorpool.h b/src/fakes/fake_executorpool.h
index 87b3f08..ea157b5 100644 (file)
--- a/src/fakes/fake_executorpool.h
+++ b/src/fakes/fake_executorpool.h
@@ -111,6 +111,11 @@ public:
     size_t getNumReadyTasks(task_type_t qType) {
         return numReadyTasks[qType];
     }
+
+    std::map<size_t, TaskQpair> getTaskLocator() {
+        return taskLocator;
+    };
+
 private:
     void cancelAll_UNLOCKED() {
         for (auto& it : taskLocator) {

diff --git a/tests/module_tests/executorpool_test.cc b/tests/module_tests/executorpool_test.cc
index f92feba..7c4ef5e 100644 (file)
--- a/tests/module_tests/executorpool_test.cc
+++ b/tests/module_tests/executorpool_test.cc
@@ -259,3 +259,29 @@ TEST_F(ExecutorPoolDynamicWorkerTest, reschedule_dead_task) {
 
     EXPECT_EQ(2, runCount);
 }
+
+/* Testing to ensure that repeatedly scheduling a task does not result in
+ * multiple entries in the taskQueue - this could cause a deadlock in
+ * _unregisterTaskable when the taskLocator is empty but duplicate tasks remain
+ * in the queue.
+ */
+TEST_F(SingleThreadedExecutorPoolTest, ignore_duplicate_schedule) {
+    ExTask task = new LambdaTask(
+            taskable, TaskId::ItemPager, 10, true, [&] { return false; });
+
+    size_t taskId = task->getId();
+
+    ASSERT_EQ(taskId, pool->schedule(task));
+    ASSERT_EQ(taskId, pool->schedule(task));
+
+    std::map<size_t, TaskQpair> taskLocator =
+            dynamic_cast<SingleThreadedExecutorPool*>(ExecutorPool::get())
+                    ->getTaskLocator();
+
+    TaskQueue* queue = taskLocator.find(taskId)->second.second;
+
+    EXPECT_EQ(1, queue->getFutureQueueSize())
+            << "Task should only appear once in the taskQueue";
+
+    pool->cancel(taskId, true);
+}

diff --git a/tests/module_tests/executorpool_test.h b/tests/module_tests/executorpool_test.h
index e7cdc33..080193f 100644 (file)
--- a/tests/module_tests/executorpool_test.h
+++ b/tests/module_tests/executorpool_test.h
@@ -23,6 +23,7 @@
 
 #include <executorpool.h>
 #include <executorthread.h>
+#include <fakes/fake_executorpool.h>
 #include <gtest/gtest.h>
 #include <taskable.h>
 #include <thread>
@@ -103,6 +104,23 @@ public:
 
 class ExecutorPoolTest : public ::testing::Test {};
 
+class SingleThreadedExecutorPoolTest : public ::testing::Test {
+public:
+    void SetUp() override {
+        SingleThreadedExecutorPool::replaceExecutorPoolWithFake();
+        pool = ExecutorPool::get();
+        pool->registerTaskable(taskable);
+    }
+
+    void TearDown() override {
+        pool->unregisterTaskable(taskable, false);
+        pool->shutdown();
+    }
+
+    ExecutorPool* pool;
+    MockTaskable taskable;
+};
+
 class ExecutorPoolDynamicWorkerTest : public ExecutorPoolTest {
 protected:
     void SetUp() override {