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