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 "ins_rules.h"
12 : #include "null_ptr_exception.h"
13 : #include "not_support_exception.h"
14 : #include "queue_wait_group_cnt_notify_manager.h"
15 : #include "queue_bcast_post_cnt_notify_manager.h"
16 : #include "cnt_notify_res_helper.h"
17 : #include "data_type.h"
18 : #include "reduce_op.h"
19 : #include "aicpu_kernel_launcher.h"
20 : #include "coll_service_device_mode.h"
21 : #include "dlprof_function_v2.h"
22 : #include "hccl_aiv_utils.h"
23 : #include "ccu_ins_group.h"
24 :
25 : #include <limits>
26 :
27 : namespace Hccl {
28 :
29 : constexpr u32 BASE_BIT = 1; // 用于左移设置二进制数的特定位
30 : constexpr u32 TOKEN_VALUE_INDEX = 2;
31 :
32 : template <typename INS_TYPE>
33 19 : inline void VerifyDataSliceIsEqual(const INS_TYPE& ins)
34 : {
35 19 : const DataSlice& localSlice = ins.GetLocalSlice();
36 19 : const DataSlice& remoteSlice = ins.GetRemoteSlice();
37 :
38 19 : if (localSlice.GetSize() != remoteSlice.GetSize()) {
39 0 : string msg = StringFormat("%s slice size is different", ins.Describe().c_str());
40 0 : THROW<NotSupportException>(msg);
41 0 : }
42 19 : }
43 :
44 : template <typename INS_TYPE>
45 6 : inline RmaBufferSlice PrepareP2PRmaBufferSlice(const INS_TYPE& ins, CommunicatorImpl& comm)
46 : {
47 6 : CollOperator op = *comm.GetCurrentCollOperator();
48 6 : const DataSlice& localSlice = ins.GetLocalSlice();
49 6 : auto buffer = comm.GetDataBufferManager().Get(op.opTag, localSlice.GetType());
50 6 : if (buffer == nullptr) {
51 4 : string msg = StringFormat(
52 4 : "%s DataBuffer is nullptr, opTag[%s], bufferType[%s]", op.opTag.c_str(), ins.Describe().c_str(),
53 4 : localSlice.GetType().Describe().c_str());
54 2 : THROW<NullPtrException>(msg);
55 2 : }
56 4 : u64 addrLocal = buffer->GetAddr() + localSlice.GetOffset();
57 8 : return RmaBufferSlice{.addr = addrLocal, .size = localSlice.GetSize(), .buf = nullptr};
58 6 : }
59 :
60 : template <typename INS_TYPE>
61 13 : inline RmaBufferSlice PrepareRmaBufferSlice(const INS_TYPE& ins, CommunicatorImpl& comm)
62 : {
63 13 : CollOperator op = *comm.GetCurrentCollOperator();
64 13 : const DataSlice& localSlice = ins.GetLocalSlice();
65 : LocalRmaBuffer* localRmaBuffer
66 13 : = comm.GetLocalRmaBufManager().Get(op.opTag, ins.GetLink()->GetLocalPort(), localSlice.GetType());
67 13 : if (localRmaBuffer == nullptr) {
68 0 : string msg = StringFormat(
69 0 : "%s LocalRmaBuffer Get is nullptr, localBufType[%u]", ins.Describe().c_str(),
70 0 : static_cast<u32>(localSlice.GetType()));
71 0 : THROW<NullPtrException>(msg);
72 0 : }
73 13 : u64 addrLocal = localRmaBuffer->GetBuf()->GetAddr() + localSlice.GetOffset();
74 :
75 26 : return RmaBufferSlice{.addr = addrLocal, .size = localSlice.GetSize(), .buf = localRmaBuffer};
76 13 : }
77 :
78 : template <typename INS_TYPE>
79 16 : inline RmtRmaBufferSlice PrepareRmtRmaBufferSlice(const INS_TYPE& ins, BaseMemTransport& transport)
80 : {
81 16 : const DataSlice& remoteSlice = ins.GetRemoteSlice();
82 :
83 16 : RemoteRmaBuffer* remoteRmaBuffer = transport.GetRmtRmaBuffer(remoteSlice.GetType());
84 16 : if (remoteRmaBuffer == nullptr) {
85 1 : string msg = StringFormat(
86 2 : "%s RemoteRmaBuffer Get is nullptr, remoteBufType[%u]", ins.Describe().c_str(),
87 1 : static_cast<u32>(remoteSlice.GetType()));
88 1 : THROW<NullPtrException>(msg);
89 1 : }
90 15 : u64 addrRemote = 0;
91 15 : if (remoteRmaBuffer != nullptr) {
92 15 : addrRemote = remoteRmaBuffer->GetAddr() + remoteSlice.GetOffset();
93 : }
94 15 : return RmtRmaBufferSlice{.addr = addrRemote, .size = remoteSlice.GetSize(), .buf = remoteRmaBuffer};
95 : }
96 :
97 : template <typename INS_TYPE>
98 31 : inline BaseMemTransport* GetTransport(const INS_TYPE& ins, CommunicatorImpl& comm)
99 : {
100 31 : CollOperator op = *comm.GetCurrentCollOperator();
101 31 : BaseMemTransport* transport = nullptr;
102 31 : if (ins.GetLink() == nullptr) {
103 0 : THROW<NullPtrException>(StringFormat("[%s] ins.GetLink() is nullptr", __func__));
104 : }
105 31 : if (op.opMode == OpMode::OPBASE) {
106 31 : transport = comm.GetMemTransportManager()->GetOpbasedTransport(*ins.GetLink());
107 0 : } else if (op.opMode == OpMode::OFFLOAD) {
108 0 : transport = comm.GetMemTransportManager()->GetOffloadTransport(op.opTag, *ins.GetLink());
109 : }
110 31 : if (transport == nullptr) {
111 14 : string msg = StringFormat(
112 14 : "%s MemTransport Get is nullptr, opTag[%s], remoteRank[%d], linkData[%s]", ins.Describe().c_str(),
113 14 : op.opTag.c_str(), ins.GetRemoteRank(), ins.GetLink()->Describe().c_str());
114 7 : THROW<NullPtrException>(msg);
115 7 : }
116 :
117 24 : return transport;
118 31 : }
119 :
120 : template <typename INS_TYPE>
121 7 : inline ReduceIn GetReduceIn(const INS_TYPE& ins)
122 : {
123 7 : return ReduceIn(ins.GetDataType(), ins.GetReduceOp());
124 : }
125 :
126 : template <typename INS_TYPE>
127 8 : inline WithNotifyIn GetFinWithNotify(const INS_TYPE& ins, BaseMemTransport& transport)
128 : {
129 8 : if (ins.GetNotifyType() == NotifyType::NORMAL) {
130 4 : return WithNotifyIn(TransportNotifyType::NORMAL, NOTIFY_INDEX_FIN);
131 4 : } else if (ins.GetNotifyType() == NotifyType::COUNTER) {
132 4 : auto desc = transport.GetRmtCntNotifyDesc();
133 : CntNotifyResHelper tool;
134 4 : u32 index = tool.GetIndex(desc, ins.GetTopicId(), NOTIFY_INDEX_FIN);
135 4 : return WithNotifyIn(TransportNotifyType::COUNT, index, ins.GetBitValue());
136 4 : } else {
137 0 : string msg = StringFormat("only support NORMAL or COUNTER notifyType, ins=%s", ins.Describe().c_str());
138 0 : MACRO_THROW(NotSupportException, msg);
139 0 : }
140 : }
141 :
142 : inline RtsNotify*
143 4 : RtsNotifyGet(QueueNotifyManager& queueNotifyManager, QId postQid, QId waitQid, u32 topicId, const string& desc)
144 : {
145 4 : auto* notify = queueNotifyManager.Get(postQid, waitQid, topicId);
146 4 : if (notify == nullptr) {
147 : string msg = StringFormat(
148 : "%s BaseLocalNotify Get nullptr, postQid[%u], waitQid[%u], topicId[%u]", desc.c_str(), postQid, waitQid,
149 2 : topicId);
150 2 : THROW<NullPtrException>(msg);
151 2 : }
152 2 : return notify;
153 : }
154 :
155 4 : inline RtsCntNotify* RtsCntNotifyGet(
156 : QueueWaitGroupCntNotifyManager& queueWaitGroupCntNotifyManager, QId waitQid, u32 topicId, const string& desc)
157 : {
158 4 : RtsCntNotify* notify = queueWaitGroupCntNotifyManager.Get(waitQid, topicId);
159 4 : if (notify == nullptr) {
160 : string msg
161 2 : = StringFormat("%s RtsCntNotify Get nullptr, waitQid[%u], topicId[%u]", desc.c_str(), waitQid, topicId);
162 2 : THROW<NullPtrException>(msg);
163 2 : }
164 2 : return notify;
165 : }
166 :
167 4 : inline Rts1ToNCntNotify* Rts1ToNCntNotifyGet(
168 : QueueBcastPostCntNotifyManager& queueBcastPostCntNotifyManager, QId postQid, u32 topicId, const string& desc)
169 : {
170 4 : Rts1ToNCntNotify* notify = queueBcastPostCntNotifyManager.Get(postQid, topicId);
171 4 : if (notify == nullptr) {
172 : string msg
173 2 : = StringFormat("%s Rts1ToNCntNotify Get nullptr, postQid[%u], topicId[%u]", desc.c_str(), postQid, topicId);
174 2 : THROW<NullPtrException>(msg);
175 2 : }
176 2 : return notify;
177 : }
178 :
179 : inline vector<RtsCntNotify*>
180 1 : LocalCntNotifyGet(ConnLocalCntNotifyManager& connLocalCntNotifyManager, u32 topicId, const string& desc)
181 : {
182 1 : u32 listSize = 2;
183 1 : auto notifyList = connLocalCntNotifyManager.Get(topicId);
184 1 : if (notifyList.size() != listSize || notifyList[0] == nullptr || notifyList[1] == nullptr) {
185 0 : string msg = StringFormat("%s LocalCntNotify Get nullptr, topicId[%u]", desc.c_str(), topicId);
186 0 : THROW<NullPtrException>(msg);
187 0 : }
188 1 : return notifyList;
189 0 : }
190 :
191 : static void
192 16 : SaveDfxTaskInfo(const CommunicatorImpl& comm, const TaskParam& taskParam, const u32 remoteRankId, bool isMaster = false)
193 : {
194 : u32 taskId;
195 : u32 streamId;
196 16 : HrtGetTaskIdAndStreamID(taskId, streamId);
197 :
198 : std::unique_ptr<TaskInfo> taskInfo = std::make_unique<TaskInfo>(
199 16 : streamId, taskId, remoteRankId, taskParam, comm.GetMirrorTaskManager().GetCurrDfxOpInfo(), isMaster);
200 :
201 48 : HCCL_INFO("Begin to AddTaskInfo: streamId[%lu], taskId[%lu], remoteRankId[%u].", streamId, taskId, remoteRankId);
202 16 : comm.GetMirrorTaskManager().AddTaskInfo(std::move(taskInfo));
203 16 : }
204 :
205 2 : void Interpret(
206 : const InsPostReady& insPostReady, CommunicatorImpl& comm, const Stream& stream,
207 : [[maybe_unused]] const OpTaskConfig& taskConfig)
208 : {
209 2 : GetTransport(insPostReady, comm)->Post(NOTIFY_INDEX_READY, stream);
210 1 : }
211 :
212 2 : void Interpret(
213 : const InsWaitReady& insWaitReady, CommunicatorImpl& comm, const Stream& stream, const OpTaskConfig& taskConfig)
214 : {
215 2 : GetTransport(insWaitReady, comm)->Wait(NOTIFY_INDEX_READY, stream, taskConfig.GetNotifyWaitTime());
216 1 : }
217 :
218 2 : void Interpret(
219 : const InsPostFin& insPostFin, CommunicatorImpl& comm, const Stream& stream,
220 : [[maybe_unused]] const OpTaskConfig& taskConfig)
221 : {
222 2 : GetTransport(insPostFin, comm)->Post(NOTIFY_INDEX_FIN, stream);
223 1 : }
224 :
225 2 : void Interpret(
226 : const InsWaitFin& insWaitFin, CommunicatorImpl& comm, const Stream& stream, const OpTaskConfig& taskConfig)
227 : {
228 2 : GetTransport(insWaitFin, comm)->Wait(NOTIFY_INDEX_FIN, stream, taskConfig.GetNotifyWaitTime());
229 1 : }
230 :
231 3 : void Interpret(
232 : const InsPostFinAck& insPostFinAck, CommunicatorImpl& comm, const Stream& stream,
233 : [[maybe_unused]] const OpTaskConfig& taskConfig)
234 : {
235 3 : GetTransport(insPostFinAck, comm)->Post(NOTIFY_INDEX_FIN_ACK, stream);
236 2 : }
237 :
238 1 : void Interpret(
239 : const InsWaitGroupFin& insWaitGroupFin, CommunicatorImpl& comm, const Stream& stream,
240 : const OpTaskConfig& taskConfig)
241 : {
242 1 : TaskParam taskParam{};
243 1 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
244 :
245 : auto notifyList = LocalCntNotifyGet(
246 1 : comm.GetConnLocalCntNotifyManager(), insWaitGroupFin.GetTopicId(), insWaitGroupFin.Describe());
247 1 : notifyList[NOTIFY_INDEX_FIN]->WaitValue(insWaitGroupFin.GetValue(), taskConfig.GetNotifyWaitTime(), stream);
248 :
249 1 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
250 1 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
251 1 : taskParam.taskPara.Notify.notifyID = notifyList[NOTIFY_INDEX_FIN]->GetId();
252 1 : taskParam.taskPara.Notify.value = insWaitGroupFin.GetValue();
253 :
254 1 : SaveDfxTaskInfo(comm, taskParam, INVALID_VALUE_RANKID); // 本地填充rmt rankId， 为0xffff
255 1 : }
256 :
257 3 : void Interpret(
258 : const InsWaitFinAck& insWaitFinAck, CommunicatorImpl& comm, const Stream& stream, const OpTaskConfig& taskConfig)
259 : {
260 3 : GetTransport(insWaitFinAck, comm)->Wait(NOTIFY_INDEX_FIN_ACK, stream, taskConfig.GetNotifyWaitTime());
261 2 : }
262 :
263 5 : void Interpret(const InsRead& insRead, CommunicatorImpl& comm, const Stream& stream, const OpTaskConfig& taskConfig)
264 : {
265 : (void)taskConfig;
266 5 : VerifyDataSliceIsEqual(insRead);
267 : RmaBufferSlice locSlice;
268 5 : if (insRead.GetLink()->GetType() == PortDeploymentType::P2P) {
269 4 : locSlice = PrepareP2PRmaBufferSlice(insRead, comm);
270 : } else {
271 1 : locSlice = PrepareRmaBufferSlice(insRead, comm);
272 : }
273 3 : auto transport = GetTransport(insRead, comm);
274 2 : RmtRmaBufferSlice rmtSlice = PrepareRmtRmaBufferSlice(insRead, *transport);
275 2 : transport->Read(locSlice, rmtSlice, stream);
276 2 : }
277 :
278 3 : void Interpret(const InsWrite& insWrite, CommunicatorImpl& comm, const Stream& stream, const OpTaskConfig& taskConfig)
279 : {
280 : (void)taskConfig;
281 3 : VerifyDataSliceIsEqual(insWrite);
282 : RmaBufferSlice locSlice;
283 3 : if (insWrite.GetLink()->GetType() == PortDeploymentType::P2P) {
284 1 : locSlice = PrepareP2PRmaBufferSlice(insWrite, comm);
285 : } else {
286 2 : locSlice = PrepareRmaBufferSlice(insWrite, comm);
287 : }
288 3 : auto transport = GetTransport(insWrite, comm);
289 3 : RmtRmaBufferSlice rmtSlice = PrepareRmtRmaBufferSlice(insWrite, *transport);
290 2 : transport->Write(locSlice, rmtSlice, stream);
291 2 : }
292 :
293 2 : void Interpret(
294 : const InsReadReduce& insReadReduce, CommunicatorImpl& comm, const Stream& stream,
295 : [[maybe_unused]] const OpTaskConfig& taskConfig)
296 : {
297 2 : VerifyDataSliceIsEqual(insReadReduce);
298 : RmaBufferSlice locSlice;
299 2 : if (insReadReduce.GetLink()->GetType() == PortDeploymentType::P2P) {
300 1 : locSlice = PrepareP2PRmaBufferSlice(insReadReduce, comm);
301 : } else {
302 1 : locSlice = PrepareRmaBufferSlice(insReadReduce, comm);
303 : }
304 2 : auto transport = GetTransport(insReadReduce, comm);
305 2 : RmtRmaBufferSlice rmtSlice = PrepareRmtRmaBufferSlice(insReadReduce, *transport);
306 2 : transport->ReadReduce(locSlice, rmtSlice, GetReduceIn(insReadReduce), stream);
307 2 : }
308 :
309 1 : void Interpret(
310 : const InsWriteReduce& insWriteReduce, CommunicatorImpl& comm, const Stream& stream,
311 : [[maybe_unused]] const OpTaskConfig& taskConfig)
312 : {
313 1 : VerifyDataSliceIsEqual(insWriteReduce);
314 : RmaBufferSlice locSlice;
315 1 : if (insWriteReduce.GetLink()->GetType() == PortDeploymentType::P2P) {
316 0 : locSlice = PrepareP2PRmaBufferSlice(insWriteReduce, comm);
317 : } else {
318 1 : locSlice = PrepareRmaBufferSlice(insWriteReduce, comm);
319 : }
320 1 : auto transport = GetTransport(insWriteReduce, comm);
321 1 : RmtRmaBufferSlice rmtSlice = PrepareRmtRmaBufferSlice(insWriteReduce, *transport);
322 1 : transport->WriteReduce(locSlice, rmtSlice, GetReduceIn(insWriteReduce), stream);
323 1 : }
324 :
325 4 : void Interpret(
326 : const InsWriteWithFin& insWriteWithFin, CommunicatorImpl& comm, const Stream& stream,
327 : [[maybe_unused]] const OpTaskConfig& taskConfig)
328 : {
329 4 : VerifyDataSliceIsEqual(insWriteWithFin);
330 4 : RmaBufferSlice locSlice = PrepareRmaBufferSlice(insWriteWithFin, comm); // InsWriteWithFin当前不支持P2P
331 4 : auto transport = GetTransport(insWriteWithFin, comm);
332 4 : RmtRmaBufferSlice rmtSlice = PrepareRmtRmaBufferSlice(insWriteWithFin, *transport);
333 4 : transport->WriteWithNotify(locSlice, rmtSlice, GetFinWithNotify(insWriteWithFin, *transport), stream);
334 4 : }
335 :
336 4 : void Interpret(
337 : const InsWriteReduceWithFin& insWriteReduceWithFin, CommunicatorImpl& comm, const Stream& stream,
338 : [[maybe_unused]] const OpTaskConfig& taskConfig)
339 : {
340 4 : VerifyDataSliceIsEqual(insWriteReduceWithFin);
341 4 : RmaBufferSlice locSlice = PrepareRmaBufferSlice(insWriteReduceWithFin, comm); // InsWriteReduceWithFin当前不支持P2P
342 4 : auto transport = GetTransport(insWriteReduceWithFin, comm);
343 4 : RmtRmaBufferSlice rmtSlice = PrepareRmtRmaBufferSlice(insWriteReduceWithFin, *transport);
344 4 : transport->WriteReduceWithNotify(
345 4 : locSlice, rmtSlice, GetReduceIn(insWriteReduceWithFin), GetFinWithNotify(insWriteReduceWithFin, *transport),
346 : stream);
347 4 : }
348 :
349 4 : void Interpret(
350 : const InsLocalPostTo& insLocalPostTo, CommunicatorImpl& comm, const Stream& stream,
351 : [[maybe_unused]] const OpTaskConfig& taskConfig)
352 : {
353 4 : TaskParam taskParam{};
354 4 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
355 4 : u32 bitValue = BASE_BIT;
356 : u64 notifyID;
357 :
358 4 : if (insLocalPostTo.GetNotifyType() == NotifyType::NORMAL) {
359 4 : auto notify = RtsNotifyGet(
360 2 : comm.GetCcuQueueNotifyManager(), insLocalPostTo.GetPostQid(), insLocalPostTo.GetWaitQid(),
361 5 : insLocalPostTo.GetTopicId(), insLocalPostTo.Describe());
362 1 : notify->Post(stream);
363 1 : notifyID = notify->GetId();
364 2 : } else if (insLocalPostTo.GetNotifyType() == NotifyType::COUNTER) {
365 4 : RtsCntNotify* notify = RtsCntNotifyGet(
366 2 : comm.GetQueueWaitGroupCntNotifyManager(), insLocalPostTo.GetWaitQid(), insLocalPostTo.GetTopicId(),
367 5 : insLocalPostTo.Describe());
368 1 : auto postQid = insLocalPostTo.GetPostQid();
369 1 : bitValue = BASE_BIT << postQid;
370 1 : notify->PostBits(bitValue, stream);
371 1 : notifyID = notify->GetId();
372 : } else {
373 : string msg = StringFormat(
374 0 : "only support NORMAL or COUNTER notifyType, %s", insLocalPostTo.GetNotifyType().Describe().c_str());
375 0 : MACRO_THROW(NotSupportException, msg);
376 0 : }
377 :
378 2 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
379 2 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
380 2 : taskParam.taskPara.Notify.notifyID = notifyID;
381 2 : taskParam.taskPara.Notify.value = bitValue;
382 :
383 2 : SaveDfxTaskInfo(comm, taskParam, INVALID_VALUE_RANKID); // 本地填充rmt rankId， 为0xffff
384 4 : }
385 :
386 4 : void Interpret(
387 : const InsLocalWaitFrom& insLocalWaitFrom, CommunicatorImpl& comm, const Stream& stream,
388 : const OpTaskConfig& taskConfig)
389 : {
390 4 : TaskParam taskParam{};
391 4 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
392 4 : u32 bitValue = BASE_BIT;
393 : u64 notifyID;
394 :
395 4 : if (insLocalWaitFrom.GetNotifyType() == NotifyType::NORMAL) {
396 4 : auto notify = RtsNotifyGet(
397 2 : comm.GetCcuQueueNotifyManager(), insLocalWaitFrom.GetPostQid(), insLocalWaitFrom.GetWaitQid(),
398 5 : insLocalWaitFrom.GetTopicId(), insLocalWaitFrom.Describe());
399 1 : notify->Wait(stream, taskConfig.GetNotifyWaitTime());
400 1 : notifyID = notify->GetId();
401 2 : } else if (insLocalWaitFrom.GetNotifyType() == NotifyType::COUNTER) {
402 4 : Rts1ToNCntNotify* notify = Rts1ToNCntNotifyGet(
403 2 : comm.GetBcastPostCntNotifyManager(), insLocalWaitFrom.GetPostQid(), insLocalWaitFrom.GetTopicId(),
404 5 : insLocalWaitFrom.Describe());
405 1 : auto waitQid = insLocalWaitFrom.GetWaitQid();
406 1 : bitValue = BASE_BIT << waitQid;
407 1 : notify->WaitBits(bitValue, taskConfig.GetNotifyWaitTime(), stream);
408 1 : notifyID = notify->GetId();
409 : } else {
410 : string msg = StringFormat(
411 0 : "only support NORMAL or COUNTER notifyType, %s", insLocalWaitFrom.GetNotifyType().Describe().c_str());
412 0 : MACRO_THROW(NotSupportException, msg);
413 0 : }
414 :
415 2 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
416 2 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
417 2 : taskParam.taskPara.Notify.notifyID = notifyID;
418 2 : taskParam.taskPara.Notify.value = bitValue;
419 :
420 2 : SaveDfxTaskInfo(comm, taskParam, INVALID_VALUE_RANKID); // 本地填充rmt rankId， 为0xffff
421 4 : }
422 :
423 2 : void Interpret(
424 : const InsLocalWaitGroup& insLocalWaitGroup, CommunicatorImpl& comm, const Stream& stream,
425 : const OpTaskConfig& taskConfig)
426 : {
427 2 : TaskParam taskParam{};
428 2 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
429 :
430 4 : RtsCntNotify* notify = RtsCntNotifyGet(
431 2 : comm.GetQueueWaitGroupCntNotifyManager(), insLocalWaitGroup.GetWaitQid(), insLocalWaitGroup.GetTopicId(),
432 5 : insLocalWaitGroup.Describe());
433 :
434 1 : u32 value = 0;
435 3 : for (auto iter = insLocalWaitGroup.Iter(); iter.HasNext(); ++iter) {
436 2 : value |= BASE_BIT << *iter;
437 : }
438 1 : notify->WaitValue(value, taskConfig.GetNotifyWaitTime(), stream);
439 :
440 1 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
441 1 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
442 1 : taskParam.taskPara.Notify.notifyID = notify->GetId();
443 1 : taskParam.taskPara.Notify.value = value;
444 :
445 1 : SaveDfxTaskInfo(comm, taskParam, INVALID_VALUE_RANKID); // 本地填充rmt rankId， 为0xffff
446 2 : }
447 :
448 2 : void Interpret(
449 : const InsLocalBcastPost& insLocalBcastPost, CommunicatorImpl& comm, const Stream& stream,
450 : [[maybe_unused]] const OpTaskConfig& taskConfig)
451 : {
452 2 : TaskParam taskParam{};
453 2 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
454 :
455 4 : Rts1ToNCntNotify* notify = Rts1ToNCntNotifyGet(
456 2 : comm.GetBcastPostCntNotifyManager(), insLocalBcastPost.GetPostQid(), insLocalBcastPost.GetTopicId(),
457 5 : insLocalBcastPost.Describe());
458 :
459 1 : u32 value = 0;
460 3 : for (auto iter = insLocalBcastPost.Iter(); iter.HasNext(); ++iter) {
461 2 : value |= BASE_BIT << *iter;
462 : }
463 1 : notify->PostValue(value, stream);
464 :
465 1 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
466 1 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
467 1 : taskParam.taskPara.Notify.notifyID = notify->GetId();
468 1 : taskParam.taskPara.Notify.value = value;
469 :
470 1 : SaveDfxTaskInfo(comm, taskParam, INVALID_VALUE_RANKID); // 本地填充rmt rankId， 为0xffff
471 2 : }
472 :
473 2 : void Interpret(
474 : const InsLocalCopy& insLocalCopy, CommunicatorImpl& comm, const Stream& stream,
475 : [[maybe_unused]] const OpTaskConfig& taskConfig)
476 : {
477 2 : if (insLocalCopy.GetSrcSlice().GetSize() == 0) {
478 1 : return;
479 : }
480 :
481 1 : TaskParam taskParam{};
482 1 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
483 :
484 1 : auto dstBuffer = comm.GetCurrentCollOperator()->GetBuffer(insLocalCopy.GetDstSlice().GetType());
485 1 : if (dstBuffer == nullptr) {
486 0 : THROW<NullPtrException>(StringFormat("LocalCopy Interpret dstBuffer ptr is null"));
487 : }
488 1 : auto srcBuffer = comm.GetCurrentCollOperator()->GetBuffer(insLocalCopy.GetSrcSlice().GetType());
489 1 : if (srcBuffer == nullptr) {
490 0 : THROW<NullPtrException>(StringFormat("LocalCopy Interpret srcBuffer ptr is null"));
491 : }
492 1 : void* dst = reinterpret_cast<void*>(dstBuffer->GetAddr() + insLocalCopy.GetDstSlice().GetOffset());
493 1 : void* src = reinterpret_cast<void*>(srcBuffer->GetAddr() + insLocalCopy.GetSrcSlice().GetOffset());
494 :
495 2 : HrtMemAsyncCopy(
496 1 : dst, insLocalCopy.GetDstSlice().GetSize(), src, insLocalCopy.GetSrcSlice().GetSize(),
497 : ACL_MEMCPY_DEVICE_TO_DEVICE, stream.GetPtr());
498 :
499 1 : taskParam.taskType = TaskParamType::TASK_SDMA;
500 1 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
501 :
502 1 : taskParam.taskPara.DMA.src = src;
503 1 : taskParam.taskPara.DMA.dst = dst;
504 1 : taskParam.taskPara.DMA.size = insLocalCopy.GetSrcSlice().GetSize();
505 1 : taskParam.taskPara.DMA.notifyID = 0; // 填充无效值
506 1 : taskParam.taskPara.DMA.linkType = DfxLinkType::ONCHIP;
507 1 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
508 :
509 1 : SaveDfxTaskInfo(comm, taskParam, comm.GetMyRank());
510 1 : }
511 :
512 2 : inline void CheckLocalReduceIns(const InsLocalReduce& ins)
513 : {
514 2 : if (ins.GetDataType() == DataType::INT64) {
515 1 : THROW<InvalidParamsException>(StringFormat(
516 : "%s LocalReduce SDMAInlineReduce dose not support INT64, "
517 : "need use TBE reduce.",
518 : __func__));
519 : }
520 1 : }
521 :
522 2 : void Interpret(
523 : const InsLocalReduce& insLocalReduce, CommunicatorImpl& comm, const Stream& stream,
524 : [[maybe_unused]] const OpTaskConfig& taskConfig)
525 : {
526 6 : HCCL_INFO("%s Instruction %s", __func__, insLocalReduce.Describe().c_str());
527 : // SDMA支持的Reduce，则使用 sdmaReduce
528 : // SDMA不支持的Reduce，则使用 TBE算子(Asend C算子）
529 :
530 2 : if (insLocalReduce.GetSrcSlice().GetSize() == 0) {
531 0 : HCCL_WARNING("%s InsLocalReduce srcSlice size is 0, return", __func__);
532 0 : return;
533 : }
534 :
535 2 : if (insLocalReduce.GetSrcSlice().GetSize() != insLocalReduce.GetDstSlice().GetSize()) {
536 0 : HCCL_WARNING("%s InsLocalReduce srcSlice size is not equal to dstSlice size, return", __func__);
537 0 : return;
538 : }
539 :
540 2 : CheckLocalReduceIns(insLocalReduce);
541 :
542 1 : TaskParam taskParam{};
543 1 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
544 :
545 1 : auto dstBuffer = comm.GetCurrentCollOperator()->GetBuffer(insLocalReduce.GetDstSlice().GetType());
546 1 : if (dstBuffer == nullptr) {
547 0 : THROW<NullPtrException>(StringFormat("LocalReduce Interpret dstBuffer ptr is null"));
548 : }
549 1 : auto srcBuffer = comm.GetCurrentCollOperator()->GetBuffer(insLocalReduce.GetSrcSlice().GetType());
550 1 : if (srcBuffer == nullptr) {
551 0 : THROW<NullPtrException>(StringFormat("LocalReduce Interpret srcBuffer ptr is null"));
552 : }
553 1 : void* dst = reinterpret_cast<void*>(dstBuffer->GetAddr() + insLocalReduce.GetDstSlice().GetOffset());
554 1 : void* src = reinterpret_cast<void*>(srcBuffer->GetAddr() + insLocalReduce.GetSrcSlice().GetOffset());
555 :
556 1 : ReduceIn reduceIn(insLocalReduce.GetDataType(), insLocalReduce.GetReduceOp());
557 :
558 : aclrtReduceKind rtReduceOp
559 1 : = static_cast<aclrtReduceKind>(static_cast<int>(RtReduceOpGet(insLocalReduce.GetReduceOp())));
560 1 : aclDataType rtDataType = static_cast<aclDataType>(static_cast<int>(RtDataTypeGet(insLocalReduce.GetDataType())));
561 2 : HrtReduceAsync(
562 1 : dst, insLocalReduce.GetDstSlice().GetSize(), src, insLocalReduce.GetSrcSlice().GetSize(), rtReduceOp,
563 : rtDataType, stream.GetPtr());
564 :
565 1 : taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
566 1 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
567 :
568 1 : taskParam.taskPara.Reduce.src = src;
569 1 : taskParam.taskPara.Reduce.dst = dst;
570 1 : taskParam.taskPara.Reduce.size = insLocalReduce.GetSrcSlice().GetSize();
571 1 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
572 1 : taskParam.taskPara.Reduce.linkType = DfxLinkType::ONCHIP;
573 1 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(insLocalReduce.GetDataType());
574 1 : taskParam.taskPara.Reduce.reduceOp = ReduceOpToHcclReduceOp(insLocalReduce.GetReduceOp());
575 :
576 1 : SaveDfxTaskInfo(comm, taskParam, comm.GetMyRank());
577 1 : }
578 :
579 : static void
580 6 : LaunchCcuTasks(vector<CcuTaskParam> params, const Stream* stream, TaskParam& taskParam, const OpTaskConfig& taskConfig)
581 : {
582 6 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
583 :
584 6 : for (auto it = params.begin(); it != params.end(); ++it) {
585 0 : rtCcuTaskInfo_t taskInfo{};
586 0 : taskInfo.dieId = it->dieId;
587 0 : taskInfo.missionId = it->missionId;
588 0 : taskInfo.instStartId = it->instStartId;
589 0 : taskInfo.instCnt = it->instCnt;
590 0 : taskInfo.key = it->key;
591 0 : taskInfo.argSize = it->argSize;
592 0 : taskInfo.timeout = taskConfig.GetNotifyWaitTime() >= std::numeric_limits<u16>::max() ?
593 0 : std::numeric_limits<u16>::max() :
594 0 : static_cast<u16>(taskConfig.GetNotifyWaitTime());
595 0 : std::copy(std::begin(it->args), std::end(it->args), std::begin(taskInfo.args));
596 :
597 0 : HCCL_INFO(
598 : "start ccu task, dieId[%u] missionId[%u] instStartId[%u] instCnt[%u], argSize[%u], timeout[%u]s",
599 : taskInfo.dieId, taskInfo.missionId, taskInfo.instStartId, taskInfo.instCnt, taskInfo.argSize,
600 : taskInfo.timeout);
601 :
602 0 : for (std::size_t i = 0; i < taskInfo.argSize; i++) { // args 大小为 13
603 0 : if (i == TOKEN_VALUE_INDEX) {
604 0 : continue;
605 : }
606 0 : HCCL_INFO("arg[%lu] = %lu", i, taskInfo.args[i]);
607 : }
608 0 : HrtCcuLaunch(taskInfo, stream->GetPtr());
609 : }
610 6 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
611 6 : }
612 :
613 6 : static void ReportCcuProfilingInfo(
614 : uint64_t execId, std::vector<CcuProfilingInfo>& streamProfilingInfo, const CommunicatorImpl& comm,
615 : TaskParam& taskParam, bool isMaster)
616 : {
617 6 : if (streamProfilingInfo.empty()) {
618 0 : HCCL_INFO("There is no ccu profiling info.");
619 0 : return;
620 : }
621 6 : taskParam.taskPara.Ccu.dieId = streamProfilingInfo[0].dieId;
622 6 : taskParam.taskPara.Ccu.missionId = streamProfilingInfo[0].missionId;
623 6 : taskParam.taskPara.Ccu.execMissionId = streamProfilingInfo[0].missionId;
624 6 : taskParam.taskPara.Ccu.instrId = streamProfilingInfo[0].instrId;
625 6 : taskParam.taskPara.Ccu.executeId = execId;
626 :
627 12 : CcuJettyMgr* ccuJettyMgr = dynamic_cast<CollServiceDeviceMode*>(comm.GetCollService())
628 : ->GetCcuInsPreprocessor()
629 : ->GetCcuComm()
630 12 : ->GetCcuJettyMgr();
631 42 : for (auto& profInfo : streamProfilingInfo) {
632 132 : for (int idx = 0; idx < CCU_MAX_CHANNEL_NUM; idx++) {
633 126 : if (profInfo.channelId[idx] == INVALID_VALUE_CHANNELID) {
634 30 : break;
635 : }
636 : profInfo.remoteRankId[idx]
637 96 : = ccuJettyMgr->GetRemoteRankIdByChannelId(profInfo.dieId, profInfo.channelId[idx]);
638 : }
639 : }
640 6 : taskParam.ccuDetailInfo = std::make_shared<std::vector<CcuProfilingInfo>>(streamProfilingInfo);
641 18 : HCCL_INFO("Begin to SaveDfxTaskInfo. taskType[%d]", static_cast<int32_t>(TaskParamType::TASK_CCU));
642 6 : SaveDfxTaskInfo(comm, taskParam, INVALID_VALUE_RANKID, isMaster);
643 : }
644 :
645 3 : static void GetCcuProfilingInfo(
646 : const CcuInstruction& ccuInstruction, const vector<vector<CcuTaskParam>>& ccuParams,
647 : std::vector<std::vector<CcuProfilingInfo>>& ccuProfilingInfo)
648 : {
649 3 : HcclResult res = CcuCtxMgr::GetProfilingInfo(
650 6 : HrtGetDevice(), *(ccuInstruction.GetTaskArg()), ccuInstruction.GetExecId(), ccuProfilingInfo);
651 3 : if (res != HcclResult::HCCL_SUCCESS) {
652 0 : string msg = StringFormat("Get ccu profiling info failed, res[%d]", res);
653 0 : THROW<NotSupportException>(msg);
654 0 : }
655 3 : if (ccuProfilingInfo.size() != ccuParams.size()) {
656 : string msg
657 0 : = StringFormat("Get ccu profiling info size error(%u-%u).", ccuProfilingInfo.size(), ccuParams.size());
658 0 : THROW<NotSupportException>(msg);
659 0 : }
660 3 : }
661 :
662 1 : static void FastLoadSaveParams(
663 : const CcuInstruction& ccuInstruction, CommunicatorImpl& comm, [[maybe_unused]] const OpTaskConfig& taskConfig,
664 : const Stream& stream, std::vector<std::vector<CcuTaskParam>>& ccuParams,
665 : std::vector<std::vector<CcuProfilingInfo>>& ccuProfilingInfo)
666 : {
667 1 : std::size_t totalSize = 0;
668 6 : for (const auto& ccuParam : ccuParams) {
669 5 : totalSize += ccuParam.size();
670 : }
671 1 : if (totalSize != 0 && comm.isEnableSuperFasterLoad()) {
672 0 : CcuInstType insType = ccuInstruction.GetInstType();
673 0 : if (ccuInstruction.GetInstType() == CcuInstType::CCU_INS_GROUP) {
674 0 : const CcuInsGroup* insGroup = dynamic_cast<const CcuInsGroup*>(&ccuInstruction);
675 0 : if (insGroup == nullptr) {
676 0 : THROW<NullPtrException>(StringFormat("%s CcuInsGroup trans failed", __func__));
677 : }
678 0 : if (insGroup->GetCcuInstructions().empty()) {
679 0 : THROW<InvalidParamsException>(StringFormat("%s insGroup CcuInstructions isEmpty", __func__));
680 : }
681 0 : insType = insGroup->GetCcuInstructions()[0]->GetInstType();
682 : }
683 0 : HCCL_RUN_INFO("current CcuInstType: %d", static_cast<int>(insType));
684 0 : comm.saveCCUParams(
685 0 : std::move(ccuParams), std::move(ccuProfilingInfo), ccuInstruction.GetExecId(), insType,
686 0 : stream.GetId() != comm.GetStreamManager().GetMaster()->GetId());
687 : }
688 1 : }
689 :
690 3 : void SubmitCcuInsGroupTasks(
691 : const CcuInstruction& ccuInstruction, CommunicatorImpl& comm, const OpTaskConfig& taskConfig, const Stream& stream,
692 : std::vector<std::vector<CcuTaskParam>>& ccuParams)
693 : {
694 3 : TaskParam taskParam = {};
695 3 : taskParam.taskType = TaskParamType::TASK_CCU;
696 3 : std::vector<std::vector<CcuProfilingInfo>> ccuProfilingInfo;
697 3 : GetCcuProfilingInfo(ccuInstruction, ccuParams, ccuProfilingInfo);
698 :
699 3 : u32 timeout = taskConfig.GetNotifyWaitTime();
700 3 : u32 reqStreamNum = ccuParams.size() - 1;
701 3 : u32 value = 0;
702 15 : for (u32 i = 0; i < reqStreamNum; ++i) {
703 12 : value |= BASE_BIT << i;
704 : }
705 :
706 : // launch LocalPostTo on stream
707 3 : Rts1ToNCntNotify* cntNotify1ToN = comm.GetCcuStreamSyncNotifyManager().GetRts1ToNCntNotify(stream.GetId());
708 3 : if (cntNotify1ToN == nullptr) {
709 3 : HCCL_ERROR("[SubmitCcuInsGroupTasks] GetRts1ToNCntNotify returned nullptr");
710 1 : return;
711 : }
712 2 : cntNotify1ToN->PostValue(value, stream);
713 :
714 : // launch ccu task
715 2 : LaunchCcuTasks(*ccuParams.begin(), &stream, taskParam, taskConfig);
716 2 : ReportCcuProfilingInfo(ccuInstruction.GetExecId(), ccuProfilingInfo[0], comm, taskParam, stream.IsMaster());
717 :
718 : // launch LocalWaitFrom on stream
719 2 : RtsCntNotify* cntNotifyNTo1 = comm.GetCcuStreamSyncNotifyManager().GetRtsNTo1CntNotify(stream.GetId());
720 2 : if (cntNotifyNTo1 == nullptr) {
721 3 : HCCL_ERROR("[SubmitCcuInsGroupTasks] GetRtsNTo1CntNotify returned nullptr");
722 1 : return;
723 : }
724 1 : cntNotifyNTo1->WaitValue(value, timeout, stream);
725 :
726 1 : auto& streamMgr = comm.GetStreamManager();
727 : // 查询当前从流持有的子从流
728 1 : auto streamIndex = streamMgr.GetStreamIndex(stream.GetId());
729 1 : auto& candidateSubSlaveStreamIndexes = streamMgr.GetSubSlaveIndexes(streamIndex);
730 5 : for (u32 ccuProfIdx = 1; ccuProfIdx <= reqStreamNum; ++ccuProfIdx) {
731 : Stream* slave;
732 4 : if (ccuProfIdx > candidateSubSlaveStreamIndexes.size()) {
733 : // 子从流不足，添加(主)从流->(子)从流对应关系, 并创建流
734 4 : streamMgr.RegisterBucket(streamIndex, streamMgr.GetSlaveIndex());
735 4 : slave = streamMgr.GetSlave();
736 : } else {
737 0 : slave = streamMgr.GetSlaveByIndex(candidateSubSlaveStreamIndexes[ccuProfIdx - 1]);
738 : }
739 :
740 : // 捕获slaveStream
741 4 : auto masterStream = comm.GetStreamManager().GetMaster();
742 4 : comm.GetStreamManager().CaptureSlaveStream(masterStream, slave); // 捕获slaveStream
743 4 : u32 bitValue = BASE_BIT << (ccuProfIdx - 1);
744 4 : cntNotify1ToN->WaitBits(bitValue, timeout, *slave);
745 :
746 : // launch ccu task
747 4 : LaunchCcuTasks(ccuParams[ccuProfIdx], slave, taskParam, taskConfig);
748 8 : ReportCcuProfilingInfo(
749 4 : ccuInstruction.GetExecId(), ccuProfilingInfo[ccuProfIdx], comm, taskParam, slave->IsMaster());
750 :
751 : // launch localPostTo on extra streams
752 4 : cntNotifyNTo1->PostBits(bitValue, *slave);
753 : }
754 1 : FastLoadSaveParams(ccuInstruction, comm, taskConfig, stream, ccuParams, ccuProfilingInfo);
755 5 : }
756 :
757 9 : static void SubmitCcuTasks(
758 : const CcuInstruction& ccuInstruction, CommunicatorImpl& comm, const OpTaskConfig& taskConfig, const Stream& stream)
759 : {
760 9 : std::vector<std::vector<CcuTaskParam>> ccuParams;
761 9 : ccuInstruction.Translate(ccuParams);
762 9 : if (ccuParams.size() == 0) {
763 18 : HCCL_INFO("There is no ccu mission ccuParams.");
764 6 : return;
765 : }
766 :
767 3 : if (ccuParams.size() > 1) {
768 3 : SubmitCcuInsGroupTasks(ccuInstruction, comm, taskConfig, stream, ccuParams);
769 3 : return;
770 : }
771 :
772 0 : TaskParam taskParam = {};
773 0 : taskParam.taskType = TaskParamType::TASK_CCU;
774 0 : std::vector<std::vector<CcuProfilingInfo>> ccuProfilingInfo;
775 0 : GetCcuProfilingInfo(ccuInstruction, ccuParams, ccuProfilingInfo);
776 :
777 : // esl 2die适配，先申请从流再启动task
778 0 : LaunchCcuTasks(*ccuParams.begin(), &stream, taskParam, taskConfig);
779 0 : ReportCcuProfilingInfo(ccuInstruction.GetExecId(), ccuProfilingInfo[0], comm, taskParam, stream.IsMaster());
780 0 : FastLoadSaveParams(ccuInstruction, comm, taskConfig, stream, ccuParams, ccuProfilingInfo);
781 9 : }
782 :
783 9 : void Interpret(
784 : const CcuInstruction& ccuInstruction, CommunicatorImpl& comm, const Stream& stream, const OpTaskConfig& taskConfig)
785 : {
786 9 : SubmitCcuTasks(ccuInstruction, comm, taskConfig, stream);
787 9 : }
788 :
789 0 : void Interpret(
790 : const AicpuInstruction& aicpuInstruction, CommunicatorImpl& comm, const Stream& stream,
791 : const OpTaskConfig& taskConfig)
792 : {
793 : (void)taskConfig;
794 :
795 0 : AicpuKernelLauncher aicpuKernelLauncher(comm);
796 0 : aicpuKernelLauncher.AicpuKernelLaunch(stream, aicpuInstruction.GetAlgName());
797 0 : }
798 :
799 1 : static void ReportAivTaskInfo(const CommunicatorImpl& comm, AivOpArgs& aivOpArgs, bool isMaster)
800 : {
801 3 : HCCL_DEBUG("Begin to SaveAivDfxTaskInfo taskType[%d]", static_cast<int32_t>(TaskParamType::TASK_AIV));
802 : // flagMem每个stream的中的任务复用，异常时只有最后一个task的信息
803 1 : TaskParam taskParam = {
804 : .taskType = TaskParamType::TASK_AIV,
805 1 : .beginTime = aivOpArgs.beginTime,
806 1 : .endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime(),
807 : .aicpuTaskId = 0,
808 : .npuDevId = 0,
809 : .isMaster = isMaster,
810 : .taskPara = {
811 : .Aiv = {
812 1 : .cmdType = aivOpArgs.cmdType,
813 1 : .tag = aivOpArgs.aivTag,
814 1 : .count = aivOpArgs.count,
815 1 : .numBlocks = aivOpArgs.numBlocks,
816 1 : .rankSize = aivOpArgs.rankSize,
817 0 : .flagMem = aivOpArgs.isOpBase ? reinterpret_cast<void *>(comm.GetAivTagBuffer()->GetAddr() + AIV_FLAG_ADDR_OFFSET):
818 2 : reinterpret_cast<void *>(comm.GetAivOffloadTagBuffer()->GetAddr() + AIV_FLAG_ADDR_OFFSET),
819 : .flagMemSize = AIV_FLAG_AREA_SIZE,
820 1 : .rank = aivOpArgs.rank,
821 1 : .sendRecvRemoteRank = aivOpArgs.sendRecvRemoteRank,
822 2 : .dataType = DataTypeToHcclDataType(aivOpArgs.dataType),
823 : }
824 : },
825 : .ccuDetailInfo = nullptr
826 3 : };
827 :
828 1 : SaveDfxTaskInfo(comm, taskParam, INVALID_VALUE_RANKID, isMaster);
829 1 : }
830 :
831 1 : void Interpret(
832 : const AivInstruction& aivInstruction, const CommunicatorImpl& comm, const Stream& stream,
833 : const OpTaskConfig& taskConfig)
834 : {
835 : (void)taskConfig;
836 1 : AivOpArgs aivOpArgs;
837 1 : aivInstruction.GetAivInsArgs(aivOpArgs);
838 :
839 1 : aivOpArgs.stream = stream.GetPtr();
840 :
841 1 : aivOpArgs.aivTag = aivOpArgs.isOpBase ? (static_cast<uint32_t>(comm.GetAivTag()) << AIV_TAG_MOVE_LEFT_BITS)
842 0 : | static_cast<uint32_t>(aivOpArgs.aivTag) :
843 1 : (static_cast<uint32_t>(comm.GetAivOffloadTag()) << AIV_TAG_MOVE_LEFT_BITS)
844 1 : | static_cast<uint32_t>(aivOpArgs.aivTag);
845 3 : HCCL_INFO("%s AivTag[%u]", __func__, aivOpArgs.aivTag);
846 1 : void* buffersInAddr = aivOpArgs.isOpBase ? reinterpret_cast<void*>(comm.GetAivTagBuffer()->GetAddr()) :
847 1 : reinterpret_cast<void*>(comm.GetAivOffloadTagBuffer()->GetAddr());
848 1 : aivOpArgs.buffersIn = buffersInAddr;
849 :
850 1 : if ((aivOpArgs.aivTag & AIV_LOW_16_BITS) == 1 && (aivOpArgs.aivTag >> AIV_TAG_MOVE_LEFT_BITS) == 1) {
851 : void* buffersInAddrSrc;
852 0 : u64 buffersIn[MAX_RANK_SIZE_] = {};
853 0 : buffersIn[comm.GetMyRank()] = comm.GetCclBuffer()->GetAddr();
854 0 : auto ubMemLink2TransportMap = comm.GetUbMemoryTransportMgr()->GetRmtRankId2RmtIpcRmaBufList();
855 0 : for (auto ubMemLink2TransportIter : ubMemLink2TransportMap) {
856 0 : auto rmtRank = ubMemLink2TransportIter.first;
857 0 : auto rmtMemBuffer = ubMemLink2TransportIter.second->GetAddr();
858 0 : buffersIn[rmtRank] = rmtMemBuffer;
859 : }
860 0 : HrtMemcpy(
861 : buffersInAddr, MAX_RANK_SIZE_ * sizeof(uint64_t), buffersIn, MAX_RANK_SIZE_ * sizeof(uint64_t),
862 : RT_MEMCPY_HOST_TO_DEVICE);
863 0 : u64 buffersOut[MAX_RANK_SIZE_] = {};
864 0 : auto ubMemLink2TransportMap_ = aivOpArgs.isOpBase ?
865 0 : comm.GetUbMemoryTransportMgr()->GetAllRankId2AivTagBufAddrList() :
866 0 : comm.GetUbMemoryTransportMgr()->GetAllRankId2AivOffloadTagBufAddrList();
867 0 : for (auto ubMemLink2TransportIter : ubMemLink2TransportMap_) {
868 0 : auto rmtRank = ubMemLink2TransportIter.first;
869 0 : auto rmtMemBuffer = ubMemLink2TransportIter.second;
870 0 : buffersOut[rmtRank] = rmtMemBuffer;
871 : }
872 0 : buffersInAddr = aivOpArgs.isOpBase ?
873 0 : reinterpret_cast<void*>(comm.GetAivTagBuffer()->GetAddr() + AIV_TAG_ADDR_OFFSET) :
874 0 : reinterpret_cast<void*>(comm.GetAivOffloadTagBuffer()->GetAddr() + AIV_TAG_ADDR_OFFSET);
875 0 : HrtMemcpy(
876 : buffersInAddr, MAX_RANK_SIZE_ * sizeof(uint64_t), buffersOut, MAX_RANK_SIZE_ * sizeof(uint64_t),
877 : RT_MEMCPY_HOST_TO_DEVICE);
878 :
879 0 : buffersInAddr = aivOpArgs.isOpBase ?
880 0 : reinterpret_cast<void*>(comm.GetAivTagBuffer()->GetAddr() + AIV_FLAG_ADDR_OFFSET) :
881 0 : reinterpret_cast<void*>(comm.GetAivOffloadTagBuffer()->GetAddr() + AIV_FLAG_ADDR_OFFSET);
882 : buffersInAddrSrc
883 0 : = aivOpArgs.isOpBase ?
884 0 : reinterpret_cast<void*>(comm.GetAivTagBuffer()->GetAddr() + AIV_FLAG_CLEAR_OFFSET) :
885 0 : reinterpret_cast<void*>(comm.GetAivOffloadTagBuffer()->GetAddr() + AIV_FLAG_CLEAR_OFFSET);
886 0 : bool isAivClearEnable = comm.GetAivClearEnable();
887 0 : if (isAivClearEnable) {
888 0 : HrtMemAsyncCopy(
889 : buffersInAddr, AIV_FLAG_AREA_SIZE, buffersInAddrSrc, AIV_FLAG_AREA_SIZE, ACL_MEMCPY_DEVICE_TO_DEVICE,
890 : stream.GetPtr());
891 : }
892 0 : }
893 :
894 1 : if (comm.GetCurrentCollOperator()->inputMem == nullptr) {
895 0 : HCCL_INFO("%s comm.GetCurrentCollOperator()->inputMem is nullptr", __func__);
896 : } else {
897 1 : u64 localInputAddr = static_cast<uint64_t>(comm.GetCurrentCollOperator()->inputMem->GetAddr());
898 1 : aivOpArgs.input += localInputAddr;
899 : }
900 :
901 1 : if (comm.GetCurrentCollOperator()->outputMem == nullptr) {
902 0 : HCCL_INFO("%s comm.GetCurrentCollOperator()->outputMem is nullptr", __func__);
903 : } else {
904 1 : u64 localOutputAddr = static_cast<uint64_t>(comm.GetCurrentCollOperator()->outputMem->GetAddr());
905 1 : aivOpArgs.output += localOutputAddr;
906 : }
907 1 : aivOpArgs.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
908 1 : ExecuteKernelLaunch(aivOpArgs);
909 1 : ReportAivTaskInfo(comm, aivOpArgs, stream.IsMaster());
910 1 : }
911 :
912 : } // namespace Hccl
|
