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