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_rpc_serverv2.h"
12 :
13 : #include <numeric>
14 : #include "log_control.h"
15 : #include "common/aicpu_hccl_common.h"
16 : #include "utils/hccl_aicpu_utils.h"
17 : #include "common/aicpu_kfc_utils.h"
18 : #include "framework/aicpu_kfc_prof.h"
19 :
20 : using namespace HcclApi;
21 : static constexpr uint16_t TURN_LEFT_SHIFT_BIT = 16;
22 :
23 : HcclResult
24 20 : AicpuKfcRpcServerV2::Init(const HcclMC2WorkSpace& workspaceInfo, const HcclApi::Mc2InitTilingInner* tilingData)
25 : {
26 : // 为提升效率，workspace 必须512 对齐
27 20 : u64 addr = workspaceInfo.workSpace;
28 20 : if (addr & 0x1ff) {
29 20 : addr = (addr & (~((uint64_t)0x1ff))) + 0x200;
30 : }
31 20 : Reset();
32 20 : blockNum_ = HcclAicpuUtils::GetBlockNum();
33 20 : CHK_PRT_RET(blockNum_ == 0U, HCCL_ERROR("Invalid block number."), HCCL_E_INTERNAL);
34 20 : HCCL_INFO(
35 : "Align hcclmsgarea from %p to %p, block number %u, current block idx %u.", workspaceInfo.workSpace, addr,
36 : blockNum_, HcclAicpuUtils::GetBlockIdx());
37 20 : if (tilingData != nullptr && tilingData->queueNum > 0U) {
38 2 : totalQueueNum_ = tilingData->commBlockNum * tilingData->queueNum;
39 2 : CHK_PRT_RET(
40 : totalQueueNum_ > LOCAL_STREAM_MAX_NUM || blockNum_ > std::min(MAX_AICPU_NUM_BLOCKS, totalQueueNum_),
41 : HCCL_ERROR(
42 : "Invalid para, comm block %u, aicpu block %u, queue number %u.", tilingData->commBlockNum, blockNum_,
43 : tilingData->queueNum),
44 : HCCL_E_INTERNAL);
45 : } else {
46 18 : totalQueueNum_ = 0U;
47 : }
48 20 : hcclMsgArea_ = reinterpret_cast<HcclMsgArea*>(addr);
49 20 : turnNumAddr_ = addr + sizeof(HcclMsgArea);
50 20 : if (turnNumAddr_ + sizeof(u32) * TILING_TURN_MAX * HCCL_MAX_RANK_NUM_V2
51 20 : > workspaceInfo.workSpace + workspaceInfo.workSpaceSize) {
52 0 : HCCL_ERROR(
53 : "Turn number addr %#llx, space for turn number is %lu, the space after workspace %#llx will "
54 : "be overwritten.",
55 : turnNumAddr_, sizeof(u32) * TILING_TURN_MAX * HCCL_MAX_RANK_NUM_V2,
56 : workspaceInfo.workSpace + workspaceInfo.workSpaceSize);
57 0 : return HCCL_E_INTERNAL;
58 : }
59 20 : uint32_t* turnNums = reinterpret_cast<uint32_t*>(turnNumAddr_);
60 20 : std::iota(&turnNums[0], &turnNums[TILING_TURN_MAX * HCCL_MAX_RANK_NUM_V2], 0);
61 20 : tilingBaseAddr_ = reinterpret_cast<u64>(tilingData);
62 20 : return HCCL_SUCCESS;
63 : }
64 :
65 50 : void AicpuKfcRpcServerV2::GetLocalQueueRange(u32& start, u32& end)
66 : {
67 50 : if (blockNum_ == 0U || totalQueueNum_ == 0U) {
68 40 : start = end = 0U;
69 40 : return;
70 : }
71 10 : const u32 base = totalQueueNum_ / blockNum_;
72 10 : const u32 remainder = totalQueueNum_ % blockNum_;
73 10 : const u32 blockIdx = HcclAicpuUtils::GetBlockIdx();
74 10 : if (blockIdx < remainder) {
75 0 : start = blockIdx * base + blockIdx;
76 0 : end = start + base;
77 : } else {
78 10 : start = blockIdx * base + remainder;
79 10 : end = start + base - 1U;
80 : }
81 : }
82 :
83 20 : void AicpuKfcRpcServerV2::Reset()
84 : {
85 20 : (void)memset_s(msgPos_, sizeof(msgPos_), 0, sizeof(msgPos_));
86 20 : msgPosForKernel_ = 0;
87 1280 : for (int8_t i = 0; i < HCCL_MAX_HANDLE_ID; i++) {
88 1260 : handleIdMsgPosition_[i] = -1;
89 : }
90 20 : (void)memset_s(isFinalize_, sizeof(isFinalize_), 0, sizeof(isFinalize_));
91 20 : (void)memset_s(barrierFlags_, sizeof(barrierFlags_), 0, sizeof(barrierFlags_));
92 20 : (void)memset_s(barrierFinishCnt_, sizeof(barrierFinishCnt_), 0, sizeof(barrierFinishCnt_));
93 20 : const u64 ts = GetCurCpuTimestamp();
94 980 : for (u32 i = 0U; i < MAX_QUE_NUM; ++i) {
95 960 : prepareTime_[i] = ts;
96 : }
97 20 : eventPrintTurn_ = 1U;
98 20 : }
99 :
100 9 : void AicpuKfcRpcServerV2::SetMsgHandlePos(uint32_t msgPos, HcclHandle handleId)
101 : {
102 9 : if (handleId >= HCCL_MAX_HANDLE_ID || handleId < 0) {
103 0 : return;
104 : }
105 9 : handleIdMsgPosition_[handleId] = msgPos;
106 : }
107 :
108 7 : int32_t AicpuKfcRpcServerV2::GetMsgHandlePos(HcclHandle handleId)
109 : {
110 7 : if (handleId >= HCCL_MAX_HANDLE_ID || handleId < 0) {
111 0 : HCCL_ERROR("[GetMsgHandlePos] invalid handleId %d", handleId);
112 0 : return -1;
113 : }
114 7 : if (handleIdMsgPosition_[handleId] < 0) {
115 5 : HCCL_WARNING("[GetMsgHandlePos] invalid handleIdMsgPosition %d", handleIdMsgPosition_[handleId]);
116 5 : return -1;
117 : }
118 2 : return handleIdMsgPosition_[handleId];
119 : }
120 :
121 1 : bool AicpuKfcRpcServerV2::IsPrintLog() const { return isPrintLog_; }
122 :
123 61 : bool AicpuKfcRpcServerV2::GetIsFinalize(u32 queueId)
124 : {
125 61 : if (queueId < MAX_QUE_NUM) {
126 28 : return isFinalize_[queueId];
127 : }
128 33 : u32 start = 0U;
129 33 : u32 end = 0U;
130 33 : GetLocalQueueRange(start, end);
131 43 : for (u32 i = start; i <= end; ++i) {
132 34 : if (!isFinalize_[i]) {
133 24 : return false;
134 : }
135 : }
136 9 : return true;
137 : }
138 :
139 10 : void AicpuKfcRpcServerV2::SetIsFinalize(u32 queueId, bool finalize) { isFinalize_[queueId] = finalize; }
140 :
141 7 : HcclMsgExt* AicpuKfcRpcServerV2::GetHcclMsgExtPtr() { return msgExt_.get(); }
142 :
143 0 : HcclMsgArea* AicpuKfcRpcServerV2::GetHcclMsgArea(void) { return hcclMsgArea_; }
144 :
145 29 : HcclMsg (*AicpuKfcRpcServerV2::GetMsgWorkSpace())[HCCL_MSG_CNT]
146 : {
147 29 : if (totalQueueNum_ == 0U) {
148 24 : return &(hcclMsgArea_->commMsg.singleMsg.sendMsgs);
149 : } else {
150 5 : return hcclMsgArea_->commMsg.multiMsg.sendMsgs;
151 : }
152 : }
153 :
154 7 : uint64_t AicpuKfcRpcServerV2::GetFinishAddr(int32_t idx) const
155 : {
156 7 : if (idx >= static_cast<int32_t>(HCCL_MSG_CNT) || hcclMsgArea_ == nullptr) {
157 3 : HCCL_ERROR("idx %d exceed max or msg area is not initialized.", idx);
158 3 : return 0;
159 : }
160 4 : return reinterpret_cast<uint64_t>(&(hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[idx].cnt));
161 : }
162 :
163 14 : uint64_t AicpuKfcRpcServerV2::GetCommitareaAddr(int32_t idx) const
164 : {
165 14 : if (idx >= static_cast<int32_t>(HCCL_MSG_CNT) || hcclMsgArea_ == nullptr) {
166 3 : HCCL_ERROR("idx %d exceed max or hcclMsgArea_ is not initialized.", idx);
167 3 : return 0;
168 : }
169 11 : return reinterpret_cast<uint64_t>(&(hcclMsgArea_->commMsg.singleMsg.commitTurnCnt[idx].cnt));
170 : }
171 :
172 12 : HcclResult AicpuKfcRpcServerV2::AddFlipTask(HcclDispatcher dispatcherPtr, hccl::Stream* stream)
173 : {
174 12 : if (!dfx::ProfilingManager::GetProfL0State()) {
175 12 : return HCCL_SUCCESS;
176 : }
177 0 : hccl::HcclSqeContext* sqeCtx = stream->GetSqeContextPtr();
178 0 : CHK_PTR_NULL(sqeCtx);
179 0 : hccl::SqeRingBuffer& buff = sqeCtx->buffer;
180 : // nextTaskId=0的时候下发PlaceHolder
181 0 : if (UNLIKELY(buff.tailSqeTaskId == 0 && buff.filpNum != 0)) {
182 0 : CHK_RET(AddRetryPreamble(dispatcherPtr, *stream));
183 : }
184 :
185 0 : return HCCL_SUCCESS;
186 : }
187 :
188 9 : HcclResult AicpuKfcRpcServerV2::AddCcoreWait(
189 : HcclDispatcher dispatcherPtr, u64 waitAddr, uint32_t turnNum, hccl::Stream* stream,
190 : bool isLast) // client commit wait
191 : {
192 9 : uint8_t* sqeBuffer = nullptr;
193 9 : uint8_t* sqeTypeAddr = nullptr;
194 9 : uint8_t* sqeDfxInfoAddr = nullptr;
195 9 : uint16_t taskId = 0U;
196 :
197 9 : CHK_RET(AddFlipTask(dispatcherPtr, stream));
198 9 : CHK_RET(stream->GetNextSqeBufferAddr(sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
199 9 : const HcclComStreamInfo& streamInfo = stream->GetHcclStreamInfo();
200 9 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_COMMIT_TIMEOUT)) {
201 0 : uint32_t* turnNums = reinterpret_cast<uint32_t*>(turnNumAddr_);
202 0 : turnNums[turnNum] = 0xFF;
203 : }
204 9 : AddOneWaitStartSqe(
205 9 : streamInfo.actualStreamId, taskId, waitAddr, turnNumAddr_ + turnNum * sizeof(u32), isLast,
206 : reinterpret_cast<rtStarsCcoreWaitStartSqe_t*>(sqeBuffer), sqeTypeAddr);
207 9 : hccl::HcclSqeContext* sqeCtx = stream->GetSqeContextPtr();
208 9 : if (sqeCtx == nullptr) {
209 0 : HCCL_ERROR("AddCcoreWait sqeCtx is nullptr");
210 0 : return HCCL_E_INTERNAL;
211 : }
212 9 : sqeCtx->buffer.addInfo[taskId % hccl::HCCL_SQE_MAX_CNT]
213 9 : = ((turnNum << TURN_LEFT_SHIFT_BIT) + static_cast<uint32_t>(isLast));
214 9 : return HCCL_SUCCESS;
215 : }
216 :
217 3 : HcclResult AicpuKfcRpcServerV2::AddCcoreNotify(
218 : HcclDispatcher dispatcherPtr, u64 recordAddr, uint32_t turnNum,
219 : hccl::Stream* stream) // client finish notify
220 : {
221 3 : uint8_t* sqeBuffer = nullptr;
222 3 : uint8_t* sqeTypeAddr = nullptr;
223 3 : uint8_t* sqeDfxInfoAddr = nullptr;
224 3 : uint16_t taskId = 0U;
225 :
226 3 : CHK_RET(AddFlipTask(dispatcherPtr, stream));
227 3 : CHK_RET(stream->GetNextSqeBufferAddr(sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
228 3 : const HcclComStreamInfo& streamInfo = stream->GetHcclStreamInfo();
229 3 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_AICORE_WAIT_TIMEOUT)) {
230 0 : uint32_t* turnNums = reinterpret_cast<uint32_t*>(turnNumAddr_);
231 0 : turnNums[turnNum] = 0;
232 : }
233 3 : AddOneWriteValueStartSqe(
234 3 : streamInfo.actualStreamId, taskId, recordAddr, turnNumAddr_ + turnNum * sizeof(u32),
235 : reinterpret_cast<rtStarsCcoreWriteValueSqe_t*>(sqeBuffer), sqeTypeAddr);
236 3 : hccl::HcclSqeContext* sqeCtx = stream->GetSqeContextPtr();
237 3 : if (sqeCtx == nullptr) {
238 3 : HCCL_ERROR("AddCcoreNotify sqeCtx is nullptr");
239 3 : return HCCL_E_INTERNAL;
240 : }
241 0 : sqeCtx->buffer.addInfo[taskId % hccl::HCCL_SQE_MAX_CNT] = turnNum;
242 0 : return HCCL_SUCCESS;
243 : }
244 :
245 3 : uint64_t AicpuKfcRpcServerV2::GetFinishAddrByHandleId(HcclHandle handleId)
246 : {
247 3 : int32_t msgPos = GetMsgHandlePos(handleId);
248 3 : if (msgPos < 0) {
249 2 : return 0;
250 : }
251 1 : return GetFinishAddr(msgPos);
252 : }
253 :
254 12 : void AicpuKfcRpcServerV2::SetMsgRepeatCnt(u8 repeatCnt)
255 : {
256 12 : repeatCnt_[msgPos_[0U]] = (totalStep_ == 0U ? repeatCnt : repeatCnt * totalStep_);
257 12 : }
258 :
259 4 : int32_t AicpuKfcRpcServerV2::GetMsgRepeatCnt(HcclHandle handleId)
260 : {
261 4 : int32_t msgPos = GetMsgHandlePos(handleId);
262 4 : if (msgPos < 0) {
263 3 : return -1;
264 : }
265 1 : return repeatCnt_[msgPos];
266 : }
267 :
268 12 : HcclResult AicpuKfcRpcServerV2::ProcessExpectPrepareMsg(uint8_t seqNum, uint8_t expectId)
269 : {
270 : // 当前无翻转场景，只需考虑单个通信域提前Finializa场景
271 12 : if (seqNum == 0 && expectId > 0) {
272 1 : return HCCL_SUCCESS;
273 : }
274 11 : if (seqNum < expectId) {
275 1 : HCCL_ERROR("curMsg seqNum %d is smaller than expect %d ignore.", seqNum, expectId);
276 1 : return HCCL_E_INTERNAL;
277 : }
278 10 : if (totalQueueNum_ == 0U && seqNum > expectId) {
279 1 : HCCL_INFO("curMsg seqNum %d is bigger than expect %d ignore.", seqNum, expectId);
280 1 : return HCCL_E_UNAVAIL;
281 : }
282 9 : return HCCL_SUCCESS;
283 : }
284 :
285 15 : void AicpuKfcRpcServerV2::SetNeedRetryFlag(bool needRetryFlag) { needReProcess_ = needRetryFlag; }
286 :
287 4 : bool AicpuKfcRpcServerV2::ReadValidMsgExtArea(int32_t idx, u32 rankSize)
288 : {
289 : #ifdef __aarch64__
290 : __asm__ __volatile__("dsb ld" : : : "memory");
291 : #endif
292 : #ifdef __amd64__
293 4 : __asm__ __volatile__("" : : : "memory");
294 : #endif
295 4 : auto& extMsgList = hcclMsgArea_->commMsg.singleMsg.paramExtMsgList[idx];
296 4 : if (hcclMsgArea_ == nullptr || extMsgList.valid != static_cast<u64>(HCCL_MSG_VALID_MASK)) {
297 2 : return false;
298 : }
299 2 : uint64_t msgExtXorCheck = AicpuKfcUtils::GenXor(&extMsgList, rankSize);
300 : static uint32_t msgExtXorCheckTurn = 0;
301 2 : if (UNLIKELY(msgExtXorCheck != extMsgList.xorCheck)) {
302 1 : if (msgExtXorCheckTurn++ % MC2_API_XORCHECK_PRINT_NUM == 0) {
303 1 : HCCL_RUN_INFO(
304 : "Extend data is modified! modified_xor:%llu, origin_xor:%llu.", msgExtXorCheck, extMsgList.xorCheck);
305 : }
306 1 : return false;
307 : }
308 1 : HCCL_INFO("hcclMsgArea xorCheck[%llu]", extMsgList.xorCheck);
309 1 : const size_t copySize = sizeof(uint64_t) * rankSize;
310 1 : (void)memcpy_s(msgExt_->sendCounts, copySize, extMsgList.sendCounts, copySize);
311 1 : (void)memcpy_s(msgExt_->sendOffset, copySize, extMsgList.sendOffset, copySize);
312 1 : (void)memcpy_s(msgExt_->recvCounts, copySize, extMsgList.recvCounts, copySize);
313 1 : (void)memcpy_s(msgExt_->recvOffset, copySize, extMsgList.recvOffset, copySize);
314 1 : (void)memcpy_s(
315 1 : msgExt_->reserved, sizeof(HcclMsgExt) - offsetof(HcclMsgExt, reserved), extMsgList.reserved,
316 : sizeof(HcclMsgExt) - offsetof(HcclMsgExt, reserved));
317 :
318 : #ifdef __aarch64__
319 : __asm__ __volatile__("dsb ld" : : : "memory");
320 : #endif
321 : #ifdef __amd64__
322 1 : __asm__ __volatile__("" : : : "memory");
323 : #endif
324 1 : extMsgList.valid = static_cast<u64>(~HCCL_MSG_VALID_MASK);
325 : #ifdef __aarch64__
326 : __asm__ __volatile__("dsb st" : : : "memory");
327 : #endif
328 1 : HCCL_INFO("reset paramExtMsgList valid value %lu", extMsgList.valid);
329 1 : return true;
330 : }
331 :
332 15 : bool AicpuKfcRpcServerV2::IsExceedLimit(HcclCMDType commType, u32 rankSize)
333 : {
334 15 : if (rankSize > HCCL_MAX_RANK_NUM_V2
335 0 : && (commType == HcclCMDType::HCCL_CMD_ALLTOALLV || commType == HcclCMDType::HCCL_CMD_ALLTOALL)) {
336 0 : HCCL_ERROR(
337 : "The number[%u] of ranks exceeds the 256p limit supported by the ALLTOALL/ALLTOALLV algorithm.", rankSize);
338 0 : return true;
339 : }
340 15 : return false;
341 : }
342 :
343 13 : bool AicpuKfcRpcServerV2::CheckDebugMode(HcclMsg* rMsg)
344 : {
345 13 : if (UNLIKELY(
346 : AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_PREPARE_TIMEOUT)
347 : && (rMsg->commType.msgType != ControlMsgType::HCCL_CMD_FINALIZE))) {
348 0 : return false;
349 : }
350 13 : if (UNLIKELY(
351 : AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_FINALIZE_TIMEOUT)
352 : && (rMsg->commType.msgType == ControlMsgType::HCCL_CMD_FINALIZE))) {
353 0 : return false;
354 : }
355 13 : return true;
356 : }
357 :
358 18 : bool AicpuKfcRpcServerV2::ReadValidMsg(HcclMsg* rMsg, HcclMsg* msg, bool needReProcess, uint32_t msgPos, u32 rankSize)
359 : {
360 : #if defined(__aarch64__)
361 : __asm__ __volatile__("dsb ld" : : : "memory");
362 : #elif defined(__amd64__)
363 18 : __asm__ __volatile__("" : : : "memory");
364 : #endif
365 : // 重处理消息
366 18 : if (needReProcess) {
367 0 : *rMsg = *msg;
368 0 : return true;
369 : }
370 18 : if (msg->addMsg.v0Msg.valid != HCCL_MSG_VALID_MASK) {
371 2 : return false;
372 : }
373 16 : if (memcpy_s(rMsg, sizeof(HcclMsg), msg, sizeof(HcclMsg)) != EOK) {
374 0 : HCCL_ERROR("rMsg memcpy_s failed!,size:%d", sizeof(HcclMsg));
375 0 : return false;
376 : }
377 16 : uint32_t msgXorCheck = AicpuKfcUtils::GenXor(rMsg);
378 : static uint32_t msgXorCheckTurn = 0;
379 16 : if (UNLIKELY(msgXorCheck != rMsg->addMsg.v0Msg.xorCheck)) {
380 2 : if (msgXorCheckTurn++ % MC2_API_XORCHECK_PRINT_NUM == 0) {
381 2 : AicpuKfcUtils::PrintMsg("Rcv src msg", *msg, true);
382 1 : AicpuKfcUtils::PrintMsg("Rcv dst msg", *rMsg, true);
383 1 : HCCL_RUN_INFO(
384 : "data is modified! modified_xor:%u, origin_xor:%u.", msgXorCheck, rMsg->addMsg.v0Msg.xorCheck);
385 : }
386 2 : return false;
387 : }
388 14 : if (UNLIKELY(IsExceedLimit(static_cast<HcclCMDType>(rMsg->commType.prepareType), rankSize))) {
389 0 : return false;
390 : }
391 14 : if (UNLIKELY(
392 : static_cast<HcclCMDType>(rMsg->commType.prepareType) == HCCL_CMD_ALLTOALLV
393 : && !ReadValidMsgExtArea(msgPos, rankSize))) {
394 1 : return false;
395 : }
396 13 : msg->addMsg.v0Msg.valid = ~HCCL_MSG_VALID_MASK;
397 13 : if (UNLIKELY(!CheckDebugMode(rMsg))) {
398 0 : return false;
399 : }
400 13 : HCCL_INFO("reset valid value 0x%x", msg->addMsg.v0Msg.valid);
401 13 : return true;
402 : }
403 :
404 17 : bool AicpuKfcRpcServerV2::ReadAddrMsg(HcclMsg* hcclMsg, HcclMsg* msgList, u32 queueIdx, u32 msgPos, u32 rankSize)
405 : {
406 17 : bool ret = ReadValidMsg(hcclMsg, &(msgList[msgPos]), needReProcess_, msgPos, rankSize);
407 17 : isPrintLog_ = false;
408 17 : if (LIKELY(ret)) {
409 13 : HCCL_DEBUG("read valid msg msgPos %u commType %u", msgPos, static_cast<uint32_t>(hcclMsg->commType.msgType));
410 13 : PrintMsg(hcclMsg, msgPos, rankSize);
411 : // Prepare 成功，打印耗时
412 13 : u64 prepareTime = GetCurCpuTimestamp();
413 13 : if (eventPrintTurn_ > 1) {
414 0 : HCCL_RUN_INFO(
415 : "[AicpuKfcRpcServerV2][ReadAddrMsg] Read HcclMsg[%u] cost[%llu ns]", msgPos,
416 : prepareTime - prepareTime_[queueIdx]);
417 : } else {
418 13 : HCCL_INFO(
419 : "[AicpuKfcRpcServerV2][ReadAddrMsg] Read HcclMsg[%u] cost[%llu ns]", msgPos,
420 : prepareTime - prepareTime_[queueIdx]);
421 : }
422 13 : prepareTime_[queueIdx] = prepareTime;
423 13 : eventPrintTurn_ = 1;
424 4 : } else if (
425 4 : GetCurCpuTimestamp() - prepareTime_[queueIdx]
426 4 : > static_cast<unsigned long long>(NSEC_PER_SEC) * MC2_API_MSG_TIMEOUT * eventPrintTurn_) {
427 : // Prepare 等待 20s
428 0 : HCCL_RUN_WARNING(
429 : "[AicpuKfcRpcServerV2][ReadAddrMsg] ReadValidMsg[%u] timeout %lus", msgPos,
430 : MC2_API_MSG_TIMEOUT * eventPrintTurn_);
431 0 : eventPrintTurn_ *= 2; // 2 is print event log times
432 0 : LogControl logControl(false, true);
433 0 : PrintAllHcclMsgArea(rankSize);
434 0 : isPrintLog_ = true;
435 0 : }
436 17 : return ret;
437 : }
438 :
439 : // reset消息区msgPos的commitTurnId
440 8 : HcclResult AicpuKfcRpcServerV2::ResetCommitTaskAdd(HcclDispatcher dispatcherPtr, hccl::Stream* stream)
441 : {
442 : // reset函数复用 AddCcoreWait,turnNum保证条件算子恒成立
443 16 : for (uint32_t i = 0; i < static_cast<uint32_t>(msgPos_[0U]); i++) {
444 8 : uint64_t waitAddr = GetCommitareaAddr(i);
445 8 : CHK_RET(AddCcoreWait(dispatcherPtr, waitAddr, 0, stream, true));
446 : }
447 8 : return HCCL_SUCCESS;
448 : }
449 :
450 9 : void AicpuKfcRpcServerV2::WriteFinishWhenAllFinalize()
451 : {
452 9 : if (hcclMsgArea_ == nullptr || totalQueueNum_ != 0U) {
453 1 : return;
454 : }
455 8 : uint32_t msgPos = GetMsgPos();
456 8 : hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt = FINALIZE_FINISH_CNT; // 用于校验的非法值
457 8 : HCCL_INFO(
458 : "Post finishedTurnCnt[%u].cnt = %lu.", msgPos, hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt);
459 : #ifdef __aarch64__
460 : __asm__ __volatile__("dsb st" : : : "memory");
461 : #endif
462 : }
463 :
464 0 : void AicpuKfcRpcServerV2::WriteRestartFlag()
465 : {
466 0 : if (totalQueueNum_ == 0U) {
467 0 : for (uint32_t i = 0; i < HCCL_MSG_CNT; i++) {
468 0 : hcclMsgArea_->commMsg.singleMsg.sendMsgs[i].addMsg.v0Msg.valid = ~HCCL_MSG_VALID_MASK;
469 0 : hcclMsgArea_->commMsg.singleMsg.commitTurnCnt[i].cnt = 0;
470 0 : hcclMsgArea_->commMsg.singleMsg.finishedTurnCnt[i].cnt = 0;
471 : }
472 : } else {
473 0 : for (uint32_t i = 0; i < MAX_QUE_NUM; i++) {
474 0 : for (uint32_t j = 0; j < HCCL_MSG_CNT; j++) {
475 0 : hcclMsgArea_->commMsg.multiMsg.sendMsgs[i][j].addMsg.v1Msg.valid = ~HCCL_MSG_VALID_MASK;
476 : }
477 : }
478 : }
479 0 : hcclMsgArea_->controlMsg.restart = 1;
480 : #ifdef __aarch64__
481 : __asm__ __volatile__("dsb st" : : : "memory");
482 : #endif
483 0 : }
484 :
485 2 : void AicpuKfcRpcServerV2::PrintAllHcclMsgArea(u32 rankSize)
486 : {
487 2 : if (totalQueueNum_ == 0U) {
488 1 : AicpuKfcUtils::PrintAllHcclMsgArea(hcclMsgArea_, rankSize, true);
489 1 : } else if (HcclAicpuUtils::GetBlockIdx() == 0U) {
490 1 : AicpuKfcUtils::PrintAllHcclMsgAreaForMulti(hcclMsgArea_, true);
491 : }
492 2 : }
493 :
494 13 : void AicpuKfcRpcServerV2::PrintMsg(HcclMsg* hcclMsg, uint32_t msgPos, u32 rankSize)
495 : {
496 13 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_PRINT_MSG)) {
497 0 : AicpuKfcUtils::PrintMsg("ReadAddrMsg msgPos " + std::to_string(msgPos), *hcclMsg, true);
498 0 : if (totalQueueNum_ == 0U) {
499 0 : AicpuKfcUtils::PrintAllHcclMsgArea(hcclMsgArea_, rankSize);
500 : } else {
501 0 : AicpuKfcUtils::PrintAllHcclMsgAreaForMulti(hcclMsgArea_);
502 : }
503 : } else {
504 13 : AicpuKfcUtils::PrintMsg("ReadAddrMsg msgPos " + std::to_string(msgPos), *hcclMsg);
505 : }
506 13 : if (AicpuKfcProf::IsDebugModeEquals(MC2_DEBUG_PRINT_BUFF)) {
507 13 : AicpuKfcUtils::PrintApiBufferByMsgPos(*hcclMsg, msgPos);
508 : }
509 13 : }
510 :
511 10 : void AicpuKfcRpcServerV2::PrintAllHcclMsgAreaData()
512 : {
513 10 : if (totalQueueNum_ != 0U) {
514 2 : return;
515 : }
516 520 : for (uint32_t i = 0; i < HCCL_MSG_CNT; ++i) {
517 512 : AicpuKfcUtils::PrintApiBufferByMsgPos(hcclMsgArea_->commMsg.singleMsg.sendMsgs[i], i);
518 : }
519 : }
520 :
521 1 : void AicpuKfcRpcServerV2::DumpBarrierInfo(u32 groupIdx, u32 sqId, u32 devId)
522 : {
523 1 : const auto& barrierInfos = barrierFlags_[groupIdx];
524 3 : for (u32 i = 0U; i < totalQueueNum_; ++i) {
525 2 : const BarrierStatus status = barrierInfos[i].status;
526 2 : HCCL_ERROR("Queue:%u, msg pos:%u, status:%u.", i, GetMsgPos(i), static_cast<u32>(status));
527 2 : if (status == BarrierStatus::SELF_BARRIER) {
528 : u32 sqHead, sqTail;
529 0 : (void)QuerySqStatusByType(devId, sqId, DRV_SQCQ_PROP_SQ_HEAD, sqHead);
530 0 : (void)QuerySqStatusByType(devId, sqId, DRV_SQCQ_PROP_SQ_TAIL, sqTail);
531 0 : HCCL_ERROR("Queue:%u, sq head:%u, tail:%u.", i, sqHead, sqTail);
532 : }
533 : }
534 1 : }
|
