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