From a10fb6f458a06e2f98b42635c29b2264e5b14724 Mon Sep 17 00:00:00 2001 From: Chunel Date: Sun, 5 Jul 2026 11:43:08 +0800 Subject: [PATCH 1/2] [perf] optimize dynamic engine fast path --- .../GDynamicEngine/GDynamicEngine.cpp | 61 ++++++++++--------- .../_GEngine/GDynamicEngine/GDynamicEngine.h | 8 +++ 2 files changed, 41 insertions(+), 28 deletions(-) diff --git a/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp b/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp index 5bf8a5b6..dd31bf4d 100644 --- a/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp +++ b/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 +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 diff --git a/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h b/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h index 92dd02d3..0877d1f4 100644 --- a/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h +++ b/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h @@ -103,6 +103,14 @@ class GDynamicEngine : public GEngine { */ CVoid prepareRun(); + /** + * 快速校验结束 + * @return + * @notice 为了方式线程快速进入内核态，短时间在用户层停留，性能加速 + */ + template + CBool isFastFinished(Pred&& check); + private: GElementPtrArr total_element_arr_ {}; // pipeline中所有的元素信息集合 GElementPtrArr front_element_arr_ {}; // 没有依赖的元素信息 From 9f06d9168f578632487c173e69da536eb84dcb54 Mon Sep 17 00:00:00 2001 From: Chunel Date: Sun, 5 Jul 2026 13:17:27 +0800 Subject: [PATCH 2/2] [chron] change pipeline state to atomic --- src/GraphCtrl/GraphElement/GElementRepository.cpp | 14 ++++++-------- src/GraphCtrl/GraphElement/GElementRepository.h | 10 +++++++--- src/GraphCtrl/GraphPipeline/GPipeline.cpp | 2 +- 3 files changed, 14 insertions(+), 12 deletions(-) diff --git a/src/GraphCtrl/GraphElement/GElementRepository.cpp b/src/GraphCtrl/GraphElement/GElementRepository.cpp index 40fc7496..9c4e9b79 100644 --- a/src/GraphCtrl/GraphElement/GElementRepository.cpp +++ b/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(element); + const auto* group = dynamic_cast(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 \ No newline at end of file +CGRAPH_NAMESPACE_END diff --git a/src/GraphCtrl/GraphElement/GElementRepository.h b/src/GraphCtrl/GraphElement/GElementRepository.h index 1b6cf170..46492673 100644 --- a/src/GraphCtrl/GraphElement/GElementRepository.h +++ b/src/GraphCtrl/GraphElement/GElementRepository.h @@ -9,6 +9,8 @@ #ifndef CGRAPH_GELEMENTREPOSITORY_H #define CGRAPH_GELEMENTREPOSITORY_H +#include + #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 cur_state_ { GElementState::NORMAL }; // 当前pipeline状态信息 + GElementPtrSet async_elements_ {}; // 所有异步执行的逻辑，到后来一次性统一获取执行结果信息 friend class GPipeline; friend class GPerf; diff --git a/src/GraphCtrl/GraphPipeline/GPipeline.cpp b/src/GraphCtrl/GraphPipeline/GPipeline.cpp index 39f97ed5..938c6c97 100644 --- a/src/GraphCtrl/GraphPipeline/GPipeline.cpp +++ b/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); }