*: Fix bug that DynamicThreadPool may not destruct tasks in time.

dbms/src/Common/DynamicThreadPool.cpp

@@ -81,10 +81,11 @@ void DynamicThreadPool::scheduledToNewDynamicThread(TaskPtr & task)
     t.detach();
 }
 
-void executeTask(const std::unique_ptr<IExecutableTask> & task)
+void DynamicThreadPool::executeTask(TaskPtr & task)
 {
     UPDATE_CUR_AND_MAX_METRIC(tiflash_thread_count, type_active_threads_of_thdpool, type_max_active_threads_of_thdpool);
     task->execute();
+    task.reset();
 }
 
 void DynamicThreadPool::fixedWork(size_t index)
@@ -124,7 +125,6 @@ void DynamicThreadPool::dynamicWork(TaskPtr initial_task)
         if (!node.task) // may be timeout or cancelled
             break;
         executeTask(node.task);
-        node.task.reset();
     }
     alive_dynamic_threads.fetch_sub(1);
 }

dbms/src/Common/DynamicThreadPool.h

@@ -64,6 +64,8 @@ class DynamicThreadPool
     void fixedWork(size_t index);
     void dynamicWork(TaskPtr initial_task);
 
+    static void executeTask(TaskPtr & task);
+
     const std::chrono::nanoseconds dynamic_auto_shrink_cooldown;
 
     std::vector<std::thread> fixed_threads;

dbms/src/Common/tests/gtest_dynamic_thread_pool.cpp

@@ -3,6 +3,8 @@
 
 namespace DB::tests
 {
+namespace
+{
 class DynamicThreadPoolTest : public ::testing::Test
 {
 };
@@ -148,4 +150,45 @@ try
 }
 CATCH
 
+struct X
+{
+    std::mutex * mu;
+    std::condition_variable * cv;
+    bool * destructed;
+
+    X(std::mutex * mu_, std::condition_variable * cv_, bool * destructed_)
+        : mu(mu_)
+        , cv(cv_)
+        , destructed(destructed_)
+    {}
+
+    ~X()
+    {
+        std::unique_lock lock(*mu);
+        *destructed = true;
+        cv->notify_all();
+    }
+};
+
+TEST_F(DynamicThreadPoolTest, testTaskDestruct)
+try
+{
+    std::mutex mu;
+    std::condition_variable cv;
+    bool destructed = false;
+
+    DynamicThreadPool pool(0, std::chrono::minutes(1));
+    auto tmp = std::make_shared<X>(&mu, &cv, &destructed);
+    pool.schedule(true, [x = tmp] {});
+    tmp.reset();
+
+    {
+        std::unique_lock lock(mu);
+        auto ret = cv.wait_for(lock, std::chrono::seconds(1), [&] { return destructed; });
+        ASSERT_TRUE(ret);
+    }
+}
+CATCH
+
+} // namespace
 } // namespace DB::tests