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