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