Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "aicpu_sharder.h"
12 :
13 : #include <atomic>
14 : #include <semaphore.h>
15 : #include <unistd.h>
16 : #ifndef _AOSCORE_
17 : #include <error.h>
18 : #endif
19 : #include <cstring>
20 : #include <algorithm>
21 : #include <cerrno>
22 :
23 : #include "aicpu_sharder_log.h"
24 : #include "aicpu_context.h"
25 :
26 : namespace aicpu {
27 : constexpr uint32_t GET_EVENT_LIMITED_NUM = 1000U;
28 :
29 3 : SharderNonBlock::SharderNonBlock()
30 3 : : cpuCoreNum_(0U),
31 3 : randomKernelScheduler_(nullptr),
32 3 : splitKernelScheduler_(nullptr),
33 3 : splitKernelGetProcesser_(nullptr),
34 3 : parallelId_(0U)
35 3 : {}
36 :
37 48 : void SharderNonBlock::Register(
38 : const uint32_t cpuCoreNum, const RandomKernelScheduler& randomKernelScheduler,
39 : const SplitKernelScheduler& splitKernelScheduler, const SplitKernelGetProcesser& splitKernelGetProcesser)
40 : {
41 48 : cpuCoreNum_ = cpuCoreNum;
42 48 : randomKernelScheduler_ = randomKernelScheduler;
43 48 : splitKernelScheduler_ = splitKernelScheduler;
44 48 : splitKernelGetProcesser_ = splitKernelGetProcesser;
45 48 : }
46 :
47 3 : void SharderNonBlock::Schedule(const Closure& aicpuClosure)
48 : {
49 3 : if (randomKernelScheduler_ == nullptr) {
50 1 : aicpuClosure();
51 1 : return;
52 : }
53 :
54 2 : const uint32_t ret = randomKernelScheduler_(aicpuClosure);
55 2 : if (ret != 0U) {
56 1 : aicpuClosure();
57 1 : AICPUE_LOGE("Schedule random kernel event failed, do it by self. ret=%u", ret);
58 : }
59 :
60 2 : return;
61 : }
62 :
63 1 : uint32_t SharderNonBlock::GetCPUNum() { return cpuCoreNum_; }
64 :
65 113 : SharderNonBlock& SharderNonBlock::GetInstance()
66 : {
67 113 : static SharderNonBlock sharderNonBlock;
68 113 : return sharderNonBlock;
69 : }
70 :
71 114 : inline int64_t SharderNonBlock::CeilMultiple(const int64_t x, const int64_t base) const
72 : {
73 114 : int64_t ret = x / base;
74 114 : if ((x % base) != 0) {
75 0 : ret++;
76 : }
77 :
78 114 : return ret;
79 : }
80 :
81 37 : void SharderNonBlock::ParallelFor(const int64_t total, const int64_t perUnitSize, const SharderWork& work)
82 : {
83 37 : uint32_t parallelId = 0U;
84 : {
85 37 : const std::lock_guard<std::mutex> lk(parallelIdMutex_);
86 38 : ++parallelId_;
87 38 : parallelId = parallelId_.load();
88 38 : }
89 :
90 38 : AICPUE_LOGI("In parallel for. parallelId=%u, total=%ld, perUnitSize=%ld", parallelId, total, perUnitSize);
91 38 : if ((total <= 0) || (work == nullptr)) {
92 0 : AICPUE_LOGE("Invalid param. parallelId=%u, total=%ld", parallelId, total);
93 0 : return;
94 : }
95 :
96 38 : if ((splitKernelScheduler_ == nullptr) || (splitKernelGetProcesser_ == nullptr) || (cpuCoreNum_ <= 1U)) {
97 0 : AICPUE_LOGI("Work itself all. parallelId=%u, cpuCoreNum=%u", parallelId, cpuCoreNum_);
98 0 : work(0, total);
99 0 : return;
100 : }
101 :
102 : // In order to ensure a smaller scheduling delay, the maximum number of slices is twice the number of CPU cores
103 38 : const int64_t maxShardNum = static_cast<int64_t>(cpuCoreNum_) * 2;
104 :
105 : // calculate shard number and block size
106 : // i.e., if total is 118, perUintSize is 2, and cpuCoreNum_ is 13
107 : // then shardNum is 24, blockSize is 5
108 38 : int64_t blockSize = std::max(int64_t{1}, std::min(total, perUnitSize));
109 38 : int64_t shardNum = CeilMultiple(total, blockSize);
110 38 : shardNum = std::min(maxShardNum, shardNum);
111 38 : blockSize = CeilMultiple(total, shardNum);
112 38 : shardNum = CeilMultiple(total, blockSize);
113 : // There is no need to submit an event if shardNum is 1
114 38 : if (shardNum == 1) {
115 1 : AICPUE_LOGI(
116 : "Executes on the current thread by shardNum is 1. parallelId=%u, total=%ld, perUnitSize=%ld", parallelId,
117 : total, perUnitSize);
118 1 : work(0, total);
119 1 : return;
120 : }
121 :
122 37 : ExecuteParallelFor(total, shardNum, blockSize, work, parallelId);
123 37 : return;
124 : }
125 :
126 37 : void SharderNonBlock::ExecuteParallelFor(
127 : const int64_t total, const int64_t shardNum, const int64_t blockSize, const SharderWork& work,
128 : const uint32_t parallelId)
129 : {
130 37 : AICPUE_LOGI(
131 : "Op parallel process start. parallelId=%u, shardNum=%ld, blockSize=%ld", parallelId, shardNum, blockSize);
132 :
133 37 : std::atomic<int64_t> cpuNumCounter(shardNum);
134 : sem_t aicpuSem;
135 37 : const int32_t semInitRet = sem_init(&aicpuSem, 0, 0U);
136 37 : if (semInitRet == -1) {
137 0 : AICPUE_LOGE("sem_init error with message: %s", strerror(errno));
138 0 : work(0, total);
139 0 : return;
140 : }
141 :
142 37 : uint32_t taskIndex = 0U;
143 37 : std::queue<aicpu::Closure> taskQueue;
144 174 : for (int64_t start = 0; start < total; start += blockSize) {
145 141 : const auto limit = std::min(start + blockSize, total);
146 0 : const Closure aicpuClosure = [&aicpuSem, &work, &cpuNumCounter, start, limit, parallelId, shardNum,
147 : taskIndex]() {
148 141 : cpuNumCounter--;
149 : // In order to ensure that user's work function exception does not affect multithread services,
150 : // exception capture is needed. Exception type is not cared here, and error log is printed.
151 142 : AICPUE_LOGI(
152 : "Start call work func. parallelId=%u, shardNum=%ld, taskIndex=%u", parallelId, shardNum, taskIndex);
153 : try {
154 142 : work(start, limit);
155 4 : } catch (std::exception& e) {
156 4 : AICPUE_LOGE(
157 : "Exception occurred in work func. parallelId=%u, shardNum=%ld, taskIndex=%u, "
158 : "exception=%s",
159 : parallelId, shardNum, taskIndex, e.what());
160 4 : }
161 :
162 142 : const int32_t semPostRet = sem_post(&aicpuSem);
163 142 : if (semPostRet == -1) {
164 0 : AICPUE_LOGE("sem_post error with message: %s", strerror(errno));
165 : }
166 142 : AICPUE_LOGI(
167 : "End call work func. parallelId=%u, shardNum=%ld, taskIndex=%u", parallelId, shardNum, taskIndex);
168 283 : };
169 :
170 139 : taskQueue.push(aicpuClosure);
171 140 : ++taskIndex;
172 140 : }
173 :
174 33 : const uint32_t ret = splitKernelScheduler_(parallelId, shardNum, taskQueue);
175 37 : if (ret != 0U) {
176 1 : AICPUE_LOGE("Submit split kernel task failed, ret=%u, parallelId=%u", ret, parallelId);
177 1 : (void)sem_destroy(&aicpuSem);
178 1 : return;
179 : }
180 :
181 36 : DoTaskItself(parallelId, cpuNumCounter, shardNum);
182 :
183 178 : for (int64_t i = 0; i < shardNum; ++i) {
184 142 : const int32_t semWaitRet = sem_wait(&aicpuSem);
185 142 : if (semWaitRet == -1) {
186 4 : AICPUE_LOGE("sem_wait error with message: %s", strerror(errno));
187 : }
188 : }
189 36 : const int32_t semDesRet = sem_destroy(&aicpuSem);
190 36 : if (semDesRet == -1) {
191 1 : AICPUE_LOGE("sem_destroy error with message: %s", strerror(errno));
192 : }
193 :
194 36 : AICPUE_LOGI("Op parallel process finished. parallelId=%u", parallelId);
195 :
196 36 : return;
197 37 : }
198 :
199 40 : void SharderNonBlock::DoTaskItself(
200 : const uint32_t parallelId, std::atomic<int64_t>& cpuNumCounter, const int64_t shardNum) const
201 : {
202 40 : uint32_t getEventCnt = 0U;
203 40 : bool ret = true;
204 40 : bool logPrintFlag = true;
205 1195 : while ((cpuNumCounter > 0) && ret) {
206 1155 : ret = splitKernelGetProcesser_();
207 1155 : ++getEventCnt;
208 1155 : if ((getEventCnt >= GET_EVENT_LIMITED_NUM) && (logPrintFlag)) {
209 1 : logPrintFlag = false;
210 1 : std::string opname("");
211 1 : (void)GetOpname(GetAicpuThreadIndex(), opname);
212 2 : AICPUE_RUN_LOGW(
213 : "Get event num has exceeded %u. parallelId=%u, cpuNumCounter=%ld, shardNum=%ld, "
214 : "opName=%s",
215 : GET_EVENT_LIMITED_NUM, parallelId, cpuNumCounter.load(), shardNum, opname.c_str());
216 1 : }
217 : }
218 40 : }
219 :
220 6 : void SharderNonBlock::ExecuteParallelForHash(
221 : const int64_t total, const int64_t cpuNums, const SharderWork& work, const uint32_t parallelId)
222 : {
223 6 : AICPUE_LOGI("Op hash parallel process start. parallelId=%u, total=%ld, cpuNums=%ld", parallelId, total, cpuNums);
224 :
225 6 : std::atomic<int64_t> cpuNumCounter(cpuNums);
226 : sem_t aicpuSem;
227 6 : const int32_t semInitRet = sem_init(&aicpuSem, 0, 0U);
228 6 : if (semInitRet == -1) {
229 1 : AICPUE_LOGE("sem_init error with message: %s", strerror(errno));
230 1 : return;
231 : }
232 :
233 5 : std::queue<aicpu::Closure> taskQueue;
234 15 : for (int64_t cur = 0; cur < cpuNums; cur++) {
235 0 : const Closure aicpuClosure = [&aicpuSem, &work, &cpuNumCounter, total, cur, parallelId]() {
236 8 : cpuNumCounter--;
237 16 : AICPUE_LOGI(
238 : "Start call work func. parallelId=%u, cur=%ld, cpuNumCounter=%ld", parallelId, cur,
239 : cpuNumCounter.load());
240 8 : work(total, cur);
241 :
242 8 : const int32_t semPostRet = sem_post(&aicpuSem);
243 8 : if (semPostRet == -1) {
244 2 : AICPUE_LOGE("sem_post error with message: %s", strerror(errno));
245 : }
246 10 : };
247 :
248 10 : taskQueue.push(aicpuClosure);
249 10 : }
250 :
251 5 : const uint32_t ret = splitKernelScheduler_(parallelId, cpuNums, taskQueue);
252 5 : if (ret != 0U) {
253 1 : AICPUE_LOGE("Submit hash split kernel task failed, ret=%u, parallelId=%u", ret, parallelId);
254 1 : (void)sem_destroy(&aicpuSem);
255 1 : return;
256 : }
257 :
258 4 : DoTaskItself(parallelId, cpuNumCounter, cpuNums);
259 :
260 12 : for (int64_t i = 0; i < cpuNums; i++) {
261 8 : const int32_t semWaitRet = sem_wait(&aicpuSem);
262 8 : if (semWaitRet == -1) {
263 4 : AICPUE_LOGE("sem_wait error with message: %s", strerror(errno));
264 : }
265 : }
266 4 : const int32_t semDesRet = sem_destroy(&aicpuSem);
267 4 : if (semDesRet == -1) {
268 1 : AICPUE_LOGE("sem_destroy error with message: %s", strerror(errno));
269 : }
270 :
271 4 : AICPUE_LOGI("Op hash parallel process finished. parallelId=%u", parallelId);
272 4 : return;
273 5 : }
274 :
275 7 : void SharderNonBlock::ParallelForHash(const int64_t total, const int64_t cpuNums, const SharderWork& work)
276 : {
277 7 : uint32_t parallelId = 0U;
278 : {
279 7 : const std::lock_guard<std::mutex> lk(parallelIdMutex_);
280 7 : ++parallelId_;
281 7 : parallelId = parallelId_.load();
282 7 : }
283 :
284 7 : if ((total <= 0) || (work == nullptr)) {
285 0 : AICPUE_LOGE("invalid param: total<=0 or work is nullptr");
286 0 : return;
287 : }
288 :
289 7 : if ((splitKernelScheduler_ == nullptr) || (splitKernelGetProcesser_ == nullptr) || (cpuCoreNum_ <= 1U)) {
290 1 : AICPUE_LOGE("schedule is nullptr or cpu core num is not enough");
291 1 : return;
292 : }
293 :
294 6 : ExecuteParallelForHash(total, cpuNums, work, parallelId);
295 6 : return;
296 : }
297 : } // namespace aicpu
298 :
299 : /**
300 : * Shards the "total" unit of work refer "perUintSize"
301 : */
302 38 : void ParallelFor(int64_t total, int64_t perUnitSize, const aicpu::SharderWork& work)
303 : {
304 38 : aicpu::SharderNonBlock::GetInstance().ParallelFor(total, perUnitSize, work);
305 38 : }
306 :
307 : /**
308 : * Get CPU number
309 : */
310 1 : uint32_t GetCPUNum() { return aicpu::SharderNonBlock::GetInstance().GetCPUNum(); }
|
