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