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