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