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) { extMsg = std::make_shared<HcclMsgExt>(); }
30 : };
31 : static constexpr u32 MAX_RETRY_CNT = 10U;
32 :
33 12 : HcclResult CreateServerList(void* args[], u32 ctxNum, std::vector<AscCommServerInfo>& serverList)
34 : {
35 12 : CHK_PRT_RET(ctxNum == 0U, HCCL_ERROR("Invalid context number."), HCCL_E_PARA);
36 33 : for (u32 i = 0U; i < ctxNum; ++i) {
37 21 : const CommKfcContext* ctx = static_cast<const CommKfcContext*>(args[i]);
38 21 : CHK_PTR_NULL(ctx);
39 21 : auto it = std::find_if(serverList.begin(), serverList.end(), [ctx](const AscCommServerInfo& server) {
40 11 : return reinterpret_cast<u64>(server.serverIns.GetMsgAreaAddr()) == ctx->apiCtx.workSpace;
41 : });
42 21 : const u32 serverIdx = it - serverList.begin();
43 21 : if (serverIdx == serverList.size()) {
44 16 : AscCommServerInfo server(serverIdx);
45 16 : CHK_SMART_PTR_NULL(server.extMsg);
46 16 : HCCL_INFO("Server for group %u is created.", serverIdx);
47 16 : serverList.emplace_back(server);
48 16 : }
49 21 : CHK_PRT_RET(
50 : serverList[serverIdx].serverIns.AddOpContext(ctx) != HCCL_SUCCESS,
51 : HCCL_ERROR("Failed to add op for group %u.", serverIdx), HCCL_E_INTERNAL);
52 : }
53 12 : return HCCL_SUCCESS;
54 : }
55 :
56 34980987 : HcclResult GetCurrentMsg(AscCommServerInfo& server)
57 : {
58 34980987 : if (server.retryCnt > 0) {
59 15 : CHK_PRT_RET(
60 : server.retryCnt > MAX_RETRY_CNT, HCCL_ERROR("Retry count %d exceeds max value.", server.retryCnt),
61 : HCCL_E_INTERNAL);
62 14 : HCCL_INFO(
63 : "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(
67 : "Process cache extended message %s at seq num %u.",
68 : AicpuKfcUtils::GetMsgSimpleStr(server.serverIns.GetRankNum(), *(server.extMsg)).c_str(), server.msgPos);
69 : }
70 14 : return HCCL_SUCCESS;
71 : }
72 :
73 34980972 : auto& msgBaseAddr = server.serverIns.GetMsgAreaAddr()->commMsg.singleMsg;
74 34980972 : HcclResult ret = AicpuKfcUtils::ReadMsgFromMemory(msgBaseAddr.sendMsgs + server.msgPos, server.msg);
75 34980972 : if (ret != HCCL_SUCCESS) {
76 34980947 : 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(
100 : groupId >= serverList.size() || groupId == curGroupIdx || handleId < 0,
101 : HCCL_ERROR(
102 : "Invalid handle id %d or group id %u, current group id %u/%u.", handleId, groupId, curGroupIdx,
103 : serverList.size()),
104 : HCCL_E_PARA);
105 3 : HcclResult ret = server.serverIns.InterGroupSync(serverList[groupId].serverIns, handleId);
106 3 : if (ret == HCCL_SUCCESS) {
107 1 : server.retryCnt = 0;
108 1 : server.msgPos = (server.msgPos + 1U) % HCCL_MSG_CNT;
109 1 : HCCL_INFO("Group %u added wait sqe for group %u handle id %d successfully.", curGroupIdx, groupId, handleId);
110 2 : } else if (ret == HCCL_E_AGAIN) {
111 2 : ++(server.retryCnt);
112 2 : HCCL_INFO("Group sync(%u-%u) will be retried at seq num %u.", curGroupIdx, groupId, server.msgPos);
113 : } else {
114 0 : HCCL_ERROR("Group sync(%u-%u) failed, handle id %d, error code %u.", groupId, handleId, ret);
115 0 : return ret;
116 : }
117 3 : return HCCL_SUCCESS;
118 : }
119 :
120 27 : HcclResult PrepareProcess(AscCommServerInfo& server, u32& expectSeqNum)
121 : {
122 27 : const u32 seqNum = static_cast<u32>(server.msg.addMsg.v1Msg.seqNum);
123 27 : if (expectSeqNum != seqNum) {
124 13 : HCCL_INFO("Expect seq id %u but receive %u.", expectSeqNum, seqNum);
125 13 : ++(server.retryCnt);
126 : } else {
127 14 : CHK_RET(server.serverIns.Orchestrate(server.msg, *(server.extMsg), server.msgPos));
128 12 : server.msgPos = (server.msgPos + 1U) % HCCL_MSG_CNT;
129 12 : ++expectSeqNum;
130 12 : server.retryCnt = 0;
131 : }
132 25 : return HCCL_SUCCESS;
133 : }
134 :
135 36107888 : HcclResult GroupServerProcess(std::vector<AscCommServerInfo>& serverList, u32 groupIdx, u32& expectSeq, u32& finishCnt)
136 : {
137 36107888 : auto& server = serverList[groupIdx];
138 36107888 : if (server.finishFlag) {
139 6 : return HCCL_SUCCESS;
140 : }
141 :
142 : HcclResult ret;
143 36107882 : if (server.finalizeFlag) {
144 1126895 : bool isFinish = false;
145 1126895 : CHK_RET(server.serverIns.IsAllTaskFinished(server.msgPos, isFinish));
146 1126894 : if (isFinish) {
147 4 : server.finishFlag = true;
148 4 : ++finishCnt;
149 4 : HCCL_INFO("Group %u is finished, total finished number %u/%u.", groupIdx, finishCnt, serverList.size());
150 : } else {
151 1126890 : ret = server.serverIns.CheckTimeOut(server.msgPos);
152 1126890 : if (ret != HCCL_SUCCESS) {
153 6 : return ret;
154 : }
155 : }
156 1126888 : return HCCL_SUCCESS;
157 : }
158 :
159 34980987 : ret = GetCurrentMsg(server);
160 34980987 : if (ret == HCCL_E_AGAIN) {
161 34980947 : ret = server.serverIns.CheckTimeOut(server.msgPos);
162 34980947 : if (ret != HCCL_SUCCESS) {
163 12 : return ret;
164 : }
165 34980935 : return HCCL_SUCCESS;
166 : }
167 40 : CHK_RET(ret);
168 :
169 38 : HCCL_INFO(
170 : "Process message for group %u, kernel index %u, message index %u.", groupIdx,
171 : static_cast<u32>(server.msg.addMsg.v1Msg.seqNum), server.msgPos);
172 38 : switch (server.msg.commType.msgType) {
173 6 : case ControlMsgType::HCCL_CMD_FINALIZE:
174 6 : CHK_RET(FinalizeProcess(server));
175 6 : break;
176 5 : case ControlMsgType::HCCL_CMD_INTER_GROUP_SYNC:
177 5 : CHK_RET(InterGroupSyncProcess(serverList, groupIdx));
178 3 : break;
179 27 : default:
180 27 : CHK_RET(PrepareProcess(server, expectSeq));
181 25 : break;
182 : }
183 34 : return HCCL_SUCCESS;
184 : }
185 : } // namespace
186 :
187 12 : u32 CommKfcDispatcher::Run(void* args[], u32 ctxNum)
188 : {
189 12 : std::vector<AscCommServerInfo> serverList{};
190 12 : CHK_RET(CreateServerList(args, ctxNum, serverList));
191 :
192 12 : u32 finishCnt = 0U;
193 12 : u32 expectSeqNum = 0U;
194 36107876 : while (finishCnt != serverList.size()) {
195 72215752 : for (u32 i = 0U; i < serverList.size(); ++i) {
196 36107888 : HcclResult ret = GroupServerProcess(serverList, i, expectSeqNum, finishCnt);
197 36107888 : CHK_RET(serverList[i].serverIns.ErrorDfxProcess(ret));
198 : }
199 : }
200 :
201 2 : return HCCL_SUCCESS;
202 12 : }
|
