feat(eventx,flow): 1.12.11, 添加了eventx::ThreadExecutor接口，实现了flow::ExecuteCmdAction动作

Author: hevake

modules/eventx/CMakeLists.txt

@@ -30,6 +30,7 @@ add_definitions(-DMODULE_ID="tbox.eventx")
 set(TBOX_LIBRARY_NAME tbox_eventx)
 
 set(TBOX_EVENTX_HEADERS
+	thread_executor.h
     thread_pool.h
     timer_pool.h
     timeout_monitor.hpp

modules/eventx/Makefile

@@ -25,6 +25,7 @@ LIB_VERSION_Y = 0
 LIB_VERSION_Z = 2
 
 HEAD_FILES = \
+	thread_executor.h \
 	thread_pool.h \
 	timer_pool.h \
 	timeout_monitor.hpp \

modules/eventx/thread_executor.h

@@ -0,0 +1,85 @@
+/*
+ *     .============.
+ *    //  M A K E  / \
+ *   //  C++ DEV  /   \
+ *  //  E A S Y  /  \/ \
+ * ++ ----------.  \/\  .
+ *  \\     \     \ /\  /
+ *   \\     \     \   /
+ *    \\     \     \ /
+ *     -============'
+ *
+ * Copyright (c) 2018 Hevake and contributors, all rights reserved.
+ *
+ * This file is part of cpp-tbox (https://github.com/cpp-main/cpp-tbox)
+ * Use of this source code is governed by MIT license that can be found
+ * in the LICENSE file in the root of the source tree. All contributing
+ * project authors may be found in the CONTRIBUTORS.md file in the root
+ * of the source tree.
+ */
+#ifndef TBOX_EVENTX_THREAD_EXECUTOR_H
+#define TBOX_EVENTX_THREAD_EXECUTOR_H
+
+#include <functional>
+#include <tbox/base/cabinet_token.h>
+
+namespace tbox {
+namespace eventx {
+
+/**
+ * 线程执行器接口
+ */
+class ThreadExecutor {
+  public:
+    using TaskToken = cabinet::Token;
+    using NonReturnFunc = std::function<void ()>;
+
+    enum class TaskStatus {
+        kWaiting,   //!< 等待中
+        kExecuting, //!< 执行中
+        kNotFound,  //!< 未找到（可能已完成）
+        kCleanup,   //!< 执行器与被清除
+    };
+
+    enum class CancelResult {
+        kSuccess,   //!< 成功
+        kExecuting, //!< 该任务正在执行
+        kNotFound,  //!< 没有找到该任务（可能已完成）
+        kCleanup,   //!< 执行器与被清除
+    };
+
+  public:
+    /**
+     * 使用worker线程执行某个函数
+     *
+     * \param backend_task      让worker线程执行的函数对象
+     *
+     * \return TaskToken        任务Token
+     */
+    virtual TaskToken execute(NonReturnFunc &&backend_task) = 0;
+    virtual TaskToken execute(const NonReturnFunc &backend_task) = 0;
+
+    /**
+     * 使用worker线程执行某个函数，并在完成之后在主线程执行指定的回调函数
+     *
+     * \param backend_task      让worker线程执行的函数对象
+     * \param main_cb           任务完成后，由主线程执行的回调函数对象
+     * \param prio              任务优先级[-2, -1, 0, 1, 2]，越小优先级越高
+     *
+     * \return TaskToken        任务Token
+     */
+    virtual TaskToken execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb) = 0;
+    virtual TaskToken execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb) = 0;
+
+
+    //! 获取任务的状态
+    virtual TaskStatus getTaskStatus(TaskToken task_token) const = 0;
+
+    //! 取消任务
+    virtual CancelResult cancel(TaskToken task_token) = 0;
+};
+
+}
+}
+
+#endif //TBOX_EVENTX_THREAD_EXECUTOR_H

modules/eventx/thread_pool.cpp

@@ -126,17 +126,7 @@ bool ThreadPool::initialize(ssize_t min_thread_num, ssize_t max_thread_num)
     return true;
 }
 
