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() : cpuCoreNum_(0U),
30 3 : randomKernelScheduler_(nullptr),
31 3 : splitKernelScheduler_(nullptr),
32 3 : splitKernelGetProcesser_(nullptr),
33 3 : parallelId_(0U)
34 : {
35 3 : }
36 :
37 48 : void SharderNonBlock::Register(const uint32_t cpuCoreNum, const RandomKernelScheduler &randomKernelScheduler,
38 : const SplitKernelScheduler &splitKernelScheduler,
39 : 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()
64 : {
65 1 : return cpuCoreNum_;
66 : }
67 :
68 113 : SharderNonBlock &SharderNonBlock::GetInstance()
69 : {
70 113 : static SharderNonBlock sharderNonBlock;
71 113 : return sharderNonBlock;
72 : }
73 :
74 114 : inline int64_t SharderNonBlock::CeilMultiple(const int64_t x, const int64_t base) const
75 : {
76 114 : int64_t ret = x / base;
77 114 : if ((x % base) != 0) {
78 0 : ret++;
79 : }
80 :
81 114 : return ret;
82 : }
83 :
84 38 : void SharderNonBlock::ParallelFor(const int64_t total, const int64_t perUnitSize, const SharderWork &work)
85 : {
86 38 : uint32_t parallelId = 0U;
87 : {
88 38 : const std::lock_guard<std::mutex> lk(parallelIdMutex_);
89 38 : ++parallelId_;
90 38 : parallelId = parallelId_.load();
91 38 : }
92 :
93 38 : AICPUE_LOGI("In parallel for. parallelId=%u, total=%ld, perUnitSize=%ld", parallelId, total, perUnitSize);
94 38 : if ((total <= 0) || (work == nullptr)) {
95 0 : AICPUE_LOGE("Invalid param. parallelId=%u, total=%ld", parallelId, total);
96 0 : return;
97 : }
98 :
99 38 : if ((splitKernelScheduler_ == nullptr) || (splitKernelGetProcesser_ == nullptr) || (cpuCoreNum_ <= 1U)) {
100 0 : AICPUE_LOGI("Work itself all. parallelId=%u, cpuCoreNum=%u", parallelId, cpuCoreNum_);
101 0 : work(0, total);
102 0 : return;
103 : }
104 :
105 : // In order to ensure a smaller scheduling delay, the maximum number of slices is twice the number of CPU cores
106 38 : const int64_t maxShardNum = static_cast<int64_t>(cpuCoreNum_) * 2;
107 :
108 : // calculate shard number and block size
109 : // i.e., if total is 118, perUintSize is 2, and cpuCoreNum_ is 13
110 : // then shardNum is 24, blockSize is 5
111 38 : int64_t blockSize = std::max(int64_t{1}, std::min(total, perUnitSize));
112 38 : int64_t shardNum = CeilMultiple(total, blockSize);
113 38 : shardNum = std::min(maxShardNum, shardNum);
114 38 : blockSize = CeilMultiple(total, shardNum);
115 38 : shardNum = CeilMultiple(total, blockSize);
116 : // There is no need to submit an event if shardNum is 1
117 38 : if (shardNum == 1) {
118 1 : AICPUE_LOGI("Executes on the current thread by shardNum is 1. parallelId=%u, total=%ld, perUnitSize=%ld",
119 : parallelId, total, perUnitSize);
120 1 : work(0, total);
121 1 : return;
122 : }
123 :
124 37 : ExecuteParallelFor(total, shardNum, blockSize, work, parallelId);
125 37 : return;
126 : }
127 :
128 37 : void SharderNonBlock::ExecuteParallelFor(const int64_t total, const int64_t shardNum,
129 : const int64_t blockSize, const SharderWork &work,
130 : const uint32_t parallelId)
131 : {
132 37 : AICPUE_LOGI("Op parallel process start. parallelId=%u, shardNum=%ld, blockSize=%ld",
133 : parallelId, shardNum, blockSize);
134 :
135 37 : std::atomic<int64_t> cpuNumCounter(shardNum);
136 : sem_t aicpuSem;
137 37 : const int32_t semInitRet = sem_init(&aicpuSem, 0, 0U);
138 37 : if (semInitRet == -1) {
139 0 : AICPUE_LOGE("sem_init error with message: %s", strerror(errno));
140 0 : work(0, total);
141 0 : return;
142 : }
143 :
144 37 : uint32_t taskIndex = 0U;
145 37 : std::queue<aicpu::Closure> taskQueue;
146 175 : for (int64_t start = 0; start < total; start += blockSize) {
147 140 : const auto limit = std::min(start + blockSize, total);
148 0 : const Closure aicpuClosure = [&aicpuSem, &work, &cpuNumCounter, start, limit,
149 : parallelId, shardNum, taskIndex]() {
150 142 : cpuNumCounter--;
151 : // In order to ensure that user's work function exception does not affect multithread services,
152 : // exception capture is needed. Exception type is not cared here, and error log is printed.
153 142 : AICPUE_LOGI("Start call work func. parallelId=%u, shardNum=%ld, taskIndex=%u",
154 : parallelId, shardNum, taskIndex);
155 : try {
156 142 : work(start, limit);
157 4 : } catch (std::exception &e) {
158 4 : AICPUE_LOGE("Exception occurred in work func. parallelId=%u, shardNum=%ld, taskIndex=%u, "
159 : "exception=%s", 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("End call work func. parallelId=%u, shardNum=%ld, taskIndex=%u",
167 : parallelId, shardNum, taskIndex);
168 282 : };
169 :
170 138 : taskQueue.push(aicpuClosure);
171 140 : ++taskIndex;
172 140 : }
173 :
174 35 : 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(const uint32_t parallelId, std::atomic<int64_t> &cpuNumCounter,
200 : const int64_t shardNum) const
201 : {
202 40 : uint32_t getEventCnt = 0U;
203 40 : bool ret = true;
204 40 : bool logPrintFlag = true;
205 1190 : while ((cpuNumCounter > 0) && ret) {
206 1150 : ret = splitKernelGetProcesser_();
207 1150 : ++getEventCnt;
208 1150 : 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("Get event num has exceeded %u. parallelId=%u, cpuNumCounter=%ld, shardNum=%ld, "
213 : "opName=%s", GET_EVENT_LIMITED_NUM, parallelId, cpuNumCounter.load(), shardNum,
214 : opname.c_str());
215 1 : }
216 : }
217 40 : }
218 :
219 6 : void SharderNonBlock::ExecuteParallelForHash(const int64_t total, const int64_t cpuNums, const SharderWork &work,
220 : const uint32_t parallelId)
221 : {
222 6 : AICPUE_LOGI("Op hash parallel process start. parallelId=%u, total=%ld, cpuNums=%ld",
223 : 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("Start call work func. parallelId=%u, cur=%ld, cpuNumCounter=%ld",
238 : parallelId, cur, cpuNumCounter.load());
239 8 : work(total, cur);
240 :
241 8 : const int32_t semPostRet = sem_post(&aicpuSem);
242 8 : if (semPostRet == -1) {
243 2 : AICPUE_LOGE("sem_post error with message: %s", strerror(errno));
244 : }
245 10 : };
246 :
247 10 : taskQueue.push(aicpuClosure);
248 10 : }
249 :
250 5 : const uint32_t ret = splitKernelScheduler_(parallelId, cpuNums, taskQueue);
251 5 : if (ret != 0U) {
252 1 : AICPUE_LOGE("Submit hash split kernel task failed, ret=%u, parallelId=%u", ret, parallelId);
253 1 : (void)sem_destroy(&aicpuSem);
254 1 : return;
255 : }
256 :
257 4 : DoTaskItself(parallelId, cpuNumCounter, cpuNums);
258 :
259 12 : for (int64_t i = 0; i < cpuNums; i++) {
260 8 : const int32_t semWaitRet = sem_wait(&aicpuSem);
261 8 : if (semWaitRet == -1) {
262 4 : AICPUE_LOGE("sem_wait error with message: %s", strerror(errno));
263 : }
264 : }
265 4 : const int32_t semDesRet = sem_destroy(&aicpuSem);
266 4 : if (semDesRet == -1) {
267 1 : AICPUE_LOGE("sem_destroy error with message: %s", strerror(errno));
268 : }
269 :
270 4 : AICPUE_LOGI("Op hash parallel process finished. parallelId=%u", parallelId);
271 4 : return;
272 5 : }
273 :
274 7 : void SharderNonBlock::ParallelForHash(const int64_t total, const int64_t cpuNums, const SharderWork &work)
275 : {
276 7 : uint32_t parallelId = 0U;
277 : {
278 7 : const std::lock_guard<std::mutex> lk(parallelIdMutex_);
279 7 : ++parallelId_;
280 7 : parallelId = parallelId_.load();
281 7 : }
282 :
283 7 : if ((total <= 0) || (work == nullptr)) {
284 0 : AICPUE_LOGE("invalid param: total<=0 or work is nullptr");
285 0 : return;
286 : }
287 :
288 7 : if ((splitKernelScheduler_ == nullptr) || (splitKernelGetProcesser_ == nullptr) || (cpuCoreNum_ <= 1U)) {
289 1 : AICPUE_LOGE("schedule is nullptr or cpu core num is not enough");
290 1 : return;
291 : }
292 :
293 6 : ExecuteParallelForHash(total, cpuNums, work, parallelId);
294 6 : return;
295 : }
296 : }
297 :
298 : /**
299 : * Shards the "total" unit of work refer "perUintSize"
300 : */
301 38 : void ParallelFor(int64_t total, int64_t perUnitSize, const aicpu::SharderWork &work)
302 : {
303 38 : aicpu::SharderNonBlock::GetInstance().ParallelFor(total, perUnitSize, work);
304 38 : }
305 :
306 : /**
307 : * Get CPU number
308 : */
309 1 : uint32_t GetCPUNum()
310 : {
311 1 : return aicpu::SharderNonBlock::GetInstance().GetCPUNum();
312 : }
|
