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(void* mc2Tiling, void** commContext, const AcceleratorState& tilingAccelerator)
185 : {
186 0 : THROW<NotSupportException>("AllocCommResource was not support in this mode.");
187 : }
188 :
189 0 : HcclResult CollServiceBase::AllocCollOpResource(CollOperator& op, const std::string& opAlgTag, void** addr)
190 : {
191 0 : HCCL_ERROR("[%s] was not support in this mode.", __func__);
192 0 : return HCCL_E_NOT_SUPPORT;
193 : }
194 :
195 0 : void CollServiceBase::GetCcuTaskInfo(void* tilingData, void* ccuTaskGroup)
196 : {
197 0 : THROW<NotSupportException>("GetCcuTaskInfo was not support in this mode.");
198 : }
199 :
200 0 : u32 CollServiceBase::GetCcuMc2ServerNum()
201 : {
202 0 : THROW<NotSupportException>("GetCcuMc2ServerNum was not support in this mode.");
203 : return 0;
204 : }
205 :
206 0 : void CollServiceBase::Resume() { THROW<NotSupportException>("Resume was not support in this mode."); }
207 :
208 8 : void CollServiceBase::WaitOpbasedTransportReady() const
209 : {
210 16 : CHECK_NULLPTR(comm, "[CollServiceBase::WaitOpbasedTransportReady] comm is nullptr!");
211 8 : CHECK_NULLPTR(
212 8 : comm->GetMemTransportManager(),
213 : "[CollServiceBase::WaitOpbasedTransportReady] comm->GetMemTransportManager is nullptr!");
214 8 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
215 :
216 8 : HcclUs startTime = std::chrono::steady_clock::now();
217 : while (true) {
218 1043841 : if (comm->GetMemTransportManager()->IsAllOpbasedTransportReady()) {
219 7 : break;
220 : }
221 1043834 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
222 1 : string timeoutMsg = StringFormat("WaitOpbasedTransportReady timeout, commId[%s].", comm->GetId().c_str());
223 7 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}), std::vector<std::string>({timeoutMsg}));
224 3 : HCCL_ERROR(timeoutMsg.c_str());
225 1 : comm->GetMemTransportManager()->DumpNotReadyTransportsOpbased();
226 1 : THROW<InternalException>(timeoutMsg);
227 1 : }
228 1043833 : }
229 8 : }
230 :
231 7 : void CollServiceBase::WaitOffloadTransportReady(const std::string& opTag) const
232 : {
233 14 : CHECK_NULLPTR(comm, "[CollServiceBase::WaitOffloadTransportReady] comm is nullptr!");
234 7 : CHECK_NULLPTR(
235 7 : comm->GetMemTransportManager(),
236 : "[CollServiceBase::WaitOffloadTransportReady] comm->GetMemTransportManager is nullptr!");
237 7 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
238 :
239 7 : HcclUs startTime = std::chrono::steady_clock::now();
240 : while (true) {
241 866671 : if (comm->GetMemTransportManager()->IsAllOffloadTransportReady(opTag)) {
242 6 : break;
243 : }
244 866665 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
245 : string timeoutMsg = StringFormat(
246 1 : "WaitOffloadTransportReady timeout, opTag[%s] commId[%s].", opTag.c_str(), comm->GetId().c_str());
247 7 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}), std::vector<std::string>({timeoutMsg}));
248 3 : HCCL_ERROR(timeoutMsg.c_str());
249 1 : comm->GetMemTransportManager()->DumpNotReadyTransportsOffload(opTag);
250 1 : THROW<InternalException>(timeoutMsg);
251 1 : }
252 866664 : }
253 7 : }
254 :
255 2 : void CollServiceBase::WaitTransportReady(const std::string& opTag) const
256 : {
257 4 : CHECK_NULLPTR(comm, "[CollServiceBase::WaitTransportReady] comm is nullptr!");
258 2 : CHECK_NULLPTR(
259 2 : comm->GetMemTransportManager(),
260 : "[CollServiceBase::WaitTransportReady] comm->GetMemTransportManager is nullptr!");
261 2 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
262 :
263 2 : HcclUs startTime = std::chrono::steady_clock::now();
264 : while (true) {
265 2 : auto op = comm->GetCurrentCollOperator();
266 2 : if (op->opMode == OpMode::OPBASE) {
267 1 : if (comm->GetMemTransportManager()->IsAllOpbasedTransportReady()) {
268 1 : break;
269 : }
270 1 : } else if (op->opMode == OpMode::OFFLOAD) {
271 1 : if (comm->GetMemTransportManager()->IsAllOffloadTransportReady(opTag)) {
272 1 : break;
273 : }
274 : }
275 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
276 0 : RPT_INPUT_ERR(
277 : true, "EI0006", std::vector<std::string>({"reason"}),
278 : std::vector<std::string>({"WaitTransportReady timeout, SOCKET_TIMEOUT."}));
279 0 : THROW<InternalException>(
280 0 : "WaitTransportReady timeout, opTag[%s] commId[%s].", opTag.c_str(), comm->GetId().c_str());
281 : }
282 0 : }
283 2 : }
284 :
285 390 : void CollServiceBase::AddOpCounterMems()
286 : {
287 1170 : HCCL_INFO("[CollServiceBase::%s] start.", __func__);
288 :
289 390 : u64 size = 4 * 3; // 第一个四字节用于计数加1, 后面两个四字节分别保存headCounter和tailCounter
290 390 : counterBuf = std::make_shared<DevBuffer>(size);
291 :
292 : // 初始化第一个四字节置1, 用于计数加1, reduce task add 1
293 390 : u64 srcSize = 4;
294 390 : float srcValue = 1;
295 390 : void* srcAddr = reinterpret_cast<void*>(counterBuf->GetAddr());
296 390 : HrtMemcpy(srcAddr, srcSize, &srcValue, srcSize, RT_MEMCPY_HOST_TO_DEVICE);
297 :
298 : // 初始化后面两个四字节置0
299 390 : u64 countMemSize = srcSize;
300 390 : float startValue = 0; // value为0表示从0开始计数
301 390 : void* headCountAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + srcSize);
302 390 : void* tailCountAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + srcSize * 2);
303 390 : HrtMemcpy(headCountAddr, countMemSize, &startValue, countMemSize, RT_MEMCPY_HOST_TO_DEVICE);
304 390 : HrtMemcpy(tailCountAddr, countMemSize, &startValue, countMemSize, RT_MEMCPY_HOST_TO_DEVICE);
305 1170 : HCCL_INFO(
306 : "[CollServiceBase::%s] end, counterBuf[%llu] srcAddr[%p] headCountAddr[%p] tailCountAddr[%p].", __func__,
307 : counterBuf->GetAddr(), srcAddr, headCountAddr, tailCountAddr);
308 390 : }
309 :
310 397 : std::pair<u32, u32> CollServiceBase::GetOpCount()
311 : {
312 1191 : HCCL_INFO("[CollServiceBase::%s] start.", __func__);
313 :
314 397 : std::pair<float, float> floatCounter;
315 397 : u64 size = 4;
316 397 : if (counterBuf->GetSize() < size * ADDR_SIZE) {
317 0 : THROW<InternalException>("counterBuf size[%zu] is less than %u bytes", counterBuf->GetSize(), size * ADDR_SIZE);
318 : }
319 397 : void* headAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + size);
320 397 : void* tailAddr = reinterpret_cast<void*>(counterBuf->GetAddr() + size * 2);
321 397 : HrtMemcpy(&floatCounter.first, size, headAddr, size, RT_MEMCPY_DEVICE_TO_HOST);
322 397 : HrtMemcpy(&floatCounter.second, size, tailAddr, size, RT_MEMCPY_DEVICE_TO_HOST);
323 :
324 397 : std::pair<u32, u32> counter;
325 397 : counter.first = static_cast<u32>(floatCounter.first);
326 397 : counter.second = static_cast<u32>(floatCounter.second);
327 :
328 1191 : HCCL_INFO("[CollServiceBase::%s] end, head:%u, tail:%u", __func__, counter.first, counter.second);
329 397 : return counter;
330 : }
331 :
332 2 : DevBuffer* CollServiceBase::GetOpCounterBuf() { return counterBuf.get(); }
333 :
334 677 : CollServiceBase::~CollServiceBase()
335 : {
336 677 : if (counterBuf == nullptr) {
337 280 : return;
338 : }
339 : // 用于图模式算子计数打印，待有心跳检测后适配删除
340 1191 : DECTOR_TRY_CATCH("CollServiceBase", {
341 : auto count = GetOpCount();
342 : HCCL_INFO("[CollServiceBase::~CollServiceBase] head:%u, tail:%u", count.first, count.second);
343 : });
344 677 : }
345 :
346 : // 功能说明：等待transport建链完成
347 : // 输入说明：string &opTag：通信域ID，唯一标记一个通信域
348 1 : bool CollServiceBase::IsAllTransportRecoveredReady(const std::string& opTag)
349 : {
350 1 : CHECK_NULLPTR(comm, "[CollServiceBase::IsAllTransportRecoveredReady] comm is nullptr!");
351 1 : auto op = comm->GetCurrentCollOperator();
352 1 : if (op->opMode == OpMode::OPBASE) {
353 1 : return comm->GetMemTransportManager()->IsAllOpbasedTransportRecoveredReady();
354 0 : } else if (op->opMode == OpMode::OFFLOAD) {
355 0 : return comm->GetMemTransportManager()->IsAllOffloadTransportRecoveredReady(op->opTag);
356 : }
357 0 : HCCL_ERROR("[CollServiceBase][IsAllTransportRecoveredReady] opMode[%d] is invalid", op->opMode);
358 0 : return false;
359 : }
360 :
361 0 : HcclResult CollServiceBase::GetSnapShotDynamicBuf(CollOperator& op, BinaryStream& buf)
362 : {
363 0 : HCCL_ERROR("[%s] not support.", __func__);
364 0 : return HCCL_E_NOT_SUPPORT;
365 : }
366 :
367 : constexpr u32 TAILADDR_OFFSET_MULTIPLIER = 2;
368 7 : void CollServiceBase::SaveMirrorDfxOpInfo()
369 : {
370 7 : auto dfxOpInfo = std::make_shared<DfxOpInfo>();
371 7 : CHECK_NULLPTR(comm, "[CollServiceBase::SaveMirrorDfxOpInfo] comm is nullptr!");
372 :
373 7 : dfxOpInfo->op_ = *comm->GetCurrentCollOperator();
374 7 : dfxOpInfo->tag_ = dfxOpInfo->op_.opTag;
375 7 : dfxOpInfo->algType_ = comm->GetCurAlgName().c_str();
376 7 : dfxOpInfo->commIndex_ = comm->GetIdIndex();
377 7 : dfxOpInfo->comm_ = comm;
378 7 : dfxOpInfo->beginTime_ = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
379 7 : dfxOpInfo->commId_ = comm->GetId();
380 7 : dfxOpInfo->opIndex_ = comm->GetOpIndex();
381 7 : u64 size = 4;
382 7 : dfxOpInfo->headOpCounterAddr_ = counterBuf->GetAddr() + size;
383 7 : dfxOpInfo->tailOpCounterAddr_ = counterBuf->GetAddr() + size * TAILADDR_OFFSET_MULTIPLIER;
384 :
385 7 : comm->GetMirrorTaskManager().SetCurrDfxOpInfo(dfxOpInfo);
386 7 : }
387 :
388 3 : void CollServiceBase::AddCountTask(bool isHead)
389 : {
390 3 : if (counterBuf == nullptr) {
391 1 : AddOpCounterMems();
392 : }
393 3 : CHECK_NULLPTR(comm, "[CollServiceBase::AddCountTask] comm is nullptr!");
394 :
395 3 : u64 size = sizeof(float);
396 3 : void* dst = isHead == true ? reinterpret_cast<void*>(counterBuf->GetAddr() + size) :
397 1 : reinterpret_cast<void*>(counterBuf->GetAddr() + size * 2);
398 3 : void* src = reinterpret_cast<void*>(counterBuf->GetAddr());
399 :
400 : // 下发reduce task
401 3 : aclrtReduceKind rtReduceOp = ACL_RT_MEMCPY_SDMA_AUTOMATIC_SUM;
402 3 : aclDataType rtDataType = ACL_FLOAT;
403 3 : aclrtStream streamPtr = comm->GetStreamManager().GetMaster()->GetPtr();
404 3 : CHECK_NULLPTR(streamPtr, "master stream pointer is nullptr!");
405 3 : HrtReduceAsync(dst, size, src, size, rtReduceOp, rtDataType, streamPtr);
406 :
407 9 : HCCL_INFO("[CollServiceBase::AddCountTask] isHead[%d] dst addr[%p] src addr[%p]", isHead, dst, src);
408 3 : }
409 :
410 1 : void CollServiceBase::ReLoadWithOpBasedMode(CollOperator& op)
411 : {
412 1 : THROW<NotSupportException>("ReLoadWithOpBasedMode was not support in this mode.");
413 : }
414 :
415 1 : void CollServiceBase::ReLoadWithOffloadMode(CollOperator& op)
416 : {
417 1 : THROW<NotSupportException>("ReLoadWithOffloadMode was not support in this mode.");
418 : }
419 :
420 11 : void CollServiceBase::AllocQueueNotify(const InsQueue& insQueue)
421 : {
422 11 : if (insQueue.SizeOfSlaves() == 0)
423 11 : return;
424 0 : AllocQNotifyForSingleQ(insQueue);
425 :
426 0 : for (auto slaveIt = insQueue.IterSlaves(); slaveIt.HasNext(); ++slaveIt) {
427 0 : AllocQNotifyForSingleQ(*slaveIt);
428 0 : }
429 : }
430 :
431 4 : void CollServiceBase::AllocQNotifyForSingleQ(const InsQueue& insQueue) const
432 : {
433 4 : auto& queueNotifyManager = comm->GetCcuQueueNotifyManager();
434 8 : for (auto it = insQueue.Iter(); it.HasNext(); ++it) {
435 4 : const Instruction& ins = *it;
436 4 : auto insType = ins.GetType();
437 4 : if (insType == InstructionType::LOCAL_POST_TO) {
438 1 : const auto& p = static_cast<const InsLocalPostTo&>(ins);
439 1 : queueNotifyManager.ApplyFor(p.GetPostQid(), p.GetWaitQid(), p.GetTopicId());
440 3 : } else if (insType == InstructionType::LOCAL_WAIT_FROM) {
441 1 : const auto& p = static_cast<const InsLocalWaitFrom&>(ins);
442 1 : queueNotifyManager.ApplyFor(p.GetPostQid(), p.GetWaitQid(), p.GetTopicId());
443 2 : } else if (insType == InstructionType::LOCAL_WAIT_GROUP) {
444 1 : auto& queueWaitGroupCntNotifyManager = comm->GetQueueWaitGroupCntNotifyManager();
445 1 : const auto& p = static_cast<const InsLocalWaitGroup&>(ins);
446 1 : queueWaitGroupCntNotifyManager.ApplyFor(p.GetWaitQid(), p.GetTopicId());
447 1 : } else if (insType == InstructionType::LOCAL_BCAST_POST) {
448 1 : auto& queueBcastPostCntNotifyManager = comm->GetBcastPostCntNotifyManager();
449 1 : const auto& p = static_cast<const InsLocalBcastPost&>(ins);
450 1 : queueBcastPostCntNotifyManager.ApplyFor(p.GetPostQid(), p.GetTopicId());
451 : }
452 4 : }
453 4 : }
454 :
455 1 : HcclResult CollServiceBase::GetAlgExecParam(bool clearEnable, u32 numBlocks, void*& commContext, u64& len)
456 : {
457 3 : HCCL_ERROR("GetAlgExecParam was not support in this mode.");
458 1 : return HCCL_E_NOT_SUPPORT;
459 : }
460 :
461 : } // namespace Hccl
|
