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