-ThreadPool::TaskToken ThreadPool::execute(NonReturnFunc &&backend_task, int prio)
-{
-    return execute(std::move(backend_task), nullptr, prio);
-}
-
-ThreadPool::TaskToken ThreadPool::execute(const NonReturnFunc &backend_task, int prio)
-{
-    return execute(backend_task, nullptr, prio);
-}
-
-ThreadPool::TaskToken ThreadPool::execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb, int prio)
+ThreadExecutor::TaskToken ThreadPool::execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb, int prio)
 {
     RECORD_SCOPE();
     TaskToken token;
@@ -180,14 +170,44 @@ ThreadPool::TaskToken ThreadPool::execute(NonReturnFunc &&backend_task, NonRetur
     return token;
 }
 
-ThreadPool::TaskToken ThreadPool::execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb, int prio)
+ThreadExecutor::TaskToken ThreadPool::execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb, int prio)
 {
     NonReturnFunc backend_task_copy(backend_task);
     NonReturnFunc main_cb_copy(main_cb);
     return execute(std::move(backend_task_copy), std::move(main_cb_copy), prio);
 }
 
-ThreadPool::TaskStatus ThreadPool::getTaskStatus(TaskToken task_token) const
+ThreadExecutor::TaskToken ThreadPool::execute(NonReturnFunc &&backend_task, int prio)
+{
+    return execute(std::move(backend_task), nullptr, prio);
+}
+
+ThreadExecutor::TaskToken ThreadPool::execute(const NonReturnFunc &backend_task, int prio)
+{
+    return execute(backend_task, nullptr, prio);
+}
+
+ThreadExecutor::TaskToken ThreadPool::execute(NonReturnFunc &&backend_task)
+{
+    return execute(std::move(backend_task), 0);
+}
+
+ThreadExecutor::TaskToken ThreadPool::execute(const NonReturnFunc &backend_task)
+{
+    return execute(backend_task, 0);
+}
+
+ThreadExecutor::TaskToken ThreadPool::execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb)
+{
+    return execute(std::move(backend_task), std::move(main_cb), 0);
+}
+
+ThreadExecutor::TaskToken ThreadPool::execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb)
+{
+    return execute(backend_task, main_cb, 0);
+}
+
+ThreadExecutor::TaskStatus ThreadPool::getTaskStatus(TaskToken task_token) const
 {
     std::lock_guard<std::mutex> lg(d_->lock);
 
@@ -200,20 +220,14 @@ ThreadPool::TaskStatus ThreadPool::getTaskStatus(TaskToken task_token) const
     return TaskStatus::kNotFound;
 }
 
-/**
- * 返回值如下：
- * 0: 取消成功
- * 1: 没有找到该任务
- * 2: 该任务正在执行
- */
-int ThreadPool::cancel(TaskToken token)
+ThreadExecutor::CancelResult ThreadPool::cancel(TaskToken token)
 {
     RECORD_SCOPE();
     std::lock_guard<std::mutex> lg(d_->lock);
 
     //! 如果正在执行
     if (d_->doing_tasks_token.find(token) != d_->doing_tasks_token.end())
-        return 2;   //! 返回正在执行
+        return CancelResult::kExecuting;   //! 返回正在执行
 
     //! 从高优先级向低优先级遍历，找出优先级最高的任务
     for (size_t i = 0; i < d_->undo_tasks_token.size(); ++i) {
@@ -223,12 +237,12 @@ int ThreadPool::cancel(TaskToken token)
             if (iter != tasks_token.end()) {
                 tasks_token.erase(iter);
                 d_->task_pool.free(d_->undo_tasks_cabinet.free(token));
-                return 0;
+                return CancelResult::kSuccess;
             }
         }
     }
 
-    return 1;   //! 返回没有找到
+    return CancelResult::kNotFound;   //! 返回没有找到
 }
 
 void ThreadPool::cleanup()

modules/eventx/thread_pool.h

@@ -23,8 +23,10 @@
 #include <limits>
 #include <functional>
 #include <array>
+
 #include <tbox/event/forward.h>
