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_kfc_batchwrite_process.h"
12 :
13 : #include "common/aicpu_hccl_common.h"
14 : #include "utils/hccl_aicpu_utils.h"
15 : #include "framework/aicpu_kfc_prof.h"
16 : #include "coll_batch_write_executor.h"
17 :
18 : using namespace hccl;
19 :
20 : ANONYMOUS_NAMESPACE_BEGIN
21 : class CommonHcclMsgRingBuffer {
22 : public:
23 : static constexpr uint8_t DEFAULT_CAPACITY = 4;
24 :
25 43 : CommonHcclMsgRingBuffer() : CommonHcclMsgRingBuffer(DEFAULT_CAPACITY)
26 43 : {}
27 :
28 43 : CommonHcclMsgRingBuffer(uint8_t capacity) : capacity_(capacity)
29 : {
30 43 : if (capacity > 0) {
31 43 : buffer_ = new CommonHcclMsg[capacity_];
32 : }
33 43 : }
34 :
35 43 : ~CommonHcclMsgRingBuffer()
36 : {
37 43 : if (capacity_ > 0 && buffer_ != nullptr) {
38 43 : delete[] buffer_;
39 43 : buffer_ = nullptr;
40 43 : capacity_ = 0;
41 : }
42 43 : }
43 :
44 1 : bool Enqueue(const CommonHcclMsg *msg)
45 : {
46 1 : if (capacity_ == 0) {
47 0 : HCCL_ERROR("capacity is zero");
48 0 : return false;
49 : }
50 1 : uint32_t curTail = tail_.load(std::memory_order_acquire);
51 1 : uint32_t nextTail = (curTail + 1) % capacity_;
52 2 : if (nextTail == head_.load(std::memory_order_acquire)) {
53 0 : HCCL_INFO("CommonHcclMsgRingBuffer queue is full.");
54 0 : return false;
55 : }
56 1 : s32 sRet = memcpy_s(&buffer_[curTail], sizeof(CommonHcclMsg), msg, sizeof(CommonHcclMsg));
57 1 : if (sRet != EOK) {
58 0 : HCCL_ERROR("memcpy_s failed, errorno[%d]", sRet);
59 0 : return false;
60 : }
61 1 : tail_.store(nextTail, std::memory_order_release);
62 1 : return true;
63 : }
64 :
65 1079 : bool Peek(CommonHcclMsg *msg)
66 : {
67 : uint32_t tempIdx;
68 : do {
69 1079 : tempIdx = head_.load(std::memory_order_acquire);
70 2158 : if (tempIdx == tail_.load(std::memory_order_acquire)) {
71 1078 : return false;
72 : }
73 1 : s32 sRet = memcpy_s(msg, sizeof(CommonHcclMsg), &buffer_[tempIdx], sizeof(CommonHcclMsg));
74 1 : if (sRet != EOK) {
75 0 : HCCL_ERROR("memcpy_s failed, errorno[%d]", sRet);
76 0 : return false;
77 : }
78 2 : } while (tempIdx != head_.load(std::memory_order_acquire)); // 确保在读取过程中head没被修改
79 1 : return true;
80 : }
81 :
82 2 : bool Dequeue()
83 : {
84 2 : uint32_t curHead = head_.load(std::memory_order_acquire);
85 4 : if (curHead == tail_.load(std::memory_order_acquire) || capacity_ == 0) {
86 1 : HCCL_INFO("CommonHcclMsgRingBuffer queue is empty.");
87 1 : return false;
88 : }
89 1 : head_.store((curHead + 1) % capacity_, std::memory_order_release);
90 1 : return true;
91 : }
92 :
93 3 : void Clear()
94 : {
95 3 : head_.store(0, std::memory_order_release);
96 3 : tail_.store(0, std::memory_order_release);
97 3 : }
98 :
99 : private:
100 : uint8_t capacity_{0};
101 : std::atomic<uint32_t> head_{0};
102 : std::atomic<uint32_t> tail_{0};
103 : CommonHcclMsg *buffer_{nullptr};
104 : };
105 :
106 : struct BatchWriteItem {
107 : uint64_t localBuf;
108 : uint64_t remoteBuf;
109 : uint64_t count;
110 : uint32_t dataType;
111 : uint32_t remoteRankId;
112 : };
113 : WqeSendSharedContect g_sharedCtx;
114 : CommonHcclMsgRingBuffer g_hcclMsgQueue;
115 : constexpr s32 PREFER_CLUSTER_ID = 0;
116 : constexpr u32 DELAY_TIME_IN_NS = 15U * 1000U;
117 : static constexpr uint64_t WQE_SEND_TIMEOUT = 15;
118 : std::mutex g_mtxForCpuCheck;
119 : #ifdef CCL_LLT
120 : // mock GetCpuId 多个线程需要放回不同的值，mock组件在多线程时不安全，会放回错误。所以在跑llt时加锁。
121 : std::mutex g_mtxForLLT;
122 : #endif
123 :
124 3 : HcclResult ConcurrentPostSendWqe(const CommonHcclMsg &commonHcclMsg, const AicpuComContext *ctx, u8 *needSendTotalNum) {
125 3 : const BatchWriteItem *item = reinterpret_cast<BatchWriteItem *>(static_cast<uintptr_t>(commonHcclMsg.sendBuffer));
126 6 : std::vector <Transport::Buffer> remoteList = {{}};
127 6 : std::vector <Transport::Buffer> local = {{}};
128 3 : int32_t cpuId = 0;
129 : {
130 : #ifdef CCL_LLT
131 3 : std::lock_guard<std::mutex> lock(g_mtxForLLT);
132 : #endif
133 3 : cpuId = HcclAicpuUtils::GetCpuId();
134 3 : }
135 3 : u32 threadId = g_sharedCtx.curThreadIdsOnCpu[cpuId];
136 3 : u32 sendWqeNum = 0;
137 70 : for (u64 i = 0; i < commonHcclMsg.dataCnt; ++i) {
138 67 : if (item->remoteRankId != ctx->rankId) {
139 64 : (*needSendTotalNum)++;
140 64 : if (item->remoteRankId % g_sharedCtx.workedThreadNum == threadId) {
141 33 : remoteList[0].addr = reinterpret_cast<void *>(item->remoteBuf);
142 33 : local[0].addr = reinterpret_cast<void *>(item->localBuf);
143 33 : remoteList[0].size = local[0].size =
144 33 : item->count * DataUnitSize(static_cast<HcclDataType>(item->dataType));
145 33 : HCCL_INFO(
146 : "Batch write item[%u]: context rankId [%u], remoteRankId[%u], sendThreadId[%ld], remoteBuf[%#llx],"
147 : " localBuf[%#llx], dataType[%u], count[%lu]",
148 : i,
149 : ctx->rankId,
150 : item->remoteRankId,
151 : threadId,
152 : item->remoteBuf,
153 : item->localBuf,
154 : item->dataType,
155 : item->count);
156 33 : CHK_RET(HcclAicpuUtils::PostSend(*ctx, item->remoteRankId, remoteList, local, true));
157 33 : sendWqeNum++;
158 : }
159 : }
160 67 : ++item;
161 : }
162 3 : g_sharedCtx.sendWqeNum[threadId] = sendWqeNum;
163 3 : HCCL_INFO("thread %u send %u wqe success.", threadId, sendWqeNum);
164 3 : return HCCL_SUCCESS;
165 3 : }
166 :
167 9 : bool CheckTimeOut(u64 startTimeStamp, u64 timeOutTime) {
168 9 : if ((GetCurCpuTimestamp() - startTimeStamp) > static_cast<unsigned long long>(NSEC_PER_SEC * timeOutTime)) {
169 0 : HCCL_ERROR("Execution TimeOut %lus...", timeOutTime);
170 0 : return true;
171 : }
172 9 : return false;
173 : }
174 :
175 2 : HcclResult WaitForSlaveCompletion(u8 needSendTotalNum) {
176 2 : HCCL_DEBUG("needsendTotalNum is %ld.", needSendTotalNum);
177 2 : u64 startTimeStamp = GetCurCpuTimestamp();
178 : while (true) {
179 11 : uint32_t sendNum = 0;
180 32 : for (uint32_t i = 0; i < g_sharedCtx.workedThreadNum; ++i) {
181 21 : HCCL_DEBUG("wait thread %ld send %ld wqe success.", i, g_sharedCtx.sendWqeNum[i]);
182 21 : sendNum += g_sharedCtx.sendWqeNum[i];
183 : }
184 11 : if (needSendTotalNum <= sendNum) {
185 5 : for (uint32_t i = 0; i < g_sharedCtx.workedThreadNum; ++i) {
186 3 : g_sharedCtx.sendWqeNum[i] = 0U;
187 : }
188 2 : HCCL_INFO("needsendTotalNum is %ld, already send %ld", needSendTotalNum, sendNum);
189 2 : return HCCL_SUCCESS;
190 : }
191 9 : if (CheckTimeOut(startTimeStamp, WQE_SEND_TIMEOUT)) {
192 0 : g_sharedCtx.taskFinishFlag.store(true, std::memory_order_release);
193 0 : HCCL_ERROR("slave thread send wqe timeout.");
194 0 : return HCCL_E_TIMEOUT;
195 : }
196 9 : }
197 : }
198 :
199 3 : void InitMultiThreadSharedCtx(int32_t cpuId) {
200 3 : g_sharedCtx.startedThreadNum = 1;
201 3 : g_hcclMsgQueue.Clear();
202 3 : g_sharedCtx.taskFinishFlag.store(false, std::memory_order_release);
203 3 : g_sharedCtx.curThreadIdsOnCpu[cpuId] = 0;
204 3 : g_sharedCtx.sendWqeNum[0] = 0;
205 27 : for (s32 i = 0; i < AICPU_CNT; ++i) {
206 24 : g_sharedCtx.curThreadIdsOnCpu[i] = 0;
207 : }
208 3 : }
209 :
210 1 : HcclResult OrchestrateSdmaSqe(const OpParam ¶m, hccl::HcclCommAicpu &comm)
211 : {
212 1 : AicpuKfcProf::SetKfcTimeLine(KfcTimeLine::HCC_EXEC_START_TIME);
213 1 : const u32 queueIdx = param.BatchWriteDataDes.queueIdx;
214 1 : auto streams = comm.GetSlaveStream();
215 1 : CHK_PRT_RET(queueIdx >= streams.size(),
216 : HCCL_ERROR("Invalid queue idx %u, stream number %u", queueIdx, streams.size()), HCCL_E_PARA);
217 1 : auto streamInfo = streams[queueIdx];
218 1 : u8 *newSqAddr = static_cast<u8 *>(param.inputPtr);
219 1 : auto &sqeBuffer = streamInfo.GetSqeContextPtr()->buffer;
220 1 : u16 &taskId = sqeBuffer.tailSqeTaskId;
221 1 : const u32 sqeCnt = param.BatchWriteDataDes.itemNum;
222 1 : const u32 depth = streamInfo.GetHcclStreamInfo().sqDepth;
223 1 : CHK_PRT_RET(sqeCnt >= depth, HCCL_ERROR("Sqe count %u reaches the sq depth %u.", sqeCnt, depth), HCCL_E_PARA);
224 : u8 sqeType;
225 2 : for (u32 i = 0U; i < sqeCnt; ++i) {
226 1 : const uint8_t *sqe = newSqAddr + i * AC_SQE_SIZE;
227 1 : AddOneMemcpySqeV1(streamInfo.id(), taskId++, nullptr, 0U, ACL_DT_UNDEFINED, ACL_RT_MEMCPY_SDMA_AUTOMATIC_SUM,
228 : nullptr, 0U, 0U, 0U, 0U, static_cast<uint8_t>(LinkType::LINK_RESERVED), sqe, &sqeType, SDMA_QOS_DEFAULT);
229 : }
230 :
231 1 : u32 &head = sqeBuffer.sqHead;
232 1 : u32 &tail = sqeBuffer.sqTail;
233 1 : u32 newTail = (tail + sqeCnt) % depth;
234 1 : HCCL_INFO("Before send sqe:%d cnt:%u head:%u curtail:%u newTail:%u.", streamInfo.sqId(),
235 : sqeCnt, head, tail, newTail);
236 1 : const u64 startUsec = GetCurCpuTimestamp();
237 1 : const u32 devId = comm.GetDevId();
238 1 : while ((tail + depth - head) % depth + sqeCnt >= depth) {
239 0 : CHK_RET(QuerySqStatusByType(devId, streamInfo.sqId(), DRV_SQCQ_PROP_SQ_HEAD, head));
240 0 : if (GetCurCpuTimestamp() - startUsec > NSEC_PER_SEC * dfx::kKfcTimeOut) {
241 0 : HCCL_ERROR("Rtsq(%u) full for more than %u seconds, head:%u.", streamInfo.sqId(), dfx::kKfcTimeOut, head);
242 0 : return HCCL_E_INTERNAL;
243 : }
244 : }
245 :
246 1 : u8 *sqAddr = static_cast<u8 *>(streamInfo.GetHcclStreamInfo().sqBaseAddr);
247 1 : const u32 left = depth - tail;
248 1 : HCCL_INFO("Before copy sqe:%d cnt:%u head:%u curtail:%u newTail:%u left:%u", streamInfo.sqId(),
249 : sqeCnt, head, tail, newTail, left);
250 1 : if (sqeCnt <= left) {
251 0 : (void)memcpy_s(sqAddr + tail * AC_SQE_SIZE, left * AC_SQE_SIZE, newSqAddr, sqeCnt * AC_SQE_SIZE);
252 : } else {
253 1 : (void)memcpy_s(sqAddr + tail * AC_SQE_SIZE, left * AC_SQE_SIZE, newSqAddr, left * AC_SQE_SIZE);
254 1 : (void)memcpy_s(sqAddr, head * AC_SQE_SIZE, newSqAddr + left * AC_SQE_SIZE, (sqeCnt - left) * AC_SQE_SIZE);
255 : }
256 : #ifdef __aarch64__
257 : __asm__ __volatile__("dsb st" : : : "memory");
258 : #endif
259 1 : if (UNLIKELY(HcclCheckLogLevel(DLOG_DEBUG))) {
260 1 : rtStarsMemcpyAsyncSqe_t *tmp = reinterpret_cast<rtStarsMemcpyAsyncSqe_t *>(sqAddr) + tail;
261 1 : for (u32 i = tail; i < newTail; ++i) {
262 0 : HCCL_DEBUG("[Sdma-BatchWrite]Orchestrated sq %u, idx %u, stream %u, task %u, data length %u, "
263 : "src addr %#llx, dst addr %#llx.", streamInfo.sqId(), i, tmp->header.rtStreamId,
264 : tmp->header.taskId, tmp->length,
265 : (static_cast<uint64_t>(tmp->src_addr_high) << 32U) | tmp->src_addr_low,
266 : (static_cast<uint64_t>(tmp->dst_addr_high) << 32U) | tmp->dst_addr_low);
267 0 : ++tmp;
268 : }
269 : }
270 :
271 1 : AicpuKfcProf::SetKfcTimeLine(KfcTimeLine::SEND_TASK_START_TIME);
272 1 : CHK_RET(ConfigSqStatusByType(devId, streamInfo.sqId(), DRV_SQCQ_PROP_SQ_TAIL, newTail));
273 1 : tail = newTail;
274 1 : AicpuKfcProf::SetKfcTimeLine(KfcTimeLine::SEND_SQE_FINISH_TIME);
275 1 : return HCCL_SUCCESS;
276 1 : }
277 : ANONYMOUS_NAMESPACE_END
278 :
279 2 : void AicpuKfcBatchwriteProcess::FinishProcess()
280 : {
281 2 : HCCL_INFO("master over task is finish.");
282 2 : g_sharedCtx.taskFinishFlag.store(true, std::memory_order_release);
283 2 : }
284 :
285 6 : AicpuServerRole AicpuKfcBatchwriteProcess::GetVerifiedServerRole(const AicpuComContext &ctx)
286 : {
287 6 : if (!ctx.multiServerFlag) {
288 2 : HCCL_INFO("Skip server start check for non-multi server scene.");
289 2 : return AicpuServerRole::MASTER;
290 : }
291 :
292 : static std::atomic<u32> opThreadIdx{0U};
293 4 : if (HcclAicpuUtils::GetCurClusterId() != PREFER_CLUSTER_ID) {
294 1 : u64 startTimestamp = GetCurCpuTimestamp();
295 2 : while (opThreadIdx.load(std::memory_order_acquire) == 0U &&
296 0 : GetCurCpuTimestamp() - startTimestamp < DELAY_TIME_IN_NS) {
297 0 : usleep(1);
298 : }
299 : }
300 :
301 4 : std::lock_guard<std::mutex> lock(g_mtxForCpuCheck);
302 4 : int32_t cpuId = HcclAicpuUtils::GetCpuId();
303 : AicpuServerRole role;
304 4 : if (opThreadIdx.fetch_add(1U, std::memory_order_acq_rel) == 0U) {
305 2 : InitMultiThreadSharedCtx(cpuId);
306 2 : HCCL_INFO("Master thread starts on cpu %d, clusterID %d", cpuId, HcclAicpuUtils::GetCurClusterId());
307 2 : role = AicpuServerRole::MASTER;
308 : } else {
309 3 : if (HcclAicpuUtils::GetCurClusterId() != PREFER_CLUSTER_ID ||
310 1 : g_sharedCtx.startedThreadNum >= MAX_BATCH_WRITE_THREAD_NUM) {
311 1 : HCCL_INFO("This is invalid thread, cluster id %d, started thread number %ld.",
312 : HcclAicpuUtils::GetCurClusterId(), g_sharedCtx.startedThreadNum);
313 1 : role = AicpuServerRole::INVALID;
314 : } else {
315 1 : g_sharedCtx.sendWqeNum[g_sharedCtx.startedThreadNum] = 0;
316 1 : g_sharedCtx.curThreadIdsOnCpu[cpuId] = g_sharedCtx.startedThreadNum++;
317 1 : HCCL_INFO("Slave thread index %u on cpu %d. clusterID %d",
318 : g_sharedCtx.curThreadIdsOnCpu[cpuId], cpuId, HcclAicpuUtils::GetCurClusterId());
319 1 : role = AicpuServerRole::SLAVE;
320 : }
321 : }
322 : // 老驱动包无法获取GetBlockNum，使用默认值6
323 4 : const u32 numBlocks = HcclAicpuUtils::GetBlockNum(6U);
324 4 : if (opThreadIdx.load(std::memory_order_acquire) == numBlocks) {
325 2 : HCCL_INFO("Clear thread index at last with block dim %u.", numBlocks);
326 : opThreadIdx.store(0U, std::memory_order_relaxed);
327 : }
328 4 : return role;
329 4 : }
330 :
331 : // 真正处理BatchWrite master 从commonHcclMsg中取消息，更新工作线程数，放到队列中。 从队列中取数据进行发送。
332 2 : HcclResult AicpuKfcBatchwriteProcess::HandleBatchWriteOperation(const CommonHcclMsg &commonHcclMsg,
333 : const AicpuComContext *ctx) {
334 2 : if (commonHcclMsg.dataCnt == 0UL || commonHcclMsg.sendBuffer == 0UL) {
335 0 : HCCL_ERROR("Get msg send buffer is nullptr or dataCnt is zero. "
336 : "Msg[commType %u, opType %u, sendBuffer %p, dataCnt %lu]",
337 : static_cast<uint32_t>(commonHcclMsg.commType),
338 : static_cast<uint32_t>(commonHcclMsg.opType), commonHcclMsg.sendBuffer, commonHcclMsg.dataCnt);
339 0 : return HCCL_E_PARA;
340 : }
341 :
342 2 : g_sharedCtx.workedThreadNum = g_sharedCtx.startedThreadNum;
343 2 : if (g_sharedCtx.workedThreadNum > 1) {
344 1 : bool success = false;
345 2 : while (!success) {
346 1 : success = g_hcclMsgQueue.Enqueue(&commonHcclMsg);
347 : }
348 : }
349 2 : u8 needSendTotalNum = 0;
350 2 : CHK_RET(ConcurrentPostSendWqe(commonHcclMsg, ctx, &needSendTotalNum));
351 2 : HCCL_DEBUG("total need send wqe num is %u", needSendTotalNum);
352 2 : CHK_RET(WaitForSlaveCompletion(needSendTotalNum));
353 2 : g_hcclMsgQueue.Dequeue();
354 2 : return HCCL_SUCCESS;
355 : }
356 :
357 1 : HcclResult AicpuKfcBatchwriteProcess::RunSlaveRpcServerForApi(AicpuComContext *ctx)
358 : {
359 1 : HCCL_INFO("----------start Slave Rpc Server For Api Hccl, ctx:%p ----------", ctx);
360 1 : if (ctx->devType != DevType::DEV_TYPE_910B) {
361 0 : HCCL_WARNING("Platform not support multi thread handle batch write, please use 910B platform.");
362 0 : return HCCL_SUCCESS;
363 : }
364 : CommonHcclMsg commonHcclMsg;
365 1 : int32_t sendSeqNum = -1;
366 1 : u32 threadId = g_sharedCtx.curThreadIdsOnCpu[HcclAicpuUtils::GetCpuId()];
367 : while (true) {
368 : #if defined(__aarch64__) || defined(__amd64__)
369 287805651 : __asm__ __volatile__("nop");
370 : #endif
371 :
372 287805651 : if (g_sharedCtx.taskFinishFlag.load(std::memory_order_acquire)) {
373 1 : HCCL_INFO("task is finish, slave process exit");
374 1 : break;
375 : }
376 287805650 : u8 needSendTotalNum = 0;
377 287805650 : if (threadId < g_sharedCtx.workedThreadNum && g_hcclMsgQueue.Peek(&commonHcclMsg) && commonHcclMsg.seqNum != sendSeqNum) {
378 1 : if (commonHcclMsg.commType == HcclCMDType::HCCL_CMD_BATCH_WRITE) {
379 1 : CHK_RET(ConcurrentPostSendWqe(commonHcclMsg, ctx, &needSendTotalNum));
380 1 : sendSeqNum = commonHcclMsg.seqNum;
381 : }
382 : }
383 287805650 : }
384 1 : return HCCL_SUCCESS;
385 : }
386 :
387 2 : HcclResult AicpuKfcBatchwriteProcess::BatchWriteProcess(hccl::OpParam &opParam, hccl::HcclCommAicpu &comm,
388 : HcclOpResParam ¶m)
389 : {
390 : static hccl::AlgResourceResponse *algResResponse = nullptr;
391 2 : if (UNLIKELY(algResResponse == nullptr || algResResponse->slaveStreams.empty())) {
392 : const std::string tag =
393 2 : comm.GetGroupName() + std::to_string(static_cast<uint8_t>(HcclCMDType::HCCL_CMD_BATCH_WRITE)) +
394 7 : std::string("_mc2") + std::string(BATCH_WRITE_ALG_NAME) + std::string("_device");
395 1 : std::unique_ptr<hccl::CollExecutorBase> executor;
396 2 : CHK_RET(comm.GetAlgResponseRes(tag, BATCH_WRITE_ALG_NAME, opParam, ¶m, executor, algResResponse));
397 1 : }
398 2 : const u64 ts = GetCurCpuTimestamp();
399 2 : while (algResResponse->slaveStreams.empty()) {
400 0 : CHK_PRT_RET(GetCurCpuTimestamp() - ts > static_cast<u64>(NSEC_PER_SEC),
401 : HCCL_ERROR("[%s]Timeout during batchwrite initialization.", __func__),
402 : HCCL_E_INTERNAL);
403 : }
404 2 : HcclResult ret = OrchestrateSdmaSqe(opParam, comm);
405 2 : AicpuKfcProf::GetCurrentAicpuProf()->workCnt++;
406 2 : return ret;
407 : }
|
