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_dispatcher.h"
12 :
13 : #include "hccl_msg.h"
14 :
15 : #include "common/aicpu_kfc_utils.h"
16 : #include "comm_kfc_aicpu_server.h"
17 :
18 : using namespace HcclApi;
19 :
20 : namespace {
21 : struct AscCommServerInfo {
22 : CommKfcAicpuServer serverIns;
23 : HcclMsg msg{};
24 : std::shared_ptr<HcclMsgExt> extMsg;
25 : u32 msgPos{0U};
26 : u32 retryCnt{0U};
27 : bool finalizeFlag{false};
28 : bool finishFlag{false};
29 16 : AscCommServerInfo(u32 groupIdx): serverIns(groupIdx) {
30 16 : extMsg = std::make_shared<HcclMsgExt>();
31 16 : }
32 : };
33 : static constexpr u32 MAX_RETRY_CNT = 10U;
34 :
35 12 : HcclResult CreateServerList(void *args[], u32 ctxNum, std::vector<AscCommServerInfo> &serverList)
36 : {
37 12 : CHK_PRT_RET(ctxNum == 0U, HCCL_ERROR("Invalid context number."), HCCL_E_PARA);
38 33 : for (u32 i = 0U; i < ctxNum; ++i) {
39 21 : const CommKfcContext *ctx = static_cast<const CommKfcContext *>(args[i]);
40 21 : CHK_PTR_NULL(ctx);
41 21 : auto it = std::find_if(serverList.begin(), serverList.end(),
42 11 : [ctx](const AscCommServerInfo &server) {
43 11 : return reinterpret_cast<u64>(server.serverIns.GetMsgAreaAddr()) == ctx->apiCtx.workSpace;
44 : });
45 21 : const u32 serverIdx = it - serverList.begin();
46 21 : if (serverIdx == serverList.size()) {
47 16 : AscCommServerInfo server(serverIdx);
48 16 : CHK_SMART_PTR_NULL(server.extMsg);
49 16 : HCCL_INFO("Server for group %u is created.", serverIdx);
50 16 : serverList.emplace_back(server);
51 16 : }
52 21 : CHK_PRT_RET(serverList[serverIdx].serverIns.AddOpContext(ctx) != HCCL_SUCCESS,
53 : HCCL_ERROR("Failed to add op for group %u.", serverIdx), HCCL_E_INTERNAL);
54 : }
55 12 : return HCCL_SUCCESS;
56 : }
57 :
58 34978633 : HcclResult GetCurrentMsg(AscCommServerInfo &server)
59 : {
60 34978633 : if (server.retryCnt > 0) {
61 15 : CHK_PRT_RET(server.retryCnt > MAX_RETRY_CNT, HCCL_ERROR("Retry count %d exceeds max value.", server.retryCnt),
62 : HCCL_E_INTERNAL);
63 14 : HCCL_INFO("Process cache message %s at seq num %u, retry count %u.",
64 : AicpuKfcUtils::GetMsgSimpleStr(server.msg).c_str(), server.msgPos, server.retryCnt);
65 14 : if (static_cast<HcclCMDType>(server.msg.commType.prepareType) == HCCL_CMD_ALLTOALLV) {
66 2 : HCCL_INFO("Process cache extended message %s at seq num %u.",
67 : AicpuKfcUtils::GetMsgSimpleStr(server.serverIns.GetRankNum(), *(server.extMsg)).c_str(),
68 : server.msgPos);
69 : }
70 14 : return HCCL_SUCCESS;
71 : }
72 :
73 34978618 : auto &msgBaseAddr = server.serverIns.GetMsgAreaAddr()->commMsg.singleMsg;
74 34978618 : HcclResult ret = AicpuKfcUtils::ReadMsgFromMemory(msgBaseAddr.sendMsgs + server.msgPos, server.msg);
75 34978618 : if (ret != HCCL_SUCCESS) {
76 34978593 : return ret;
77 : }
78 :
79 25 : if (static_cast<HcclCMDType>(server.msg.commType.prepareType) == HCCL_CMD_ALLTOALLV) {
80 6 : ret = AicpuKfcUtils::ReadMsgFromMemory(
81 6 : msgBaseAddr.paramExtMsgList + server.msgPos, server.serverIns.GetRankNum(), *(server.extMsg));
82 : }
83 :
84 25 : return ret;
85 : }
86 :
87 6 : HcclResult FinalizeProcess(AscCommServerInfo &server)
88 : {
89 6 : CHK_RET(server.serverIns.Finalize(server.msgPos));
90 6 : server.finalizeFlag = true;
91 6 : return HCCL_SUCCESS;
92 : }
93 :
94 5 : HcclResult InterGroupSyncProcess(std::vector<AscCommServerInfo> &serverList, u32 curGroupIdx)
95 : {
96 5 : auto &server = serverList[curGroupIdx];
97 5 : const u32 groupId = static_cast<u32>(server.msg.addMsg.v0Msg.commDepGroupID);
98 5 : const HcclHandle handleId = server.msg.addMsg.v0Msg.commDepHandleID;
99 5 : CHK_PRT_RET(groupId >= serverList.size() || groupId == curGroupIdx || handleId < 0,
100 : HCCL_ERROR("Invalid handle id %d or group id %u, current group id %u/%u.",
101 : handleId, groupId, curGroupIdx, serverList.size()),
102 : HCCL_E_PARA);
103 3 : HcclResult ret = server.serverIns.InterGroupSync(serverList[groupId].serverIns, handleId);
104 3 : if (ret == HCCL_SUCCESS) {
105 1 : server.retryCnt = 0;
106 1 : server.msgPos = (server.msgPos + 1U) % HCCL_MSG_CNT;
107 1 : HCCL_INFO("Group %u added wait sqe for group %u handle id %d successfully.", curGroupIdx, groupId, handleId);
108 2 : } else if (ret == HCCL_E_AGAIN) {
109 2 : ++(server.retryCnt);
110 2 : HCCL_INFO("Group sync(%u-%u) will be retried at seq num %u.", curGroupIdx, groupId, server.msgPos);
111 : } else {
112 0 : HCCL_ERROR("Group sync(%u-%u) failed, handle id %d, error code %u.", groupId, handleId, ret);
113 0 : return ret;
114 : }
115 3 : return HCCL_SUCCESS;
116 : }
117 :
118 27 : HcclResult PrepareProcess(AscCommServerInfo &server, u32 &expectSeqNum)
119 : {
120 27 : const u32 seqNum = static_cast<u32>(server.msg.addMsg.v1Msg.seqNum);
121 27 : if (expectSeqNum != seqNum) {
122 13 : HCCL_INFO("Expect seq id %u but receive %u.", expectSeqNum, seqNum);
123 13 : ++(server.retryCnt);
124 : } else {
125 14 : CHK_RET(server.serverIns.Orchestrate(server.msg, *(server.extMsg), server.msgPos));
126 12 : server.msgPos = (server.msgPos + 1U) % HCCL_MSG_CNT;
127 12 : ++expectSeqNum;
128 12 : server.retryCnt = 0;
129 : }
130 25 : return HCCL_SUCCESS;
131 : }
132 :
133 36084375 : HcclResult GroupServerProcess(std::vector<AscCommServerInfo> &serverList, u32 groupIdx, u32 &expectSeq, u32 &finishCnt)
134 : {
135 36084375 : auto &server = serverList[groupIdx];
136 36084375 : if (server.finishFlag) {
137 6 : return HCCL_SUCCESS;
138 : }
139 :
140 : HcclResult ret;
141 36084369 : if (server.finalizeFlag) {
142 1105736 : bool isFinish = false;
143 1105736 : CHK_RET(server.serverIns.IsAllTaskFinished(server.msgPos, isFinish));
144 1105735 : if (isFinish) {
145 4 : server.finishFlag = true;
146 4 : ++finishCnt;
147 4 : HCCL_INFO("Group %u is finished, total finished number %u/%u.", groupIdx, finishCnt, serverList.size());
148 : } else {
149 1105731 : ret = server.serverIns.CheckTimeOut(server.msgPos);
150 1105731 : if (ret != HCCL_SUCCESS) {
151 6 : return ret;
152 : }
153 : }
154 1105729 : return HCCL_SUCCESS;
155 : }
156 :
157 34978633 : ret = GetCurrentMsg(server);
158 34978633 : if (ret == HCCL_E_AGAIN) {
159 34978593 : ret = server.serverIns.CheckTimeOut(server.msgPos);
160 34978593 : if (ret != HCCL_SUCCESS) {
161 12 : return ret;
162 : }
163 34978581 : return HCCL_SUCCESS;
164 : }
165 40 : CHK_RET(ret);
166 :
167 38 : HCCL_INFO("Process message for group %u, kernel index %u, message index %u.",
168 : groupIdx, static_cast<u32>(server.msg.addMsg.v1Msg.seqNum), server.msgPos);
169 38 : switch (server.msg.commType.msgType) {
170 6 : case ControlMsgType::HCCL_CMD_FINALIZE:
171 6 : CHK_RET(FinalizeProcess(server));
172 6 : break;
173 5 : case ControlMsgType::HCCL_CMD_INTER_GROUP_SYNC:
174 5 : CHK_RET(InterGroupSyncProcess(serverList, groupIdx));
175 3 : break;
176 27 : default:
177 27 : CHK_RET(PrepareProcess(server, expectSeq));
178 25 : break;
179 : }
180 34 : return HCCL_SUCCESS;
181 : }
182 : }
183 :
184 12 : u32 CommKfcDispatcher::Run(void *args[], u32 ctxNum)
185 : {
186 12 : std::vector<AscCommServerInfo> serverList{};
187 12 : CHK_RET(CreateServerList(args, ctxNum, serverList));
188 :
189 12 : u32 finishCnt = 0U;
190 12 : u32 expectSeqNum = 0U;
191 36084363 : while (finishCnt != serverList.size()) {
192 72168726 : for (u32 i = 0U; i < serverList.size(); ++i) {
193 36084375 : HcclResult ret = GroupServerProcess(serverList, i, expectSeqNum, finishCnt);
194 36084375 : CHK_RET(serverList[i].serverIns.ErrorDfxProcess(ret));
195 : }
196 : }
197 :
198 2 : return HCCL_SUCCESS;
199 12 : }
|