-#include <tbox/base/cabinet_token.h>
+
+#include "thread_executor.h"
 
 namespace tbox {
 namespace eventx {
@@ -42,10 +44,8 @@ namespace eventx {
 /**
  * 线程池类
  */
-class ThreadPool {
+class ThreadPool : public ThreadExecutor {
   public:
-    using TaskToken = cabinet::Token;
-
     /**
      * 构造函数
      *
@@ -67,47 +67,45 @@ class ThreadPool {
     using NonReturnFunc = std::function<void ()>;
 
     /**
-     * 使用worker线程执行某个函数
+     * 使用worker线程执行某个函数，并在完成之后在主线程执行指定的回调函数
      *
      * \param backend_task      让worker线程执行的函数对象
+     * \param main_cb           任务完成后，由主线程执行的回调函数对象
      * \param prio              任务优先级[-2, -1, 0, 1, 2]，越小优先级越高
      *
      * \return TaskToken        任务Token
      */
-    TaskToken execute(NonReturnFunc &&backend_task, int prio = 0);
-    TaskToken execute(const NonReturnFunc &backend_task, int prio = 0);
+    TaskToken execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb, int prio);
+    TaskToken execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb, int prio);
 
     /**
-     * 使用worker线程执行某个函数，并在完成之后在主线程执行指定的回调函数
+     * 使用worker线程执行某个函数
      *
      * \param backend_task      让worker线程执行的函数对象
-     * \param main_cb           任务完成后，由主线程执行的回调函数对象
      * \param prio              任务优先级[-2, -1, 0, 1, 2]，越小优先级越高
      *
      * \return TaskToken        任务Token
      */
-    TaskToken execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb, int prio = 0);
-    TaskToken execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb, int prio = 0);
-
-    enum class TaskStatus {
-        kWaiting,   //! 等待中
-        kExecuting, //! 执行中
-        kNotFound   //! 未找到（可能已完成）
-    };
-
-    //! 获取任务的状态
-    TaskStatus getTaskStatus(TaskToken task_token) const;
+    TaskToken execute(NonReturnFunc &&backend_task, int prio);
+    TaskToken execute(const NonReturnFunc &backend_task, int prio);
 
     /**
-     * 取消任务
+     * 使用worker线程执行某个函数
      *
-     * \return task_token       任务Token
+     * \param backend_task      让worker线程执行的函数对象
+     * \param main_cb           任务完成后，由主线程执行的回调函数对象
      *
-     * \return  int     0: 成功
-     *                  1: 没有找到该任务（已执行）
-     *                  2: 该任务正在执行
+     * \return TaskToken        任务Token
      */
-    int cancel(TaskToken task_token);
+    virtual TaskToken execute(NonReturnFunc &&backend_task) override;
+    virtual TaskToken execute(const NonReturnFunc &backend_task) override;
+    virtual TaskToken execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb) override;
+    virtual TaskToken execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb) override;
+
+    //! 获取任务的状态
+    virtual TaskStatus getTaskStatus(TaskToken task_token) const override;
+    //! 取消任务
+    virtual CancelResult cancel(TaskToken task_token) override;
 
     /**
      * 清理资源，并等待所有的worker线程结束

modules/eventx/thread_pool_test.cpp

@@ -133,11 +133,11 @@ TEST(ThreadPool, cancel_task) {
 
     std::this_thread::sleep_for(std::chrono::milliseconds(1500));
 
-    EXPECT_EQ(tp->cancel(task_ids[0]), 1);  //! 第一个任务已完成
-    EXPECT_EQ(tp->cancel(task_ids[1]), 2);  //! 第二个任务正在执行
-    EXPECT_EQ(tp->cancel(task_ids[2]), 0);  //! 第三个任务可正常取消
+    EXPECT_EQ(tp->cancel(task_ids[0]), ThreadPool::CancelResult::kNotFound);    //! 第一个任务已完成
+    EXPECT_EQ(tp->cancel(task_ids[1]), ThreadPool::CancelResult::kExecuting);   //! 第二个任务正在执行
+    EXPECT_EQ(tp->cancel(task_ids[2]), ThreadPool::CancelResult::kSuccess);     //! 第三个任务可正常取消
     ThreadPool::TaskToken invalid_token(100, 1);
-    EXPECT_EQ(tp->cancel(invalid_token), 1);  //! 任务不存在
+    EXPECT_EQ(tp->cancel(invalid_token), ThreadPool::CancelResult::kNotFound);  //! 任务不存在
 
     loop->exitLoop(std::chrono::seconds(4));
     loop->runLoop();