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_base.h"
12 : #include "communicator_impl.h"
13 : #include "env_config_v2.h"
14 : #include "dlprof_function_v2.h"
15 : namespace Hccl {
16 :
17 : constexpr u32 ADDR_SIZE = 2;
18 :
19 17 : void CollServiceBase::RegisterOpBufToBufMgr(CollOperator& op)
20 : {
21 17 : CHECK_NULLPTR(comm, "[CollServiceBase::RegisterOpBufToBufMgr] comm is nullptr!");
22 17 : DataBufManager& dataBufManager = comm->GetDataBufferManager();
23 17 : dataBufManager.Register(op.opTag, BufferType::INPUT, op.inputMem);
24 17 : dataBufManager.Register(op.opTag, BufferType::OUTPUT, op.outputMem);
25 17 : dataBufManager.Register(op.opTag, BufferType::SCRATCH, op.scratchMem);
26 17 : }
27 :
28 194 : void CollServiceBase::RegisterCclLocRmaBuffer() const // 注册CCL buffer
29 : {
30 194 : if (comm->GetRankSize() == 1) {
31 11 : return;
32 : }
33 366 : CHECK_NULLPTR(comm, "[CollServiceBase::RegisterCclLocRmaBuffer] comm is nullptr!");
34 183 : CHECK_NULLPTR(comm->GetRankGraph(), "[CollServiceBase::RegisterCclLocRmaBuffer] comm->newVirtualTopo is nullptr!");
35 183 : auto myself = comm->GetRankGraph()->GetPeer(comm->GetMyRank());
36 183 : if (myself == nullptr) {
37 0 : THROW<NullPtrException>(StringFormat("CollServiceAiCpuImpl::Init ptr is null"));
38 : }
39 183 : bool p2pRegistered = false;
40 183 : const auto& interfacesMap = myself->GetInterfacesMap();
41 355 : for (const auto& pair : interfacesMap) {
42 172 : const auto& ifaceVec = pair.second;
43 346 : for (const auto& connIface : ifaceVec) {
44 174 : std::set<LinkProtocol> protocols = connIface->GetLinkProtocols();
45 174 : if (protocols.find(LinkProtocol::HCCS) != protocols.end()
46 174 : || protocols.find(LinkProtocol::PCIE) != protocols.end()) {
47 0 : if (p2pRegistered) {
48 0 : break;
49 : }
50 0 : p2pRegistered = true;
51 : }
52 174 : auto& rmaBufManager = comm->GetLocalRmaBufManager();
53 522 : HCCL_INFO("rmaBufManager reg");
54 174 : PortData portData(comm->GetMyRank(), *connIface);
55 522 : HCCL_INFO("rmaBufManager reg portData[%s]", portData.Describe().c_str());
56 174 : if (portData.GetType() == PortDeploymentType::HOST_NET && portData.GetProto() == LinkProtoType::RDMA) {
57 0 : HCCL_WARNING("HostNet RDMA port will not be reged here"); // 1825不支持重复注册MR
58 0 : continue;
59 0 : }
60 174 : rmaBufManager.Reg(comm->GetId(), BufferType::SCRATCH, comm->GetCclBuffer(), portData, *(protocols.begin()));
61 174 : }
62 : }
63 183 : }
64 :
65 6 : void CollServiceBase::RegisterCclBuffer(const std::vector<LinkData>& links) const
66 : {
67 18 : HCCL_INFO("RegisterCclBuffer reg links.size(%u)", links.size());
68 6 : CHECK_NULLPTR(comm, "[CollServiceBase::RegisterCclBuffer] comm is nullptr!");
69 9 : for (auto& link : links) {
70 3 : PortData portData = link.GetLocalPort();
71 9 : HCCL_INFO("RegisterCclBuffer reg portData[%s]", portData.Describe().c_str());
72 :
73 3 : auto& rmaBufManager = comm->GetLocalRmaBufManager();
74 9 : HCCL_INFO("RegisterCclBuffer reg");
75 3 : if (rmaBufManager.Get(comm->GetId(), portData, BufferType::SCRATCH) != nullptr) {
76 3 : HCCL_WARNING(
77 : "RegisterCclBuffer has reged, optag(%s) portData[%s]", comm->GetId().c_str(),
78 : portData.Describe().c_str());
79 1 : continue;
80 1 : }
81 2 : rmaBufManager.Reg(comm->GetId(), BufferType::SCRATCH, comm->GetCclBuffer(), portData, link.GetLinkProtocol());
82 : }
83 6 : }
84 :
85 8 : void CollServiceBase::RegisterOpbasedStream(unique_ptr<Stream> stream)
86 : {
87 8 : CHECK_NULLPTR(comm, "[CollServiceBase::RegisterOpbasedStream] comm is nullptr!");
88 8 : StreamManager& sm = comm->GetStreamManager();
89 8 : CHECK_NULLPTR(sm.opbase, "[CollServiceBase::RegisterOpbasedStream] sm.opbase is nullptr!");
90 8 : sm.opbase->RegisterMaster(std::move(stream));
91 8 : }
92 :
93 4 : void CollServiceBase::RegisterOpbasedLocalRmaBuf(const std::string& opTag) const
94 : {
95 8 : std::vector<BufferType> bufTypes = {BufferType::INPUT, BufferType::OUTPUT, BufferType::SCRATCH};
96 4 : std::unordered_map<BufferType, shared_ptr<DevBuffer>, std::EnumClassHash> devBuffers;
97 4 : CHECK_NULLPTR(comm, "[CollServiceBase::RegisterOpbasedLocalRmaBuf] comm is nullptr!");
98 4 : DataBufManager& dataBufManager = comm->GetDataBufferManager();
99 16 : for (auto& bufType : bufTypes) {
100 12 : auto dataBuf = dataBufManager.Get(opTag, bufType);
101 12 : if (dataBuf != nullptr) {
102 6 : devBuffers[bufType] = DevBuffer::Create(dataBuf->GetAddr(), dataBuf->GetSize());
103 : } else {
104 18 : HCCL_WARNING("dataBuf[type=%s] is nullptr", bufType.Describe().c_str());
105 : }
106 : }
107 :
108 8 : CHECK_NULLPTR(
109 8 : comm->GetRankGraph(), "[CollServiceBase::RegisterOpbasedLocalRmaBuf] comm->newVirtualTopo is nullptr!");
110 4 : auto myself = comm->GetRankGraph()->GetPeer(comm->GetMyRank());
111 4 : if (myself == nullptr) {
112 0 : THROW<NullPtrException>(StringFormat("CollServiceDefaultImpl::Init ptr is null"));
113 : }
114 4 : auto& localRmaBufManager = comm->GetLocalRmaBufManager();
115 4 : const auto& interfacesMap = myself->GetInterfacesMap();
116 8 : for (const auto& pair : interfacesMap) {
117 4 : const auto& ifaceVec = pair.second;
118 8 : for (const auto& connIface : ifaceVec) {
119 4 : PortData portData(comm->GetMyRank(), *connIface);
120 4 : std::set<LinkProtocol> protocols = connIface->GetLinkProtocols();
121 10 : for (auto& devBuf : devBuffers) {
122 6 : if (localRmaBufManager.Get(comm->GetId(), portData, devBuf.first) != nullptr) {
123 12 : HCCL_WARNING(
124 : "RegisterOpbasedLocalRmaBuf has reged, bufferType[%s], optag[%s] portData[%s]",
125 : devBuf.first.Describe().c_str(), comm->GetId().c_str(), portData.Describe().c_str());
126 4 : continue;
127 4 : }
128 2 : if (devBuf.first != BufferType::SCRATCH && portData.GetType() == PortDeploymentType::P2P) {
129 0 : HCCL_WARNING("Input and Output Mem will not be reged at P2P");
130 0 : continue;
131 0 : }
132 2 : localRmaBufManager.Reg(opTag, devBuf.first, devBuf.second, portData, *(protocols.begin()));
133 : }
134 4 : }
135 : }
136 4 : }
137 :
138 4 : void CollServiceBase::RegisterOffloadLocalRmaBuf(const std::string& opTag) const
139 : {
140 8 : std::vector<BufferType> bufTypes = {BufferType::INPUT, BufferType::OUTPUT, BufferType::SCRATCH};
141 4 : std::unordered_map<BufferType, shared_ptr<DevBuffer>, std::EnumClassHash> devBuffers;
142 4 : CHECK_NULLPTR(comm, "[CollServiceBase::RegisterOffloadLocalRmaBuf] comm is nullptr!");
143 4 : DataBufManager& dataBufManager = comm->GetDataBufferManager();
144 16 : for (auto& bufType : bufTypes) {
145 12 : auto dataBuf = dataBufManager.Get(opTag, bufType);
146 12 : if (dataBuf != nullptr) {
147 8 : devBuffers[bufType] = DevBuffer::Create(dataBuf->GetAddr(), dataBuf->GetSize());
148 : } else {
149 12 : HCCL_WARNING("dataBuf[type=%s] is nullptr", bufType.Describe().c_str());
150 : }
151 : }
152 :
153 8 : CHECK_NULLPTR(
154 8 : comm->GetRankGraph(), "[CollServiceBase::RegisterOffloadLocalRmaBuf] comm->newVirtualTopo is nullptr!");
155 4 : auto myself = comm->GetRankGraph()->GetPeer(comm->GetMyRank());
156 4 : if (myself == nullptr) {
157 0 : THROW<NullPtrException>(StringFormat("CollServiceDefaultImpl::Init ptr is null"));
158 : }
159 4 : auto& localRmaBufManager = comm->GetLocalRmaBufManager();
160 4 : const auto& interfacesMap = myself->GetInterfacesMap();
161 14 : for (const auto& pair : interfacesMap) {
162 10 : const auto& ifaceVec = pair.second;
163 26 : for (const auto& connIface : ifaceVec) {
164 16 : PortData portData(comm->GetMyRank(), *connIface);
165 16 : std::set<LinkProtocol> protocols = connIface->GetLinkProtocols();
166 56 : for (auto& devBuf : devBuffers) {
167 120 : HCCL_INFO("CollServiceBase::RegisterOffloadLocalRmaBuf, devBuf[%s]", devBuf.second->Describe().c_str());
168 40 : localRmaBufManager.Reg(opTag, devBuf.first, devBuf.second, portData, *(protocols.begin()));
169 : }
170 16 : }
171 : }
172 4 : }
173 :
174 7 : void CollServiceBase::RegisterOffloadMasterStream(const std::string& opTag, unique_ptr<Stream> stream) const
175 : {
176 7 : CHECK_NULLPTR(comm, "[CollServiceBase::RegisterOffloadMasterStream] comm is nullptr!");
177 7 : StreamManager& sm = comm->GetStreamManager();
178 6 : CHECK_NULLPTR(sm.offload, "[CollServiceBase::RegisterOffloadMasterStream] sm.offload is nullptr!");
179 6 : sm.offload->RegisterMaster(opTag, std::move(stream));
180 6 : }
181 :
182 677 : CollServiceBase::CollServiceBase(CommunicatorImpl* comm) : comm(comm) {}
183 :
184 0 : void CollServiceBase::AllocCommResource(
185 : [[maybe_unused]] void* mc2Tiling, [[maybe_unused]] void** commContext,
186 : [[maybe_unused]] const AcceleratorState& tilingAccelerator)
187 : {
188 0 : THROW<NotSupportException>("AllocCommResource was not support in this mode.");
189 : }
190 :
191 0 : HcclResult CollServiceBase::AllocCollOpResource(
192 : [[maybe_unused]] CollOperator& op, [[maybe_unused]] const std::string& opAlgTag, [[maybe_unused]] void** addr)
193 : {
194 0 : HCCL_ERROR("[%s] was not support in this mode.", __func__);
195 0 : return HCCL_E_NOT_SUPPORT;
196 : }
197 :
198 0 : void CollServiceBase::GetCcuTaskInfo([[maybe_unused]] void* tilingData, [[maybe_unused]] void* ccuTaskGroup)
199 : {
200 0 : THROW<NotSupportException>("GetCcuTaskInfo was not support in this mode.");
201 : }
202 :
203 0 : u32 CollServiceBase::GetCcuMc2ServerNum()
204 : {
205 0 : THROW<NotSupportException>("GetCcuMc2ServerNum was not support in this mode.");
206 : return 0;
207 : }
208 :
209 0 : void CollServiceBase::Resume() { THROW<NotSupportException>("Resume was not support in this mode."); }
210 :
211 8 : void CollServiceBase::WaitOpbasedTransportReady() const
212 : {
213 16 : CHECK_NULLPTR(comm, "[CollServiceBase::WaitOpbasedTransportReady] comm is nullptr!");
214 8 : CHECK_NULLPTR(
215 8 : comm->GetMemTransportManager(),
216 : "[CollServiceBase::WaitOpbasedTransportReady] comm->GetMemTransportManager is nullptr!");
217 8 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
218 :
219 8 : HcclUs startTime = std::chrono::steady_clock::now();
220 : while (true) {
221 1119573 : if (comm->GetMemTransportManager()->IsAllOpbasedTransportReady()) {
222 7 : break;
223 : }
224 1119566 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
225 1 : string timeoutMsg = StringFormat("WaitOpbasedTransportReady timeout, commId[%s].", comm->GetId().c_str());
226 7 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}), std::vector<std::string>({timeoutMsg}));
227 3 : HCCL_ERROR(timeoutMsg.c_str());
228 1 : comm->GetMemTransportManager()->DumpNotReadyTransportsOpbased();
229 1 : THROW<InternalException>(timeoutMsg);
230 1 : }
231 1119565 : }
232 8 : }
233 :
234 7 : void CollServiceBase::WaitOffloadTransportReady(const std::string& opTag) const
235 : {
236 14 : CHECK_NULLPTR(comm, "[CollServiceBase::WaitOffloadTransportReady] comm is nullptr!");
237 7 : CHECK_NULLPTR(
238 7 : comm->GetMemTransportManager(),
239 : "[CollServiceBase::WaitOffloadTransportReady] comm->GetMemTransportManager is nullptr!");
240 7 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
241 :
242 7 : HcclUs startTime = std::chrono::steady_clock::now();
243 : while (true) {
244 916514 : if (comm->GetMemTransportManager()->IsAllOffloadTransportReady(opTag)) {
245 6 : break;
246 : }
247 916508 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
248 : string timeoutMsg = StringFormat(
249 1 : "WaitOffloadTransportReady timeout, opTag[%s] commId[%s].", opTag.c_str(), comm->GetId().c_str());
250 7 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}), std::vector<std::string>({timeoutMsg}));
251 3 : HCCL_ERROR(timeoutMsg.c_str());
252 1 : comm->GetMemTransportManager()->DumpNotReadyTransportsOffload(opTag);
253 1 : THROW<InternalException>(timeoutMsg);
254 1 : }
255 916507 : }
256 7 : }
257 :
258 2 : void CollServiceBase::WaitTransportReady(const std::string& opTag) const
259 : {
260 4 : CHECK_NULLPTR(comm, "[CollServiceBase::WaitTransportReady] comm is nullptr!");
261 2 : CHECK_NULLPTR(
262 2 : comm->GetMemTransportManager(),
263 : "[CollServiceBase::WaitTransportReady] comm->GetMemTransportManager is nullptr!");
264 2 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
265 :
266 2 : HcclUs startTime = std::chrono::steady_clock::now();
267 : while (true) {
268 2 : auto op = comm->GetCurrentCollOperator();
269 2 : if (op->opMode == OpMode::OPBASE) {
270 1 : if (comm->GetMemTransportManager()->IsAllOpbasedTransportReady()) {
271 1 : break;
272 : }
273 1 : } else if (op->opMode == OpMode::OFFLOAD) {
274 1 : if (comm->GetMemTransportManager()->IsAllOffloadTransportReady(opTag)) {
275 1 : break;
276 : }
277 : }
278 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
279 0 : RPT_INPUT_ERR(
280 : true, "EI0006", std::vector<std::string>({"reason"}),
281 : std::vector<std::string>({"WaitTransportReady timeout, SOCKET_TIMEOUT."}));
282 0 : THROW<InternalException>(
283 0 : "WaitTransportReady timeout, opTag[%s] commId[%s].", opTag.c_str(), comm->GetId().c_str());
284 : }
285 0 : }
286 2 : }
287 :
288 390 : void CollServiceBase::AddOpCounterMems()
289 : {
290 1170 : HCCL_INFO("[CollServiceBase::%s] start.", __func__);
291 :
292 390 : u64 size = 4 * 3; // 第一个四字节用于计数加1, 后面两个四字节分别保存headCounter和tailCounter
293 390 : counterBuf = std::make_shared<DevBuffer>(size);
294 :
295 : // 初始化第一个四字节置1, 用于计数加1, reduce task add 1
296 390 : u64 srcSize = 4;
297 390 : float srcValue = 1;
298 390 : void* srcAddr = reinterpret_cast<void*>(counterBuf->GetAddr());
299 390 : HrtMemcpy(srcAddr, srcSize, &srcValue, srcSize, RT_MEMCPY_HOST_TO_DEVICE);
300 :
301 : // 初始化后面两个四字节置0
302 390 : u64 countMemSize = srcSize;
303 390 : float startValue = 0; // value为0表示从0开始计数
304 390 : void* headCountAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + srcSize);
305 390 : void* tailCountAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + srcSize * 2);
306 390 : HrtMemcpy(headCountAddr, countMemSize, &startValue, countMemSize, RT_MEMCPY_HOST_TO_DEVICE);
307 390 : HrtMemcpy(tailCountAddr, countMemSize, &startValue, countMemSize, RT_MEMCPY_HOST_TO_DEVICE);
308 1170 : HCCL_INFO(
309 : "[CollServiceBase::%s] end, counterBuf[%llu] srcAddr[%p] headCountAddr[%p] tailCountAddr[%p].", __func__,
310 : counterBuf->GetAddr(), srcAddr, headCountAddr, tailCountAddr);
311 390 : }
312 :
313 397 : std::pair<u32, u32> CollServiceBase::GetOpCount()
314 : {
315 1191 : HCCL_INFO("[CollServiceBase::%s] start.", __func__);
316 :
317 397 : std::pair<float, float> floatCounter;
318 397 : u64 size = 4;
319 397 : if (counterBuf->GetSize() < size * ADDR_SIZE) {
320 0 : THROW<InternalException>("counterBuf size[%zu] is less than %u bytes", counterBuf->GetSize(), size * ADDR_SIZE);
321 : }
322 397 : void* headAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + size);
323 397 : void* tailAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + size * 2);
324 397 : HrtMemcpy(&floatCounter.first, size, headAddr, size, RT_MEMCPY_DEVICE_TO_HOST);
325 397 : HrtMemcpy(&floatCounter.second, size, tailAddr, size, RT_MEMCPY_DEVICE_TO_HOST);
326 :
327 397 : std::pair<u32, u32> counter;
328 397 : counter.first = static_cast<u32>(floatCounter.first);
329 397 : counter.second = static_cast<u32>(floatCounter.second);
330 :
331 1191 : HCCL_INFO("[CollServiceBase::%s] end, head:%u, tail:%u", __func__, counter.first, counter.second);
332 397 : return counter;
333 : }
334 :
335 2 : DevBuffer* CollServiceBase::GetOpCounterBuf() { return counterBuf.get(); }
336 :
337 677 : CollServiceBase::~CollServiceBase()
338 : {
339 677 : if (counterBuf == nullptr) {
340 280 : return;
341 : }
342 : // 用于图模式算子计数打印，待有心跳检测后适配删除
343 1191 : DECTOR_TRY_CATCH("CollServiceBase", {
344 : auto count = GetOpCount();
345 : HCCL_INFO("[CollServiceBase::~CollServiceBase] head:%u, tail:%u", count.first, count.second);
346 : });
347 677 : }
348 :
349 : // 功能说明：等待transport建链完成
350 : // 输入说明：string &opTag：通信域ID，唯一标记一个通信域
351 1 : bool CollServiceBase::IsAllTransportRecoveredReady([[maybe_unused]] const std::string& opTag)
352 : {
353 1 : CHECK_NULLPTR(comm, "[CollServiceBase::IsAllTransportRecoveredReady] comm is nullptr!");
354 1 : auto op = comm->GetCurrentCollOperator();
355 1 : if (op->opMode == OpMode::OPBASE) {
356 1 : return comm->GetMemTransportManager()->IsAllOpbasedTransportRecoveredReady();
357 0 : } else if (op->opMode == OpMode::OFFLOAD) {
358 0 : return comm->GetMemTransportManager()->IsAllOffloadTransportRecoveredReady(op->opTag);
359 : }
360 0 : HCCL_ERROR("[CollServiceBase][IsAllTransportRecoveredReady] opMode[%d] is invalid", op->opMode);
361 0 : return false;
362 : }
363 :
364 0 : HcclResult CollServiceBase::GetSnapShotDynamicBuf([[maybe_unused]] CollOperator& op, [[maybe_unused]] BinaryStream& buf)
365 : {
366 0 : HCCL_ERROR("[%s] not support.", __func__);
367 0 : return HCCL_E_NOT_SUPPORT;
368 : }
369 :
370 : constexpr u32 TAILADDR_OFFSET_MULTIPLIER = 2;
371 7 : void CollServiceBase::SaveMirrorDfxOpInfo()
372 : {
373 7 : auto dfxOpInfo = std::make_shared<DfxOpInfo>();
374 7 : CHECK_NULLPTR(comm, "[CollServiceBase::SaveMirrorDfxOpInfo] comm is nullptr!");
375 :
376 7 : dfxOpInfo->op_ = *comm->GetCurrentCollOperator();
377 7 : dfxOpInfo->tag_ = dfxOpInfo->op_.opTag;
378 7 : dfxOpInfo->algType_ = comm->GetCurAlgName().c_str();
379 7 : dfxOpInfo->commIndex_ = comm->GetIdIndex();
380 7 : dfxOpInfo->comm_ = comm;
381 7 : dfxOpInfo->beginTime_ = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
382 7 : dfxOpInfo->commId_ = comm->GetId();
383 7 : dfxOpInfo->opIndex_ = comm->GetOpIndex();
384 7 : u64 size = 4;
385 7 : dfxOpInfo->headOpCounterAddr_ = counterBuf->GetAddr() + size;
386 7 : dfxOpInfo->tailOpCounterAddr_ = counterBuf->GetAddr() + size * TAILADDR_OFFSET_MULTIPLIER;
387 :
388 7 : comm->GetMirrorTaskManager().SetCurrDfxOpInfo(dfxOpInfo);
389 7 : }
390 :
391 3 : void CollServiceBase::AddCountTask(bool isHead)
392 : {
393 3 : if (counterBuf == nullptr) {
394 1 : AddOpCounterMems();
395 : }
396 3 : CHECK_NULLPTR(comm, "[CollServiceBase::AddCountTask] comm is nullptr!");
397 :
398 3 : u64 size = sizeof(float);
399 3 : void* dst = isHead == true ? reinterpret_cast<void*>(counterBuf->GetAddr() + size) :
400 1 : reinterpret_cast<void*>(counterBuf->GetAddr() + size * 2);
401 3 : void* src = reinterpret_cast<void*>(counterBuf->GetAddr());
402 :
403 : // 下发reduce task
404 3 : aclrtReduceKind rtReduceOp = ACL_RT_MEMCPY_SDMA_AUTOMATIC_SUM;
405 3 : aclDataType rtDataType = ACL_FLOAT;
406 3 : aclrtStream streamPtr = comm->GetStreamManager().GetMaster()->GetPtr();
407 3 : CHECK_NULLPTR(streamPtr, "master stream pointer is nullptr!");
408 3 : HrtReduceAsync(dst, size, src, size, rtReduceOp, rtDataType, streamPtr);
409 :
410 9 : HCCL_INFO("[CollServiceBase::AddCountTask] isHead[%d] dst addr[%p] src addr[%p]", isHead, dst, src);
411 3 : }
412 :
413 1 : void CollServiceBase::ReLoadWithOpBasedMode([[maybe_unused]] CollOperator& op)
414 : {
415 1 : THROW<NotSupportException>("ReLoadWithOpBasedMode was not support in this mode.");
416 : }
417 :
418 1 : void CollServiceBase::ReLoadWithOffloadMode([[maybe_unused]] CollOperator& op)
419 : {
420 1 : THROW<NotSupportException>("ReLoadWithOffloadMode was not support in this mode.");
421 : }
422 :
423 11 : void CollServiceBase::AllocQueueNotify(const InsQueue& insQueue)
424 : {
425 11 : if (insQueue.SizeOfSlaves() == 0)
426 11 : return;
427 0 : AllocQNotifyForSingleQ(insQueue);
428 :
429 0 : for (auto slaveIt = insQueue.IterSlaves(); slaveIt.HasNext(); ++slaveIt) {
430 0 : AllocQNotifyForSingleQ(*slaveIt);
431 0 : }
432 : }
433 :
434 4 : void CollServiceBase::AllocQNotifyForSingleQ(const InsQueue& insQueue) const
435 : {
436 4 : auto& queueNotifyManager = comm->GetCcuQueueNotifyManager();
437 8 : for (auto it = insQueue.Iter(); it.HasNext(); ++it) {
438 4 : const Instruction& ins = *it;
439 4 : auto insType = ins.GetType();
440 4 : if (insType == InstructionType::LOCAL_POST_TO) {
441 1 : const auto& p = static_cast<const InsLocalPostTo&>(ins);
442 1 : queueNotifyManager.ApplyFor(p.GetPostQid(), p.GetWaitQid(), p.GetTopicId());
443 3 : } else if (insType == InstructionType::LOCAL_WAIT_FROM) {
444 1 : const auto& p = static_cast<const InsLocalWaitFrom&>(ins);
445 1 : queueNotifyManager.ApplyFor(p.GetPostQid(), p.GetWaitQid(), p.GetTopicId());
446 2 : } else if (insType == InstructionType::LOCAL_WAIT_GROUP) {
447 1 : auto& queueWaitGroupCntNotifyManager = comm->GetQueueWaitGroupCntNotifyManager();
448 1 : const auto& p = static_cast<const InsLocalWaitGroup&>(ins);
449 1 : queueWaitGroupCntNotifyManager.ApplyFor(p.GetWaitQid(), p.GetTopicId());
450 1 : } else if (insType == InstructionType::LOCAL_BCAST_POST) {
451 1 : auto& queueBcastPostCntNotifyManager = comm->GetBcastPostCntNotifyManager();
452 1 : const auto& p = static_cast<const InsLocalBcastPost&>(ins);
453 1 : queueBcastPostCntNotifyManager.ApplyFor(p.GetPostQid(), p.GetTopicId());
454 : }
455 4 : }
456 4 : }
457 :
458 1 : HcclResult CollServiceBase::GetAlgExecParam(
459 : [[maybe_unused]] bool clearEnable, [[maybe_unused]] u32 numBlocks, [[maybe_unused]] void*& commContext,
460 : [[maybe_unused]] u64& len)
461 : {
462 3 : HCCL_ERROR("GetAlgExecParam was not support in this mode.");
463 1 : return HCCL_E_NOT_SUPPORT;
464 : }
465 :
466 : } // namespace Hccl
|
