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 "hccl_process.h"
12 : #include <cstdint>
13 : #include "fsm/state_base.h"
14 : #include "common/bqs_log.h"
15 : #include "queue_manager.h"
16 : #include "statistic_manager.h"
17 : #include "entity_manager.h"
18 : #include "router_server.h"
19 : #include "profile_manager.h"
20 :
21 : namespace dgw {
22 : namespace {
23 : // call hccl api max count when processing one event
24 : constexpr uint32_t GET_DATA_THRESHOLD = 100U;
25 : // supply hccl events
26 : const std::vector<uint32_t> g_supplyEvents = {
27 : EVENT_RECV_REQUEST_MSG, EVENT_SEND_COMPLETION_MSG, EVENT_RECV_COMPLETION_MSG};
28 : // link setup timeout gap
29 : constexpr float64_t LINK_SET_UP_TIMEOUE = 60000000.0;
30 : } // namespace
31 :
32 15 : HcclProcess& HcclProcess::GetInstance()
33 : {
34 : static HcclProcess instance;
35 15 : instance.Init();
36 15 : return instance;
37 : }
38 :
39 16 : void HcclProcess::Init()
40 : {
41 16 : if (inited_) {
42 14 : return;
43 : }
44 2 : oneTrackEventEnabled_ = false;
45 2 : inited_ = true;
46 : }
47 :
48 3 : FsmStatus HcclProcess::ProcessRecvRequestEvent(
49 : const event_info& event, const uint32_t deviceId, const uint32_t resIndex)
50 : {
51 3 : if (oneTrackEventEnabled_) {
52 1 : return FsmStatus::FSM_SUCCESS;
53 : }
54 2 : auto ret = FsmStatus::FSM_SUCCESS;
55 2 : auto& recvRequestEventAtomicFlag =
56 : (resIndex == 0U) ? recvRequestEventAtomicFlag_ : recvRequestEventAtomicFlagExtra_;
57 2 : if (!recvRequestEventAtomicFlag.test_and_set()) {
58 2 : DGW_LOG_INFO("Begin to process recv request event.");
59 : // init profiling data
60 2 : const uint64_t eventBegin = bqs::ProfileManager::GetInstance(resIndex).GetCpuTick();
61 2 : const uint64_t schedDelay = static_cast<uint64_t>(event.comm.sched_timestamp - event.comm.submit_timestamp);
62 2 : const uint64_t schedTimes = bqs::StatisticManager::GetInstance().HcclMpiRecvRequestEventStat();
63 2 : bqs::ProfileManager::GetInstance(resIndex).InitMarkerForRecvReqEvent(schedTimes, schedDelay);
64 : // process recv request
65 : const std::function<FsmStatus(const ChannelEntityPtr&, uint32_t&)> probeFunc =
66 6 : [this](const ChannelEntityPtr& entity, uint32_t& probeCount) -> FsmStatus {
67 2 : return ProbeCommChannel(entity, probeCount);
68 2 : };
69 2 : ret = EntityManager::Instance(resIndex).ProbeSrcCommChannel(probeFunc);
70 : // print profiling data
71 2 : bqs::ProfileManager::GetInstance(resIndex).DoMarkerForRecvReqEvent(eventBegin);
72 : // clear event working flag
73 : recvRequestEventAtomicFlag.clear();
74 :
75 : // ack event: if reply action in atomic lock, it maybe cause lost event
76 2 : (void)ReplyHcclEvent(event, deviceId);
77 : // postprocess: check and supply recv request event
78 2 : (void)EntityManager::Instance(resIndex).SupplyRecvRequestEvent();
79 2 : } else {
80 0 : bqs::StatisticManager::GetInstance().HcclMpiRecvReqFalseAwakenStat();
81 0 : ret = FsmStatus::FSM_FAILED;
82 0 : DGW_LOG_INFO("Recv request event is being processed by other thread.");
83 : }
84 2 : return ret;
85 : }
86 :
87 4 : FsmStatus HcclProcess::ProcessSendCompletionEvent(
88 : const event_info& event, const uint32_t deviceId, const uint32_t resIndex)
89 : {
90 4 : if (oneTrackEventEnabled_) {
91 1 : return FsmStatus::FSM_SUCCESS;
92 : }
93 3 : auto ret = FsmStatus::FSM_SUCCESS;
94 3 : auto& sendCompEventAtomicFlag = (resIndex == 0U) ? sendCompEventAtomicFlag_ : sendCompEventAtomicFlagExtra_;
95 3 : if (!sendCompEventAtomicFlag.test_and_set()) {
96 3 : DGW_LOG_INFO("Begin to process send completion event.");
97 : // init profiling data
98 3 : const uint64_t eventBegin = bqs::ProfileManager::GetInstance(resIndex).GetCpuTick();
99 3 : const uint64_t schedDelay = static_cast<uint64_t>(event.comm.sched_timestamp - event.comm.submit_timestamp);
100 3 : const uint64_t schedTimes = bqs::StatisticManager::GetInstance().HcclMpiSendCompEventStat();
101 3 : bqs::ProfileManager::GetInstance(resIndex).InitMarkerForSendCompEvent(schedTimes, schedDelay);
102 : // process send comppletion
103 : const std::function<FsmStatus(CommChannels&, uint32_t&, uint32_t&)> testSomeFunc =
104 9 : [this](CommChannels& channels, uint32_t& totalCompCount, uint32_t& resIndexTmp) -> FsmStatus {
105 3 : return TestSomeCommChannels(channels, false, totalCompCount, resIndexTmp);
106 3 : };
107 3 : ret = EntityManager::Instance(resIndex).TestSomeCommChannels(testSomeFunc, false);
108 : // print profiling data
109 3 : bqs::ProfileManager::GetInstance(resIndex).DoMarkerForSendCompEvent(eventBegin);
110 : // clear event working flag
111 : sendCompEventAtomicFlag.clear();
112 :
113 : // ack event: if reply action in atomic lock, it maybe cause lost event
114 3 : (void)ReplyHcclEvent(event, deviceId);
115 3 : DGW_LOG_INFO("reply event[%u], deviceId[%u] success.", event.comm.event_id, deviceId);
116 3 : } else {
117 0 : bqs::StatisticManager::GetInstance().HcclMpiSendCompFalseAwakenStat();
118 0 : ret = FsmStatus::FSM_FAILED;
119 0 : DGW_LOG_INFO("Send completion event is being processed by other thread.");
120 : }
121 3 : return ret;
122 : }
123 :
124 5 : FsmStatus HcclProcess::ProcessRecvCompletionEvent(
125 : const event_info& event, const uint32_t deviceId, const uint32_t resIndex)
126 : {
127 5 : auto ret = FsmStatus::FSM_SUCCESS;
128 5 : auto& recvCompEventAtomicFlag = (resIndex == 0U) ? recvCompEventAtomicFlag_ : recvCompEventAtomicFlagExtra_;
129 5 : if (!recvCompEventAtomicFlag.test_and_set()) {
130 5 : DGW_LOG_INFO("Begin to process recv completion event.deviceId[%u]", deviceId);
131 : // init profiling data
132 5 : const uint64_t eventBegin = bqs::ProfileManager::GetInstance(resIndex).GetCpuTick();
133 5 : const uint64_t schedDelay = static_cast<uint64_t>(event.comm.sched_timestamp - event.comm.submit_timestamp);
134 5 : const uint64_t schedTimes = bqs::StatisticManager::GetInstance().HcclMpiRecvCompEventStat();
135 5 : bqs::ProfileManager::GetInstance(resIndex).InitMarkerForRecvCompEvent(schedTimes, schedDelay);
136 : // process recv completion event
137 : const std::function<FsmStatus(CommChannels&, uint32_t&, uint32_t&)> testSomeFunc =
138 15 : [this](CommChannels& channels, uint32_t& totalCompCount, uint32_t& resIndexTmp) -> FsmStatus {
139 5 : return TestSomeCommChannels(channels, true, totalCompCount, resIndexTmp);
140 5 : };
141 5 : ret = EntityManager::Instance(resIndex).TestSomeCommChannels(testSomeFunc, true);
142 :
143 5 : if (oneTrackEventEnabled_) {
144 : // process send comppletion
145 : const std::function<FsmStatus(CommChannels&, uint32_t&, uint32_t&)> testSomeSendFunc =
146 9 : [this](CommChannels& channels, uint32_t& totalCompCount, uint32_t& resIndexTmp) -> FsmStatus {
147 3 : return TestSomeCommChannels(channels, false, totalCompCount, resIndexTmp);
148 3 : };
149 3 : (void)EntityManager::Instance(resIndex).TestSomeCommChannels(testSomeSendFunc, false);
150 :
151 : // process recv request
152 : const std::function<FsmStatus(const ChannelEntityPtr&, uint32_t&)> probeFunc =
153 9 : [this](const ChannelEntityPtr& entity, uint32_t& probeCount) -> FsmStatus {
154 3 : return ProbeCommChannel(entity, probeCount);
155 3 : };
156 3 : (void)EntityManager::Instance(resIndex).ProbeSrcCommChannel(probeFunc);
157 3 : (void)EntityManager::Instance(resIndex).SupplyOneTrackEvent();
158 3 : }
159 :
160 : // clear event working flag
161 : recvCompEventAtomicFlag.clear();
162 : // ack event: if reply action in atomic lock, it maybe cause lost event
163 5 : (void)ReplyHcclEvent(event, deviceId);
164 : // print profiling data
165 5 : bqs::ProfileManager::GetInstance(resIndex).DoMarkerForRecvCompEvent(eventBegin);
166 5 : DGW_LOG_INFO("reply event[%u], deviceId[%u] success.", event.comm.event_id, deviceId);
167 5 : } else {
168 0 : bqs::StatisticManager::GetInstance().HcclMpiRecvCompFalseAwakenStat();
169 0 : ret = FsmStatus::FSM_FAILED;
170 0 : DGW_LOG_INFO("Send completion event is being processed by other thread.");
171 : }
172 5 : return ret;
173 : }
174 :
175 1 : FsmStatus HcclProcess::ProcessCongestionReliefEvent(
176 : const event_info& event, const uint32_t deviceId, const uint32_t resIndex) const
177 : {
178 : (void)deviceId;
179 : (void)event;
180 : (void)resIndex;
181 1 : DGW_LOG_ERROR("WARNING! Receive congestion relief event!");
182 1 : bqs::StatisticManager::GetInstance().HcclMpiF2nfEventStat();
183 1 : return FsmStatus::FSM_SUCCESS;
184 : }
185 :
186 11 : FsmStatus HcclProcess::TestSomeCommChannels(
187 : CommChannels& channels, const bool isSrc, uint32_t& totalCompCount, const uint32_t resIndex) const
188 : {
189 11 : auto& entities = channels.entities;
190 11 : auto& requests = channels.requests;
191 : // check request count
192 11 : if (entities.size() > requests.capacity()) {
193 0 : DGW_LOG_ERROR(
194 : "WARNING: Please check requests capacity[%zu] which is less than entities size[%zu].", requests.capacity(),
195 : entities.size());
196 0 : return FsmStatus::FSM_FAILED;
197 : }
198 :
199 11 : auto ret = FsmStatus::FSM_SUCCESS;
200 11 : uint32_t reqCount = 0U;
201 11 : totalCompCount = 0U;
202 24 : while (reqCount < GET_DATA_THRESHOLD) {
203 : // fill requests
204 24 : bool allNullReq = true;
205 24 : size_t index = 0UL;
206 45 : for (auto iter = entities.begin(); iter != entities.end(); ++iter) {
207 21 : const RequestInfo* const hcclReq = (*iter)->FrontUncompReq();
208 21 : if (((*iter)->linkStatus_ == ChannelLinkStatus::ABNORMAL) || (hcclReq == nullptr)) {
209 8 : requests[index++] = HCCL_REQUEST_NULL;
210 : } else {
211 13 : if (!(hcclReq->isLink)) {
212 8 : requests[index++] = hcclReq->req;
213 8 : allNullReq = false;
214 8 : DGW_LOG_DEBUG("Prepare to testsome req of entity[%s].", (*iter)->ToString().c_str());
215 : } else {
216 5 : if (PreProcessSetUplinkReq(hcclReq) == FsmStatus::FSM_SUCCESS) {
217 5 : requests[index++] = hcclReq->req;
218 5 : allNullReq = false;
219 : } else {
220 0 : requests[index++] = HCCL_REQUEST_NULL;
221 0 : (*iter)->linkStatus_ = ChannelLinkStatus::ABNORMAL;
222 0 : DGW_LOG_ERROR("entity[%s] link setup timeout.", (*iter)->ToString().c_str());
223 : }
224 : }
225 : }
226 : }
227 24 : if (allNullReq) {
228 11 : DGW_LOG_DEBUG("Not exist any request which need to be tested.");
229 11 : break;
230 : }
231 : // call HcclTestSome
232 13 : int32_t compCount = 0;
233 13 : auto& compIndices = channels.compIndices;
234 13 : auto& compStatus = channels.compStatus;
235 13 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex).GetCpuTick();
236 26 : const auto hcclRet = HcclTestSome(
237 13 : static_cast<int32_t>(entities.size()), requests.data(), &compCount, compIndices.data(), compStatus.data());
238 26 : bqs::ProfileManager::GetInstance(resIndex).AddHcclTestSomeCost(
239 13 : bqs::ProfileManager::GetInstance(resIndex).GetCpuTick() - begin, isSrc);
240 13 : if (hcclRet == static_cast<int32_t>(HCCL_E_IN_STATUS)) {
241 1 : DGW_LOG_INFO("Test some is unreachable, ret is [%d].", hcclRet);
242 12 : } else if (hcclRet != static_cast<int32_t>(HCCL_SUCCESS)) {
243 0 : DGW_LOG_ERROR("Failed to test some, ret is [%d].", hcclRet);
244 0 : ret = FsmStatus::FSM_FAILED;
245 0 : break;
246 : }
247 13 : reqCount++;
248 :
249 13 : if (compCount == 0) {
250 0 : DGW_LOG_INFO("Not exist test completed request.");
251 0 : break;
252 : }
253 13 : totalCompCount += static_cast<uint32_t>(compCount);
254 : // check result
255 13 : (void)ProcTestSomeResults(compCount, channels, hcclRet);
256 : }
257 11 : if ((!isSrc) && (totalCompCount != 0U)) {
258 3 : (void)EntityManager::Instance(resIndex).SupplyEvent(static_cast<uint32_t>(EVENT_QUEUE_FULL_TO_NOT_FULL));
259 3 : DGW_LOG_INFO("Success to trigger tag f2nf event.");
260 : }
261 11 : DGW_LOG_INFO("Test some comm channels success count is %u.", reqCount);
262 11 : return ret;
263 : }
264 :
265 13 : FsmStatus HcclProcess::ProcTestSomeResults(const int32_t compCount, CommChannels& channels, int32_t hcclRet) const
266 : {
267 26 : for (size_t i = 0UL; i < static_cast<size_t>(compCount); i++) {
268 13 : const size_t reqIdx = static_cast<size_t>(channels.compIndices[i]);
269 13 : HcclStatus& status = channels.compStatus[i];
270 13 : ChannelEntityPtr& entity = channels.entities[reqIdx];
271 : // check status
272 13 : if (status.error != 0) {
273 1 : DGW_LOG_ERROR(
274 : "Comm channel[%s] test some failed, status:[rank:%d, tag:%d, error:%d], hcclRet[%d].",
275 : entity->ToString().c_str(), status.srcRank, status.tag, status.error, hcclRet);
276 1 : if (((status.error == static_cast<int32_t>(HCCL_E_TCP_TRANSFER)) ||
277 1 : (status.error == static_cast<int32_t>(HCCL_E_ROCE_TRANSFER))) &&
278 : (hcclRet == HCCL_E_IN_STATUS)) {
279 1 : entity->linkStatus_ = ChannelLinkStatus::ABNORMAL;
280 1 : DGW_LOG_RUN_INFO("set entity link status is abnormal.");
281 : }
282 1 : continue;
283 : }
284 : // process completed request
285 12 : (void)entity->ProcessCompReq();
286 : }
287 13 : return FsmStatus::FSM_SUCCESS;
288 : }
289 :
290 5 : FsmStatus HcclProcess::ProbeCommChannel(const ChannelEntityPtr& entity, uint32_t& probeCount) const
291 : {
292 5 : if (entity == nullptr) {
293 0 : probeCount = 0U;
294 0 : return FsmStatus::FSM_FAILED;
295 : }
296 5 : HcclMessage msg = nullptr;
297 5 : uint64_t dataCount = 0UL;
298 :
299 5 : uint32_t reqTotalCount = 0U;
300 5 : uint32_t envelopeCacheCount = 0U;
301 5 : uint64_t probeTick = 0UL;
302 17 : while (reqTotalCount < GET_DATA_THRESHOLD) {
303 17 : auto ret = entity->Probe(dataCount, msg, probeTick);
304 : // uncompReqQueue_ full, cache envelope, then continue probe
305 17 : if (ret == FsmStatus::FSM_CACHED) {
306 4 : envelopeCacheCount++;
307 4 : reqTotalCount++;
308 4 : continue;
309 : }
310 : // probe failed, perhaps current channel have no envelope, quit the loop
311 13 : if (ret != FsmStatus::FSM_SUCCESS) {
312 5 : break;
313 : }
314 : // probe success, alloc mbuf and call HcclImrecv to get request
315 8 : reqTotalCount++;
316 8 : ret = entity->ReceiveData(msg, dataCount, probeTick);
317 8 : if (ret != FsmStatus::FSM_SUCCESS) {
318 0 : break;
319 : }
320 : }
321 5 : probeCount = reqTotalCount;
322 5 : DGW_LOG_INFO(
323 : "Probe comm channel success, total count is [%u], envelope cached count is [%u], entity:[%s].", reqTotalCount,
324 : envelopeCacheCount, entity->ToString().c_str());
325 5 : return FsmStatus::FSM_SUCCESS;
326 : }
327 :
328 0 : FsmStatus HcclProcess::SupplyEvents(const uint32_t resIndex) const
329 : {
330 : // supply F2NF event
331 0 : (void)EntityManager::Instance(resIndex).SupplyEvent(static_cast<uint32_t>(EVENT_QUEUE_FULL_TO_NOT_FULL));
332 :
333 0 : if (!bqs::RouterServer::GetInstance().GetCallHcclFlag()) {
334 0 : return FsmStatus::FSM_SUCCESS;
335 : }
336 : // supply hccl event
337 0 : if (oneTrackEventEnabled_) {
338 0 : (void)EntityManager::Instance(resIndex).SupplyEvent(EVENT_RECV_COMPLETION_MSG);
339 : } else {
340 0 : for (const auto eventId : g_supplyEvents) {
341 0 : (void)EntityManager::Instance(resIndex).SupplyEvent(eventId);
342 : };
343 : }
344 0 : DGW_LOG_INFO("Supply event success.");
345 0 : return FsmStatus::FSM_SUCCESS;
346 : }
347 :
348 10 : FsmStatus HcclProcess::ReplyHcclEvent(const event_info& event, const uint32_t deviceId) const
349 : {
350 10 : const uint32_t eventId = static_cast<uint32_t>(event.comm.event_id);
351 : // ack event
352 : const auto drvRet =
353 10 : halEschedAckEvent(deviceId, static_cast<EVENT_ID>(eventId), event.comm.subevent_id, nullptr, 0U);
354 10 : if (drvRet != DRV_ERROR_NONE) {
355 0 : DGW_LOG_ERROR("Failed to reply event[%u], deviceId[%u], ret is %d.", eventId, deviceId, drvRet);
356 0 : return FsmStatus::FSM_FAILED;
357 : }
358 10 : DGW_LOG_INFO("reply event[%u], deviceId[%u] success.", eventId, deviceId);
359 :
360 : // statistic callback count
361 10 : switch (eventId) {
362 2 : case dgw::EVENT_RECV_REQUEST_MSG: {
363 2 : bqs::StatisticManager::GetInstance().HcclMpiRecvReqCallbackStat();
364 2 : break;
365 : }
366 3 : case dgw::EVENT_SEND_COMPLETION_MSG: {
367 3 : bqs::StatisticManager::GetInstance().HcclMpiSendCompCallbackStat();
368 3 : break;
369 : }
370 5 : case dgw::EVENT_RECV_COMPLETION_MSG: {
371 5 : bqs::StatisticManager::GetInstance().HcclMpiRecvCompCallbackStat();
372 5 : break;
373 : }
374 0 : default: {
375 0 : DGW_LOG_ERROR("Unsupported event[%u].", eventId);
376 0 : break;
377 : }
378 : }
379 10 : return FsmStatus::FSM_SUCCESS;
380 : }
381 :
382 7 : FsmStatus HcclProcess::PreProcessSetUplinkReq(const RequestInfo* const hcclReq) const
383 : {
384 7 : const uint64_t curTick = bqs::ProfileManager::GetInstance().GetCpuTick();
385 7 : if (curTick >= hcclReq->startTick) {
386 6 : const auto timeCost = bqs::ProfileManager::GetInstance().GetTimeCost(curTick - hcclReq->startTick);
387 6 : if (timeCost >= LINK_SET_UP_TIMEOUE) {
388 1 : DGW_LOG_ERROR(
389 : "curtick:%lu, setuptick:%lu, threshold:%.2fus, linkSetUp timeout:%.2fus.", curTick, hcclReq->startTick,
390 : LINK_SET_UP_TIMEOUE, timeCost);
391 1 : return FsmStatus::FSM_FAILED;
392 : }
393 5 : return FsmStatus::FSM_SUCCESS;
394 : }
395 1 : DGW_LOG_ERROR("cur tick:%lu is smaller than SetUpTick:%lu.", curTick, hcclReq->startTick);
396 1 : return FsmStatus::FSM_FAILED;
397 : }
398 : } // namespace dgw
|
