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 "coll_service_default_impl.h"
12 : #include <unordered_set>
13 : #include <sstream>
14 : #include <string>
15 : #include <chrono>
16 : #include "connections_builder.h"
17 : #include "communicator_impl.h"
18 : #include "dev_ub_connection.h"
19 : #include "stl_util.h"
20 : #include "exception_util.h"
21 :
22 : using HcclUs = std::chrono::steady_clock::time_point;
23 :
24 : namespace Hccl {
25 :
26 5 : void CollServiceDefaultImpl::LoadWithOpBasedModeNoRegister(CollOperator& op)
27 : {
28 5 : shared_ptr<InsQueue> insQueue;
29 5 : insQueue = OrchestrateWithIns(op);
30 :
31 5 : AllocQueueNotify(*insQueue);
32 :
33 5 : vector<LinkData> links = insQueue->GetUniqueLinks();
34 :
35 : // Socket建链
36 5 : comm->GetSocketManager().BatchCreateSockets(links);
37 : // 建立RmaConnection并建链
38 5 : auto connBuilderPair = connectionsBuilders.emplace(comm->GetId(), make_unique<ConnectionsBuilder>(*comm));
39 5 : connBuilderPair.first->second->BatchBuild(comm->GetId(), links);
40 :
41 5 : AllocNotifies(links);
42 :
43 5 : AllocLocCntNotifies(*insQueue);
44 :
45 5 : comm->GetMemTransportManager()->BatchBuildOpbasedTransports(links);
46 5 : WaitOpbasedTransportReady();
47 :
48 5 : SaveMirrorDfxOpInfo();
49 :
50 5 : Interpreter interpreter(*comm);
51 5 : interpreter.Submit(*insQueue);
52 :
53 5 : UpdateUbCiIfNeed(op.opTag);
54 5 : }
55 :
56 5 : void CollServiceDefaultImpl::LoadWithOpBasedMode(CollOperator& op, unique_ptr<Stream> stream)
57 : {
58 15 : HCCL_INFO("LoadWithOpBasedMode START");
59 15 : HCCL_INFO("RegisterOpbasedBuf start");
60 5 : RegisterOpBufToBufMgr(op);
61 5 : RegisterOpbasedStream(std::move(stream));
62 :
63 5 : LoadWithOpBasedModeNoRegister(op);
64 15 : HCCL_INFO("LoadWithOpBasedMode END");
65 5 : }
66 :
67 7 : void CollServiceDefaultImpl::UpdateUbCiIfNeed(const std::string& opTag)
68 : {
69 21 : HCCL_INFO("CollServiceDefaultImpl::UpdateUbCiIfNeed start, opTag[%s]", opTag.c_str());
70 7 : if (updatingUbCiEvent == nullptr) {
71 18 : HCCL_INFO("updatingUbCiEvent is null");
72 6 : std::vector<DevUbConnection*> devUbConns = GetStarsPollUbConns(comm->GetRmaConnManager().GetOpTagConns(opTag));
73 18 : HCCL_INFO("starsPoll devUbConns size: %lu", devUbConns.size());
74 6 : if (IfNeedUpdatingUbCi(devUbConns)) {
75 3 : HCCL_INFO("need update ub ci");
76 1 : ubCiUpdaterMgr->SaveConnsCi(opTag);
77 3 : HCCL_INFO("ubCiUpdaterMgr saveConnsCi finished");
78 1 : updatingUbCiEvent = make_unique<MaskEvent>();
79 1 : updatingUbCiEvent->Record(*(comm->GetStreamManager().opbase->GetMaster()));
80 3 : HCCL_INFO("submit event record finished");
81 : }
82 18 : HCCL_INFO("need not update ub ci");
83 6 : } else {
84 3 : HCCL_INFO("updatingUbCiEvent is not nullptr");
85 1 : auto status = updatingUbCiEvent->QueryStatus();
86 1 : if (status == HrtEventStatus::EVENT_RECORDED) {
87 3 : HCCL_INFO("updatingUbCiEvent status is EVENT_RECORDED");
88 1 : ubCiUpdaterMgr->UpdateConnsCi(opTag);
89 3 : HCCL_INFO("ubCiUpdaterMgr updateConnsCi finished");
90 1 : updatingUbCiEvent = nullptr;
91 3 : HCCL_INFO("updatingUbCiEvent reset as nullptr");
92 : }
93 3 : HCCL_INFO("updatingUbCiEvent status is %u", static_cast<u32>(status));
94 : }
95 7 : }
96 :
97 2 : void CollServiceDefaultImpl::LoadWithOffloadModeNoRegister(CollOperator& op)
98 : {
99 2 : RegisterOffloadLocalRmaBuf(op.opTag);
100 :
101 2 : shared_ptr<InsQueue> insQueue;
102 2 : insQueue = OrchestrateWithIns(op);
103 :
104 2 : vector<LinkData> links = insQueue->GetUniqueLinks();
105 :
106 : // Socket建链
107 2 : comm->GetSocketManager().BatchCreateSockets(links);
108 : // 建立RmaConnection并建链
109 2 : auto connBuilderPair = connectionsBuilders.emplace(op.opTag, make_unique<ConnectionsBuilder>(*comm));
110 2 : connBuilderPair.first->second->BatchBuild(op.opTag, links);
111 :
112 2 : AllocNotifies(links);
113 :
114 2 : AllocLocCntNotifies(*insQueue);
115 :
116 2 : comm->GetMemTransportManager()->BatchBuildOffloadTransports(op.opTag, links);
117 2 : WaitOffloadTransportReady(op.opTag);
118 6 : HCCL_INFO("Offload Interprete start");
119 :
120 2 : SaveMirrorDfxOpInfo();
121 :
122 : // 下发head算子执行计数器task
123 2 : AddCountTask(true);
124 :
125 2 : Interpreter interpreter(*comm);
126 2 : interpreter.Submit(*insQueue);
127 6 : HCCL_INFO("Offload Interprete end");
128 :
129 : // 下发tail算子执行计数器task
130 2 : AddCountTask(false);
131 :
132 : // 基于opTag+link找到connection; connection提供方法AddNop; 调用HCCP提供的AddNop(qpHandle)接口
133 2 : AddNop(op.opTag, links);
134 2 : }
135 :
136 2 : void CollServiceDefaultImpl::LoadWithOffloadMode(CollOperator& op, std::unique_ptr<Stream> stream)
137 : {
138 6 : HCCL_INFO("LoadWithOffloadMode START");
139 6 : HCCL_INFO("RegisterOffloadBuf start");
140 2 : RegisterOpBufToBufMgr(op);
141 :
142 2 : RegisterOffloadMasterStream(op.opTag, std::move(stream));
143 :
144 2 : LoadWithOffloadModeNoRegister(op);
145 :
146 6 : HCCL_INFO("LoadWithOffloadMode END");
147 2 : }
148 :
149 1 : shared_ptr<PrimQueue> CollServiceDefaultImpl::OrchestrateWithPrim(const CollAlgOperator& op) const
150 : {
151 1 : u64 tmpMemSize = comm->GetBufferSize();
152 1 : CollAlgParams params{};
153 1 : auto primQueue = make_shared<PrimQueue>();
154 :
155 1 : params.opMode = op.opMode;
156 1 : params.maxTmpMemSize = tmpMemSize;
157 :
158 3 : HCCL_INFO("orchestrate with Prim start");
159 1 : HcclResult errCode = comm->GetCollAlgComponent()->Orchestrate(op, params, comm->GetCurAlgName(), primQueue);
160 3 : HCCL_INFO("orchestrate with Prim end");
161 :
162 1 : if (errCode != HcclResult::HCCL_SUCCESS) {
163 0 : auto msg = StringFormat("Error occurs when call collAlgComponent.orchestrate(), error code: %d", errCode);
164 0 : THROW<InternalException>(msg);
165 0 : }
166 :
167 1 : return primQueue;
168 1 : }
169 :
170 9 : shared_ptr<InsQueue> CollServiceDefaultImpl::OrchestrateWithIns(const CollAlgOperator& op) const
171 : {
172 9 : u64 tmpMemSize = 0;
173 : // 图模式部分算子不需要scratchMem
174 9 : if (op.scratchMem != nullptr) {
175 9 : tmpMemSize = op.scratchMem->GetSize();
176 : }
177 9 : CollAlgParams params{};
178 9 : auto insQueue = make_shared<InsQueue>();
179 :
180 9 : params.opMode = op.opMode;
181 9 : params.maxTmpMemSize = tmpMemSize;
182 :
183 27 : HCCL_INFO("orchestrate with Ins start");
184 9 : HcclResult errCode = comm->GetCollAlgComponent()->Orchestrate(op, params, comm->GetCurAlgName(), insQueue);
185 27 : HCCL_INFO("orchestrate with Ins end");
186 :
187 9 : if (errCode != HcclResult::HCCL_SUCCESS) {
188 1 : auto msg = StringFormat("Error occurs when call collAlgComponent.orchestrate(), error code: %d", errCode);
189 1 : THROW<InternalException>(msg);
190 1 : }
191 8 : return insQueue;
192 10 : }
193 :
194 7 : void CollServiceDefaultImpl::AllocNotifies(const vector<LinkData>& links)
195 : {
196 7 : vector<LinkData> pendingLinks;
197 7 : for (auto& link : links) {
198 0 : if (Contain(availableLinks, link)) {
199 0 : continue;
200 : }
201 0 : pendingLinks.emplace_back(link);
202 : }
203 :
204 7 : if (pendingLinks.empty()) {
205 7 : return;
206 : }
207 :
208 0 : for (auto& link : pendingLinks) {
209 : // 待修改: 申请数量
210 0 : comm->GetConnLocalNotifyManager().ApplyFor(link.GetRemoteRankId(), link);
211 : }
212 :
213 0 : availableLinks.insert(pendingLinks.begin(), pendingLinks.end());
214 7 : }
215 :
216 0 : void CollServiceDefaultImpl::AllocOneLocCntNotify(const Instruction& ins) const
217 : {
218 0 : HCCL_INFO("AllocOneLocCntNotify %s begin", ins.Describe().c_str());
219 0 : vector<LinkData> links;
220 0 : const InsWaitGroupFin& insWaitGroupFin = reinterpret_cast<const InsWaitGroupFin&>(ins);
221 0 : for (auto iter = insWaitGroupFin.Iter(); iter.HasNext(); ++iter) {
222 0 : links.push_back(*iter);
223 : }
224 0 : comm->GetConnLocalCntNotifyManager().ApplyFor(insWaitGroupFin.GetTopicId(), links);
225 0 : HCCL_INFO("AllocOneLocCntNotify %s end", ins.Describe().c_str());
226 0 : }
227 :
228 7 : void CollServiceDefaultImpl::AllocLocCntNotifies(const InsQueue& insQueue) const
229 : {
230 7 : for (auto ins = insQueue.Iter(); ins.HasNext(); ++ins) {
231 0 : if (ins->GetType() == InstructionType::WAIT_GROUP_FIN) {
232 0 : AllocOneLocCntNotify(*ins);
233 : }
234 7 : }
235 :
236 7 : for (auto slaveIter = insQueue.IterSlaves(); slaveIter.HasNext(); ++slaveIter) {
237 0 : for (auto iterSlave = slaveIter->Iter(); iterSlave.HasNext(); ++iterSlave) {
238 0 : if (iterSlave->GetType() == InstructionType::WAIT_GROUP_FIN) {
239 0 : AllocOneLocCntNotify(*iterSlave);
240 : }
241 0 : }
242 7 : }
243 7 : }
244 :
245 194 : void CollServiceDefaultImpl::Init()
246 : {
247 194 : ubCiUpdaterMgr = make_unique<UbCiUpdaterManager>(&comm->GetRmaConnManager());
248 194 : primTranslator = make_unique<PrimTranslator>();
249 194 : RegisterCclLocRmaBuffer();
250 194 : }
251 :
252 3 : void CollServiceDefaultImpl::AddNop(const std::string& opTag, const vector<LinkData>& linkDataVec) const
253 : {
254 4 : for (auto& linkData : linkDataVec) {
255 1 : auto conn = comm->GetRmaConnManager().Get(opTag, linkData);
256 1 : Stream* mainStream = comm->GetStreamManager().offload->GetMaster(opTag);
257 1 : if (conn == nullptr) {
258 0 : THROW<NullPtrException>(StringFormat("CollServiceDefaultImpl::AddNop ptr is null"));
259 : }
260 1 : conn->AddNop(*mainStream);
261 : }
262 3 : }
263 :
264 1 : void CollServiceDefaultImpl::RecoverTransport(
265 : [[maybe_unused]] vector<LinkData>& links, [[maybe_unused]] vector<std::pair<LinkGroup, u32>> linkGroupPair)
266 : {
267 1 : THROW<NotSupportException>(StringFormat("CollServiceDefaultImpl::RecoverTransport not support yet."));
268 : }
269 :
270 0 : void CollServiceDefaultImpl::ReLoadWithOpBasedMode(CollOperator& op)
271 : {
272 0 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
273 0 : LoadWithOpBasedModeNoRegister(op);
274 0 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
275 0 : }
276 :
277 0 : void CollServiceDefaultImpl::ReLoadWithOffloadMode(CollOperator& op)
278 : {
279 0 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
280 0 : LoadWithOffloadModeNoRegister(op);
281 0 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
282 0 : }
283 :
284 : } // namespace Hccl
|
