Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 : #include "group_schedule_mgr.h"
11 : #include "log.h"
12 : #include "coll_alg_utils.h"
13 : #include "hccl_comm_pub.h"
14 :
15 : thread_local int32_t hcclP2pTaskNums;
16 : thread_local std::vector<HcclComm> hcclGroupCommListV2;
17 :
18 : constexpr uint32_t NUM = 1U;
19 : constexpr uint32_t BIT_MAX = 31U;
20 :
21 : namespace {
22 1 : uint32_t pow2Up(uint32_t n)
23 : {
24 1 : if (n > (NUM << BIT_MAX)) {
25 0 : return 0;
26 : }
27 :
28 1 : uint32_t power = 1;
29 2 : while (power < n) {
30 1 : power = power << 1U;
31 : }
32 1 : return power;
33 : }
34 : } // namespace
35 :
36 : namespace hccl {
37 :
38 2 : void ClearHcclGroupCommList()
39 : {
40 2 : hcclGroupCommListV2.clear();
41 2 : }
42 :
43 54 : std::vector<HcclComm> &GetHcclGroupCommList()
44 : {
45 54 : return hcclGroupCommListV2;
46 : }
47 :
48 3 : int32_t GetHcclP2pTaskNums()
49 : {
50 3 : return hcclP2pTaskNums;
51 : }
52 :
53 54 : void SetHcclP2pTaskNums(int32_t targetP2pTaskNums)
54 : {
55 54 : hcclP2pTaskNums = targetP2pTaskNums;
56 54 : }
57 :
58 158 : GroupScheduleMgr::~GroupScheduleMgr()
59 : {
60 158 : }
61 :
62 3 : HcclResult GroupScheduleMgr::GetUsrStream(aclrtStream &usrStream)
63 : {
64 3 : CHK_PTR_NULL(this->usrStream_);
65 2 : usrStream = this->usrStream_;
66 2 : return HCCL_SUCCESS;
67 : }
68 :
69 4 : HcclResult GroupScheduleMgr::SetUsrStream(const aclrtStream &usrStream)
70 : {
71 4 : CHK_PTR_NULL(usrStream);
72 3 : this->usrStream_ = usrStream;
73 3 : return HCCL_SUCCESS;
74 : }
75 :
76 1 : HcclResult GroupScheduleMgr::InitGroupPlanner(HcclComm comm)
77 : {
78 1 : CHK_PTR_NULL(comm);
79 1 : hccl::hcclComm *hcclComm = static_cast<hccl::hcclComm *>(comm);
80 1 : CHK_PTR_NULL(hcclComm);
81 1 : hccl::CollComm *collComm = hcclComm->GetCollComm();
82 1 : CHK_PTR_NULL(collComm);
83 1 : constexpr uint32_t netLayerServer = 0;
84 1 : uint32_t *serverSizeList = nullptr;
85 1 : uint32_t serverNum = 0;
86 1 : CHK_RET(HcclRankGraphGetInstSizeListByLayer(comm, netLayerServer, &serverSizeList, &serverNum));
87 1 : CHK_PTR_NULL(serverSizeList);
88 1 : this->userRank_ = collComm->GetMyRankId();
89 1 : this->rankSize_ = collComm->GetRankSize();
90 1 : this->nTasksP2p_ = 0;
91 1 : this->serverNum_ = serverNum;
92 5 : for (uint32_t serverIdx = 0, rankIdx = 0; serverIdx < serverNum; serverIdx++) {
93 4 : this->serverToRankSize_[serverIdx] = serverSizeList[serverIdx];
94 12 : for (uint32_t localIdx = 0; localIdx < serverSizeList[serverIdx]; localIdx++) {
95 8 : this->serverToRankList_[serverIdx].emplace_back(rankIdx++);
96 : }
97 : }
98 1 : HCCL_INFO("[InitGroupPlanner] ranksize:%u, serverNum:%u nTaskP2p:%d", this->rankSize_, this->serverNum_,
99 : this->nTasksP2p_);
100 :
101 1 : return HCCL_SUCCESS;
102 : }
103 :
104 1 : HcclResult GroupScheduleMgr::GetCurLocalRank(uint32_t &localRank)
105 : {
106 1 : uint32_t curServerIdx = 0;
107 1 : for (uint32_t cumulativeRank = 0, serverIdx = 0; serverIdx < this->serverNum_; serverIdx++) {
108 1 : cumulativeRank += this->serverToRankSize_.at(serverIdx);
109 1 : if (this->userRank_ < cumulativeRank) {
110 1 : curServerIdx = serverIdx;
111 1 : break;
112 : }
113 : }
114 :
115 1 : u32 curLocalRank = 0;
116 1 : for (u32 rankIdx: this->serverToRankList_.at(curServerIdx)) {
117 1 : if (this->userRank_ == rankIdx) {
118 1 : break;
119 : }
120 0 : curLocalRank++;
121 : }
122 :
123 1 : if (curLocalRank >= this->serverToRankSize_.at(curServerIdx)) {
124 0 : HCCL_ERROR("[getCurLocalRank] is invalid:%u", curLocalRank);
125 0 : return HCCL_E_INTERNAL;
126 : }
127 :
128 1 : localRank = curLocalRank;
129 1 : return HCCL_SUCCESS;
130 : }
131 :
132 1 : HcclResult GroupScheduleMgr::CalculateGroupSize()
133 : {
134 1 : if (this->serverNum_ == 0) {
135 0 : return HCCL_E_INTERNAL;
136 : }
137 :
138 1 : if (this->serverNum_ == 1) {
139 0 : this->groupSize_ = this->rankSize_;
140 0 : return HCCL_SUCCESS;
141 : }
142 :
143 1 : std::vector<uint32_t> serverRankSizeList;
144 5 : for (const auto &pair : this->serverToRankSize_) {
145 4 : serverRankSizeList.emplace_back(pair.second);
146 : }
147 1 : this->groupSize_ = hccl::CalGCD(serverRankSizeList);
148 1 : return HCCL_SUCCESS;
149 1 : }
150 :
151 1 : uint32_t GroupScheduleMgr::GenerateP2pSchedule(const std::vector<uint32_t> &groupToServer,
152 : const std::vector<uint32_t> &groupToLocalRankBase, uint32_t curGroupIdx, uint32_t curGroupLocalRankIdx)
153 : {
154 1 : this->p2pSchedule_.resize(this->rankSize_);
155 1 : uint32_t round = 0;
156 1 : uint32_t groupRound = 0;
157 1 : uint32_t groupDelta = 0;
158 1 : uint32_t nGroupsPow2 = pow2Up(this->nGroups_);
159 1 : if (nGroupsPow2 == 0) {
160 0 : return 0;
161 : }
162 :
163 : do {
164 2 : if (groupDelta < this->nGroups_) {
165 2 : uint32_t sendGroupIdx = (curGroupIdx + groupDelta) % this->nGroups_;
166 2 : uint32_t recvGroupIdx = (curGroupIdx - groupDelta + this->nGroups_) % this->nGroups_;
167 2 : uint32_t sendServerIdx = groupToServer[sendGroupIdx];
168 2 : uint32_t recvServerIdx = groupToServer[recvGroupIdx];
169 :
170 6 : for (uint32_t delta = 0; delta < this->groupSize_; delta++) {
171 : uint32_t sendLocalIdx
172 4 : = groupToLocalRankBase[sendGroupIdx] + (curGroupLocalRankIdx + delta) % this->groupSize_;
173 4 : uint32_t recvLocalIdx = groupToLocalRankBase[recvGroupIdx]
174 4 : + (curGroupLocalRankIdx - delta + this->groupSize_) % this->groupSize_;
175 :
176 4 : this->p2pSchedule_[round].sendRank = this->serverToRankList_.at(sendServerIdx)[sendLocalIdx];
177 4 : this->p2pSchedule_[round].recvRank = this->serverToRankList_.at(recvServerIdx)[recvLocalIdx];
178 4 : round++;
179 : }
180 : }
181 2 : groupRound++;
182 2 : groupDelta = (groupDelta + groupRound) & (nGroupsPow2 - 1);
183 2 : } while (groupRound != nGroupsPow2);
184 :
185 1 : return round;
186 : }
187 :
188 1 : HcclResult GroupScheduleMgr::HcclP2pSchedulerGenerate()
189 : {
190 1 : uint32_t curLocalRank = 0; // 当前rank所在server的局部排序号
191 1 : CHK_RET(GetCurLocalRank(curLocalRank));
192 1 : CHK_RET(CalculateGroupSize());
193 1 : if (this->groupSize_ == 0) {
194 0 : HCCL_ERROR("[HcclP2pSchedulerGenerate] groupSize is zero");
195 0 : return HCCL_E_INTERNAL;
196 : }
197 :
198 1 : uint32_t curGroupLocalRankIdx = curLocalRank % this->groupSize_;
199 1 : uint32_t curGroupIdx = this->userRank_ / this->groupSize_;
200 1 : this->nGroups_ = this->rankSize_ / this->groupSize_;
201 1 : HCCL_INFO("[HcclP2pSchedulerGenerate] userRank:%u, localRank:%u, groupSize:%u, nGroups:%u", this->userRank_,
202 : curLocalRank, this->groupSize_, this->nGroups_);
203 :
204 2 : std::vector<uint32_t> groupToServer(this->nGroups_);
205 1 : std::vector<uint32_t> groupToLocalRankBase(this->nGroups_);
206 1 : uint32_t groupIdx = 0;
207 5 : for (const auto &pair : this->serverToRankList_) {
208 4 : uint32_t serverIdx = pair.first;
209 4 : uint32_t localRankSize = this->serverToRankSize_.at(serverIdx);
210 4 : uint32_t localGroupSize = localRankSize / this->groupSize_;
211 8 : for (uint32_t localGroupIdx = 0; localGroupIdx < localGroupSize; localGroupIdx++) {
212 4 : groupToServer[groupIdx] = serverIdx;
213 4 : groupToLocalRankBase[groupIdx] = localGroupIdx * this->groupSize_;
214 4 : groupIdx++;
215 : }
216 : }
217 :
218 1 : uint32_t round = GenerateP2pSchedule(groupToServer, groupToLocalRankBase, curGroupIdx, curGroupLocalRankIdx);
219 1 : if (this->rankSize_ != round) {
220 0 : HCCL_ERROR("[HcclP2pSchedulerGenerate] round:%u is not equal to rankSize:%u", round, this->rankSize_);
221 0 : return HCCL_E_INTERNAL;
222 : }
223 :
224 1 : HCCL_INFO("[HcclP2pSchedulerGenerate] schedule generated, round:%u", round);
225 1 : return HCCL_SUCCESS;
226 1 : }
227 :
228 3 : HcclResult GroupScheduleMgr::AppendGroupP2pTask(HcclComm comm, const HcclP2pTask &task, const HcclOpP2pDesc &p2pDesc)
229 : {
230 3 : CHK_PTR_NULL(comm);
231 2 : if (hcclP2pTaskNums == MAX_P2P_TASK_NUM) {
232 1 : HCCL_ERROR("[hcclGroupAddP2pTask] P2pTaskNums is out of %d", MAX_P2P_TASK_NUM);
233 1 : return HCCL_E_INTERNAL;
234 : }
235 1 : if (this->nTasksP2p_ == -1) {
236 1 : CHK_RET(InitGroupPlanner(comm));
237 1 : CHK_RET(HcclP2pSchedulerGenerate());
238 1 : this->peers_.resize(this->rankSize_);
239 : }
240 :
241 1 : if (p2pDesc.cmdType == HcclCMDType::HCCL_CMD_SEND) {
242 1 : this->peers_[p2pDesc.remoteRank].sendQue.emplace_back(task);
243 : } else {
244 0 : this->peers_[p2pDesc.remoteRank].recvQue.emplace_back(task);
245 : }
246 1 : this->nTasksP2p_ += 1;
247 1 : hcclP2pTaskNums++;
248 1 : auto itComm = std::find(hcclGroupCommListV2.begin(), hcclGroupCommListV2.end(), comm);
249 1 : if (itComm == hcclGroupCommListV2.end()) {
250 1 : hcclGroupCommListV2.emplace_back(comm);
251 : }
252 :
253 1 : return HCCL_SUCCESS;
254 : }
255 :
256 1 : HcclResult GroupScheduleMgr::GetP2pTaskSchedule(
257 : std::vector<HcclP2pTask> &sortedSendQue, std::vector<HcclP2pTask> &sortedRecvQue)
258 : {
259 1 : HCCL_INFO("[HcclP2pTaskSchedule] nTaskP2p:%d", this->nTasksP2p_);
260 :
261 1 : uint32_t epoch = 0;
262 1 : uint32_t maxEpochNum = this->nTasksP2p_;
263 1 : while (this->nTasksP2p_ > 0) {
264 0 : for (uint32_t round = 0; round < this->rankSize_; round++) {
265 0 : uint32_t sendRank = this->p2pSchedule_[round].sendRank;
266 0 : uint32_t recvRank = this->p2pSchedule_[round].recvRank;
267 :
268 0 : if (!this->peers_[sendRank].sendQue.empty()) {
269 0 : auto &sendTask = this->peers_[sendRank].sendQue.front();
270 0 : sortedSendQue.emplace_back(sendTask);
271 0 : this->peers_[sendRank].sendQue.pop_front();
272 0 : this->nTasksP2p_--;
273 : }
274 :
275 0 : if (!this->peers_[recvRank].recvQue.empty()) {
276 0 : auto &recvTask = this->peers_[recvRank].recvQue.front();
277 0 : sortedRecvQue.emplace_back(recvTask);
278 0 : this->peers_[recvRank].recvQue.pop_front();
279 0 : this->nTasksP2p_--;
280 : }
281 : }
282 :
283 0 : epoch++;
284 0 : if (epoch > maxEpochNum) {
285 0 : HCCL_ERROR("[GetP2pTaskSchedule] epoch:%u is more than max epoch:%u", epoch, maxEpochNum);
286 0 : return HCCL_E_INTERNAL;
287 : }
288 : }
289 1 : HCCL_INFO("[HcclP2pTaskSchedule] done, use epochs:%u, sendQueSize:%u, recvQueSize:%u ", epoch,
290 : static_cast<uint32_t>(sortedSendQue.size()), static_cast<uint32_t>(sortedRecvQue.size()));
291 :
292 1 : return HCCL_SUCCESS;
293 : }
294 : } // namespace hccl
|
