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