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 <algorithm>
12 : #include "ins_to_sqe_rule.h"
13 : #include "null_ptr_exception.h"
14 : #include "ins_executor.h"
15 : #include "communicator_impl_lite_manager.h"
16 :
17 : namespace Hccl {
18 :
19 : constexpr u64 FOUR_BYTES = 4;
20 : constexpr u32 LAUNCH_PRINT_INTERVAL = 20;
21 :
22 0 : void InsExecutor::Execute(const InsQueue& insQueue)
23 : {
24 0 : StreamLiteMgr* streamLiteMgr = resMgrFetcher_->GetStreamLiteMgr();
25 0 : int slaveStreamIndex = 0;
26 0 : for (auto slaveIter = insQueue.IterSlaves(); slaveIter.HasNext(); ++slaveIter) {
27 0 : ExecuteSingleQue(*slaveIter, streamLiteMgr->GetSlave(slaveStreamIndex++));
28 0 : }
29 0 : CHECK_NULLPTR(streamLiteMgr->GetMaster(), "[Execute]master stream is nullptr!");
30 0 : ExecuteSingleQue(insQueue, streamLiteMgr->GetMaster(), true);
31 0 : }
32 :
33 8 : void InsExecutor::AddOpCounter(const StreamLite& stream, bool isHead) const
34 : {
35 8 : CHECK_NULLPTR(resMgrFetcher_, "[InsExecutor::AddOpCounter] resMgrFetcher_ is nullptr!");
36 8 : auto rtsq = stream.GetRtsq();
37 8 : if (rtsq == nullptr) {
38 0 : HCCL_ERROR("[InsExecutor::%s] stream.GetRtsq() is nullptr", __func__);
39 8 : return;
40 : }
41 8 : u64 counterSrcAddr = resMgrFetcher_->GetCounterAddr();
42 8 : if (counterSrcAddr == 0) {
43 24 : HCCL_ERROR("InsExecutor::%s counter addr is null.", __func__);
44 8 : return;
45 : }
46 0 : u64 dstAddr = isHead == true ? counterSrcAddr + FOUR_BYTES : counterSrcAddr + FOUR_BYTES * 2;
47 0 : u64 count = FOUR_BYTES;
48 0 : HCCL_INFO("%s AddOpCounter start", __func__);
49 0 : auto taskId = rtsq->GetTaskId();
50 0 : rtsq->SdmaReduce(counterSrcAddr, dstAddr, count, 0, ReduceIn(DataType::FP32, ReduceOp::SUM));
51 :
52 0 : TaskParam taskParam{};
53 0 : taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
54 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
55 0 : taskParam.taskPara.Reduce.src = reinterpret_cast<void*>(counterSrcAddr);
56 0 : taskParam.taskPara.Reduce.dst = reinterpret_cast<void*>(dstAddr);
57 0 : taskParam.taskPara.Reduce.size = count;
58 0 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
59 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::ONCHIP;
60 0 : taskParam.taskPara.Reduce.reduceOp = HcclReduceOp::HCCL_REDUCE_SUM;
61 0 : taskParam.taskPara.Reduce.dataType = HcclDataType::HCCL_DATA_TYPE_FP32;
62 0 : auto taskInfo = std::make_unique<TaskInfo>(stream.GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
63 0 : resMgrFetcher_->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
64 0 : }
65 :
66 5 : void InsExecutor::ExecuteV82(const InsQueue& insQueue, bool isMc2)
67 : {
68 : // InsQueue 非空已经在外部进行了校验
69 5 : if (resMgrFetcher_ == nullptr) {
70 0 : THROW<NullPtrException>(StringFormat("InsExecutor::%s resMgrFetcher is null, isMc2 %d.", __func__, isMc2));
71 : return;
72 : }
73 5 : StreamLiteMgr* streamLiteMgr = resMgrFetcher_->GetStreamLiteMgr();
74 5 : if (streamLiteMgr == nullptr) {
75 0 : THROW<NullPtrException>(StringFormat("InsExecutor::%s streamLiteMgr is null, isMc2 %d.", __func__, isMc2));
76 : return;
77 : }
78 : // 先下主流上的notify wait任务，包括和host同步和op计数任务
79 5 : StreamLite* masterStream = streamLiteMgr->GetMaster();
80 5 : if (masterStream == nullptr) {
81 2 : THROW<NullPtrException>(StringFormat("InsExecutor::%s masterStream is null, isMc2 %d.", __func__, isMc2));
82 : return;
83 : }
84 4 : ReportMainStreamTask(*masterStream, MainStreamTaskType::HEAD);
85 4 : auto deviceWaitNotifyId = resMgrFetcher_->GetHostDeviceSyncNotifyLiteMgr()->GetDeviceWaitNotify()->GetId();
86 12 : HCCL_INFO("InsExecutor::%s GetDeviceWaitNotify id %u", __func__, deviceWaitNotifyId);
87 4 : if (!isMc2) {
88 4 : auto taskId = masterStream->GetRtsq()->GetTaskId();
89 4 : masterStream->GetRtsq()->NotifyWait(deviceWaitNotifyId);
90 4 : TaskParam taskParam{};
91 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
92 4 : taskParam.beginTime = ProfGetCurCpuTimestamp();
93 4 : taskParam.taskPara.Notify.notifyID = deviceWaitNotifyId;
94 4 : taskParam.taskPara.Notify.value = 1;
95 4 : auto taskInfo = std::make_unique<TaskInfo>(masterStream->GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
96 4 : resMgrFetcher_->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
97 4 : }
98 4 : AddOpCounter(*masterStream, true);
99 :
100 : // 将主流和从流上的Task分别下发执行
101 4 : ExecuteAllQueues950(insQueue, streamLiteMgr);
102 :
103 : // 下主流上的notify record任务，包括和host同步和op计数任务
104 4 : AddOpCounter(*masterStream, false);
105 4 : ReportMainStreamTask(*masterStream, MainStreamTaskType::TAIL);
106 4 : auto hostWaitNotifyId = resMgrFetcher_->GetHostDeviceSyncNotifyLiteMgr()->GetHostWaitNotify()->GetId();
107 12 : HCCL_INFO("InsExecutor::%s GetHostWaitNotify id %u", __func__, hostWaitNotifyId);
108 4 : if (!isMc2) {
109 4 : auto taskId = masterStream->GetRtsq()->GetTaskId();
110 4 : masterStream->GetRtsq()->NotifyRecordLoc(hostWaitNotifyId);
111 4 : TaskParam taskParam{};
112 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
113 4 : taskParam.beginTime = ProfGetCurCpuTimestamp();
114 4 : taskParam.taskPara.Notify.notifyID = hostWaitNotifyId;
115 4 : taskParam.taskPara.Notify.value = 1;
116 4 : auto taskInfo = std::make_unique<TaskInfo>(masterStream->GetSqId(), taskId, INVALID_VALUE_RANKID, taskParam);
117 4 : resMgrFetcher_->GetMirrorTaskMgrLite()->AddTaskInfo(std::move(taskInfo));
118 4 : }
119 4 : masterStream->GetRtsq()->LaunchTask();
120 : }
121 :
122 8 : void InsExecutor::ReportMainStreamTask(const StreamLite& stream, MainStreamTaskType type) const
123 : {
124 8 : FlagTaskInfo flagTaskInfo;
125 8 : flagTaskInfo.taskId = stream.GetRtsq()->GetTaskId();
126 8 : flagTaskInfo.type = type;
127 24 : HCCL_INFO("[%s] TaskInfo taskId %u", __func__, flagTaskInfo.taskId);
128 8 : ProfilingHandlerLite::GetInstance().ReportMainStreamTask(flagTaskInfo);
129 8 : }
130 :
131 4 : void InsExecutor::ExecuteAllQueues950(const InsQueue& insQueue, StreamLiteMgr* streamLiteMgr)
132 : {
133 12 : HCCL_INFO("InsExecutor::%s start", __func__);
134 4 : list<InsQueue::Iterator> slaveQueueIters;
135 4 : std::set<u32> slaveStreamIndexSet;
136 :
137 4 : bool isMasterInsIterEnd = false;
138 : // 用于判断一轮下发task过程中，是否有成功下发Task，以及开始记时
139 4 : bool isLaunchTask = false;
140 4 : auto startTime = std::chrono::steady_clock::now();
141 4 : auto timeoutValue = CommunicatorImplLiteMgr::GetInstance().GetEnvConfig().hcclExecTimeout + 20;
142 4 : auto timeout = std::chrono::seconds(timeoutValue);
143 4 : const std::chrono::seconds printInterval(LAUNCH_PRINT_INTERVAL); // 打印间隔30s
144 4 : auto lastPrintTime = std::chrono::steady_clock::now() - printInterval;
145 4 : InsQueue::Iterator masterQueueIter = insQueue.Iter();
146 4 : StreamLite* masterStream = streamLiteMgr->GetMaster();
147 4 : CHK_PRT_THROW(
148 : masterStream == nullptr, HCCL_ERROR("[InsExecutor::%s] masterStream is null.", __func__), InternalException,
149 : "masterStream is null");
150 : // 将准备下发到从流的subInsQueue的迭代器都存到迭代器数组内部
151 8 : for (auto slaveQueueIter = insQueue.IterSlaves(); slaveQueueIter.HasNext(); ++slaveQueueIter) {
152 4 : slaveQueueIters.emplace_back((*slaveQueueIter).Iter());
153 4 : }
154 4 : u32 maxSlaveQueuesSize = slaveQueueIters.size();
155 : // 创建和slaveQueueIter相对应的从流索引容器，用于后续下发任务一一对应
156 8 : for (u32 slaveStreamIndex = 0; slaveStreamIndex < slaveQueueIters.size(); ++slaveStreamIndex) {
157 4 : slaveStreamIndexSet.insert(slaveStreamIndex);
158 : }
159 : // 遍历迭代器数组，一个流上的InsQueue去下一个任务
160 21 : while (!slaveQueueIters.empty() || !isMasterInsIterEnd) {
161 : // 遍历从流InsQueue，每一次下发一个Task
162 17 : ExecuteSlaveQueue950(slaveQueueIters, streamLiteMgr, isLaunchTask, slaveStreamIndexSet);
163 : // 每一次循环，下发一次主流Task
164 17 : if (!isMasterInsIterEnd) {
165 15 : ExecuteMasterQueue950(masterQueueIter, masterStream, isMasterInsIterEnd, isLaunchTask);
166 : }
167 :
168 17 : CheckPreStreamSync(streamLiteMgr, maxSlaveQueuesSize);
169 : // 如果没有下发任务就开始记录超时时间
170 17 : if (isLaunchTask) {
171 17 : startTime = std::chrono::steady_clock::now();
172 0 : } else if (std::chrono::steady_clock::now() - lastPrintTime >= printInterval) {
173 0 : HCCL_INFO("[ExecuteAllQueues950]All Rtsq Queues full, wait for executor");
174 0 : lastPrintTime = std::chrono::steady_clock::now();
175 : }
176 17 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
177 0 : auto msg = StringFormat("[ExecuteAllQueues950]All Rtsq Queues full, timeout[%u s]", timeoutValue);
178 0 : HCCL_ERROR("%s", msg.c_str());
179 0 : THROW<InternalException>(msg);
180 0 : }
181 : }
182 12 : HCCL_INFO("InsExecutor::%s success", __func__);
183 4 : }
184 :
185 17 : void InsExecutor::ExecuteSlaveQueue950(
186 : list<InsQueue::Iterator>& slaveQueueIters, StreamLiteMgr* streamLiteMgr, bool& isLaunchTask,
187 : std::set<u32>& slaveStreamIndexSet)
188 : {
189 17 : auto slaveStreamIndexIter = slaveStreamIndexSet.begin();
190 17 : isLaunchTask = false;
191 39 : for (auto slaveQueueIter = slaveQueueIters.begin(); slaveQueueIter != slaveQueueIters.end();) {
192 22 : StreamLite* slaveStream = streamLiteMgr->GetSlave(*slaveStreamIndexIter);
193 22 : if (UNLIKELY(slaveStream == nullptr)) {
194 0 : THROW<NullPtrException>(StringFormat("InsExecutor::%s slaveStream is null,", __func__));
195 : }
196 22 : if (slaveStream->GetRtsq() == nullptr) {
197 0 : THROW<NullPtrException>(StringFormat(
198 : "InsExecutor::%s GetRtsq returned null for slaveStream Id(%u)", __func__, slaveStream->GetId()));
199 : }
200 : // 判断rtsq队列中的空间是否充足
201 22 : bool isRtsqQueueSpaceSufficient = slaveStream->GetRtsq()->IsRtsqQueueSpaceSufficient();
202 : // 判断当前从流是否有Int64类型reduce算子，是否需要等其他流任务下发完成
203 22 : bool isPreStreamSync = slaveStream->GetRtsq()->GetPreStreamSyncStatus();
204 22 : if (isRtsqQueueSpaceSufficient && !isPreStreamSync) {
205 19 : if (slaveQueueIter->HasNext()) {
206 45 : HCCL_INFO(
207 : "[ExecuteAllQueues950]InsExecutor::%s slave stream InsQueue start %s SqId(%u) stream Id(%u)",
208 : __func__, (*slaveQueueIter)->Describe().c_str(), slaveStream->GetSqId(), slaveStream->GetId());
209 15 : Interpret(**slaveQueueIter, *slaveStream, resMgrFetcher_);
210 : // 给迭代器内部Iter指向这条流上的InsQueue里下一个task
211 15 : ++(*slaveQueueIter);
212 : // 将迭代器指向下一条流上的InsQueue
213 15 : ++slaveQueueIter;
214 : // 将对应的流的索引执行下一条流
215 15 : ++slaveStreamIndexIter;
216 : } else {
217 : // 如果这个InsQueue上没有下一个task了，就擦掉容器内的对应迭代器
218 12 : HCCL_INFO(
219 : "[ExecuteAllQueues950]InsExecutor::%s slave stream Id(%u) Interpret insQueue finish", __func__,
220 : slaveStream->GetId());
221 4 : slaveQueueIter = slaveQueueIters.erase(slaveQueueIter);
222 : // 擦掉对应的从流索引，避免下任务下错从流
223 4 : slaveStreamIndexIter = slaveStreamIndexSet.erase(slaveStreamIndexIter);
224 4 : slaveStream->GetRtsq()->LaunchTask();
225 12 : HCCL_INFO(
226 : "[ExecuteAllQueues950]InsExecutor::%s slave stream Id(%u) launch task finish", __func__,
227 : slaveStream->GetId());
228 : }
229 19 : isLaunchTask = true;
230 19 : } else {
231 : // Rtsq上位置不足，先跳去下一个
232 3 : ++slaveQueueIter;
233 : // 将对应的流的索引执行下一条流
234 3 : ++slaveStreamIndexIter;
235 : }
236 22 : isPreStreamSyncExist_ = slaveStream->GetRtsq()->GetPreStreamSyncStatus() || isPreStreamSyncExist_;
237 : }
238 17 : }
239 :
240 15 : void InsExecutor::ExecuteMasterQueue950(
241 : InsQueue::Iterator& masterQueueIter, StreamLite* masterStream, bool& isMasterInsIterEnd, bool& isLaunchTask)
242 : {
243 : // 判断rtsq队列中的空间是否充足
244 15 : bool isRtsqQueueSpaceSufficient = masterStream->GetRtsq()->IsRtsqQueueSpaceSufficient();
245 : // 判断当前主流是否有Int64类型reduce算子，是否需要等其他流任务下发完成
246 15 : bool isPreStreamSync = masterStream->GetRtsq()->GetPreStreamSyncStatus();
247 15 : if (isRtsqQueueSpaceSufficient && !isPreStreamSync) {
248 11 : if (masterQueueIter.HasNext()) {
249 21 : HCCL_INFO(
250 : "[ExecuteAllQueues950]InsExecutor::%s master stream InsQueue start %s SqId(%u) stream Id(%u)", __func__,
251 : masterQueueIter->Describe().c_str(), masterStream->GetSqId(), masterStream->GetId());
252 7 : Interpret(*masterQueueIter, *masterStream, resMgrFetcher_);
253 7 : ++masterQueueIter;
254 4 : } else if (!masterQueueIter.HasNext() && !isMasterInsIterEnd) {
255 12 : HCCL_INFO(
256 : "[ExecuteAllQueues950]InsExecutor::%s master stream Id(%u) Interpret insQueue finish", __func__,
257 : masterStream->GetId());
258 4 : isMasterInsIterEnd = true;
259 4 : masterStream->GetRtsq()->LaunchTask();
260 : }
261 11 : isLaunchTask = true;
262 : }
263 15 : isPreStreamSyncExist_ = masterStream->GetRtsq()->GetPreStreamSyncStatus() || isPreStreamSyncExist_;
264 15 : }
265 :
266 17 : void InsExecutor::CheckPreStreamSync(StreamLiteMgr* streamLiteMgr, u32 slaveQueuesSize)
267 : {
268 17 : if (!isPreStreamSyncExist_) {
269 12 : return;
270 : }
271 5 : u32 preStreamSyncValue = 0;
272 15 : for (u32 slaveStreamIndex = 0; slaveStreamIndex < slaveQueuesSize; ++slaveStreamIndex) {
273 10 : StreamLite* slaveStream = streamLiteMgr->GetSlave(slaveStreamIndex);
274 10 : if (slaveStream == nullptr) {
275 0 : THROW<NullPtrException>(StringFormat("InsExecutor::%s slaveStream is null,", __func__));
276 : }
277 10 : if (slaveStream->GetRtsq() == nullptr) {
278 0 : THROW<NullPtrException>(StringFormat(
279 : "InsExecutor::%s GetRtsq returned null for slaveStream Id(%u)", __func__, slaveStream->GetId()));
280 : }
281 10 : if (slaveStream->GetRtsq()->GetPreStreamSyncStatus()) {
282 5 : ++preStreamSyncValue;
283 : }
284 : }
285 5 : StreamLite* masterStream = streamLiteMgr->GetMaster();
286 5 : if (masterStream->GetRtsq()->GetPreStreamSyncStatus()) {
287 5 : ++preStreamSyncValue;
288 : }
289 5 : if (preStreamSyncValue == slaveQueuesSize + 1) {
290 3 : for (u32 slaveStreamIndex = 0; slaveStreamIndex < slaveQueuesSize; ++slaveStreamIndex) {
291 2 : StreamLite* slaveStream = streamLiteMgr->GetSlave(slaveStreamIndex);
292 2 : CHK_RET_THROW(
293 : InternalException, StringFormat("[InsExecutor][%s] SetPreStreamSyncFin", __func__),
294 : slaveStream->GetRtsq()->SetPreStreamSyncFin());
295 : }
296 1 : masterStream->GetRtsq()->SetPreStreamSyncFin();
297 1 : CHK_RET_THROW(
298 : InternalException, StringFormat("[InsExecutor][%s] SetPreStreamSyncFin", __func__),
299 : masterStream->GetRtsq()->SetPreStreamSyncFin());
300 1 : isPreStreamSyncExist_ = false;
301 : }
302 : }
303 :
304 0 : void InsExecutor::ExecuteSingleQue(const InsQueue& insQueue, const StreamLite* streamLite, const bool isMaster)
305 : {
306 0 : sqeMgr->Begin(streamLite->GetSqId());
307 0 : if (isMaster) {
308 0 : HcclNotifyWaitSqe waitSqe;
309 0 : waitSqe.Config(
310 0 : streamLite->GetSqId(), 0, resMgrFetcher_->GetHostDeviceSyncNotifyLiteMgr()->GetDeviceWaitNotify()->GetId());
311 0 : sqeMgr->Add(streamLite->GetSqId(), &waitSqe);
312 0 : }
313 :
314 0 : for (auto iter = insQueue.Iter(); iter.HasNext(); ++iter) {
315 0 : HCCL_INFO("InsExecutor::%s start %s", __func__, iter->Describe().c_str());
316 0 : vector<std::unique_ptr<HcclSqe>> sqeItems = Interpret(*iter, streamLite->GetSqId(), resMgrFetcher_);
317 0 : std::for_each(sqeItems.begin(), sqeItems.end(), [streamLite, this](std::unique_ptr<HcclSqe>& sqeItem) {
318 0 : sqeMgr->Add(streamLite->GetSqId(), sqeItem.get());
319 0 : });
320 0 : }
321 :
322 0 : if (isMaster) {
323 0 : HcclNotifyRecordSqe recordSqe;
324 0 : recordSqe.Config(
325 0 : streamLite->GetSqId(), 0, resMgrFetcher_->GetHostDeviceSyncNotifyLiteMgr()->GetHostWaitNotify()->GetId());
326 0 : sqeMgr->Add(streamLite->GetSqId(), &recordSqe);
327 0 : }
328 :
329 0 : sqeMgr->Commit(streamLite->GetSqId());
330 0 : }
331 :
332 88 : InsExecutor::InsExecutor(ResMgrFetcher* resMgrFetcher) : resMgrFetcher_(resMgrFetcher)
333 : {
334 88 : CHECK_NULLPTR(resMgrFetcher, "[InsExecutor] resMgrFetcher is nullptr!");
335 88 : sqeMgr = make_unique<SqeMgr>(resMgrFetcher->GetDevPhyId());
336 88 : }
337 : } // namespace Hccl
|
