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 <regex>
12 : #include "ins_to_sqe_rule.h"
13 : #include "null_ptr_exception.h"
14 : #include "internal_exception.h"
15 : #include "invalid_params_exception.h"
16 : #include "mem_transport_lite.h"
17 : #include "sal.h"
18 : #include "task_info.h"
19 : #include "lite_res_mgr_fetcher.h"
20 : #include "timeout_exception.h"
21 : #include "not_support_exception.h"
22 :
23 : namespace Hccl {
24 :
25 : constexpr u32 BASE_BIT = 1; // 用于左移设置二进制数的特定位
26 :
27 33 : template <typename INS_TYPE> MemTransportLite &GetTransportLite(const INS_TYPE &ins, ResMgrFetcher *resMgrFetcher)
28 : {
29 33 : if (ins.GetLink() == nullptr) {
30 0 : THROW<NullPtrException>(StringFormat("%s ins.GetLink() is nullptr", ins.Describe().c_str()));
31 : }
32 33 : MemTransportLite *transport = nullptr;
33 33 : if (resMgrFetcher->GetCurrentOp().opMode == OpMode::OPBASE) {
34 2 : transport = resMgrFetcher->GetTransportLiteMgr()->GetOpbase(*ins.GetLink()); // 单算子，采用 GetOpBase
35 31 : } else if (resMgrFetcher->GetCurrentOp().opMode == OpMode::OFFLOAD) {
36 : // 图下沉算子，需要采用 GetOffload(opTag, linkData) 获取transport
37 : transport
38 31 : = resMgrFetcher->GetTransportLiteMgr()->GetOffload(resMgrFetcher->GetCurrentOp().opTag, *ins.GetLink());
39 : }
40 :
41 33 : if (UNLIKELY(transport == nullptr)) {
42 10 : string msg = StringFormat("%s MemTransportLite Get is nullptr, remoteRank[%d], linkData[%s]",
43 15 : ins.Describe().c_str(), ins.GetRemoteRank(), ins.GetLink()->Describe().c_str());
44 5 : THROW<NullPtrException>(msg);
45 5 : }
46 28 : return *transport;
47 : }
48 :
49 10 : template <typename INS_TYPE> RmaBufferLite GetLocRmaBufferLite(const INS_TYPE &ins, ResMgrFetcher *resMgrFetcher)
50 : {
51 10 : auto lite = resMgrFetcher->GetRmaBufferLite(ins.GetLocalSlice().GetType());
52 10 : if (UNLIKELY(lite == nullptr)) {
53 0 : string msg = StringFormat("[%s] lite Get nullptr", __func__);
54 0 : THROW<NullPtrException>(msg);
55 0 : }
56 10 : Buffer buf(lite->GetAddr(), lite->GetSize());
57 10 : auto range = buf.Range(ins.GetLocalSlice().GetOffset(), ins.GetLocalSlice().GetSize());
58 20 : return RmaBufferLite(range.GetAddr(), range.GetSize(), lite->GetTokenId(), lite->GetTokenValue());
59 10 : }
60 :
61 : template <typename INS_TYPE>
62 10 : Buffer GetRmtBuffer(const INS_TYPE &ins, MemTransportLite &transport, ResMgrFetcher *resMgrFetcher)
63 : {
64 : (void)resMgrFetcher;
65 10 : auto buf = transport.GetRmtBuffer(ins.GetRemoteSlice().GetType());
66 20 : return buf.Range(ins.GetRemoteSlice().GetOffset(), ins.GetRemoteSlice().GetSize());
67 10 : }
68 :
69 4 : template <typename INS_TYPE> NotifyLite &GetNotifyLite(const INS_TYPE &ins, ResMgrFetcher *resMgrFetcher)
70 : {
71 4 : auto notify = resMgrFetcher->GetQueueNotifyLiteMgr()->Get(ins.GetPostQid(), ins.GetWaitQid(), ins.GetTopicId());
72 4 : if (UNLIKELY(notify == nullptr)) {
73 4 : string msg = StringFormat("%s NotifyLite Get nullptr, postQid[%d], waitQid[%d], topicId[%d]",
74 4 : ins.Describe().c_str(), ins.GetPostQid(), ins.GetWaitQid(), ins.GetTopicId());
75 2 : THROW<NullPtrException>(msg);
76 2 : }
77 2 : return *notify;
78 : }
79 :
80 4 : template <typename INS_TYPE> Cnt1tonNotifyLite &GetCnt1toNNotifyLite(const INS_TYPE &ins, ResMgrFetcher *resMgrFetcher)
81 : {
82 4 : auto notify = resMgrFetcher->GetCnt1tonNotifyLiteMgr()->Get(ins.GetPostQid(), ins.GetTopicId());
83 4 : if (UNLIKELY(notify == nullptr)) {
84 2 : string msg = StringFormat("%s Cnt1tonNotifyLite Get nullptr, postQid[%d], topicId[%d]", ins.Describe().c_str(),
85 : ins.GetPostQid(), ins.GetTopicId());
86 2 : THROW<NullPtrException>(msg);
87 2 : }
88 2 : return *notify;
89 : }
90 :
91 4 : template <typename INS_TYPE> CntNto1NotifyLite &GetCntNto1NotifyLite(const INS_TYPE &ins, ResMgrFetcher *resMgrFetcher)
92 : {
93 4 : auto notify = resMgrFetcher->GetCntNto1NotifyLiteMgr()->Get(ins.GetWaitQid(), ins.GetTopicId());
94 4 : if (UNLIKELY(notify == nullptr)) {
95 2 : string msg = StringFormat("%s CntNto1NotifyLite Get nullptr, waitQid[%d], topicId[%d]", ins.Describe().c_str(),
96 : ins.GetWaitQid(), ins.GetTopicId());
97 2 : THROW<NullPtrException>(msg);
98 2 : }
99 2 : return *notify;
100 : }
101 :
102 4 : void Interpret(const InsLocalPostTo &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
103 : {
104 4 : auto taskId = stream.GetRtsq()->GetTaskId();
105 4 : u32 value = 1;
106 4 : u32 notifyId = 0;
107 4 : if (ins.GetNotifyType() == NotifyType::NORMAL) {
108 2 : auto ¬ify = GetNotifyLite(ins, resMgrFetcher);
109 1 : notifyId = notify.GetId();
110 1 : stream.GetRtsq()->NotifyRecordLoc(notify.GetId());
111 2 : } else if (ins.GetNotifyType() == NotifyType::COUNTER) {
112 2 : auto ¬ify = GetCntNto1NotifyLite(ins, resMgrFetcher);
113 1 : notifyId = notify.GetId();
114 1 : value = BASE_BIT << (ins.GetPostQid());
115 1 : stream.GetRtsq()->CntNto1NotifyRecord(notify.GetId(), value);
116 : } else {
117 : std::string msg
118 0 : = StringFormat("only support NORMAL or COUNTER notifyType, %s", ins.GetNotifyType().Describe().c_str());
119 0 : MACRO_THROW(NotSupportException, msg);
120 0 : }
121 :
122 2 : TaskParam taskParam{};
123 2 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
124 2 : taskParam.beginTime = ProfGetCurCpuTimestamp();
125 2 : taskParam.taskPara.Notify.notifyID = notifyId;
126 2 : taskParam.taskPara.Notify.value = value;
127 2 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
128 2 : resMgrFetcher->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
129 2 : }
130 :
131 4 : void Interpret(const InsLocalWaitFrom &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
132 : {
133 4 : auto taskId = stream.GetRtsq()->GetTaskId();
134 4 : u32 value = 1;
135 4 : u32 notifyId = 0;
136 4 : if (ins.GetNotifyType() == NotifyType::NORMAL) {
137 2 : auto ¬ify = GetNotifyLite(ins, resMgrFetcher);
138 1 : notifyId = notify.GetId();
139 1 : stream.GetRtsq()->NotifyWait(notify.GetId());
140 2 : } else if (ins.GetNotifyType() == NotifyType::COUNTER) {
141 2 : auto ¬ify = GetCnt1toNNotifyLite(ins, resMgrFetcher);
142 1 : notifyId = notify.GetId();
143 1 : value = BASE_BIT << (ins.GetWaitQid());
144 1 : stream.GetRtsq()->Cnt1toNNotifyWait(notify.GetId(), value);
145 : } else {
146 : std::string msg
147 0 : = StringFormat("only support NORMAL or COUNTER notifyType, %s", ins.GetNotifyType().Describe().c_str());
148 0 : MACRO_THROW(NotSupportException, msg);
149 0 : }
150 :
151 2 : TaskParam taskParam {};
152 2 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
153 2 : taskParam.beginTime = ProfGetCurCpuTimestamp();
154 2 : taskParam.taskPara.Notify.notifyID = notifyId;
155 2 : taskParam.taskPara.Notify.value = value;
156 2 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
157 2 : resMgrFetcher->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
158 2 : }
159 :
160 2 : void Interpret(const InsLocalCopy &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
161 : {
162 2 : if (UNLIKELY(ins.GetSrcSlice().GetSize() == 0)) {
163 1 : return;
164 : }
165 :
166 : // 传入数据大小不能超过 u32最大值， 需要进行切分
167 1 : u64 u32Max = UINT32_MAX;
168 1 : double countSplitingTimes = static_cast<double>(ins.GetSrcSlice().GetSize()) / static_cast<double>(u32Max);
169 1 : u64 splitingTimes = static_cast<int>(std::ceil(countSplitingTimes));
170 1 : u64 src = resMgrFetcher->GetRmaBufferLite(ins.GetSrcSlice().GetType())->GetAddr() + ins.GetSrcSlice().GetOffset();
171 1 : u64 dst = resMgrFetcher->GetRmaBufferLite(ins.GetDstSlice().GetType())->GetAddr() + ins.GetDstSlice().GetOffset();
172 1 : u64 blockSize = u32Max;
173 1 : u64 offset = u32Max;
174 2 : for (u64 i = 0; i < splitingTimes; i++) {
175 : // 处理尾块数据
176 1 : if(i == splitingTimes - 1) {
177 1 : blockSize = ins.GetSrcSlice().GetSize() - u32Max * (splitingTimes - 1);
178 1 : offset = 0;
179 : }
180 :
181 1 : auto taskId = stream.GetRtsq()->GetTaskId();
182 1 : stream.GetRtsq()->SdmaCopy(src, dst, blockSize, 0); // 待确认， PART_ID是否固定设置为 0
183 3 : HCCL_INFO("InsLocalCopy srcA:0x%llx dstA:0x%llx,size=0x%llx", src, dst, blockSize);
184 1 : TaskParam taskParam{};
185 1 : taskParam.taskType = TaskParamType::TASK_SDMA;
186 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
187 1 : taskParam.taskPara.DMA.src = reinterpret_cast<void *>(src);
188 1 : taskParam.taskPara.DMA.dst = reinterpret_cast<void *>(dst);
189 1 : taskParam.taskPara.DMA.size = blockSize;
190 1 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
191 1 : taskParam.taskPara.DMA.linkType = DfxLinkType::ONCHIP;
192 1 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
193 1 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
194 1 : resMgrFetcher->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
195 1 : src += offset;
196 1 : dst += offset;
197 1 : }
198 : }
199 :
200 2 : void Interpret(const InsLocalCopyExtend &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
201 : {
202 2 : if (UNLIKELY(ins.GetSrcBuffer().GetSize() == 0)) {
203 3 : HCCL_WARNING("%s insLocalCopyExtend srcBuffer size is 0, return", __func__);
204 1 : return;
205 : }
206 :
207 : // 传入数据大小不能超过 u32最大值， 需要进行切分
208 1 : u64 u32Max = UINT32_MAX;
209 1 : double countSplitingTimes = static_cast<double>(ins.GetSrcBuffer().GetSize()) / static_cast<double>(u32Max);
210 1 : u64 splitingTimes = static_cast<int>(std::ceil(countSplitingTimes));
211 1 : u64 src = ins.GetSrcBuffer().GetAddr();
212 1 : u64 dst = ins.GetDstBuffer().GetAddr();
213 1 : u64 blockSize = u32Max;
214 1 : u64 offset = u32Max;
215 2 : for (u64 i = 0; i < splitingTimes; i++) {
216 : // 处理尾块数据
217 1 : if(i == splitingTimes - 1) {
218 1 : blockSize = ins.GetSrcBuffer().GetSize() - u32Max * (splitingTimes - 1);
219 1 : offset = 0;
220 : }
221 :
222 1 : auto taskId = stream.GetRtsq()->GetTaskId();
223 1 : stream.GetRtsq()->SdmaCopy(src, dst, blockSize, 0); // 待确认， PART_ID是否固定设置为 0
224 3 : HCCL_INFO("InsLocalCopyExtend srcA:0x%llx dstA:0x%llx,size=0x%llx", src, dst, blockSize);
225 1 : TaskParam taskParam{};
226 1 : taskParam.taskType = TaskParamType::TASK_SDMA;
227 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
228 1 : taskParam.taskPara.DMA.src = reinterpret_cast<void *>(src);
229 1 : taskParam.taskPara.DMA.dst = reinterpret_cast<void *>(dst);
230 1 : taskParam.taskPara.DMA.size = blockSize;
231 1 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
232 1 : taskParam.taskPara.DMA.linkType = DfxLinkType::ONCHIP;
233 1 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
234 1 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
235 1 : resMgrFetcher->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
236 1 : src += offset;
237 1 : dst += offset;
238 1 : }
239 : }
240 :
241 1 : inline void AicpuCheckLocalReduceIns(const InsLocalReduce &ins)
242 : {
243 1 : if (UNLIKELY(ins.GetDataType() == DataType::INT64)) {
244 0 : THROW<InvalidParamsException>(StringFormat("%s LocalReduce SDMA InlineReduce dose not support INT64, need use TBE.",
245 : __func__));
246 : }
247 1 : }
248 :
249 1 : void Interpret(const InsLocalReduce &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
250 : {
251 : // SDMA支持的Reduce，则使用 sdmaReduce
252 : // SDMA不支持的Reduce，则使用 TBE算子(Asend C算子）
253 :
254 1 : if (UNLIKELY(ins.GetSrcSlice().GetSize() == 0)) {
255 0 : HCCL_WARNING("%s InsLocalReduce srcSlice size is 0, return", __func__);
256 0 : return;
257 : }
258 :
259 1 : if (UNLIKELY(ins.GetSrcSlice().GetSize() != ins.GetDstSlice().GetSize())) {
260 0 : HCCL_WARNING("%s InsLocalReduce srcSlice size is not equal to dstSlice size, return", __func__);
261 0 : return;
262 : }
263 :
264 1 : AicpuCheckLocalReduceIns(ins);
265 1 : RmaBufferLite* srcPtr = resMgrFetcher->GetRmaBufferLite(ins.GetSrcSlice().GetType());
266 1 : RmaBufferLite* dstPtr = resMgrFetcher->GetRmaBufferLite(ins.GetDstSlice().GetType());
267 1 : u64 srcOffset = ins.GetSrcSlice().GetOffset();
268 1 : u64 dstOffset = ins.GetDstSlice().GetOffset();
269 1 : if (UNLIKELY((srcPtr->GetSize() < srcOffset) && (dstPtr->GetSize() < dstOffset))) {
270 0 : THROW<InvalidParamsException>(StringFormat(
271 : "Interpret: offset exceeds memSize, srcPtr size[%llu], srcOffset[%llu], dstPtr size[%llu], dstOffset[%llu]",
272 : srcPtr->GetSize(), srcOffset, dstPtr->GetSize(), dstOffset));
273 : }
274 1 : u64 src = srcPtr->GetAddr() + srcOffset;
275 1 : u64 dst = dstPtr->GetAddr() + dstOffset;
276 1 : ReduceIn reduceIn(ins.GetDataType(), ins.GetReduceOp());
277 :
278 1 : auto taskId = stream.GetRtsq()->GetTaskId();
279 1 : stream.GetRtsq()->SdmaReduce(src, dst, ins.GetSrcSlice().GetSize(), 0, reduceIn); // 待确认， PART_ID是否固定设置为 0
280 :
281 3 : HCCL_INFO("InsLocalReduce srcA:0x%llx dstA:0x%llx,size=0x%llx", src, dst, ins.GetSrcSlice().GetSize());
282 1 : TaskParam taskParam{};
283 1 : taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
284 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
285 1 : taskParam.taskPara.Reduce.src = reinterpret_cast<void *>(src);
286 1 : taskParam.taskPara.Reduce.dst = reinterpret_cast<void *>(dst);
287 1 : taskParam.taskPara.Reduce.size = ins.GetSrcSlice().GetSize();
288 1 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
289 1 : taskParam.taskPara.Reduce.linkType = DfxLinkType::ONCHIP;
290 1 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(ins.GetDataType());
291 1 : taskParam.taskPara.Reduce.reduceOp = ReduceOpToHcclReduceOp(ins.GetReduceOp());
292 1 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
293 1 : resMgrFetcher->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
294 1 : }
295 :
296 2 : void Interpret(const InsLocalWaitGroup &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
297 : {
298 2 : auto taskId = stream.GetRtsq()->GetTaskId();
299 2 : auto ¬ify = GetCntNto1NotifyLite(ins, resMgrFetcher);
300 1 : u32 value = 0;
301 1 : u32 offsetNum = 32;
302 3 : for (auto iter = ins.Iter(); iter.HasNext(); ++iter) {
303 2 : if (UNLIKELY(*iter >= offsetNum)) {
304 0 : THROW<InternalException>("Invalid iter value: %d. Must be in [0, 31].", *iter);
305 : }
306 2 : value |= BASE_BIT << *iter;
307 : }
308 3 : HCCL_INFO("InsLocalBcastPost notifyId=%u, value %u", notify.GetId(), value);
309 1 : stream.GetRtsq()->CntNto1NotifyWait(notify.GetId(), value);
310 :
311 1 : TaskParam taskParam{};
312 1 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
313 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
314 1 : taskParam.taskPara.Notify.notifyID = notify.GetId();
315 1 : taskParam.taskPara.Notify.value = value;
316 1 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
317 1 : resMgrFetcher->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
318 1 : }
319 :
320 2 : void Interpret(const InsLocalBcastPost &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
321 : {
322 2 : auto taskId = stream.GetRtsq()->GetTaskId();
323 2 : auto ¬ify = GetCnt1toNNotifyLite(ins, resMgrFetcher);
324 1 : u32 value = 0;
325 3 : for (auto iter = ins.Iter(); iter.HasNext(); ++iter) {
326 2 : value |= BASE_BIT << *iter;
327 : }
328 3 : HCCL_INFO("InsLocalBcastPost notifyId=%u, value %u", notify.GetId(), value);
329 1 : stream.GetRtsq()->Cnt1toNNotifyRecord(notify.GetId(), value);
330 :
331 1 : TaskParam taskParam {};
332 1 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
333 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
334 1 : taskParam.taskPara.Notify.notifyID = notify.GetId();
335 1 : taskParam.taskPara.Notify.value = value;
336 1 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
337 1 : resMgrFetcher->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
338 1 : }
339 :
340 2 : void Interpret(const InsPostReady &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
341 : {
342 2 : auto &transport = GetTransportLite(ins, resMgrFetcher);
343 1 : transport.Post(NOTIFY_INDEX_READY, stream);
344 1 : }
345 :
346 4 : void Interpret(const InsWaitReady &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
347 : {
348 4 : auto &transport = GetTransportLite(ins, resMgrFetcher);
349 2 : transport.Wait(NOTIFY_INDEX_READY, stream);
350 2 : }
351 :
352 2 : void Interpret(const InsPostFin &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
353 : {
354 2 : auto &transport = GetTransportLite(ins, resMgrFetcher);
355 1 : transport.Post(NOTIFY_INDEX_FIN, stream);
356 1 : }
357 :
358 2 : void Interpret(const InsWaitFin &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
359 : {
360 2 : auto &transport = GetTransportLite(ins, resMgrFetcher);
361 1 : transport.Wait(NOTIFY_INDEX_FIN, stream);
362 1 : }
363 :
364 0 : void Interpret(const InsPostFinAck &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
365 : {
366 0 : auto &transport = GetTransportLite(ins, resMgrFetcher);
367 0 : transport.Post(NOTIFY_INDEX_FIN_ACK, stream);
368 0 : }
369 :
370 0 : void Interpret(const InsWaitFinAck &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
371 : {
372 0 : auto &transport = GetTransportLite(ins, resMgrFetcher);
373 0 : transport.Wait(NOTIFY_INDEX_FIN_ACK, stream);
374 0 : }
375 :
376 1 : void Interpret(const InsRead &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
377 : {
378 1 : if (UNLIKELY(ins.GetLocalSlice().GetSize() == 0 && ins.GetRemoteSlice().GetSize() == 0)) {
379 0 : HCCL_WARNING("%s InsRead localSlice size is 0 and remoteSlice size is 0, return", __func__);
380 0 : return;
381 1 : } else if (UNLIKELY(ins.GetLocalSlice().GetSize() != ins.GetRemoteSlice().GetSize())) {
382 0 : THROW<InvalidParamsException>(StringFormat("%s InsRead either localSlice size or remoteSlice size is not zero",
383 : __func__));
384 : }
385 :
386 1 : auto &transport = GetTransportLite(ins, resMgrFetcher);
387 1 : transport.Read(GetLocRmaBufferLite(ins, resMgrFetcher), GetRmtBuffer(ins, transport, resMgrFetcher), stream);
388 : }
389 :
390 1 : void Interpret(const InsReadReduce &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
391 : {
392 1 : if (UNLIKELY(ins.GetLocalSlice().GetSize() == 0 && ins.GetRemoteSlice().GetSize() == 0)) {
393 0 : HCCL_WARNING("%s InsReadReduce localSlice size is 0 and remoteSlice size is 0, return", __func__);
394 0 : return;
395 1 : } else if (UNLIKELY(ins.GetLocalSlice().GetSize() != ins.GetRemoteSlice().GetSize())) {
396 0 : THROW<InvalidParamsException>(StringFormat("%s InsReadReduce either localSlice size or remoteSlice size "
397 : "is not zero", __func__));
398 : }
399 :
400 1 : auto &transport = GetTransportLite(ins, resMgrFetcher);
401 1 : transport.ReadReduce(GetLocRmaBufferLite(ins, resMgrFetcher), GetRmtBuffer(ins, transport, resMgrFetcher),
402 2 : ReduceIn(ins.GetDataType(), ins.GetReduceOp()), stream);
403 : }
404 :
405 6 : void Interpret(const InsBatchRead &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
406 : {
407 6 : auto &transport = GetTransportLite(ins, resMgrFetcher);
408 6 : std::vector<RmaBufferLite> locRmaBufferLites;
409 6 : std::vector<Buffer> rmtBuffers;
410 6 : std::vector<BaseTransportLiteImpl::TransferOp> transferOp;
411 6 : if (UNLIKELY(!ins.Iter().HasNext())) {
412 2 : THROW<InvalidParamsException>(StringFormat("[%s] the number of InsBatchRead is zero.", __func__));
413 : }
414 :
415 9 : for (auto iter = ins.Iter(); iter.HasNext(); ++iter) {
416 6 : if (iter->GetType() == InstructionType::READ) {
417 3 : const InsRead &insRead = dynamic_cast<const InsRead &>(*iter);
418 3 : if (UNLIKELY(insRead.GetLocalSlice().GetSize() == 0 && insRead.GetRemoteSlice().GetSize() == 0)) {
419 3 : HCCL_WARNING("%s InsRead in InsBatchRead localSlice size is 0 and remoteSlice size is 0, return",
420 : __func__);
421 1 : continue;
422 3 : } else if (UNLIKELY(insRead.GetLocalSlice().GetSize() != insRead.GetRemoteSlice().GetSize())) {
423 2 : THROW<InvalidParamsException>(StringFormat("%s InsRead in InsBatchRead either localSlice size or "
424 : "remoteSlice size is not zero", __func__));
425 : }
426 1 : locRmaBufferLites.push_back(GetLocRmaBufferLite(insRead, resMgrFetcher));
427 1 : rmtBuffers.push_back(GetRmtBuffer(insRead, transport, resMgrFetcher));
428 1 : transferOp.push_back({TransferType(TransferType::READ), ReduceIn(DataType::INVALID, ReduceOp::INVALID)});
429 3 : } else if (iter->GetType() == InstructionType::READ_REDUCE) {
430 3 : const InsReadReduce &insReadReduce = dynamic_cast<const InsReadReduce &>(*iter);
431 3 : if (UNLIKELY(insReadReduce.GetLocalSlice().GetSize() == 0 && insReadReduce.GetRemoteSlice().GetSize() == 0)) {
432 3 : HCCL_WARNING("%s InsReadReduce in InsBatchRead localSlice size is 0 and remoteSlice size is 0, return",
433 : __func__);
434 1 : continue;
435 3 : } else if (UNLIKELY(insReadReduce.GetLocalSlice().GetSize() != insReadReduce.GetRemoteSlice().GetSize())) {
436 2 : THROW<InvalidParamsException>(StringFormat("%s InsReadReduce in InsBatchRead either localSlice size or "
437 : "remoteSlice size is not 0", __func__));
438 : }
439 1 : locRmaBufferLites.push_back(GetLocRmaBufferLite(insReadReduce, resMgrFetcher));
440 1 : rmtBuffers.push_back(GetRmtBuffer(insReadReduce, transport, resMgrFetcher));
441 1 : transferOp.push_back({TransferType(TransferType::READ),
442 : ReduceIn(insReadReduce.GetDataType(), insReadReduce.GetReduceOp())});
443 : }
444 5 : }
445 :
446 3 : if (UNLIKELY(locRmaBufferLites.empty())) {
447 2 : return;
448 : }
449 1 : transport.BatchTransfer(locRmaBufferLites, rmtBuffers, transferOp, stream);
450 16 : }
451 :
452 0 : void Interpret(const InsReadExtend &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
453 : {
454 0 : DataBuffer localBuffer = ins.GetLocalBuffer();
455 0 : if (UNLIKELY(localBuffer.GetSize() == 0)) {
456 0 : HCCL_WARNING("%s insReadExtend localSlice size is 0, return", __func__);
457 0 : return;
458 : }
459 0 : DataBuffer remoteBuffer = ins.GetRemoteBuffer();
460 0 : u64 scratchAddr = resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetAddr();
461 0 : u64 scratchSize = resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetSize();
462 0 : HCCL_INFO("%s scratchAddr = %llu, scratchSize = %llu", __func__, scratchAddr, scratchSize);
463 0 : RmaBufferLite loc(localBuffer.GetAddr(), localBuffer.GetSize(),
464 0 : resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetTokenId(),
465 0 : resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetTokenValue());
466 0 : Buffer rmt(remoteBuffer.GetAddr(), remoteBuffer.GetSize());
467 0 : HCCL_INFO("%s RmaBufferLite = %s, Buffer = %s", __func__, loc.Describe().c_str(), rmt.Describe().c_str());
468 0 : auto &transport = GetTransportLite(ins, resMgrFetcher);
469 0 : transport.Read(loc, rmt, stream);
470 0 : }
471 :
472 2 : void Interpret(const InsWrite &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
473 : {
474 2 : if (UNLIKELY(ins.GetLocalSlice().GetSize() == 0 && ins.GetRemoteSlice().GetSize() == 0)) {
475 0 : HCCL_WARNING("%s InsWrite localSlice size is 0 and remoteSlice size is 0, return", __func__);
476 0 : return;
477 2 : } else if (UNLIKELY(ins.GetLocalSlice().GetSize() != ins.GetRemoteSlice().GetSize())) {
478 2 : THROW<InvalidParamsException>(StringFormat("%s InsWrite either localSlice size or remoteSlice size is not zero",
479 : __func__));
480 : }
481 :
482 1 : auto &transport = GetTransportLite(ins, resMgrFetcher);
483 1 : transport.Write(GetLocRmaBufferLite(ins, resMgrFetcher), GetRmtBuffer(ins, transport, resMgrFetcher), stream);
484 : }
485 :
486 6 : void Interpret(const InsBatchWrite &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
487 : {
488 6 : auto &transport = GetTransportLite(ins, resMgrFetcher);
489 6 : std::vector<RmaBufferLite> locRmaBufferLites;
490 6 : std::vector<Buffer> rmtBuffers;
491 6 : std::vector<BaseTransportLiteImpl::TransferOp> transferOp;
492 6 : if (UNLIKELY(!ins.Iter().HasNext())) {
493 2 : THROW<InvalidParamsException>(StringFormat("[%s] the number of InsBatchWrite is zero.", __func__));
494 : }
495 :
496 9 : for (auto iter = ins.Iter(); iter.HasNext(); ++iter) {
497 6 : if (iter->GetType() == InstructionType::WRITE) {
498 3 : const InsWrite &insWrite = dynamic_cast<const InsWrite &>(*iter);
499 3 : if (UNLIKELY(insWrite.GetLocalSlice().GetSize() == 0 && insWrite.GetRemoteSlice().GetSize() == 0)) {
500 3 : HCCL_WARNING("%s InsWrite in InsBatchWrite localSlice size is 0 and remoteSlice size is 0, return",
501 : __func__);
502 1 : continue;
503 3 : } else if (UNLIKELY(insWrite.GetLocalSlice().GetSize() != insWrite.GetRemoteSlice().GetSize())) {
504 2 : THROW<InvalidParamsException>(StringFormat("%s InsWrite in InsBatchWrite either localSlice size or "
505 : "remoteSlice size is not zero", __func__));
506 : }
507 1 : locRmaBufferLites.push_back(GetLocRmaBufferLite(insWrite, resMgrFetcher));
508 1 : rmtBuffers.push_back(GetRmtBuffer(insWrite, transport, resMgrFetcher));
509 1 : transferOp.push_back({TransferType(TransferType::WRITE), ReduceIn(DataType::INVALID, ReduceOp::INVALID)});
510 3 : } else if (iter->GetType() == InstructionType::WRITE_REDUCE) {
511 3 : const InsWriteReduce &insWriteReduce = dynamic_cast<const InsWriteReduce &>(*iter);
512 3 : if (UNLIKELY(insWriteReduce.GetLocalSlice().GetSize() == 0 && insWriteReduce.GetRemoteSlice().GetSize() == 0)) {
513 3 : HCCL_WARNING("%s InsWriteReduce in InsBatchWrite localSlice size is 0 and remoteSlice size is 0, "
514 : "return", __func__);
515 1 : continue;
516 3 : } else if (UNLIKELY(insWriteReduce.GetLocalSlice().GetSize() != insWriteReduce.GetRemoteSlice().GetSize())) {
517 2 : THROW<InvalidParamsException>(StringFormat("%s InsWriteReduce in InsBatchWrite either localSlice size "
518 : "or remoteSlice size is not 0", __func__));
519 : }
520 1 : locRmaBufferLites.push_back(GetLocRmaBufferLite(insWriteReduce, resMgrFetcher));
521 1 : rmtBuffers.push_back(GetRmtBuffer(insWriteReduce, transport, resMgrFetcher));
522 1 : transferOp.push_back({TransferType(TransferType::WRITE),
523 : ReduceIn(insWriteReduce.GetDataType(), insWriteReduce.GetReduceOp())});
524 : }
525 5 : }
526 :
527 3 : if (UNLIKELY(locRmaBufferLites.empty())) {
528 2 : return;
529 : }
530 1 : transport.BatchTransfer(locRmaBufferLites, rmtBuffers, transferOp, stream);
531 16 : }
532 :
533 2 : void Interpret(const InsWriteExtend &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
534 : {
535 2 : DataBuffer localBuffer = ins.GetLocalBuffer();
536 2 : if (UNLIKELY(localBuffer.GetSize() == 0)) {
537 3 : HCCL_WARNING("%s insWriteExtend localSlice size is 0, return", __func__);
538 1 : return;
539 : }
540 1 : DataBuffer remoteBuffer = ins.GetRemoteBuffer();
541 1 : u64 scratchAddr = resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetAddr();
542 1 : u64 scratchSize = resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetSize();
543 3 : HCCL_INFO("%s scratchAddr = %llu, scratchSize = %llu", __func__, scratchAddr, scratchSize);
544 1 : RmaBufferLite loc(localBuffer.GetAddr(), localBuffer.GetSize(),
545 1 : resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetTokenId(),
546 2 : resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetTokenValue());
547 1 : Buffer rmt(remoteBuffer.GetAddr(), remoteBuffer.GetSize());
548 3 : HCCL_INFO("%s RmaBufferLite = %s, Buffer = %s", __func__, loc.Describe().c_str(), rmt.Describe().c_str());
549 1 : auto &transport = GetTransportLite(ins, resMgrFetcher);
550 1 : transport.Write(loc, rmt, stream);
551 2 : }
552 :
553 2 : void Interpret(const InsWriteWithFin &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
554 : {
555 2 : auto &transport = GetTransportLite(ins, resMgrFetcher);
556 :
557 2 : if (UNLIKELY(ins.GetLocalSlice().GetSize() == 0)) {
558 3 : HCCL_WARNING("%s insWriteWithFin localSlice size is 0, transform to insPostFin", __func__);
559 1 : transport.Post(NOTIFY_INDEX_FIN, stream);
560 1 : return;
561 : }
562 :
563 1 : transport.WriteWithNotify(GetLocRmaBufferLite(ins, resMgrFetcher), GetRmtBuffer(ins, transport, resMgrFetcher),
564 2 : WithNotifyIn(TransportNotifyType::NORMAL, NOTIFY_INDEX_FIN), stream);
565 : }
566 :
567 2 : void Interpret(const InsWriteWithFinExtend &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
568 : {
569 2 : DataBuffer localBuffer = ins.GetLocalBuffer();
570 2 : DataBuffer remoteBuffer = ins.GetRemoteBuffer();
571 2 : auto &transport = GetTransportLite(ins, resMgrFetcher);
572 :
573 2 : if (UNLIKELY(localBuffer.GetSize() == 0)) {
574 3 : HCCL_WARNING("%s insWriteWithFinExtend localBuffer size is 0, transform to insPostFin", __func__);
575 1 : transport.Post(NOTIFY_INDEX_FIN, stream);
576 1 : return;
577 : }
578 1 : u64 scratchAddr = resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetAddr();
579 1 : u64 scratchSize = resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetSize();
580 3 : HCCL_INFO("%s scratchAddr = %llu, scratchSize = %llu", __func__, scratchAddr, scratchSize);
581 :
582 1 : RmaBufferLite loc(localBuffer.GetAddr(), localBuffer.GetSize(),
583 1 : resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetTokenId(),
584 2 : resMgrFetcher->GetRmaBufferLite(BufferType::SCRATCH)->GetTokenValue());
585 1 : Buffer rmt(remoteBuffer.GetAddr(), remoteBuffer.GetSize());
586 3 : HCCL_INFO("%s RmaBufferLite = %s, Buffer = %s", __func__, loc.Describe().c_str(), rmt.Describe().c_str());
587 :
588 1 : transport.WriteWithNotify(loc, rmt, WithNotifyIn(TransportNotifyType::NORMAL, NOTIFY_INDEX_FIN), stream);
589 3 : }
590 :
591 2 : void Interpret(const InsWriteReduce &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
592 : {
593 2 : if (UNLIKELY(ins.GetLocalSlice().GetSize() == 0 && ins.GetRemoteSlice().GetSize() == 0)) {
594 0 : HCCL_WARNING("%s InsWriteReduce localSlice size is 0 and remoteSlice size is 0, return", __func__);
595 0 : return;
596 2 : } else if (UNLIKELY(ins.GetLocalSlice().GetSize() != ins.GetRemoteSlice().GetSize())) {
597 2 : THROW<InvalidParamsException>(StringFormat("%s InsWriteReduce either localSlice size or remoteSlice size "
598 : "is not zero", __func__));
599 : }
600 :
601 1 : auto &transport = GetTransportLite(ins, resMgrFetcher);
602 1 : transport.WriteReduce(GetLocRmaBufferLite(ins, resMgrFetcher), GetRmtBuffer(ins, transport, resMgrFetcher),
603 2 : ReduceIn(ins.GetDataType(), ins.GetReduceOp()), stream);
604 : }
605 :
606 2 : void Interpret(const InsWriteReduceWithFin &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
607 : {
608 2 : auto &transport = GetTransportLite(ins, resMgrFetcher);
609 :
610 2 : if (UNLIKELY(ins.GetLocalSlice().GetSize() == 0)) {
611 3 : HCCL_WARNING("%s insWriteReduceWithFin localSlice size is 0, transform to insPostFin", __func__);
612 1 : transport.Post(NOTIFY_INDEX_FIN, stream);
613 1 : return;
614 : }
615 :
616 1 : transport.WriteReduceWithNotify(GetLocRmaBufferLite(ins, resMgrFetcher),
617 2 : GetRmtBuffer(ins, transport, resMgrFetcher),
618 1 : ReduceIn(ins.GetDataType(), ins.GetReduceOp()),
619 2 : WithNotifyIn(TransportNotifyType::NORMAL, NOTIFY_INDEX_FIN), stream);
620 : }
621 :
622 0 : void Interpret(const InsBatchOneSidedRead &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
623 : {
624 0 : auto &transport = GetTransportLite(ins, resMgrFetcher);
625 0 : transport.BatchOneSidedRead(ins.GetLocalSlice(), ins.GetRemoteSlice(), stream);
626 0 : }
627 :
628 0 : void Interpret(const InsBatchOneSidedWrite &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
629 : {
630 0 : auto &transport = GetTransportLite(ins, resMgrFetcher);
631 0 : transport.BatchOneSidedWrite(ins.GetLocalSlice(), ins.GetRemoteSlice(), stream);
632 0 : }
633 :
634 : using InsToSqeRule91095 = std::function<void(const Instruction &, const StreamLite &, ResMgrFetcher *resMgrFetcher)>;
635 :
636 29 : template <class InsType> InsToSqeRule91095 Rule91095()
637 : {
638 51 : return [](const Instruction &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher) {
639 22 : return Interpret(static_cast<const InsType &>(ins), stream, resMgrFetcher);
640 29 : };
641 : }
642 :
643 1 : void Interpret(const InsStreamSync &insStreamSync, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
644 : {
645 : (void)insStreamSync;
646 1 : constexpr uint64_t NANOSECOND_TO_SECOND = 1000000000U;
647 1 : const uint64_t kPrintSqInterval = 30U;
648 1 : uint32_t head = 0;
649 1 : uint32_t tail = 0;
650 1 : u32 timeout = resMgrFetcher->GetExecTimeOut() + 10;
651 1 : u64 startUsec = GetCurAicpuTimestamp();
652 1 : u64 lastUsec = startUsec;
653 1 : u32 sqId = stream.GetSqId();
654 1 : tail = stream.GetRtsq()->QuerySqTail();
655 3 : HCCL_INFO("StreamSync aicpu stream sqid[%d] tail[%u]", sqId, tail);
656 : do {
657 1 : head = stream.GetRtsq()->QuerySqHead();
658 1 : u64 curUsec = GetCurAicpuTimestamp();
659 1 : if (UNLIKELY(curUsec - startUsec > NANOSECOND_TO_SECOND * timeout)) {
660 : string msg = StringFormat("stream sync timeout %lus. curhead:%u, curtail:%u, sqId:%u",
661 0 : timeout, head, tail, sqId);
662 0 : THROW<TimeoutException>(msg);
663 0 : }
664 :
665 : // 等待下发阶段，每隔30s打印一次状态
666 1 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * kPrintSqInterval) {
667 0 : lastUsec = curUsec;
668 0 : HCCL_INFO("[StreamSync]Current state. sqid:%d, head:%u, tail:%u",
669 : sqId, head, tail);
670 : }
671 1 : } while (head != tail);
672 1 : }
673 :
674 3 : void Interpret(const InsPreStreamSync &insPreStreamSync, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
675 : {
676 3 : HcclResult ret = stream.GetRtsq()->SetPreStreamSyncReady();
677 3 : stream.GetRtsq()->LaunchTask();
678 3 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
679 0 : string msg = StringFormat("[Interpret]SetPreStreamSyncReady failed");
680 0 : THROW<InternalException>(msg);
681 0 : }
682 3 : }
683 :
684 1 : void Interpret(const InsAicpuReduce &insAicpuReduce, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
685 : {
686 : //使用aicpu进行reduce运算，支持int64 uint64 fp64
687 1 : if (UNLIKELY(insAicpuReduce.GetSrcSlice().GetSize() == 0)) {
688 0 : HCCL_WARNING("%s InsAicpuReduce srcSlice size is 0, return", __func__);
689 0 : return;
690 : }
691 :
692 1 : if (UNLIKELY(insAicpuReduce.GetSrcSlice().GetSize() != insAicpuReduce.GetDstSlice().GetSize())) {
693 0 : HCCL_WARNING("%s InsAicpuReduce srcSlice size is not equal to dstSlice size, return", __func__);
694 0 : return;
695 : }
696 :
697 1 : RmaBufferLite* srcPtr = resMgrFetcher->GetRmaBufferLite(insAicpuReduce.GetSrcSlice().GetType());
698 1 : RmaBufferLite* dstPtr = resMgrFetcher->GetRmaBufferLite(insAicpuReduce.GetDstSlice().GetType());
699 1 : u64 srcOffset = insAicpuReduce.GetSrcSlice().GetOffset();
700 1 : u64 dstOffset = insAicpuReduce.GetDstSlice().GetOffset();
701 1 : if (UNLIKELY((srcPtr->GetSize() < srcOffset) && (dstPtr->GetSize() < dstOffset))) {
702 0 : THROW<InvalidParamsException>(StringFormat(
703 : "Interpret: offset exceeds memSize, srcPtr size[%llu], srcOffset[%llu], dstPtr size[%llu], dstOffset[%llu]",
704 : srcPtr->GetSize(), srcOffset, dstPtr->GetSize(), dstOffset));
705 : }
706 1 : void *dst = reinterpret_cast<void *>(dstPtr->GetAddr() + insAicpuReduce.GetDstSlice().GetOffset());
707 1 : void *src = reinterpret_cast<void *>(srcPtr->GetAddr() + insAicpuReduce.GetSrcSlice().GetOffset());
708 1 : insAicpuReduce.RunAicpuReduce(dst, insAicpuReduce.GetDstSlice().GetSize(), src, insAicpuReduce.GetSrcSlice().GetSize(),
709 : insAicpuReduce.GetDataType(), insAicpuReduce.GetReduceOp());
710 3 : HCCL_INFO("InsAicpuReduce srcA:0x%p dstA:0x%p, size=0x%llx", src, dst, insAicpuReduce.GetSrcSlice().GetSize());
711 1 : auto taskId = stream.GetRtsq()->GetTaskId();
712 1 : TaskParam taskParam{};
713 1 : taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
714 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
715 1 : taskParam.taskPara.Reduce.src = reinterpret_cast<void *>(src);
716 1 : taskParam.taskPara.Reduce.dst = reinterpret_cast<void *>(dst);
717 1 : taskParam.taskPara.Reduce.size = insAicpuReduce.GetSrcSlice().GetSize();
718 1 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
719 1 : taskParam.taskPara.Reduce.linkType = DfxLinkType::ONCHIP;
720 1 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(insAicpuReduce.GetDataType());
721 1 : taskParam.taskPara.Reduce.reduceOp = ReduceOpToHcclReduceOp(insAicpuReduce.GetReduceOp());
722 1 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
723 1 : resMgrFetcher->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
724 1 : }
725 :
726 : const std::unordered_map<InstructionType, InsToSqeRule91095, std::EnumClassHash> insRule91095Map{
727 : {InstructionType::LOCAL_COPY, Rule91095<InsLocalCopy>()},
728 : {InstructionType::LOCAL_POST_TO, Rule91095<InsLocalPostTo>()},
729 : {InstructionType::LOCAL_WAIT_FROM, Rule91095<InsLocalWaitFrom>()},
730 : {InstructionType::LOCAL_BCAST_POST, Rule91095<InsLocalBcastPost>()},
731 : {InstructionType::LOCAL_WAIT_GROUP, Rule91095<InsLocalWaitGroup>()},
732 : {InstructionType::WAIT_READY, Rule91095<InsWaitReady>()},
733 : {InstructionType::POST_READY, Rule91095<InsPostReady>()},
734 : {InstructionType::WAIT_FIN, Rule91095<InsWaitFin>()},
735 : {InstructionType::POST_FIN, Rule91095<InsPostFin>()},
736 : {InstructionType::WRITE, Rule91095<InsWrite>()},
737 : {InstructionType::WRITE_REDUCE, Rule91095<InsWriteReduce>()},
738 : {InstructionType::BATCH_WRITE, Rule91095<InsBatchWrite>()},
739 : {InstructionType::BATCH_READ, Rule91095<InsBatchRead>()},
740 : {InstructionType::READ, Rule91095<InsRead>()},
741 : {InstructionType::READ_REDUCE, Rule91095<InsReadReduce>()},
742 : {InstructionType::READ_EXTEND, Rule91095<InsReadExtend>()},
743 : {InstructionType::WRITE_REDUCE_WITH_FIN, Rule91095<InsWriteReduceWithFin>()},
744 : {InstructionType::WRITE_WITH_FIN, Rule91095<InsWriteWithFin>()},
745 : {InstructionType::LOCAL_COPY_EXTEND, Rule91095<InsLocalCopyExtend>()},
746 : {InstructionType::WRITE_EXTEND, Rule91095<InsWriteExtend>()},
747 : {InstructionType::WRITE_WITH_FIN_EXTEND, Rule91095<InsWriteWithFinExtend>()},
748 : {InstructionType::BATCH_ONE_SIDED_WRITE, Rule91095<InsBatchOneSidedWrite>()},
749 : {InstructionType::BATCH_ONE_SIDED_READ, Rule91095<InsBatchOneSidedRead>()},
750 : {InstructionType::LOCAL_REDUCE, Rule91095<InsLocalReduce>()},
751 : {InstructionType::POST_FIN_ACK, Rule91095<InsPostFinAck>()},
752 : {InstructionType::WAIT_FIN_ACK, Rule91095<InsWaitFinAck>()},
753 : {InstructionType::STREAM_SYNC, Rule91095<InsStreamSync>()},
754 : {InstructionType::PRE_STREAM_SYNC, Rule91095<InsPreStreamSync>()},
755 : {InstructionType::AICPU_REDUCE, Rule91095<InsAicpuReduce>()}
756 : };
757 :
758 22 : void Interpret(const Instruction &ins, const StreamLite &stream, ResMgrFetcher *resMgrFetcher)
759 : {
760 66 : HCCL_INFO("%s Instruction %s", __func__, ins.Describe().c_str());
761 22 : auto iter = insRule91095Map.find(ins.GetType());
762 22 : if (iter != insRule91095Map.end()) {
763 22 : auto &rule = iter->second;
764 44 : return rule(ins, stream, resMgrFetcher);
765 : }
766 0 : THROW<InternalException>(
767 0 : StringFormat("%s: invalid instruction type[%u]", __func__, ins.GetType()));
768 : }
769 :
770 : } // namespace Hccl
|
