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 = {EVENT_RECV_REQUEST_MSG,
27 : EVENT_SEND_COMPLETION_MSG, EVENT_RECV_COMPLETION_MSG};
28 : // link setup timeout gap
29 : constexpr float64_t LINK_SET_UP_TIMEOUE = 60000000.0;
30 : }
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(const event_info &event, const uint32_t deviceId,
49 : 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 = (resIndex == 0U) ?
56 : 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(const event_info &event, const uint32_t deviceId,
88 : 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(const event_info &event, const uint32_t deviceId,
125 : 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(const event_info &event, const uint32_t deviceId,
176 : 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(CommChannels &channels, const bool isSrc, uint32_t &totalCompCount,
187 : 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("WARNING: Please check requests capacity[%zu] which is less than entities size[%zu].",
194 : requests.capacity(), entities.size());
195 0 : return FsmStatus::FSM_FAILED;
196 : }
197 :
198 11 : auto ret = FsmStatus::FSM_SUCCESS;
199 11 : uint32_t reqCount = 0U;
200 11 : totalCompCount = 0U;
201 24 : while (reqCount < GET_DATA_THRESHOLD) {
202 : // fill requests
203 24 : bool allNullReq = true;
204 24 : size_t index = 0UL;
205 45 : for (auto iter = entities.begin(); iter != entities.end(); ++iter) {
206 21 : const RequestInfo * const hcclReq = (*iter)->FrontUncompReq();
207 21 : if (((*iter)->linkStatus_ == ChannelLinkStatus::ABNORMAL) || (hcclReq == nullptr)) {
208 8 : requests[index++] = HCCL_REQUEST_NULL;
209 : } else {
210 13 : if (!(hcclReq->isLink)) {
211 8 : requests[index++] = hcclReq->req;
212 8 : allNullReq = false;
213 8 : DGW_LOG_DEBUG("Prepare to testsome req of entity[%s].", (*iter)->ToString().c_str());
214 : } else {
215 5 : if (PreProcessSetUplinkReq(hcclReq) == FsmStatus::FSM_SUCCESS) {
216 5 : requests[index++] = hcclReq->req;
217 5 : allNullReq = false;
218 : } else {
219 0 : requests[index++] = HCCL_REQUEST_NULL;
220 0 : (*iter)->linkStatus_ = ChannelLinkStatus::ABNORMAL;
221 0 : DGW_LOG_ERROR("entity[%s] link setup timeout.", (*iter)->ToString().c_str());
222 : }
223 : }
224 : }
225 : }
226 24 : if (allNullReq) {
227 11 : DGW_LOG_DEBUG("Not exist any request which need to be tested.");
228 11 : break;
229 : }
230 : // call HcclTestSome
231 13 : int32_t compCount = 0;
232 13 : auto &compIndices = channels.compIndices;
233 13 : auto &compStatus = channels.compStatus;
234 13 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex).GetCpuTick();
235 13 : const auto hcclRet = HcclTestSome(static_cast<int32_t>(entities.size()), requests.data(),
236 : &compCount, compIndices.data(), compStatus.data());
237 26 : bqs::ProfileManager::GetInstance(resIndex).AddHcclTestSomeCost(
238 13 : bqs::ProfileManager::GetInstance(resIndex).GetCpuTick() - begin, isSrc);
239 13 : if (hcclRet == static_cast<int32_t>(HCCL_E_IN_STATUS)) {
240 1 : DGW_LOG_INFO("Test some is unreachable, ret is [%d].", hcclRet);
241 12 : } else if (hcclRet != static_cast<int32_t>(HCCL_SUCCESS)) {
242 0 : DGW_LOG_ERROR("Failed to test some, ret is [%d].", hcclRet);
243 0 : ret = FsmStatus::FSM_FAILED;
244 0 : break;
245 : }
246 13 : reqCount++;
247 :
248 13 : if (compCount == 0) {
249 0 : DGW_LOG_INFO("Not exist test completed request.");
250 0 : break;
251 : }
252 13 : totalCompCount += static_cast<uint32_t>(compCount);
253 : // check result
254 13 : (void)ProcTestSomeResults(compCount, channels, hcclRet);
255 : }
256 11 : if ((!isSrc) && (totalCompCount != 0U)) {
257 3 : (void)EntityManager::Instance(resIndex).SupplyEvent(static_cast<uint32_t>(EVENT_QUEUE_FULL_TO_NOT_FULL));
258 3 : DGW_LOG_INFO("Success to trigger tag f2nf event.");
259 : }
260 11 : DGW_LOG_INFO("Test some comm channels success count is %u.", reqCount);
261 11 : return ret;
262 : }
263 :
264 13 : FsmStatus HcclProcess::ProcTestSomeResults(const int32_t compCount, CommChannels &channels, int32_t hcclRet) const
265 : {
266 26 : for (size_t i = 0UL; i < static_cast<size_t>(compCount); i++) {
267 13 : const size_t reqIdx = static_cast<size_t>(channels.compIndices[i]);
268 13 : HcclStatus &status = channels.compStatus[i];
269 13 : ChannelEntityPtr &entity = channels.entities[reqIdx];
270 : // check status
271 13 : if (status.error != 0) {
272 1 : DGW_LOG_ERROR("Comm channel[%s] test some failed, status:[rank:%d, tag:%d, error:%d], hcclRet[%d].",
273 : entity->ToString().c_str(), status.srcRank, status.tag, status.error, hcclRet);
274 1 : if (((status.error == static_cast<int32_t>(HCCL_E_TCP_TRANSFER)) ||
275 1 : (status.error == static_cast<int32_t>(HCCL_E_ROCE_TRANSFER))) && (hcclRet == HCCL_E_IN_STATUS)) {
276 1 : entity->linkStatus_ = ChannelLinkStatus::ABNORMAL;
277 1 : DGW_LOG_RUN_INFO("set entity link status is abnormal.");
278 : }
279 1 : continue;
280 : }
281 : // process completed request
282 12 : (void)entity->ProcessCompReq();
283 : }
284 13 : return FsmStatus::FSM_SUCCESS;
285 : }
286 :
287 5 : FsmStatus HcclProcess::ProbeCommChannel(const ChannelEntityPtr &entity, uint32_t &probeCount) const
288 : {
289 5 : if (entity == nullptr) {
290 0 : probeCount = 0U;
291 0 : return FsmStatus::FSM_FAILED;
292 : }
293 5 : HcclMessage msg = nullptr;
294 5 : uint64_t dataCount = 0UL;
295 :
296 5 : uint32_t reqTotalCount = 0U;
297 5 : uint32_t envelopeCacheCount = 0U;
298 5 : uint64_t probeTick = 0UL;
299 17 : while (reqTotalCount < GET_DATA_THRESHOLD) {
300 17 : auto ret = entity->Probe(dataCount, msg, probeTick);
301 : // uncompReqQueue_ full, cache envelope, then continue probe
302 17 : if (ret == FsmStatus::FSM_CACHED) {
303 4 : envelopeCacheCount++;
304 4 : reqTotalCount++;
305 4 : continue;
306 : }
307 : // probe failed, perhaps current channel have no envelope, quit the loop
308 13 : if (ret != FsmStatus::FSM_SUCCESS) {
309 5 : break;
310 : }
311 : // probe success, alloc mbuf and call HcclImrecv to get request
312 8 : reqTotalCount++;
313 8 : ret = entity->ReceiveData(msg, dataCount, probeTick);
314 8 : if (ret != FsmStatus::FSM_SUCCESS) {
315 0 : break;
316 : }
317 : }
318 5 : probeCount = reqTotalCount;
319 5 : DGW_LOG_INFO("Probe comm channel success, total count is [%u], envelope cached count is [%u], entity:[%s].",
320 : reqTotalCount, envelopeCacheCount, entity->ToString().c_str());
321 5 : return FsmStatus::FSM_SUCCESS;
322 : }
323 :
324 0 : FsmStatus HcclProcess::SupplyEvents(const uint32_t resIndex) const
325 : {
326 : // supply F2NF event
327 0 : (void)EntityManager::Instance(resIndex).SupplyEvent(static_cast<uint32_t>(EVENT_QUEUE_FULL_TO_NOT_FULL));
328 :
329 0 : if (!bqs::RouterServer::GetInstance().GetCallHcclFlag()) {
330 0 : return FsmStatus::FSM_SUCCESS;
331 : }
332 : // supply hccl event
333 0 : if (oneTrackEventEnabled_) {
334 0 : (void)EntityManager::Instance(resIndex).SupplyEvent(EVENT_RECV_COMPLETION_MSG);
335 : } else {
336 0 : for (const auto eventId : g_supplyEvents) {
337 0 : (void)EntityManager::Instance(resIndex).SupplyEvent(eventId);
338 : };
339 : }
340 0 : DGW_LOG_INFO("Supply event success.");
341 0 : return FsmStatus::FSM_SUCCESS;
342 : }
343 :
344 10 : FsmStatus HcclProcess::ReplyHcclEvent(const event_info &event, const uint32_t deviceId) const
345 : {
346 10 : const uint32_t eventId = static_cast<uint32_t>(event.comm.event_id);
347 : // ack event
348 20 : const auto drvRet = halEschedAckEvent(deviceId,
349 10 : static_cast<EVENT_ID>(eventId), event.comm.subevent_id, nullptr, 0U);
350 10 : if (drvRet != DRV_ERROR_NONE) {
351 0 : DGW_LOG_ERROR("Failed to reply event[%u], deviceId[%u], ret is %d.", eventId, deviceId, drvRet);
352 0 : return FsmStatus::FSM_FAILED;
353 : }
354 10 : DGW_LOG_INFO("reply event[%u], deviceId[%u] success.", eventId, deviceId);
355 :
356 : // statistic callback count
357 10 : switch (eventId) {
358 2 : case dgw::EVENT_RECV_REQUEST_MSG: {
359 2 : bqs::StatisticManager::GetInstance().HcclMpiRecvReqCallbackStat();
360 2 : break;
361 : }
362 3 : case dgw::EVENT_SEND_COMPLETION_MSG: {
363 3 : bqs::StatisticManager::GetInstance().HcclMpiSendCompCallbackStat();
364 3 : break;
365 : }
366 5 : case dgw::EVENT_RECV_COMPLETION_MSG: {
367 5 : bqs::StatisticManager::GetInstance().HcclMpiRecvCompCallbackStat();
368 5 : break;
369 : }
370 0 : default: {
371 0 : DGW_LOG_ERROR("Unsupported event[%u].", eventId);
372 0 : break;
373 : }
374 : }
375 10 : return FsmStatus::FSM_SUCCESS;
376 : }
377 :
378 7 : FsmStatus HcclProcess::PreProcessSetUplinkReq(const RequestInfo * const hcclReq) const
379 : {
380 7 : const uint64_t curTick = bqs::ProfileManager::GetInstance().GetCpuTick();
381 7 : if (curTick >= hcclReq->startTick) {
382 6 : const auto timeCost = bqs::ProfileManager::GetInstance().GetTimeCost(curTick - hcclReq->startTick);
383 6 : if (timeCost >= LINK_SET_UP_TIMEOUE) {
384 1 : DGW_LOG_ERROR("curtick:%lu, setuptick:%lu, threshold:%.2fus, linkSetUp timeout:%.2fus.", curTick,
385 : hcclReq->startTick, LINK_SET_UP_TIMEOUE, timeCost);
386 1 : return FsmStatus::FSM_FAILED;
387 : }
388 5 : return FsmStatus::FSM_SUCCESS;
389 : }
390 1 : DGW_LOG_ERROR("cur tick:%lu is smaller than SetUpTick:%lu.", curTick, hcclReq->startTick);
391 1 : return FsmStatus::FSM_FAILED;
392 : }
393 : } // namespace dgw
|
