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 56 : std::vector<HcclComm>& GetHcclGroupCommList() { return hcclGroupCommListV2; }
42 :
43 3 : int32_t GetHcclP2pTaskNums() { return hcclP2pTaskNums; }
44 :
45 56 : void SetHcclP2pTaskNums(int32_t targetP2pTaskNums) { hcclP2pTaskNums = targetP2pTaskNums; }
46 :
47 254 : 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 2 : HcclResult GroupScheduleMgr::GetCurLocalRank(uint32_t& localRank)
93 : {
94 2 : uint32_t curServerIdx = 0;
95 3 : for (uint32_t cumulativeRank = 0, serverIdx = 0; serverIdx < this->serverNum_; serverIdx++) {
96 2 : cumulativeRank += this->serverToRankSize_.at(serverIdx);
97 2 : if (this->userRank_ < cumulativeRank) {
98 1 : curServerIdx = serverIdx;
99 1 : break;
100 : }
101 : }
102 :
103 2 : u32 curLocalRank = 0;
104 4 : for (u32 rankIdx : this->serverToRankList_.at(curServerIdx)) {
105 3 : if (this->userRank_ == rankIdx) {
106 1 : break;
107 : }
108 2 : curLocalRank++;
109 : }
110 :
111 2 : if (curLocalRank >= this->serverToRankSize_.at(curServerIdx)) {
112 1 : HCCL_ERROR(
113 : "[getCurLocalRank] curLocalRank[%u] is invalid, valid range [0, %u)", curLocalRank,
114 : this->serverToRankSize_.at(curServerIdx));
115 1 : return HCCL_E_INTERNAL;
116 : }
117 :
118 1 : localRank = curLocalRank;
119 1 : return HCCL_SUCCESS;
120 : }
121 :
122 1 : HcclResult GroupScheduleMgr::CalculateGroupSize()
123 : {
124 1 : if (this->serverNum_ == 0) {
125 0 : return HCCL_E_INTERNAL;
126 : }
127 :
128 1 : if (this->serverNum_ == 1) {
129 0 : this->groupSize_ = this->rankSize_;
130 0 : return HCCL_SUCCESS;
131 : }
132 :
133 1 : std::vector<uint32_t> serverRankSizeList;
134 5 : for (const auto& pair : this->serverToRankSize_) {
135 4 : serverRankSizeList.emplace_back(pair.second);
136 : }
137 1 : this->groupSize_ = hccl::CalGCD(serverRankSizeList);
138 1 : return HCCL_SUCCESS;
139 1 : }
140 :
141 1 : uint32_t GroupScheduleMgr::GenerateP2pSchedule(
142 : const std::vector<uint32_t>& groupToServer, const std::vector<uint32_t>& groupToLocalRankBase, uint32_t curGroupIdx,
143 : uint32_t curGroupLocalRankIdx)
144 : {
145 1 : this->p2pSchedule_.resize(this->rankSize_);
146 1 : uint32_t round = 0;
147 1 : uint32_t groupRound = 0;
148 1 : uint32_t groupDelta = 0;
149 1 : uint32_t nGroupsPow2 = pow2Up(this->nGroups_);
150 1 : if (nGroupsPow2 == 0) {
151 0 : return 0;
152 : }
153 :
154 : do {
155 2 : if (groupDelta < this->nGroups_) {
156 2 : uint32_t sendGroupIdx = (curGroupIdx + groupDelta) % this->nGroups_;
157 2 : uint32_t recvGroupIdx = (curGroupIdx - groupDelta + this->nGroups_) % this->nGroups_;
158 2 : uint32_t sendServerIdx = groupToServer[sendGroupIdx];
159 2 : uint32_t recvServerIdx = groupToServer[recvGroupIdx];
160 :
161 6 : for (uint32_t delta = 0; delta < this->groupSize_; delta++) {
162 : uint32_t sendLocalIdx
163 4 : = groupToLocalRankBase[sendGroupIdx] + (curGroupLocalRankIdx + delta) % this->groupSize_;
164 4 : uint32_t recvLocalIdx = groupToLocalRankBase[recvGroupIdx]
165 4 : + (curGroupLocalRankIdx - delta + this->groupSize_) % this->groupSize_;
166 :
167 4 : this->p2pSchedule_[round].sendRank = this->serverToRankList_.at(sendServerIdx)[sendLocalIdx];
168 4 : this->p2pSchedule_[round].recvRank = this->serverToRankList_.at(recvServerIdx)[recvLocalIdx];
169 4 : round++;
170 : }
171 : }
172 2 : groupRound++;
173 2 : groupDelta = (groupDelta + groupRound) & (nGroupsPow2 - 1);
174 2 : } while (groupRound != nGroupsPow2);
175 :
176 1 : return round;
177 : }
178 :
179 1 : HcclResult GroupScheduleMgr::HcclP2pSchedulerGenerate()
180 : {
181 1 : uint32_t curLocalRank = 0; // 当前rank所在server的局部排序号
182 1 : CHK_RET(GetCurLocalRank(curLocalRank));
183 1 : CHK_RET(CalculateGroupSize());
184 1 : if (this->groupSize_ == 0) {
185 0 : HCCL_ERROR("[HcclP2pSchedulerGenerate] groupSize is zero");
186 0 : return HCCL_E_INTERNAL;
187 : }
188 :
189 1 : uint32_t curGroupLocalRankIdx = curLocalRank % this->groupSize_;
190 1 : uint32_t curGroupIdx = this->userRank_ / this->groupSize_;
191 1 : this->nGroups_ = this->rankSize_ / this->groupSize_;
192 1 : HCCL_INFO(
193 : "[HcclP2pSchedulerGenerate] userRank:%u, localRank:%u, groupSize:%u, nGroups:%u", this->userRank_, curLocalRank,
194 : this->groupSize_, this->nGroups_);
195 :
196 2 : std::vector<uint32_t> groupToServer(this->nGroups_);
197 1 : std::vector<uint32_t> groupToLocalRankBase(this->nGroups_);
198 1 : uint32_t groupIdx = 0;
199 5 : for (const auto& pair : this->serverToRankList_) {
200 4 : uint32_t serverIdx = pair.first;
201 4 : uint32_t localRankSize = this->serverToRankSize_.at(serverIdx);
202 4 : uint32_t localGroupSize = localRankSize / this->groupSize_;
203 8 : for (uint32_t localGroupIdx = 0; localGroupIdx < localGroupSize; localGroupIdx++) {
204 4 : groupToServer[groupIdx] = serverIdx;
205 4 : groupToLocalRankBase[groupIdx] = localGroupIdx * this->groupSize_;
206 4 : groupIdx++;
207 : }
208 : }
209 :
210 1 : uint32_t round = GenerateP2pSchedule(groupToServer, groupToLocalRankBase, curGroupIdx, curGroupLocalRankIdx);
211 1 : if (this->rankSize_ != round) {
212 0 : HCCL_ERROR("[HcclP2pSchedulerGenerate] round:%u is not equal to rankSize:%u", round, this->rankSize_);
213 0 : return HCCL_E_INTERNAL;
214 : }
215 :
216 1 : HCCL_INFO("[HcclP2pSchedulerGenerate] schedule generated, round:%u", round);
217 1 : return HCCL_SUCCESS;
218 1 : }
219 :
220 3 : HcclResult GroupScheduleMgr::AppendGroupP2pTask(HcclComm comm, const HcclP2pTask& task, const HcclOpP2pDesc& p2pDesc)
221 : {
222 3 : CHK_PTR_NULL(comm);
223 2 : if (hcclP2pTaskNums == MAX_P2P_TASK_NUM) {
224 1 : HCCL_ERROR("[hcclGroupAddP2pTask] P2pTaskNums is out of %d", MAX_P2P_TASK_NUM);
225 1 : return HCCL_E_INTERNAL;
226 : }
227 1 : if (this->nTasksP2p_ == -1) {
228 1 : CHK_RET(InitGroupPlanner(comm));
229 1 : CHK_RET(HcclP2pSchedulerGenerate());
230 1 : this->peers_.resize(this->rankSize_);
231 : }
232 :
233 1 : if (p2pDesc.cmdType == HcclCMDType::HCCL_CMD_SEND) {
234 1 : this->peers_[p2pDesc.remoteRank].sendQue.emplace_back(task);
235 : } else {
236 0 : this->peers_[p2pDesc.remoteRank].recvQue.emplace_back(task);
237 : }
238 1 : this->nTasksP2p_ += 1;
239 1 : hcclP2pTaskNums++;
240 1 : auto itComm = std::find(hcclGroupCommListV2.begin(), hcclGroupCommListV2.end(), comm);
241 1 : if (itComm == hcclGroupCommListV2.end()) {
242 1 : hcclGroupCommListV2.emplace_back(comm);
243 : }
244 :
245 1 : return HCCL_SUCCESS;
246 : }
247 :
248 : HcclResult
249 1 : GroupScheduleMgr::GetP2pTaskSchedule(std::vector<HcclP2pTask>& sortedSendQue, std::vector<HcclP2pTask>& sortedRecvQue)
250 : {
251 1 : HCCL_INFO("[HcclP2pTaskSchedule] nTaskP2p:%d", this->nTasksP2p_);
252 :
253 1 : uint32_t epoch = 0;
254 1 : uint32_t maxEpochNum = this->nTasksP2p_;
255 1 : while (this->nTasksP2p_ > 0) {
256 0 : for (uint32_t round = 0; round < this->rankSize_; round++) {
257 0 : uint32_t sendRank = this->p2pSchedule_[round].sendRank;
258 0 : uint32_t recvRank = this->p2pSchedule_[round].recvRank;
259 :
260 0 : if (!this->peers_[sendRank].sendQue.empty()) {
261 0 : auto& sendTask = this->peers_[sendRank].sendQue.front();
262 0 : sortedSendQue.emplace_back(sendTask);
263 0 : this->peers_[sendRank].sendQue.pop_front();
264 0 : this->nTasksP2p_--;
265 : }
266 :
267 0 : if (!this->peers_[recvRank].recvQue.empty()) {
268 0 : auto& recvTask = this->peers_[recvRank].recvQue.front();
269 0 : sortedRecvQue.emplace_back(recvTask);
270 0 : this->peers_[recvRank].recvQue.pop_front();
271 0 : this->nTasksP2p_--;
272 : }
273 : }
274 :
275 0 : epoch++;
276 0 : if (epoch > maxEpochNum) {
277 0 : HCCL_ERROR("[GetP2pTaskSchedule] epoch:%u is more than max epoch:%u", epoch, maxEpochNum);
278 0 : return HCCL_E_INTERNAL;
279 : }
280 : }
281 1 : HCCL_INFO(
282 : "[HcclP2pTaskSchedule] done, use epochs:%u, sendQueSize:%u, recvQueSize:%u ", epoch,
283 : static_cast<uint32_t>(sortedSendQue.size()), static_cast<uint32_t>(sortedRecvQue.size()));
284 :
285 1 : return HCCL_SUCCESS;
286 : }
287 : } // namespace hccl
|
