diff --git a/src/GraphCtrl/GraphStage/GStage.cpp b/src/GraphCtrl/GraphStage/GStage.cpp index d0d21979..70de58dd 100644 --- a/src/GraphCtrl/GraphStage/GStage.cpp +++ b/src/GraphCtrl/GraphStage/GStage.cpp @@ -11,6 +11,7 @@ CGRAPH_NAMESPACE_BEGIN CVoid GStage::launch(GStageParamPtr param) { + (void)param; } diff --git a/src/UtilsCtrl/ThreadPool/Queue/UAtomicPriorityQueue.h b/src/UtilsCtrl/ThreadPool/Queue/UAtomicPriorityQueue.h deleted file mode 100644 index 552f4f5f..00000000 --- a/src/UtilsCtrl/ThreadPool/Queue/UAtomicPriorityQueue.h +++ /dev/null @@ -1,94 +0,0 @@ -/*************************** -@Author: Chunel -@Contact: chunel@foxmail.com -@File: UAtomicPriorityQueue.h -@Time: 2022/10/1 21:40 -@Desc: 线程安全的优先队列。因为 priority_queue和queue的弹出方式不一致，故暂时不做合并 -***************************/ - -#ifndef CGRAPH_UATOMICPRIORITYQUEUE_H -#define CGRAPH_UATOMICPRIORITYQUEUE_H - -#include - -#include "UQueueObject.h" - -CGRAPH_NAMESPACE_BEGIN - -template -class UAtomicPriorityQueue : public UQueueObject { -public: - UAtomicPriorityQueue() = default; - - /** - * 尝试弹出 - * @param value - * @return - */ - CBool tryPop(T& value) { - CBool result = false; - if (mutex_.try_lock()) { - if (!priority_queue_.empty()) { - value = std::move(*priority_queue_.top()); - priority_queue_.pop(); - result = true; - } - mutex_.unlock(); - } - - return result; - } - - - /** - * 尝试弹出多个任务 - * @param values - * @param maxPoolBatchSize - * @return - */ - CBool tryPop(std::vector& values, int maxPoolBatchSize) { - CBool result = false; - if (mutex_.try_lock()) { - while (!priority_queue_.empty() && maxPoolBatchSize-- > 0) { - values.emplace_back(std::move(*priority_queue_.top())); - priority_queue_.pop(); - result = true; - } - mutex_.unlock(); - } - - return result; - } - - - /** - * 传入数据 - * @param value - * @param priority 任务优先级，数字排序 - * @return - */ - CVoid push(T&& value, int priority) { - std::unique_ptr task(c_make_unique(std::move(value), priority)); - CGRAPH_LOCK_GUARD lk(mutex_); - priority_queue_.push(std::move(task)); - } - - - /** - * 判定队列是否为空 - * @return - */ - CBool empty() { - CGRAPH_LOCK_GUARD lk(mutex_); - return priority_queue_.empty(); - } - - CGRAPH_NO_ALLOWED_COPY(UAtomicPriorityQueue) - -private: - std::priority_queue > priority_queue_; // 优先队列信息，根据重要级别决定先后执行顺序 -}; - -CGRAPH_NAMESPACE_END - -#endif //CGRAPH_UATOMICPRIORITYQUEUE_H diff --git a/src/UtilsCtrl/ThreadPool/Queue/UQueueInclude.h b/src/UtilsCtrl/ThreadPool/Queue/UQueueInclude.h index 359308be..aa226330 100644 --- a/src/UtilsCtrl/ThreadPool/Queue/UQueueInclude.h +++ b/src/UtilsCtrl/ThreadPool/Queue/UQueueInclude.h @@ -11,7 +11,6 @@ #include "UAtomicQueue.h" #include "UWorkStealingQueue.h" -#include "UAtomicPriorityQueue.h" #include "UAtomicRingBufferQueue.h" #include "ULockFreeRingBufferQueue.h" diff --git a/src/UtilsCtrl/ThreadPool/Task/UTask.h b/src/UtilsCtrl/ThreadPool/Task/UTask.h index 5963c8de..7d2e9500 100644 --- a/src/UtilsCtrl/ThreadPool/Task/UTask.h +++ b/src/UtilsCtrl/ThreadPool/Task/UTask.h @@ -35,10 +35,9 @@ class UTask : public CStruct { public: template::type, UTask>::value, int>::type = 0> - explicit UTask(F&& func, const CInt priority = 0, const CBool owner = true) - : impl_(new TaskDerided(std::forward(func))) - , priority_(priority) - , owner_(owner) { + explicit UTask(F&& func, const CBool owner = true) + : impl_(new TaskDerided(std::forward(func))), + owner_(owner) { } CVoid operator()() const { @@ -52,7 +51,6 @@ class UTask : public CStruct { explicit UTask(const UTask* task) { if (likely(task)) { impl_ = task->impl_; - priority_ = task->priority_; owner_ = false; } } @@ -60,7 +58,6 @@ class UTask : public CStruct { explicit UTask(UTask* task) { if (likely(task)) { impl_ = task->impl_; - priority_ = task->priority_; owner_ = false; } } @@ -73,15 +70,6 @@ class UTask : public CStruct { UTask(UTask&& task) noexcept: impl_(task.impl_), - priority_(task.priority_), - owner_(task.owner_) { - task.impl_ = nullptr; - task.owner_ = false; - } - - UTask(UTask&& task, const int priority) noexcept: - impl_(task.impl_), - priority_(priority), owner_(task.owner_) { task.impl_ = nullptr; task.owner_ = false; @@ -94,7 +82,6 @@ class UTask : public CStruct { } impl_ = task.impl_; - priority_ = task.priority_; owner_ = task.owner_; task.impl_ = nullptr; @@ -104,21 +91,10 @@ class UTask : public CStruct { return *this; } - CBool operator>(const UTask& task) const { - return priority_ < task.priority_; // 新加入的，放到后面 - } - - CBool operator<(const UTask& task) const { - return priority_ >= task.priority_; - } - CGRAPH_NO_ALLOWED_COPY(UTask) private: - friend class UThreadPool; - TaskBased* impl_ { nullptr }; - CInt priority_ { 0 }; // 任务的优先级信息 CBool owner_ { true }; // impl_ 是否归属当前对象 }; diff --git a/src/UtilsCtrl/ThreadPool/Thread/UThreadBase.h b/src/UtilsCtrl/ThreadPool/Thread/UThreadBase.h index 70a6834d..4237b392 100644 --- a/src/UtilsCtrl/ThreadPool/Thread/UThreadBase.h +++ b/src/UtilsCtrl/ThreadPool/Thread/UThreadBase.h @@ -26,7 +26,7 @@ class UThreadBase : public UThreadObject { is_init_ = false; is_running_.store(false, std::memory_order_relaxed); pool_task_queue_ = nullptr; - pool_priority_task_queue_ = nullptr; + pool_long_time_task_queue_ = nullptr; config_ = nullptr; total_task_num_ = 0; } @@ -55,7 +55,7 @@ class UThreadBase : public UThreadObject { CBool result = pool_task_queue_->tryPop(task); if (!result && CGRAPH_THREAD_TYPE_SECONDARY == type_) { // 如果辅助线程没有获取到的话，还需要再尝试从长时间任务队列中，获取一次 - result = pool_priority_task_queue_->tryPop(task); + result = pool_long_time_task_queue_->tryPop(task); } return result; } @@ -69,7 +69,7 @@ class UThreadBase : public UThreadObject { virtual CBool popPoolTask(UTaskArrRef tasks) { CBool result = pool_task_queue_->tryPop(tasks, config_->max_pool_batch_size_); if (!result && CGRAPH_THREAD_TYPE_SECONDARY == type_) { - result = pool_priority_task_queue_->tryPop(tasks, 1); // 从优先队列里，最多pop出来一个 + result = pool_long_time_task_queue_->tryPop(tasks, 1); // 从长时间任务队列里，最多pop出来一个 } return result; @@ -245,9 +245,9 @@ class UThreadBase : public UThreadObject { CInt type_ = 0; // 用于区分线程类型（主线程、辅助线程） CULong total_task_num_ = 0; // 处理的任务的数字 - UAtomicQueue* pool_task_queue_ { nullptr }; // 用于存放线程池中的普通任务 - UAtomicPriorityQueue* pool_priority_task_queue_ { nullptr }; // 用于存放线程池中的包含优先级任务的队列，仅辅助线程可以执行 - UThreadPoolConfigPtr config_ { nullptr }; // 配置参数信息 + UAtomicQueue* pool_task_queue_ { nullptr }; // 用于存放线程池中的普通任务 + UAtomicQueue* pool_long_time_task_queue_ { nullptr }; // 用于存放线程池中的长时间任务，仅辅助线程可以执行 + UThreadPoolConfigPtr config_ { nullptr }; // 配置参数信息 std::thread thread_; // 线程类 std::mutex mutex_; diff --git a/src/UtilsCtrl/ThreadPool/Thread/UThreadSecondary.h b/src/UtilsCtrl/ThreadPool/Thread/UThreadSecondary.h index 88a9e0f7..532e744b 100644 --- a/src/UtilsCtrl/ThreadPool/Thread/UThreadSecondary.h +++ b/src/UtilsCtrl/ThreadPool/Thread/UThreadSecondary.h @@ -38,19 +38,19 @@ class UThreadSecondary : public UThreadBase { /** * 设置pool的信息 * @param poolTaskQueue - * @param poolPriorityTaskQueue + * @param poolLongTimeTaskQueue * @param config * @return */ CStatus setThreadPoolInfo(UAtomicQueue* poolTaskQueue, - UAtomicPriorityQueue* poolPriorityTaskQueue, + UAtomicQueue* poolLongTimeTaskQueue, const UThreadPoolConfigPtr config) { CGRAPH_FUNCTION_BEGIN CGRAPH_ASSERT_INIT(false) // 初始化之前，设置参数 - CGRAPH_ASSERT_NOT_NULL(poolTaskQueue, poolPriorityTaskQueue, config) + CGRAPH_ASSERT_NOT_NULL(poolTaskQueue, poolLongTimeTaskQueue, config) this->pool_task_queue_ = poolTaskQueue; - this->pool_priority_task_queue_ = poolPriorityTaskQueue; + this->pool_long_time_task_queue_ = poolLongTimeTaskQueue; this->config_ = config; CGRAPH_FUNCTION_END } diff --git a/src/UtilsCtrl/ThreadPool/UThreadPool.cpp b/src/UtilsCtrl/ThreadPool/UThreadPool.cpp index 8b14b486..b9d0699f 100644 --- a/src/UtilsCtrl/ThreadPool/UThreadPool.cpp +++ b/src/UtilsCtrl/ThreadPool/UThreadPool.cpp @@ -57,6 +57,7 @@ CStatus UThreadPool::init() { thread_record_map_.clear(); thread_record_map_[std::hash{}(std::this_thread::get_id())] = CGRAPH_MAIN_THREAD_ID; task_queue_.setup(); + long_time_task_queue_.setup(); primary_threads_.reserve(config_.default_thread_size_); for (int i = 0; i < config_.default_thread_size_; i++) { auto* pt = CGRAPH_SAFE_MALLOC_COBJECT(UThreadPrimary); // 创建核心线程数 @@ -78,7 +79,7 @@ CStatus UThreadPool::init() { /** * 策略更新： - * 初始化的时候，也可以创建n个辅助线程。目的是为了配合仅使用 pool中 priority_queue 的场景 + * 初始化的时候，也可以创建n个辅助线程。目的是为了配合仅使用长时间任务队列的场景 * 一般情况下，建议为0。 */ status = createSecondaryThread(config_.secondary_thread_size_); @@ -138,7 +139,7 @@ CVoid UThreadPool::envokeTask(UTask&& task, const CIndex 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); + long_time_task_queue_.push(std::move(task)); } else if (CGRAPH_TRIGGER_ALL_THREAD_STRATEGY == realIndex) { task_queue_.push(std::move(task)); (void)wakeupAllThread(); @@ -186,6 +187,7 @@ CStatus UThreadPool::destroy() { // secondary 线程是智能指针，不需要delete task_queue_.reset(); + long_time_task_queue_.reset(); for (auto &st : secondary_threads_) { CGRAPH_ASSERT_NOT_NULL(st.get()); status += st->destroy(); @@ -254,7 +256,7 @@ CStatus UThreadPool::createSecondaryThread(const CInt size) { CGRAPH_LOCK_GUARD lock(st_mutex_); for (int i = 0; i < realSize; i++) { auto ptr = CGRAPH_MAKE_UNIQUE_COBJECT(UThreadSecondary) - ptr->setThreadPoolInfo(&task_queue_, &priority_task_queue_, &config_); + ptr->setThreadPoolInfo(&task_queue_, &long_time_task_queue_, &config_); status += ptr->init(); secondary_threads_.emplace_back(std::move(ptr)); } @@ -279,8 +281,8 @@ CVoid UThreadPool::monitor() { const bool busy = !primary_threads_.empty() && std::all_of(primary_threads_.begin(), primary_threads_.end(), [](UThreadPrimaryPtr ptr) { return ptr && ptr->is_running_.load(std::memory_order_relaxed); }); - // 如果忙碌或者priority_task_queue_中有任务，则需要添加 secondary线程 - if (busy || !priority_task_queue_.empty()) { + // 如果忙碌或者 long_time_task_queue_ 中有任务，则需要添加 secondary线程 + if (busy || !long_time_task_queue_.empty()) { createSecondaryThread(1); } diff --git a/src/UtilsCtrl/ThreadPool/UThreadPool.h b/src/UtilsCtrl/ThreadPool/UThreadPool.h index fac28db3..dc404f71 100644 --- a/src/UtilsCtrl/ThreadPool/UThreadPool.h +++ b/src/UtilsCtrl/ThreadPool/UThreadPool.h @@ -87,19 +87,6 @@ class UThreadPool : public UThreadObject { auto commitWithTid(FunctionType&& func, CIndex tid, CBool enable, CBool lockable) -> std::future::type>()())>; - /** - * 根据优先级，执行任务 - * @tparam FunctionType - * @param func - * @param priority 优先级别。自然序从大到小依次执行 - * @return - * @notice 建议，priority 范围在 [-100, 100] 之间 - */ - template - auto commitWithPriority(const FunctionType& func, - int priority) - -> std::future()())>; - /** * 异步执行任务 * @tparam FunctionType @@ -220,7 +207,7 @@ class UThreadPool : public UThreadObject { CBool is_init_ { false }; // 是否初始化 std::atomic cur_index_ { 0 }; // 记录被轮询到的线程index的位置 UAtomicQueue task_queue_ {}; // 用于存放普通任务 - UAtomicPriorityQueue priority_task_queue_; // 运行时间较长的任务队列，仅在辅助线程中执行 + UAtomicQueue long_time_task_queue_ {}; // 运行时间较长的任务队列，仅在辅助线程中执行 std::vector primary_threads_; // 记录所有的主线程 std::list> secondary_threads_; // 用于记录所有的辅助线程 UThreadPoolConfig config_; // 线程池设置值 diff --git a/src/UtilsCtrl/ThreadPool/UThreadPool.inl b/src/UtilsCtrl/ThreadPool/UThreadPool.inl index ef3cb1e5..bc1b6b4e 100644 --- a/src/UtilsCtrl/ThreadPool/UThreadPool.inl +++ b/src/UtilsCtrl/ThreadPool/UThreadPool.inl @@ -37,23 +37,6 @@ auto UThreadPool::commitWithTid(FunctionType&& func, CIndex tid, CBool enable, C } -template -auto UThreadPool::commitWithPriority(const FunctionType& func, int priority) --> std::future()())> { - using ResultType = decltype(std::declval()()); - - std::packaged_task task(func); - std::future result(task.get_future()); - - if (secondary_threads_.empty()) { - createSecondaryThread(1); // 如果没有开启辅助线程，则直接开启一个 - } - - priority_task_queue_.push(UTask(std::move(task)), priority); - return result; -} - - template::type, UTask>::value, int>::type> CVoid UThreadPool::execute(FunctionType&& task, const CIndex index) {