[pull] main from ChunelFeng:main

src/GraphCtrl/GraphElement/GElementRepository.cpp

@@ -7,7 +7,6 @@
 ***************************/
 
 #include "GElementRepository.h"
-#include "GNode/GNodeInclude.h"
 #include "GGroup/GGroupInclude.h"
 
 CGRAPH_NAMESPACE_BEGIN
@@ -61,7 +60,7 @@ CStatus GElementRepository::reset() {
 
 CStatus GElementRepository::pushAllState(const GElementState& state) {
     CGRAPH_FUNCTION_BEGIN
-    if (cur_state_ == state) {
+    if (cur_state_.load(std::memory_order_acquire) == state) {
         return status;    // 避免重复赋值
     }
 
@@ -72,7 +71,7 @@ CStatus GElementRepository::pushAllState(const GElementState& state) {
             cur->suspend_locker_.cv_.notify_one();
         }
     }
-    cur_state_ = state;    // 记录当前的状态信息
+    cur_state_.store(state, std::memory_order_release);    // 记录当前的状态信息
     CGRAPH_FUNCTION_END
 }
 
@@ -93,7 +92,7 @@ CVoid GElementRepository::fetchAll(GElementManagerCPtr em) {
 CVoid GElementRepository::fetch(GElementPtr element) {
     elements_.insert(element);
     if (element->isGGroup()) {
-        auto group = dynamic_cast<GGroupPtr>(element);
+        const auto* group = dynamic_cast<GGroupPtr>(element);
         CGRAPH_ASSERT_NOT_NULL_THROW_ERROR(group)
         for (auto* cur : group->children_) {
             fetch(cur);
@@ -105,10 +104,9 @@ CVoid GElementRepository::fetch(GElementPtr element) {
 CBool GElementRepository::isCancelState() const {
     /**
      * 因为每次执行的时候，都需要判断一下这个状态是否为 cancel
-     * 且理论上不会出现多线程问题
-     * 故这一层的 cur_state_ 就不设置为atomic类型的了
+     * 外部线程可能通过 pipeline->cancel() 修改状态，故这里通过 atomic 读取
      */
-    return GElementState::CANCEL == cur_state_;
+    return GElementState::CANCEL == cur_state_.load(std::memory_order_acquire);
 }
 
 
@@ -162,4 +160,4 @@ GElementRepository::~GElementRepository() {
     }
 }
 
-CGRAPH_NAMESPACE_END
+CGRAPH_NAMESPACE_END

src/GraphCtrl/GraphElement/GElementRepository.h

@@ -9,6 +9,8 @@
 #ifndef CGRAPH_GELEMENTREPOSITORY_H
 #define CGRAPH_GELEMENTREPOSITORY_H
 
+#include <atomic>
+
 #include "GElementObject.h"
 #include "GElement.h"
 #include "GElementManager.h"
@@ -85,12 +87,14 @@ class GElementRepository : public GElementObject {
 
     CStatus run() final;
 
+    GElementRepository() = default;
+
     ~GElementRepository() override;
 
 private:
-    GElementPtrSet elements_ {};                                    // 用于记录所有的element信息
-    GElementState cur_state_ { GElementState::NORMAL };             // 当前状态信息
-    GElementPtrSet async_elements_ {};                              // 所有异步执行的逻辑，到后来一次性统一获取执行结果信息
+    GElementPtrSet elements_ {};                                            // 用于记录所有的element信息
+    std::atomic<GPipelineState> cur_state_ { GElementState::NORMAL };    // 当前pipeline状态信息
+    GElementPtrSet async_elements_ {};                                      // 所有异步执行的逻辑，到后来一次性统一获取执行结果信息
 
     friend class GPipeline;
     friend class GPerf;

src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp

@@ -204,25 +204,20 @@ CVoid GDynamicEngine::afterElementRun(GElementPtr element) {
 
 
 CVoid GDynamicEngine::fatWait() {
-    const auto epoch = thread_pool_->getConfig().pipeline_wait_busy_epoch_;
-    for (CInt i = 0; i < epoch; i++) {
-        // 对于 pipeline 运行耗时很短的情况，在 进入 cv.wait() 之前，轮询等到一会
-        // 如果遇到 isErr() 的情况，还是走进入 mutex 的逻辑
-        if (finished_end_size_.load(std::memory_order_acquire) >= total_end_size_) {
-            return;
-        }
-        CGRAPH_YIELD();
+    if (!isFastFinished([this] {
+        // 如果快速结束了，则不进入 lock 中
+        return finished_end_size_.load(std::memory_order_relaxed) >= total_end_size_;
+    })) {
+        CGRAPH_UNIQUE_LOCK lock(locker_.mtx_);
+        locker_.cv_.wait(lock, [this] {
+            /**
+             * 遇到以下条件之一，结束执行：
+             * 1，执行结束
+             * 2，状态异常
+             */
+            return (finished_end_size_.load(std::memory_order_relaxed) >= total_end_size_) || cur_status_.isErr();
+        });
     }
-
-    CGRAPH_UNIQUE_LOCK lock(locker_.mtx_);
-    locker_.cv_.wait(lock, [this] {
-        /**
-         * 遇到以下条件之一，结束执行：
-         * 1，执行结束
-         * 2，状态异常
-         */
-        return (finished_end_size_.load(std::memory_order_acquire) >= total_end_size_) || cur_status_.isErr();
-    });
 }
 
 
@@ -250,18 +245,12 @@ CVoid GDynamicEngine::parallelRunAll() {
         (void)thread_pool_->wakeupAllThread();
     }
 
-    const auto epoch = thread_pool_->getConfig().pipeline_wait_busy_epoch_;
-    for (CInt i = 0; i < epoch; i++) {
-        if (parallel_run_num_ >= total_end_size_) {
-            return;
-        }
-        CGRAPH_YIELD();
-    }
-
-    {
+    if (!isFastFinished([this] {
+       return parallel_run_num_ >= total_end_size_;
+    })) {
         CGRAPH_UNIQUE_LOCK lock(locker_.mtx_);
         locker_.cv_.wait(lock, [this] {
-            return (parallel_run_num_ >= total_end_size_ || cur_status_.isErr());
+            return parallel_run_num_ >= total_end_size_ || cur_status_.isErr();
         });
     }
 }
@@ -314,4 +303,20 @@ CVoid GDynamicEngine::prepareRun() {
     }
 }
 
+
+template<typename Pred>
+CBool GDynamicEngine::isFastFinished(Pred&& check) {
+    CBool result = false;
+    const auto epoch = thread_pool_->getConfig().pipeline_wait_busy_epoch_;
+    for (auto i = 0; i < epoch; i++) {
+        if (check()) {
+            result = true;
+            break;
+        }
+        CGRAPH_YIELD();
+    }
+
+    return result;
+}
+
 CGRAPH_NAMESPACE_END

src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h

@@ -103,6 +103,14 @@ class GDynamicEngine : public GEngine {
      */
     CVoid prepareRun();
 
+    /**
+     * 快速校验结束
+     * @return
+     * @notice 为了方式线程快速进入内核态，短时间在用户层停留，性能加速
+     */
+    template<typename Pred>
+    CBool isFastFinished(Pred&& check);
+
 private:
     GElementPtrArr total_element_arr_ {};                                                // pipeline中所有的元素信息集合
     GElementPtrArr front_element_arr_ {};                                                // 没有依赖的元素信息

src/GraphCtrl/GraphPipeline/GPipeline.cpp

@@ -294,7 +294,7 @@ CStatus GPipeline::makeSerial() {
 
 
 GPipelineState GPipeline::getCurState() const {
-    return repository_.cur_state_;
+    return repository_.cur_state_.load(std::memory_order_acquire);
 }