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 "comm_kfc_aicpu_server.h"
12 : #include <numeric>
13 : #include "log.h"
14 : #include "common/aicpu_kfc_utils.h"
15 : #include "hccl_mc2_ex.h"
16 :
17 : using namespace HcclApi;
18 : namespace {
19 : static constexpr u64 TIMEOUT_ERROR_THRESHOLD = 960UL;
20 : static const std::vector<HcclCMDType> SUPPORT_OP_LIST {
21 : HCCL_CMD_ALLREDUCE, HCCL_CMD_ALLGATHER, HCCL_CMD_REDUCE_SCATTER, HCCL_CMD_ALLTOALLV, HCCL_CMD_ALLTOALL
22 : };
23 :
24 277 : void FormatOpData(const HcclMsg &msg, HcclMsgExt &extMsg, u32 rankNum, u32 repeat, HcclOpData &data)
25 : {
26 277 : if (repeat == 0U) {
27 12 : data.opType = static_cast<HcclCMDType>(msg.commType.prepareType);
28 12 : data.reduceOp = static_cast<HcclReduceOp>(msg.opType);
29 12 : data.dataType = data.outputDataType = static_cast<HcclDataType>(msg.addMsg.v1Msg.hcclDataType);
30 12 : data.dataCount = msg.dataCnt;
31 12 : if (data.opType == HCCL_CMD_ALLTOALLV) {
32 2 : data.all2AllVDataDes.sendType = data.all2AllVDataDes.recvType = data.dataType;
33 2 : data.all2AllVDataDes.sendCounts = reinterpret_cast<void *>(reinterpret_cast<uintptr_t>(extMsg.sendCounts));
34 2 : data.all2AllVDataDes.recvCounts = reinterpret_cast<void *>(reinterpret_cast<uintptr_t>(extMsg.recvCounts));
35 2 : data.all2AllVDataDes.sdispls = reinterpret_cast<void *>(reinterpret_cast<uintptr_t>(extMsg.sendOffset));
36 2 : data.all2AllVDataDes.rdispls = reinterpret_cast<void *>(reinterpret_cast<uintptr_t>(extMsg.recvOffset));
37 10 : } else if (data.opType == HCCL_CMD_ALLTOALL) {
38 2 : data.all2AllDataDes.sendType = data.all2AllDataDes.recvType = data.dataType;
39 2 : data.all2AllDataDes.sendCount = data.all2AllDataDes.recvCount = data.dataCount;
40 : } else {
41 8 : data.dataDes.dataType = data.dataType;
42 8 : data.dataDes.dataCount = data.dataCount;
43 8 : data.dataDes.strideCount = msg.strideCount;
44 : }
45 265 : } else if (data.opType == HCCL_CMD_ALLTOALLV) {
46 6 : for (u32 i = 0U; i < rankNum; ++i) {
47 4 : extMsg.sendOffset[i] += extMsg.sendCounts[i];
48 4 : extMsg.recvOffset[i] += extMsg.recvCounts[i];
49 4 : HCCL_INFO("Formatted alltoallv info: repeat %u, rank id %u, send offset %llu, recv offset %llu.", repeat, i,
50 : static_cast<u64 *>(data.all2AllVDataDes.sdispls)[i],
51 : static_cast<u64 *>(data.all2AllVDataDes.rdispls)[i]);
52 : }
53 : }
54 277 : const u64 offset = data.dataCount * DataUnitSize(data.dataType);
55 277 : data.input = msg.sendBuffer + offset * repeat;
56 277 : data.output = msg.recvBuffer + offset * repeat;
57 277 : HCCL_INFO("Formatted op info: repeat index %u, op type %u, reduce type %u, data type %u, "
58 : "data count %llu, input addr %#llx, output addr %#llx.", static_cast<u32>(repeat),
59 : static_cast<u32>(data.opType), static_cast<u32>(data.reduceOp),
60 : static_cast<u32>(data.dataType), data.dataCount, data.input, data.output);
61 277 : }
62 : }
63 :
64 21 : HcclResult CommKfcAicpuServer::AddOpContext(const CommKfcContext *ctx)
65 : {
66 21 : CHK_PTR_NULL(ctx);
67 21 : if (ctxToOpHandle_.find(ctx->hcclContext) != ctxToOpHandle_.end()) {
68 0 : HCCL_INFO("Group %u: ctx %#llx is already added.", groupIdx_, ctx->hcclContext);
69 0 : return HCCL_SUCCESS;
70 : }
71 :
72 21 : CHK_PRT_RET(msgArea_ != nullptr && reinterpret_cast<u64>(msgArea_) != ctx->apiCtx.workSpace,
73 : HCCL_ERROR("Group %u: message area addr should be %#llx, not %#llx.",
74 : groupIdx_, msgArea_, ctx->apiCtx.workSpace),
75 : HCCL_E_PARA);
76 21 : void *opHandle = nullptr;
77 21 : HcclResult ret = HcclGetCommHandleByCtx(reinterpret_cast<void *>(ctx->hcclContext), &opHandle);
78 21 : CHK_PRT_RET(ret != HCCL_SUCCESS || opHandle == nullptr,
79 : HCCL_ERROR("Group %u: failed to get op handle by HCCL ctx %#llx.", groupIdx_, ctx->hcclContext),
80 : HCCL_E_PARA);
81 21 : ctxToOpHandle_[ctx->hcclContext] = opHandle;
82 21 : if (msgArea_ == nullptr) {
83 16 : msgArea_ = reinterpret_cast<HcclMsgArea *>(ctx->apiCtx.workSpace);
84 16 : turnNumsAddr_ = reinterpret_cast<u64>(msgArea_ + 1);
85 16 : rankNum_ = ctx->apiCtx.rankNum;
86 16 : std::iota(reinterpret_cast<u32 *>(turnNumsAddr_),
87 16 : reinterpret_cast<u32 *>(turnNumsAddr_) + UINT8_MAX + 1U, 0U);
88 16 : KeepAlive();
89 : }
90 21 : HCCL_INFO("Group %u: add op handle %#llx, HCCL context %#llx, message area address %#llx.",
91 : groupIdx_, opHandle, ctx->hcclContext, msgArea_);
92 21 : return HCCL_SUCCESS;
93 : }
94 :
95 14 : HcclResult CommKfcAicpuServer::Orchestrate(const HcclMsg &msg, HcclMsgExt &extMsg, u32 msgPos)
96 : {
97 14 : KeepAlive();
98 14 : CHK_PTR_NULL(msgArea_);
99 14 : auto handleIter = ctxToOpHandle_.find(reinterpret_cast<uintptr_t>(msg.addMsg.v1Msg.ccOpTilingData));
100 14 : CHK_PRT_RET(
101 : handleIter == ctxToOpHandle_.end(),
102 : HCCL_ERROR("Group %u: op handle %#llx is not added by host.", groupIdx_, msg.addMsg.v1Msg.ccOpTilingData),
103 : HCCL_E_PARA);
104 13 : const auto opIter = std::find(SUPPORT_OP_LIST.begin(), SUPPORT_OP_LIST.end(),
105 13 : static_cast<HcclCMDType>(msg.commType.prepareType));
106 13 : CHK_PRT_RET(opIter == SUPPORT_OP_LIST.end(),
107 : HCCL_ERROR("Unsupported comm type %u.", static_cast<u32>(msg.commType.prepareType)),
108 : HCCL_E_PARA);
109 12 : const HcclHandle handle = msg.addMsg.v1Msg.selfHandleID;
110 12 : CHK_PRT_RET(handle < 0, HCCL_ERROR("Group %u: invalid handle id %d.", groupIdx_, handle), HCCL_E_INTERNAL);
111 12 : const u32 repeatCnt = static_cast<u32>(msg.addMsg.v1Msg.repeatCnt);
112 12 : const u64 waitAddr = reinterpret_cast<u64>(&(msgArea_->commMsg.singleMsg.commitTurnCnt[msgPos].cnt));
113 12 : const u64 recordAddr = reinterpret_cast<u64>(&(msgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt));
114 :
115 12 : HcclOpData data{};
116 12 : void *opHandle = handleIter->second;
117 289 : for (u32 i = 0U; i < repeatCnt; ++i) {
118 277 : FormatOpData(msg, extMsg, rankNum_, i, data);
119 277 : const u32 turnIdx = i + 1U;
120 277 : CHK_RET(HcclLaunchCcoreWait(opHandle, waitAddr, turnIdx, turnNumsAddr_, turnIdx == repeatCnt));
121 277 : CHK_RET(HcclLaunchOp(opHandle, &data));
122 277 : CHK_RET(HcclLaunchCcorePost(opHandle, recordAddr, turnIdx, turnNumsAddr_));
123 : }
124 12 : SetMsgPosByHandle(handle, msgPos);
125 12 : SetRepeatByHandle(handle, repeatCnt);
126 12 : return HCCL_SUCCESS;
127 : }
128 :
129 6 : HcclResult CommKfcAicpuServer::Finalize(u32 msgPos)
130 : {
131 6 : KeepAlive();
132 6 : return HCCL_SUCCESS;
133 : }
134 :
135 1105736 : HcclResult CommKfcAicpuServer::IsAllTaskFinished(u32 msgPos, bool &isFinish)
136 : {
137 1105736 : CHK_PTR_NULL(msgArea_);
138 :
139 1105736 : isFinish = false;
140 : // opHandles_能保证不为空，同一个通信域检查任何一个ophandle即可
141 1105736 : void *firstHandle = ctxToOpHandle_.begin()->second;
142 1105736 : if (HcclCheckFinishByStream(firstHandle) != HCCL_SUCCESS) {
143 1105731 : return HCCL_SUCCESS;
144 : }
145 :
146 : HcclTaskStatus status;
147 5 : if (HcclGetTaskStatus(firstHandle, &status) != HCCL_SUCCESS || status != HcclTaskStatus::HCCL_NORMAL_STATUS) {
148 1 : HCCL_ERROR("Group %u: abnormal task status %u.", groupIdx_, static_cast<u32>(status));
149 1 : return HCCL_E_INTERNAL;
150 : }
151 :
152 4 : msgArea_->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt = FINALIZE_FINISH_CNT;
153 : #ifdef __aarch64__
154 : __asm__ __volatile__("dsb st" : : : "memory");
155 : #endif
156 4 : isFinish = true;
157 4 : HCCL_INFO("Group %u: all task is finished at message pos %u.", groupIdx_, msgPos);
158 13 : for (auto it: ctxToOpHandle_) {
159 9 : CHK_RET(HcclReleaseComm(it.second));
160 9 : HCCL_INFO("Group %u: Op handle %#llx is released, HCCL context %#llxx.", groupIdx_, it.second, it.first);
161 : }
162 4 : return HCCL_SUCCESS;
163 : }
164 :
165 3 : HcclResult CommKfcAicpuServer::InterGroupSync(const CommKfcAicpuServer &otherServer, HcclHandle handle)
166 : {
167 3 : KeepAlive();
168 : u32 msgPos, repeat;
169 3 : HcclResult ret = otherServer.GetServerInfoForSync(handle, msgPos, repeat);
170 3 : if (ret != HCCL_SUCCESS) {
171 2 : HCCL_INFO("Group %u: group sync info is not obtained, return code %u.", groupIdx_, ret);
172 2 : return ret;
173 : }
174 1 : CHK_PRT_RET(msgPos >= HCCL_MSG_CNT, HCCL_ERROR("Group %u: invalid message index %u.", groupIdx_, msgPos),
175 : HCCL_E_PARA);
176 :
177 1 : HcclMsgArea *msgArea = otherServer.GetMsgAreaAddr();
178 1 : CHK_PTR_NULL(msgArea);
179 1 : const u64 waitAddr = reinterpret_cast<u64>(&(msgArea->commMsg.singleMsg.finishedTurnCnt[msgPos].cnt));
180 1 : HCCL_INFO("Group %u: group sync for handle %d: message index %u, finish count %u.",
181 : groupIdx_, handle, msgPos, repeat);
182 1 : return HcclLaunchCcoreWait(ctxToOpHandle_.begin()->second, waitAddr, repeat, turnNumsAddr_, false);
183 : }
184 :
185 36084324 : HcclResult CommKfcAicpuServer::CheckTimeOut(u32 msgPos)
186 : {
187 36084324 : if (!IsTimeout()) {
188 36084306 : return HCCL_SUCCESS;
189 : }
190 18 : const bool error = (timeout_ >= TIMEOUT_ERROR_THRESHOLD);
191 : HcclResult ret;
192 18 : if (error) {
193 3 : HCCL_ERROR("Group %u: timeout %u seconds at message pos %u.", groupIdx_, timeout_, msgPos);
194 3 : ret = HCCL_E_TIMEOUT;
195 : } else {
196 15 : HCCL_RUN_INFO("Group %u: timeout %u seconds at message pos %u.", groupIdx_, timeout_, msgPos);
197 15 : ret = HCCL_E_AGAIN;
198 : }
199 18 : timeout_ *= 2U;
200 18 : return ret;
201 : }
202 :
203 3 : HcclResult CommKfcAicpuServer::GetServerInfoForSync(HcclHandle handle, u32 &msgPos, u32 &repeat) const
204 : {
205 3 : CHK_PRT_RET(handle < 0, HCCL_ERROR("Group %u: invalid handle id %d.", groupIdx_, handle), HCCL_E_PARA);
206 3 : auto it = handleIdToMsgPos_.find(handle);
207 3 : if (it == handleIdToMsgPos_.end()) {
208 2 : HCCL_INFO("Group %u: handle %d in this group is not ready.", groupIdx_, handle);
209 2 : return HCCL_E_AGAIN;
210 : }
211 1 : msgPos = it->second;
212 :
213 1 : it = handleIdToRepeat_.find(handle);
214 1 : CHK_PRT_RET(it == handleIdToRepeat_.end(),
215 : HCCL_ERROR("Group %u: handle %d in this group is not ready.", groupIdx_, handle),
216 : HCCL_E_INTERNAL);
217 1 : repeat = it->second;
218 1 : return HCCL_SUCCESS;
219 : }
220 :
221 36084375 : HcclResult CommKfcAicpuServer::ErrorDfxProcess(HcclResult errorCode)
222 : {
223 36084375 : void *firstHandle = ctxToOpHandle_.begin()->second;
224 36084375 : if (errorCode == HCCL_SUCCESS) {
225 36084350 : return errorCode;
226 25 : } else if (errorCode == HCCL_E_AGAIN) {
227 15 : AicpuKfcUtils::PrintAllHcclMsgArea(msgArea_, rankNum_);
228 15 : HcclPrintTaskExceptionAllComm(firstHandle);
229 15 : errorCode = HCCL_SUCCESS;
230 : } else {
231 10 : AicpuKfcUtils::PrintAllHcclMsgArea(msgArea_, rankNum_, true);
232 10 : HcclPrintTaskExceptionAllComm(firstHandle);
233 : }
234 25 : return errorCode;
235 : }
|
