1 /*
2 * Copyright (c) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15 #include "thread_pool.h"
16 #include <cstring>
17 #include "script_utils.h"
18
19 namespace uscript {
20 static ThreadPool* g_threadPool = nullptr;
21 static std::mutex g_initMutex;
22
CreateThreadPool(int number)23 ThreadPool* ThreadPool::CreateThreadPool(int number)
24 {
25 USCRIPT_CHECK(number > 1, return nullptr, "Invalid number %d", number);
26 std::lock_guard<std::mutex> lock(g_initMutex);
27 if (g_threadPool != nullptr) {
28 return g_threadPool;
29 }
30 if (g_threadPool == nullptr) {
31 g_threadPool = new ThreadPool();
32 g_threadPool->Init(number);
33 }
34 return g_threadPool;
35 }
36
Destroy()37 void ThreadPool::Destroy()
38 {
39 if (g_threadPool == nullptr) {
40 return;
41 }
42 std::lock_guard<std::mutex> lock(g_initMutex);
43 delete g_threadPool;
44 g_threadPool = nullptr;
45 }
46
Init(int32_t numberThread)47 void ThreadPool::Init(int32_t numberThread)
48 {
49 threadNumber_ = numberThread;
50 taskQueue_.resize(THREAD_POOL_MAX_TASKS);
51 for (size_t t = 0; t < taskQueue_.size(); ++t) {
52 taskQueue_[t].available = true;
53 for (int32_t i = 0; i < threadNumber_; ++i) {
54 taskQueue_[t].subTaskFlag.emplace_back(new std::atomic_bool { false });
55 }
56 }
57 // Create workers
58 for (int32_t threadIndex = 1; threadIndex < threadNumber_; ++threadIndex) {
59 workers_.emplace_back(std::thread(ThreadPool::ThreadExecute, this, threadIndex));
60 }
61 }
62
ThreadRun(int32_t threadIndex)63 void ThreadPool::ThreadRun(int32_t threadIndex)
64 {
65 while (!stop_) {
66 for (int32_t k = 0; k < THREAD_POOL_MAX_TASKS; ++k) {
67 if (*taskQueue_[k].subTaskFlag[threadIndex]) {
68 taskQueue_[k].task.processor(threadIndex);
69 *taskQueue_[k].subTaskFlag[threadIndex] = false;
70 }
71 }
72 std::this_thread::yield();
73 }
74 printf("ThreadPool::ThreadRun %d exit \n", threadIndex);
75 }
76
~ThreadPool()77 ThreadPool::~ThreadPool()
78 {
79 {
80 std::lock_guard<std::mutex> lock(queueMutex_);
81 stop_ = true;
82 }
83 for (auto& worker : workers_) {
84 worker.join();
85 }
86 for (auto& task : taskQueue_) {
87 for (auto c : task.subTaskFlag) {
88 delete c;
89 }
90 }
91 }
92
AddTask(Task && task)93 void ThreadPool::AddTask(Task &&task)
94 {
95 if (g_threadPool != nullptr) {
96 g_threadPool->AddNewTask(std::move(task));
97 }
98 }
99
AddNewTask(Task && task)100 void ThreadPool::AddNewTask(Task &&task)
101 {
102 int32_t index = AcquireWorkIndex();
103 USCRIPT_LOGI("ThreadPool::AddNewTask %d ", index);
104
105 // If there are no multi-works
106 // Do not need to enqueue, execute it immediately
107 if (task.workSize <= 1 || index < 0) {
108 for (int32_t i = 0; i < task.workSize; ++i) {
109 task.processor(i);
110 }
111 return;
112 }
113
114 int32_t workSize = task.workSize;
115 // If too much works, Create new task to do the work.
116 if (workSize > threadNumber_) {
117 Task newTask;
118 newTask.workSize = threadNumber_;
119 newTask.processor = [workSize, &task, this](int tId) {
120 for (int v = tId; v < workSize; v += threadNumber_) {
121 task.processor(v);
122 }
123 };
124 RunTask(std::move(newTask), index);
125 } else {
126 RunTask(std::move(task), index);
127 }
128
129 // Works done. make this task available
130 std::lock_guard<std::mutex> lock(queueMutex_);
131 taskQueue_[index].available = true;
132 }
133
AcquireWorkIndex()134 int32_t ThreadPool::AcquireWorkIndex()
135 {
136 std::lock_guard<std::mutex> lock(queueMutex_);
137 for (int32_t i = 0; i < THREAD_POOL_MAX_TASKS; ++i) {
138 if (taskQueue_[i].available) {
139 taskQueue_[i].available = false;
140 return i;
141 }
142 }
143 return -1;
144 }
145
RunTask(Task && task,int32_t index)146 void ThreadPool::RunTask(Task &&task, int32_t index)
147 {
148 taskQueue_[index].task = std::move(task);
149 int32_t workSize = task.workSize;
150 // Mark each task should be executed
151 for (int32_t i = 1; i < workSize; ++i) {
152 *taskQueue_[index].subTaskFlag[i] = true;
153 }
154
155 // Execute first task
156 taskQueue_[index].task.processor(0);
157 bool complete = true;
158 do {
159 std::this_thread::yield();
160 complete = true;
161 // 检查是否所有子任务执行结束
162 for (int32_t i = 1; i < workSize; ++i) {
163 if (*taskQueue_[index].subTaskFlag[i]) {
164 complete = false;
165 break;
166 }
167 }
168 } while (!complete);
169 }
170 } // namespace uscript
171