From 2e2b02ab90070d036940c0e7c8a5922d4dad4906 Mon Sep 17 00:00:00 2001 From: Chunel Date: Sun, 9 Aug 2026 23:28:48 +0800 Subject: [PATCH] [perf] optimize common task run, no new task everytime --- src/GraphCtrl/GraphElement/GElement.cpp | 1 + src/GraphCtrl/GraphElement/GElement.h | 7 +- .../GDynamicEngine/GDynamicEngine.cpp | 16 ++++- .../_GEngine/GDynamicEngine/GDynamicEngine.h | 6 ++ src/UtilsCtrl/ThreadPool/Task/UTask.h | 72 +++++++++++++++---- src/UtilsCtrl/ThreadPool/UThreadPool.cpp | 21 ++++++ src/UtilsCtrl/ThreadPool/UThreadPool.h | 19 ++++- src/UtilsCtrl/ThreadPool/UThreadPool.inl | 18 ++--- 8 files changed, 126 insertions(+), 34 deletions(-) diff --git a/src/GraphCtrl/GraphElement/GElement.cpp b/src/GraphCtrl/GraphElement/GElement.cpp index e908412f..b4c21aad 100644 --- a/src/GraphCtrl/GraphElement/GElement.cpp +++ b/src/GraphCtrl/GraphElement/GElement.cpp @@ -14,6 +14,7 @@ CGRAPH_NAMESPACE_BEGIN GElement::~GElement() { + CGRAPH_DELETE_PTR(run_task_) CGRAPH_DELETE_PTR(perf_info_) CGRAPH_DELETE_PTR(aspect_manager_) for (auto& param : local_params_) { diff --git a/src/GraphCtrl/GraphElement/GElement.h b/src/GraphCtrl/GraphElement/GElement.h index 70e669c9..f7d64993 100644 --- a/src/GraphCtrl/GraphElement/GElement.h +++ b/src/GraphCtrl/GraphElement/GElement.h @@ -455,10 +455,11 @@ class GElement : public GElementObject, CBool is_marco_ { false }; // 微任务 /** 执行期间相关信息 */ - GElementParamMap local_params_; // 用于记录当前element的内部参数 + GElementParamMap local_params_ {}; // 用于记录当前element的内部参数 GAspectManagerPtr aspect_manager_ { nullptr }; // 整体流程的切面管理类 UThreadPoolPtr thread_pool_ { nullptr }; // 用于执行的线程池信息 GPerfInfo* perf_info_ { nullptr }; // 用于perf的信息 + UTask* run_task_ { nullptr }; // 用于记录 调度执行函数 CBool is_prepared_ { false }; // 判断是否已经执行过 prepareRun() 方法 /** 图相关信息 */ @@ -469,8 +470,8 @@ class GElement : public GElementObject, std::vector children_ {}; // 子节点，适用于group类型 /** 异步执行相关信息 */ - std::future async_result_; // 用于记录当前节点的异步执行情况 - UCvMutex suspend_locker_; // 控制停止执行锁信息 + std::future async_result_ {}; // 用于记录当前节点的异步执行情况 + UCvMutex suspend_locker_ {}; // 控制停止执行锁信息 friend class GNode; friend class GGroup; diff --git a/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp b/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp index 29aa8c55..fd2fe6da 100644 --- a/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp +++ b/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.cpp @@ -95,6 +95,7 @@ CVoid GDynamicEngine::analysisDagType() { analysisParallelMatrix(); } else { dag_type_ = internal::GEngineDagType::COMMON; + makeCommonTask(); } } @@ -121,6 +122,16 @@ CVoid GDynamicEngine::analysisParallelMatrix() { } +CVoid GDynamicEngine::makeCommonTask() { + for (auto* element : total_element_arr_) { + CGRAPH_DELETE_PTR(element->run_task_) + element->run_task_ = new UTask([this, element] { + this->innerExec(element); + }); + } +} + + CVoid GDynamicEngine::process(GElementPtr element, const CBool affinity) { if (unlikely(cur_status_.isErr())) { return; @@ -130,8 +141,7 @@ CVoid GDynamicEngine::process(GElementPtr element, const CBool affinity) { // 如果 affinity=true，表示用当前的线程，执行这个逻辑。以便增加亲和性 innerExec(element); } else { - thread_pool_->execute([this, element] { - this->innerExec(element); }, element->binding_index_); + thread_pool_->execute(element->run_task_, element->binding_index_); } } @@ -234,7 +244,7 @@ CVoid GDynamicEngine::parallelRunAll() { } } else { // 仅有一个任务的情况，无法使用 executeWithTid 函数，故走这边的逻辑 - const auto& element = curArr.front(); + auto* element = curArr.front(); thread_pool_->execute([this, element] { parallelRunOne(element); }, element->binding_index_); } diff --git a/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h b/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h index 0877d1f4..d7a05fc7 100644 --- a/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h +++ b/src/GraphCtrl/GraphElement/_GEngine/GDynamicEngine/GDynamicEngine.h @@ -42,6 +42,12 @@ class GDynamicEngine : public GEngine { */ CVoid analysisParallelMatrix(); + /** + * 创建 common run 的情况下，element对应的任务 + * @return + */ + CVoid makeCommonTask(); + /** * 动态图运行 * @param diff --git a/src/UtilsCtrl/ThreadPool/Task/UTask.h b/src/UtilsCtrl/ThreadPool/Task/UTask.h index 93343317..5963c8de 100644 --- a/src/UtilsCtrl/ThreadPool/Task/UTask.h +++ b/src/UtilsCtrl/ThreadPool/Task/UTask.h @@ -10,7 +10,6 @@ #define CGRAPH_UTASK_H #include -#include #include #include @@ -36,28 +35,72 @@ class UTask : public CStruct { public: template::type, UTask>::value, int>::type = 0> - explicit UTask(F&& func, const int priority = 0) + explicit UTask(F&& func, const CInt priority = 0, const CBool owner = true) : impl_(new TaskDerided(std::forward(func))) - , priority_(priority) {} + , priority_(priority) + , owner_(owner) { + } CVoid operator()() const { - // impl_ 理论上不可能为空 - impl_->call(); + if (likely(impl_)) { + impl_->call(); + } } explicit UTask() = default; + explicit UTask(const UTask* task) { + if (likely(task)) { + impl_ = task->impl_; + priority_ = task->priority_; + owner_ = false; + } + } + + explicit UTask(UTask* task) { + if (likely(task)) { + impl_ = task->impl_; + priority_ = task->priority_; + owner_ = false; + } + } + + ~UTask() override { + if (owner_) { + CGRAPH_DELETE_PTR(impl_); + } + } + UTask(UTask&& task) noexcept: - impl_(std::move(task.impl_)), - priority_(task.priority_) {} + impl_(task.impl_), + priority_(task.priority_), + owner_(task.owner_) { + task.impl_ = nullptr; + task.owner_ = false; + } UTask(UTask&& task, const int priority) noexcept: - impl_(std::move(task.impl_)), - priority_(priority) {} + impl_(task.impl_), + priority_(priority), + owner_(task.owner_) { + task.impl_ = nullptr; + task.owner_ = false; + } + + UTask& operator=(UTask&& task) noexcept { + if (this != &task) { + if (owner_) { + CGRAPH_DELETE_PTR(impl_); + } + + impl_ = task.impl_; + priority_ = task.priority_; + owner_ = task.owner_; + + task.impl_ = nullptr; + task.owner_ = false; + } - UTask &operator=(UTask&& task) noexcept { - impl_ = std::move(task.impl_); - priority_ = task.priority_; return *this; } @@ -72,8 +115,11 @@ class UTask : public CStruct { CGRAPH_NO_ALLOWED_COPY(UTask) private: - std::unique_ptr impl_ { nullptr }; + friend class UThreadPool; + + TaskBased* impl_ { nullptr }; CInt priority_ { 0 }; // 任务的优先级信息 + CBool owner_ { true }; // impl_ 是否归属当前对象 }; diff --git a/src/UtilsCtrl/ThreadPool/UThreadPool.cpp b/src/UtilsCtrl/ThreadPool/UThreadPool.cpp index 3233d757..6e497d8f 100644 --- a/src/UtilsCtrl/ThreadPool/UThreadPool.cpp +++ b/src/UtilsCtrl/ThreadPool/UThreadPool.cpp @@ -127,6 +127,27 @@ CStatus UThreadPool::submit(CGRAPH_DEFAULT_CONST_FUNCTION_REF func, const CMSec } +CVoid UThreadPool::execute(const UTask* task, const CIndex index) { + envokeTask(UTask(task), index); +} + + +CVoid UThreadPool::envokeTask(UTask&& task, const CIndex index) { + const CIndex realIndex = dispatch(index); + + if (likely(realIndex >= 0 && realIndex < config_.default_thread_size_)) { + primary_threads_[realIndex]->pushTask(std::move(task)); + } else if (CGRAPH_LONG_TIME_TASK_STRATEGY == realIndex) { + priority_task_queue_.push(std::move(task), CGRAPH_LONG_TIME_TASK_STRATEGY); + } else if (CGRAPH_TRIGGER_ALL_THREAD_STRATEGY == realIndex) { + task_queue_.push(std::move(task)); + (void)wakeupAllThread(); + } else { + task_queue_.push(std::move(task)); + } +} + + CIndex UThreadPool::getThreadIndex(const CSize tid) { int index = CGRAPH_SECONDARY_THREAD_COMMON_ID; const auto result = thread_record_map_.find(tid); diff --git a/src/UtilsCtrl/ThreadPool/UThreadPool.h b/src/UtilsCtrl/ThreadPool/UThreadPool.h index 801f49a3..90533bc9 100644 --- a/src/UtilsCtrl/ThreadPool/UThreadPool.h +++ b/src/UtilsCtrl/ThreadPool/UThreadPool.h @@ -106,10 +106,20 @@ class UThreadPool : public UThreadObject { * @param task * @param index */ - template + template::type, UTask>::value, int>::type = 0> CVoid execute(FunctionType&& task, CIndex index = CGRAPH_DEFAULT_TASK_STRATEGY); + /** + * 异步执行已封装的任务。队列仅借用 task 中的执行内容，不接管其生命周期。 + * task 必须存活到本次异步执行结束，并且可以被重复提交。 + * @param task + * @param index + */ + CVoid execute(const UTask* task, + CIndex index = CGRAPH_DEFAULT_TASK_STRATEGY); + /** * 异步写入特定thread id，执行信息 * @tparam FunctionType @@ -197,6 +207,13 @@ class UThreadPool : public UThreadObject { */ CVoid monitor(); + /** + * 将任务投递到线程池内部 + * @param task + * @param index + */ + CVoid envokeTask(UTask&& task, CIndex index); + CGRAPH_NO_ALLOWED_COPY(UThreadPool) private: diff --git a/src/UtilsCtrl/ThreadPool/UThreadPool.inl b/src/UtilsCtrl/ThreadPool/UThreadPool.inl index d5124fef..3c43e3cb 100644 --- a/src/UtilsCtrl/ThreadPool/UThreadPool.inl +++ b/src/UtilsCtrl/ThreadPool/UThreadPool.inl @@ -54,25 +54,15 @@ auto UThreadPool::commitWithPriority(const FunctionType& func, int priority) } -template +template::type, UTask>::value, int>::type> CVoid UThreadPool::execute(FunctionType&& task, const CIndex index) { - const CIndex realIndex = dispatch(index); - - if (likely(realIndex >= 0 && realIndex < config_.default_thread_size_)) { - primary_threads_[realIndex]->pushTask(UTask(std::forward(task))); - } else if (CGRAPH_LONG_TIME_TASK_STRATEGY == realIndex) { - priority_task_queue_.push(UTask(std::forward(task)), CGRAPH_LONG_TIME_TASK_STRATEGY); - } else if (CGRAPH_TRIGGER_ALL_THREAD_STRATEGY == realIndex) { - task_queue_.push(UTask(std::forward(task))); - (void)wakeupAllThread(); - } else { - task_queue_.push(UTask(std::forward(task))); - } + envokeTask(UTask(std::forward(task)), index); } template -CVoid UThreadPool::executeWithTid(FunctionType&& task, CIndex tid, CBool enable, CBool lockable) { +CVoid UThreadPool::executeWithTid(FunctionType&& task, const CIndex tid, const CBool enable, const CBool lockable) { if (likely(tid >= 0 && tid < config_.default_thread_size_)) { primary_threads_[tid]->pushTask(UTask(std::forward(task)), enable, lockable); } else {