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 "channel_entity.h"
12 : #include <algorithm>
13 : #include "bqs_status.h"
14 : #include "bqs_util.h"
15 : #include "msprof_manager.h"
16 : #include "profile_manager.h"
17 : #include "queue_manager.h"
18 : #include "schedule_config.h"
19 :
20 : namespace dgw {
21 :
22 : namespace {
23 : // probe comm channel failed
24 : constexpr int32_t PROBE_COMM_CHANNEL_FAILED = 0;
25 : // comm channel queue name prefix
26 : constexpr const char_t* COMM_CHANNEL_QUEUE_NAME_PREFIX = "CommChannelQueue_";
27 : // request process completed time cost threshold (us) maybe 500000us
28 : constexpr float64_t REQ_COMP_TIME_COST_THRESHOLD = 500000.0;
29 : // envelope processed time cost threshold (us)
30 : constexpr float64_t ENVELOPE_PROC_TIME_COST_THRESHOLD = 500000.0;
31 : // count threshold for print error (first improbe and testsome cost too long time, no need check)
32 : const uint64_t COUNT_THRESHOLD_FOR_PRINT_ERROR = 10UL;
33 : const uint32_t CHECK_SEND_COMPLETION_INTERVAL_US = 100U;
34 : const uint32_t CHECK_SEND_COMPLETION_LIMIT_US = 100000U; // 100ms
35 : } // namespace
36 :
37 56 : ChannelEntity::ChannelEntity(const EntityMaterial& material, const uint32_t resIndex)
38 : : SimpleEntity(material, resIndex),
39 56 : linkStatus_(ChannelLinkStatus::UNCONNECTED),
40 56 : channelPtr_(material.channel),
41 56 : compReqQueueId_(0U),
42 56 : cachedReqCount_(0U),
43 56 : maxCachedReqCount_(0U),
44 56 : mbufDataSend_(false),
45 56 : compReqCount_(0UL),
46 56 : procEnvelopeCount_(0UL)
47 : {
48 56 : if (channelPtr_ != nullptr) {
49 50 : (void)entityDesc_.append(", ").append(channelPtr_->ToString());
50 : }
51 56 : }
52 :
53 56 : ChannelEntity::~ChannelEntity() { DGW_LOG_RUN_INFO("Success to destruct tag entity[%s].", entityDesc_.c_str()); }
54 :
55 37 : FsmStatus ChannelEntity::Init(const FsmState state, const EntityDirection direction)
56 : {
57 37 : if (channelPtr_ == nullptr) {
58 1 : DGW_LOG_ERROR("channelPtr_ is nullptr in comm channel entity[%s].", entityDesc_.c_str());
59 1 : return FsmStatus::FSM_FAILED;
60 : }
61 :
62 36 : (void)SimpleEntity::Init(state, direction);
63 :
64 : // calculate maxCachedReqCount_
65 36 : maxCachedReqCount_ = channelPtr_->GetLocalTagDepth() * 2U;
66 : // init uncompleted request queue
67 36 : const uint32_t uncompQueDepth = channelPtr_->GetLocalTagDepth() * 2U + 1U;
68 36 : auto ret = uncompReqQueue_.Init(uncompQueDepth);
69 36 : if (ret != FsmStatus::FSM_SUCCESS) {
70 2 : return ret;
71 : }
72 :
73 : // only src tag need envelope chached queue and completed request queue
74 : // dst tag need try to establish a link with peer tag
75 34 : if (direction == EntityDirection::DIRECTION_RECV) {
76 25 : ret = SendDataForLink();
77 : } else {
78 : // init envelope cached queue
79 9 : const uint32_t cacheQueDepth = channelPtr_->GetPeerTagDepth() * 2U + 1U;
80 9 : ret = cachedEnvelopeQueue_.Init(cacheQueDepth);
81 9 : if (ret != FsmStatus::FSM_SUCCESS) {
82 1 : return ret;
83 : }
84 :
85 8 : ret = CreateAndSubscribeCompletedQueue();
86 8 : if (ret != FsmStatus::FSM_SUCCESS) {
87 1 : return ret;
88 : }
89 7 : (void)entityDesc_.append(", compReqQueue:").append(std::to_string(compReqQueueId_));
90 : }
91 32 : if (ret != FsmStatus::FSM_SUCCESS) {
92 1 : return ret;
93 : }
94 :
95 31 : linkStatus_ = dgw::ChannelLinkStatus::UNCONNECTED;
96 : // add unlink tag count
97 31 : const uint32_t unlinkTagCount = bqs::StatisticManager::GetInstance().AddUnlinkCount();
98 31 : bqs::StatisticManager::GetInstance().AddTagCount();
99 31 : DGW_LOG_RUN_INFO(
100 : "Success to init entity:[%s], current unlink tag count is [%u].", entityDesc_.c_str(), unlinkTagCount);
101 31 : return FsmStatus::FSM_SUCCESS;
102 : }
103 :
104 10 : FsmStatus ChannelEntity::CreateAndSubscribeCompletedQueue()
105 : {
106 : // create and subscribe completed request queue
107 10 : std::string queueName(COMM_CHANNEL_QUEUE_NAME_PREFIX);
108 10 : (void)queueName.append(std::to_string(id_)).append("_");
109 10 : const uint32_t compQueDepth = channelPtr_->GetLocalTagDepth() + 1U;
110 : auto bqsRet =
111 10 : bqs::QueueManager::GetInstance().CreateQueue(queueName.c_str(), compQueDepth, compReqQueueId_, deviceId_);
112 10 : if (bqsRet != bqs::BqsStatus::BQS_STATUS_OK) {
113 1 : DGW_LOG_ERROR(
114 : "Create completed queue failed, queueName[%s], ret[%d].", queueName.c_str(), static_cast<int32_t>(bqsRet));
115 1 : return FsmStatus::FSM_FAILED;
116 : }
117 9 : const auto subscriber = GetSubscriber();
118 9 : if (subscriber == nullptr) {
119 1 : return FsmStatus::FSM_FAILED;
120 : }
121 8 : bqsRet = subscriber->Subscribe(compReqQueueId_);
122 8 : if (bqsRet != bqs::BqsStatus::BQS_STATUS_OK) {
123 1 : DGW_LOG_ERROR(
124 : "Subscribe completed queue failed, queueName[%s], queueId[%u], ret[%d].", queueName.c_str(),
125 : compReqQueueId_, static_cast<int32_t>(bqsRet));
126 1 : return FsmStatus::FSM_FAILED;
127 : }
128 7 : return FsmStatus::FSM_SUCCESS;
129 10 : }
130 :
131 38 : FsmStatus ChannelEntity::Uninit()
132 : {
133 : // clear mbuf
134 74 : while (!uncompReqQueue_.IsEmpty()) {
135 37 : RequestInfo* const uncompReq = uncompReqQueue_.Front();
136 37 : if (uncompReq == nullptr) {
137 1 : DGW_LOG_ERROR("Failed to get front from uncompleted req queue, entity:[%s].", entityDesc_.c_str());
138 1 : break;
139 : }
140 36 : const auto mbuf = uncompReq->mbuf;
141 36 : if (mbuf != nullptr) {
142 5 : (void)halMbufFree(mbuf);
143 5 : if (direction_ == EntityDirection::DIRECTION_RECV) {
144 4 : statInfo_.freeMbufTimes++;
145 : }
146 5 : DGW_LOG_RUN_INFO("Success to free mbuf for entity[%s] when uninit entity.", entityDesc_.c_str());
147 : }
148 36 : if (uncompReqQueue_.Pop() == 0) {
149 1 : DGW_LOG_ERROR("Failed to pop from uncompleted req queue, entity:[%s].", entityDesc_.c_str());
150 : } else {
151 35 : statInfo_.uncompReqQueuePopTimes++;
152 35 : DGW_LOG_RUN_INFO("Success to pop from uncompleted req queue when uninit entity:[%s].", entityDesc_.c_str());
153 : }
154 : }
155 :
156 38 : uncompReqQueue_.Uninit();
157 38 : if (direction_ == EntityDirection::DIRECTION_SEND) {
158 13 : cachedEnvelopeQueue_.Uninit();
159 13 : const auto subscriber = GetSubscriber();
160 13 : if (subscriber == nullptr) {
161 6 : return FsmStatus::FSM_FAILED;
162 : }
163 7 : subscriber->Unsubscribe(compReqQueueId_);
164 7 : (void)bqs::QueueManager::GetInstance().DestroyQueue(compReqQueueId_, deviceId_);
165 : }
166 32 : bqs::StatisticManager::GetInstance().ReduceTagCount();
167 32 : if (hostGroupId_ == INVALID_GROUP_ID) {
168 29 : (void)CommChannelManager::GetInstance().DeleteCommChannel(*channelPtr_);
169 : }
170 32 : Dump();
171 32 : return FsmStatus::FSM_SUCCESS;
172 : }
173 :
174 18 : FsmStatus ChannelEntity::Probe(uint64_t& dataCount, HcclMessage& msg, uint64_t& probeTick)
175 : {
176 18 : bool cachedEnvelopeQueEmpty = true;
177 : // check cached envelope queue empty
178 18 : if (!cachedEnvelopeQueue_.IsEmpty()) {
179 : // no need check uncompReqQue full
180 9 : if (AddCachedReqCount()) {
181 4 : const auto info = cachedEnvelopeQueue_.Front();
182 4 : msg = info->msg;
183 4 : dataCount = info->dataSize;
184 4 : probeTick = info->probeTick;
185 4 : (void)cachedEnvelopeQueue_.Pop();
186 4 : DGW_LOG_INFO(
187 : "Get cached envelope for comm channel[%s], rest envelope size is [%u].", entityDesc_.c_str(),
188 : cachedEnvelopeQueue_.Size());
189 4 : return FsmStatus::FSM_SUCCESS;
190 : }
191 5 : if (cachedEnvelopeQueue_.IsFull()) {
192 3 : DGW_LOG_INFO(
193 : "Cached req count of comm channel[%s] is up to [%u] and cachedEnvelopeQueue is up to [%u],"
194 : "then skip probe.",
195 : entityDesc_.c_str(), maxCachedReqCount_, cachedEnvelopeQueue_.Size());
196 3 : return FsmStatus::FSM_FAILED;
197 : }
198 2 : cachedEnvelopeQueEmpty = false;
199 2 : DGW_LOG_INFO(
200 : "Cached req count of comm channel[%s] is up to [%u], try to probe channel, then cache envelope.",
201 : entityDesc_.c_str(), maxCachedReqCount_);
202 : }
203 :
204 11 : uint64_t probeSuccTick = 0U;
205 11 : const auto probeRet = DoProbe(dataCount, msg, probeSuccTick);
206 11 : if (probeRet != FsmStatus::FSM_SUCCESS) {
207 2 : return probeRet;
208 : }
209 :
210 : // cachedEnvelopeQueue_ not empty: cache envelope
211 : // cachedEnvelopeQueue_ empty: if cached req count up to max, cache envelope
212 9 : if ((!cachedEnvelopeQueEmpty) || (!AddCachedReqCount())) {
213 5 : EnvelopeInfo info = {.msg = msg, .dataSize = dataCount, .probeTick = probeSuccTick};
214 5 : if (cachedEnvelopeQueue_.Push(info) != 1) {
215 1 : DGW_LOG_ERROR(
216 : "Unhandle error! cached req count of channel[%s] is up to max[%u], but cache envelope failed!"
217 : " Current cache envelope count is [%u].",
218 : entityDesc_.c_str(), maxCachedReqCount_, cachedEnvelopeQueue_.Size());
219 1 : return FsmStatus::FSM_FAILED;
220 : }
221 4 : DGW_LOG_RUN_INFO(
222 : "Cached req count of channel[%s] is up to max[%u], cache envelope info, current count is [%u].",
223 : entityDesc_.c_str(), maxCachedReqCount_, cachedEnvelopeQueue_.Size());
224 4 : return FsmStatus::FSM_CACHED;
225 : }
226 4 : probeTick = probeSuccTick;
227 4 : return FsmStatus::FSM_SUCCESS;
228 : }
229 :
230 13 : FsmStatus ChannelEntity::DoProbe(uint64_t& dataCount, HcclMessage& msg, uint64_t& probeSuccTick)
231 : {
232 : // probe src tag
233 13 : DGW_LOG_DEBUG("Begin to probe comm channel[%s].", entityDesc_.c_str());
234 13 : HcclStatus status = {};
235 13 : int32_t probeFlag = PROBE_COMM_CHANNEL_FAILED;
236 13 : const uint64_t probeBegin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
237 13 : auto hcclRet = HcclImprobe(
238 13 : static_cast<int32_t>(channelPtr_->GetPeerRankId()), static_cast<int32_t>(channelPtr_->GetPeerTagId()),
239 13 : channelPtr_->GetHandle(), &probeFlag, &msg, &status);
240 26 : bqs::ProfileManager::GetInstance(resIndex_).AddHcclImprobeCost(
241 13 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - probeBegin);
242 13 : statInfo_.hcclImprobeTotalTimes++;
243 13 : if (hcclRet != static_cast<int32_t>(HCCL_SUCCESS)) {
244 1 : statInfo_.hcclImprobeFailTimes++;
245 1 : DGW_LOG_ERROR("Failed to probe comm channel[%s], ret is [%d].", entityDesc_.c_str(), hcclRet);
246 1 : return FsmStatus::FSM_FAILED;
247 : }
248 12 : if (probeFlag == PROBE_COMM_CHANNEL_FAILED) {
249 2 : DGW_LOG_DEBUG("No data in comm channel[%s], flag is [%d].", entityDesc_.c_str(), probeFlag);
250 2 : return FsmStatus::FSM_FAILED;
251 : }
252 10 : probeSuccTick = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
253 10 : DGW_LOG_DEBUG("Success to probe comm channel[%s].", entityDesc_.c_str());
254 :
255 : // get count
256 10 : int32_t count = 0;
257 10 : const uint64_t getCountBegin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
258 10 : hcclRet = HcclGetCount(&status, HCCL_DATA_TYPE_INT8, &count);
259 20 : bqs::ProfileManager::GetInstance(resIndex_).AddHcclGetCountCost(
260 10 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - getCountBegin);
261 10 : if (hcclRet != static_cast<int32_t>(HCCL_SUCCESS)) {
262 1 : DGW_LOG_ERROR("Failed to get count from comm channel[%s], ret is [%d].", entityDesc_.c_str(), hcclRet);
263 1 : return FsmStatus::FSM_FAILED;
264 : }
265 9 : dataCount = static_cast<uint64_t>(count);
266 :
267 : // check link message
268 9 : if (dataCount == 0UL) {
269 3 : DGW_LOG_RUN_INFO("Success to get link message from comm channel[%s].", entityDesc_.c_str());
270 : } else {
271 6 : statInfo_.hcclImprobeSuccTimes++;
272 6 : DGW_LOG_DEBUG("Success to get data count[%lu] from comm channel[%s].", dataCount, entityDesc_.c_str());
273 : }
274 9 : return FsmStatus::FSM_SUCCESS;
275 : }
276 :
277 13 : FsmStatus ChannelEntity::AllocMbuf(Mbuf*& mbufPtr, void*& headBuf, void*& dataBuf, const uint64_t dataLen)
278 : {
279 13 : bqs::ProfInfo reportData = {};
280 13 : if (bqs::BqsMsprofManager::GetInstance().IsStartProfling()) {
281 1 : reportData.type = static_cast<uint32_t>(bqs::DgwProfInfoType::ALLOC_MBUF);
282 1 : reportData.itemId = transId_;
283 1 : reportData.timeStamp = bqs::GetTimeStamp();
284 : }
285 :
286 13 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
287 13 : int32_t ret = halMbufAlloc(dataLen, &mbufPtr);
288 26 : bqs::ProfileManager::GetInstance(resIndex_).AddMbufAllocCost(
289 13 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - begin);
290 13 : bqs::BqsMsprofManager::GetInstance().ReportApiPerf(reportData);
291 13 : if (ret != static_cast<int32_t>(DRV_ERROR_NONE)) {
292 1 : DGW_LOG_ERROR("Failed to call halMbufAlloc, dataLen:[%lu], ret=[%d].", dataLen, ret);
293 1 : return FsmStatus::FSM_FAILED;
294 : }
295 12 : bqs::StatisticManager::GetInstance().MbufAllocStat(dataLen);
296 :
297 12 : ret = halMbufSetDataLen(mbufPtr, dataLen);
298 12 : if (ret != static_cast<int32_t>(DRV_ERROR_NONE)) {
299 1 : DGW_LOG_ERROR("Failed to call halMbufSetDataLen, ret=[%d].", ret);
300 1 : (void)halMbufFree(mbufPtr);
301 1 : return FsmStatus::FSM_FAILED;
302 : }
303 :
304 11 : uint32_t headerSize = 0U;
305 11 : ret = halMbufGetPrivInfo(mbufPtr, &headBuf, &headerSize);
306 11 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) || (headBuf == nullptr)) {
307 2 : DGW_LOG_ERROR("Failed to call halMbufGetPrivInfo, ret=[%d].", ret);
308 2 : (void)halMbufFree(mbufPtr);
309 2 : return FsmStatus::FSM_FAILED;
310 : }
311 9 : hcclData_.mbufHeadSize = static_cast<uint64_t>(headerSize);
312 :
313 9 : ret = halMbufGetBuffAddr(mbufPtr, &dataBuf);
314 9 : if ((ret != static_cast<int32_t>(DRV_ERROR_NONE)) || (dataBuf == nullptr)) {
315 1 : DGW_LOG_ERROR("Failed to call halMbufGetBuffAddr, ret=[%d].", ret);
316 1 : (void)halMbufFree(mbufPtr);
317 1 : return FsmStatus::FSM_FAILED;
318 : }
319 8 : DGW_LOG_DEBUG("Success to alloc mbuf, dataLen:[%lu].", dataLen);
320 8 : return FsmStatus::FSM_SUCCESS;
321 : }
322 :
323 9 : FsmStatus ChannelEntity::ReceiveData(HcclMessage& msg, const uint64_t dataCount, const uint64_t probeTick)
324 : {
325 : // process link message
326 9 : if (dataCount == 0UL) {
327 2 : return ReceiveDataForLink(msg);
328 : }
329 :
330 7 : procEnvelopeCount_++;
331 14 : const auto timeCost = bqs::ProfileManager::GetInstance(resIndex_).GetTimeCost(
332 7 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - probeTick);
333 7 : if ((timeCost > ENVELOPE_PROC_TIME_COST_THRESHOLD) && (procEnvelopeCount_ > COUNT_THRESHOLD_FOR_PRINT_ERROR)) {
334 1 : DGW_LOG_RUN_INFO(
335 : "Time cost to process envelope is %.2fus, count:[%lu], entity:[%s].", timeCost, procEnvelopeCount_,
336 : entityDesc_.c_str());
337 : }
338 :
339 7 : bool isMbufData = true;
340 : {
341 : // no need lock, no parallel scenarios
342 7 : if (hcclData_.dataSize == 0UL) {
343 5 : hcclData_.dataSize = dataCount;
344 5 : isMbufData = true;
345 : } else {
346 2 : hcclData_.headSize = dataCount;
347 2 : isMbufData = false;
348 : }
349 : }
350 :
351 7 : if (isMbufData) {
352 5 : return ReceiveMbufData(msg);
353 : }
354 2 : return ReceiveMbufHead(msg);
355 : }
356 :
357 6 : FsmStatus ChannelEntity::ReceiveDataForLink(HcclMessage& msg)
358 : {
359 : HcclRequest request;
360 6 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
361 6 : const auto hcclRet = HcclImrecv(nullptr, 0, HCCL_DATA_TYPE_INT8, &msg, &request);
362 12 : bqs::ProfileManager::GetInstance(resIndex_).AddHcclImrecvCost(
363 6 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - begin);
364 6 : if (hcclRet != HCCL_SUCCESS) {
365 : // unable to handle irecv error
366 1 : bqs::StatisticManager::GetInstance().HcclMpiRecvFailStat();
367 1 : DGW_LOG_ERROR(
368 : "Fail to call HcclImrecv to recv link zero data, entity:[%s], ret:[%d].", entityDesc_.c_str(), hcclRet);
369 1 : return FsmStatus::FSM_FAILED;
370 : }
371 5 : bqs::StatisticManager::GetInstance().HcclMpiRecvSuccStat();
372 :
373 : // save request, unable to handle enqueue failure
374 5 : RequestInfo req = {
375 : .req = request,
376 : .isLink = true,
377 : .mbuf = nullptr,
378 5 : .startTick = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick()};
379 5 : const int32_t count = uncompReqQueue_.Push(req);
380 5 : if (count == 0) {
381 1 : DGW_LOG_ERROR(
382 : "Unhandled error! Failed to enqueue uncompleted request for link establishment, entity[%s].",
383 : entityDesc_.c_str());
384 1 : return FsmStatus::FSM_FAILED;
385 : }
386 4 : DGW_LOG_RUN_INFO("Success to receive zero data for link establishment, entity:[%s].", entityDesc_.c_str());
387 4 : return FsmStatus::FSM_SUCCESS;
388 : }
389 :
390 10 : FsmStatus ChannelEntity::ReceiveMbufData(HcclMessage& msg)
391 : {
392 10 : Mbuf* mbuf = nullptr;
393 10 : void* headBuf = nullptr;
394 10 : void* dataBuf = nullptr;
395 10 : const uint64_t dataSize = hcclData_.dataSize;
396 10 : const auto ret = AllocMbuf(mbuf, headBuf, dataBuf, dataSize);
397 10 : if (ret != FsmStatus::FSM_SUCCESS) {
398 1 : return ret;
399 : }
400 9 : statInfo_.allocMbufTimes++;
401 : // record mbuf and headBuf
402 9 : hcclData_.mbuf = mbuf;
403 9 : hcclData_.headBuf = headBuf;
404 :
405 : // call hccl irecv api
406 : HcclRequest request;
407 9 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
408 9 : const auto hcclRet = HcclImrecv(dataBuf, static_cast<int32_t>(dataSize), HCCL_DATA_TYPE_INT8, &msg, &request);
409 18 : bqs::ProfileManager::GetInstance(resIndex_).AddHcclImrecvCost(
410 9 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - begin);
411 9 : if (hcclRet != static_cast<int32_t>(HCCL_SUCCESS)) {
412 1 : DGW_LOG_ERROR("HcclImrecv fail for entity:[%s], ret:[%d].", entityDesc_.c_str(), hcclRet);
413 1 : statInfo_.hcclImrecvFailTimes++;
414 : // unable to handle irecv error
415 1 : bqs::StatisticManager::GetInstance().HcclMpiRecvFailStat();
416 1 : return FsmStatus::FSM_FAILED;
417 : }
418 8 : statInfo_.hcclImrecvSuccTimes++;
419 8 : bqs::StatisticManager::GetInstance().HcclMpiRecvSuccStat();
420 8 : DGW_LOG_INFO(
421 : "Success to call HcclImrecv to recv data, data size:[%lu], "
422 : "entity:[%s]",
423 : dataSize, entityDesc_.c_str());
424 :
425 : // save request, unable to handle enqueue failure
426 8 : RequestInfo req = {
427 : .req = request,
428 : .isLink = false,
429 : .mbuf = nullptr,
430 8 : .startTick = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick()};
431 8 : const int32_t count = uncompReqQueue_.Push(req);
432 8 : if (count == 0) {
433 1 : DGW_LOG_ERROR("Unhandled error! Failed to enqueue uncompleted request for entity[%s].", entityDesc_.c_str());
434 1 : return FsmStatus::FSM_FAILED;
435 : }
436 7 : statInfo_.uncompReqQueuePushTimes++;
437 7 : DGW_LOG_INFO("Success to enqueue uncompleted request and mbuf for entity[%s]", entityDesc_.c_str());
438 7 : return FsmStatus::FSM_SUCCESS;
439 : }
440 :
441 5 : FsmStatus ChannelEntity::ReceiveMbufHead(HcclMessage& msg)
442 : {
443 : // call hccl irecv api
444 : HcclRequest request;
445 5 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
446 10 : const auto hcclRet = HcclImrecv(
447 5 : hcclData_.headBuf, static_cast<int32_t>(hcclData_.mbufHeadSize), HCCL_DATA_TYPE_INT8, &msg, &request);
448 10 : bqs::ProfileManager::GetInstance(resIndex_).AddHcclImrecvCost(
449 5 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - begin);
450 5 : Mbuf* const mbuf = hcclData_.mbuf;
451 5 : if (hcclRet != static_cast<int32_t>(HCCL_SUCCESS)) {
452 1 : DGW_LOG_ERROR("HcclImrecv fail for entity:[%s], ret:[%d].", entityDesc_.c_str(), hcclRet);
453 1 : statInfo_.hcclImrecvFailTimes++;
454 : // unable to handle irecv error
455 1 : if (mbuf != nullptr) {
456 1 : DGW_LOG_INFO("Free Mbuf for entity[%s].", entityDesc_.c_str());
457 1 : (void)halMbufFree(mbuf);
458 : }
459 1 : bqs::StatisticManager::GetInstance().HcclMpiRecvFailStat();
460 1 : return FsmStatus::FSM_FAILED;
461 : }
462 4 : statInfo_.hcclImrecvSuccTimes++;
463 4 : bqs::StatisticManager::GetInstance().HcclMpiRecvSuccStat();
464 :
465 : // clear hcclData
466 4 : hcclData_.headSize = 0UL;
467 4 : hcclData_.dataSize = 0UL;
468 4 : hcclData_.mbuf = nullptr;
469 4 : hcclData_.headBuf = nullptr;
470 4 : hcclData_.mbufHeadSize = 0UL;
471 : // save request, unable to handle enqueue failure
472 4 : RequestInfo req = {
473 : .req = request,
474 : .isLink = false,
475 : .mbuf = mbuf,
476 4 : .startTick = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick()};
477 4 : const int32_t count = uncompReqQueue_.Push(req);
478 4 : if (count == 0) {
479 1 : return FsmStatus::FSM_FAILED;
480 : }
481 3 : statInfo_.uncompReqQueuePushTimes++;
482 3 : DGW_LOG_INFO("Success to enqueue uncompleted request and mbuf for entity[%s].", entityDesc_.c_str());
483 3 : return FsmStatus::FSM_SUCCESS;
484 : }
485 :
486 10 : FsmStatus ChannelEntity::DoSendData(Mbuf* const mbuf)
487 : {
488 10 : if (linkStatus_ == ChannelLinkStatus::ABNORMAL) {
489 1 : DGW_LOG_ERROR("channel is abnormal send data failed.");
490 1 : return FsmStatus::FSM_ERROR_PENDING;
491 : }
492 9 : bqs::ProfInfo reportData = {};
493 9 : if (bqs::BqsMsprofManager::GetInstance().IsStartProfling()) {
494 2 : reportData.type = static_cast<uint32_t>(bqs::DgwProfInfoType::HCCL_TRANS_DATA);
495 2 : reportData.itemId = transId_;
496 2 : reportData.timeStamp = bqs::GetTimeStamp();
497 : }
498 18 : bqs::ScopeGuard profGuard([&reportData]() { bqs::BqsMsprofManager::GetInstance().ReportApiPerf(reportData); });
499 : // After recovery, if the data filed of mbuf has been sent, it will not be sent again
500 : // first, send data field of mbuf; then, send head field of mbuf
501 9 : if (!mbufDataSend_) {
502 8 : const FsmStatus sendDataRet = SendMbufData(mbuf);
503 8 : if (sendDataRet != FsmStatus::FSM_SUCCESS) {
504 3 : return sendDataRet;
505 : }
506 5 : mbufDataSend_ = true;
507 : }
508 :
509 6 : const FsmStatus sendHeadRet = SendMbufHead(mbuf);
510 6 : if (sendHeadRet != FsmStatus::FSM_SUCCESS) {
511 1 : return sendHeadRet;
512 : }
513 : // set status for next data
514 5 : mbufDataSend_ = false;
515 5 : return FsmStatus::FSM_SUCCESS;
516 9 : }
517 :
518 16 : FsmStatus ChannelEntity::SendDataWithHccl(void* const dataBuf, const int32_t dataLen, Mbuf* const mbufToRecord)
519 : {
520 16 : HcclRequest req = nullptr;
521 16 : HcclComm handle = channelPtr_->GetHandle();
522 16 : const int32_t rankId = static_cast<int32_t>(channelPtr_->GetPeerRankId());
523 16 : const int32_t tagId = static_cast<int32_t>(channelPtr_->GetPeerTagId());
524 16 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
525 16 : const auto hcclRet = HcclIsend(dataBuf, dataLen, HCCL_DATA_TYPE_INT8, rankId, tagId, handle, &req);
526 32 : bqs::ProfileManager::GetInstance(resIndex_).AddHcclIsendCost(
527 16 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - begin);
528 16 : if (hcclRet == static_cast<int32_t>(HCCL_E_AGAIN)) {
529 3 : statInfo_.hcclIsendFullTimes++;
530 3 : bqs::StatisticManager::GetInstance().HcclMpiSendFullStat();
531 3 : DGW_LOG_WARN(
532 : "Failed to call HcclIsendWithEvent to send data for mbuf, tag full, entity:[%s], ret=[%d]",
533 : entityDesc_.c_str(), hcclRet);
534 3 : return FsmStatus::FSM_DEST_FULL;
535 : }
536 13 : if (hcclRet != static_cast<int32_t>(HCCL_SUCCESS)) {
537 2 : statInfo_.hcclIsendFailTimes++;
538 2 : bqs::StatisticManager::GetInstance().HcclMpiSendFailStat();
539 2 : DGW_LOG_ERROR("entity:[%s] fail to send data with hccl.", entityDesc_.c_str());
540 2 : return FsmStatus::FSM_ERROR_PENDING;
541 : }
542 11 : statInfo_.hcclIsendSuccTimes++;
543 11 : bqs::StatisticManager::GetInstance().HcclMpiSendSuccStat();
544 :
545 : // cache request
546 11 : RequestInfo reqInfo = {
547 : .req = req,
548 : .isLink = false,
549 : .mbuf = mbufToRecord,
550 11 : .startTick = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick()};
551 11 : const int32_t count = uncompReqQueue_.Push(reqInfo);
552 11 : if (count == 0) {
553 1 : DGW_LOG_ERROR("entity:[%s] fail to push req into uncompReqQueue.", entityDesc_.c_str());
554 1 : return FsmStatus::FSM_ERROR_PENDING;
555 : }
556 10 : statInfo_.uncompReqQueuePushTimes++;
557 10 : return FsmStatus::FSM_SUCCESS;
558 : }
559 :
560 11 : FsmStatus ChannelEntity::SendMbufData(Mbuf* const mbuf)
561 : {
562 : // check uncompleted req queue full
563 11 : if (uncompReqQueue_.IsFull()) {
564 2 : DGW_LOG_RUN_INFO("Uncompleted request queue of dst entity:[%s] is full.", entityDesc_.c_str());
565 2 : return FsmStatus::FSM_DEST_FULL;
566 : }
567 :
568 9 : uint64_t dataLen = 0UL;
569 9 : auto drvRet = halMbufGetDataLen(mbuf, &dataLen);
570 9 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) || (dataLen == 0U)) {
571 9 : drvRet = halMbufGetBuffSize(mbuf, &dataLen);
572 9 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
573 1 : DGW_LOG_ERROR("Fail to get buff size for mbuf, entity:[%s], ret=[%d]", entityDesc_.c_str(), drvRet);
574 1 : return FsmStatus::FSM_FAILED;
575 : }
576 : }
577 :
578 8 : void* dataBuf = nullptr;
579 8 : drvRet = halMbufGetBuffAddr(mbuf, &dataBuf);
580 8 : if ((drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) || (dataBuf == nullptr)) {
581 1 : DGW_LOG_ERROR("Fail to get buff addr for mbuf, entity:[%s], ret=[%d]", entityDesc_.c_str(), drvRet);
582 1 : return FsmStatus::FSM_FAILED;
583 : }
584 :
585 7 : DGW_LOG_INFO("Tag[%u] HcclIsend data[%lu]", channelPtr_->GetPeerTagId(), dataLen);
586 7 : const auto sendRet = SendDataWithHccl(dataBuf, static_cast<int32_t>(dataLen), nullptr);
587 7 : if (sendRet != FsmStatus::FSM_SUCCESS) {
588 2 : DGW_LOG_ERROR("Tag[%u] HcclIsend data[%lu] fail", channelPtr_->GetPeerTagId(), dataLen);
589 2 : return sendRet;
590 : }
591 :
592 5 : DGW_LOG_INFO(
593 : "Success to call HcclIsend to send data for mbuf, entity:[%s], len:[%lu].", entityDesc_.c_str(), dataLen);
594 5 : return FsmStatus::FSM_SUCCESS;
595 : }
596 :
597 8 : FsmStatus ChannelEntity::SendMbufHead(Mbuf* const mbuf)
598 : {
599 : // check uncompleted req queue full
600 8 : if (uncompReqQueue_.IsFull()) {
601 1 : DGW_LOG_RUN_INFO("Uncompleted request queue of dst entity:[%s] is full.", entityDesc_.c_str());
602 1 : return FsmStatus::FSM_DEST_FULL;
603 : }
604 :
605 7 : uint32_t headSize = 0U;
606 7 : void* headBuf = nullptr;
607 7 : const auto drvRet = halMbufGetPrivInfo(mbuf, &headBuf, &headSize);
608 7 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
609 1 : DGW_LOG_ERROR("Failed to get head info from mbuf, ret[%d].", drvRet);
610 1 : return FsmStatus::FSM_FAILED;
611 : }
612 :
613 6 : const auto sendRet = SendDataWithHccl(headBuf, static_cast<int32_t>(headSize), mbuf);
614 6 : if (sendRet != FsmStatus::FSM_SUCCESS) {
615 1 : DGW_LOG_ERROR("Tag[%u] HcclIsend head fail", channelPtr_->GetPeerTagId());
616 1 : return sendRet;
617 : }
618 :
619 5 : DGW_LOG_INFO(
620 : "Success to call HcclIsend to send head for mbuf, entity:[%s], len:[%u].", entityDesc_.c_str(), headSize);
621 5 : return FsmStatus::FSM_SUCCESS;
622 : }
623 :
624 27 : FsmStatus ChannelEntity::SendDataForLink()
625 : {
626 : HcclRequest req;
627 27 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
628 27 : const auto hcclRet = HcclIsend(
629 27 : nullptr, 0, HCCL_DATA_TYPE_INT8, static_cast<int32_t>(channelPtr_->GetPeerRankId()),
630 27 : static_cast<int32_t>(channelPtr_->GetPeerTagId()), channelPtr_->GetHandle(), &req);
631 54 : bqs::ProfileManager::GetInstance(resIndex_).AddHcclIsendCost(
632 27 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - begin);
633 27 : if (hcclRet != HCCL_SUCCESS) {
634 2 : DGW_LOG_ERROR(
635 : "Failed to call HcclIsend to send zero data for link establishment, entity:[%s], ret=[%d]",
636 : entityDesc_.c_str(), hcclRet);
637 2 : bqs::StatisticManager::GetInstance().HcclMpiSendFailStat();
638 2 : return FsmStatus::FSM_FAILED;
639 : }
640 25 : bqs::StatisticManager::GetInstance().HcclMpiSendSuccStat();
641 :
642 : // cache request
643 25 : RequestInfo reqInfo = {
644 : .req = req,
645 : .isLink = true,
646 : .mbuf = nullptr,
647 25 : .startTick = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick()};
648 25 : const int32_t count = uncompReqQueue_.Push(reqInfo);
649 25 : if (count == 0) {
650 1 : return FsmStatus::FSM_FAILED;
651 : }
652 24 : DGW_LOG_INFO("Success to send zero data for link establishment, entity:[%s].", entityDesc_.c_str());
653 24 : return FsmStatus::FSM_SUCCESS;
654 : }
655 :
656 14 : FsmStatus ChannelEntity::ProcessCompReq()
657 : {
658 14 : RequestInfo* const uncompReq = uncompReqQueue_.Front();
659 14 : if (uncompReq == nullptr) {
660 1 : DGW_LOG_ERROR("Failed to get front from uncompleted req queue, entity:[%s].", entityDesc_.c_str());
661 1 : return FsmStatus::FSM_FAILED;
662 : }
663 13 : const bool isSrc = (direction_ == EntityDirection::DIRECTION_SEND);
664 13 : const auto mbuf = uncompReq->mbuf;
665 13 : const auto req = uncompReq->req;
666 13 : const auto isLink = uncompReq->isLink;
667 13 : const auto reqProcTickCost = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - uncompReq->startTick;
668 13 : const auto reqProcCost = bqs::ProfileManager::GetInstance(resIndex_).AddReqProcCompCost(reqProcTickCost, isSrc);
669 :
670 : // process link request
671 13 : if (isLink) {
672 4 : return ProcessLinkRequest(req, reqProcCost);
673 : }
674 :
675 : // process request of data send/receive
676 9 : compReqCount_++;
677 9 : if ((reqProcCost > REQ_COMP_TIME_COST_THRESHOLD) && (compReqCount_ > COUNT_THRESHOLD_FOR_PRINT_ERROR)) {
678 1 : DGW_LOG_RUN_INFO(
679 : "Time cost to complete request is %.2fus, count:[%lu], entity:[%s], isSrc[%d].", reqProcCost, compReqCount_,
680 : entityDesc_.c_str(), static_cast<int32_t>(isSrc));
681 : }
682 :
683 9 : statInfo_.hcclTestSomeSuccTimes++;
684 : // pop request: pop failed, unhandled error
685 9 : const int32_t count = uncompReqQueue_.Pop();
686 9 : if (count == 0) {
687 1 : DGW_LOG_ERROR("Failed to pop request from uncompleted req queue, entity:[%s].", entityDesc_.c_str());
688 : } else {
689 8 : statInfo_.uncompReqQueuePopTimes++;
690 8 : DGW_LOG_DEBUG("Success to pop request from uncompleted req queue, entity:[%s].", entityDesc_.c_str());
691 : }
692 : // no need to process when mbuf is nullptr
693 9 : if (mbuf == nullptr) {
694 : // data
695 6 : UpdateStatisticForBody(reqProcTickCost);
696 6 : DGW_LOG_DEBUG("Mbuf is nullptr, no need to process!");
697 6 : return FsmStatus::FSM_SUCCESS;
698 : }
699 : // head
700 3 : UpdateStatisticForHead(reqProcTickCost);
701 :
702 3 : return isSrc ? ProcessReceiveCompletion(mbuf) : ProcessSendCompletion(mbuf);
703 : }
704 :
705 6 : void ChannelEntity::UpdateStatisticForBody(const uint64_t reqProcTickCost)
706 : {
707 6 : if (reqProcTickCost > statInfo_.maxCompletionGapTickForBody) {
708 5 : statInfo_.maxCompletionGapTickForBody = reqProcTickCost;
709 : }
710 6 : if ((reqProcTickCost < statInfo_.minCompletionGapTickForBody) || statInfo_.totalCompletionCountForBody == 0U) {
711 6 : statInfo_.minCompletionGapTickForBody = reqProcTickCost;
712 : }
713 6 : statInfo_.totalCompletionGapTickForBody += reqProcTickCost;
714 6 : ++statInfo_.totalCompletionCountForBody;
715 6 : }
716 :
717 3 : void ChannelEntity::UpdateStatisticForHead(const uint64_t reqProcTickCost)
718 : {
719 3 : if (reqProcTickCost > statInfo_.maxCompletionGapTickForHead) {
720 3 : statInfo_.maxCompletionGapTickForHead = reqProcTickCost;
721 : }
722 3 : if ((reqProcTickCost < statInfo_.minCompletionGapTickForHead) || (statInfo_.totalCompletionCountForHead == 0U)) {
723 3 : statInfo_.minCompletionGapTickForHead = reqProcTickCost;
724 : }
725 3 : statInfo_.totalCompletionGapTickForHead += reqProcTickCost;
726 3 : ++statInfo_.totalCompletionCountForHead;
727 3 : }
728 :
729 21 : RequestInfo* ChannelEntity::FrontUncompReq() { return uncompReqQueue_.Front(); }
730 :
731 17 : bool ChannelEntity::AddCachedReqCount()
732 : {
733 17 : cachedReqCountLock.Lock();
734 17 : if (ScheduleConfig::GetInstance().IsStopped(schedCfgKey_)) {
735 1 : cachedReqCount_ = 0U;
736 1 : DGW_LOG_INFO("Entity[%s] modify cachedReqCount to zero for schedule_stopped", entityDesc_.c_str());
737 1 : cachedReqCountLock.Unlock();
738 1 : return true;
739 : }
740 :
741 16 : if (cachedReqCount_ >= maxCachedReqCount_) {
742 8 : cachedReqCountLock.Unlock();
743 8 : DGW_LOG_INFO(
744 : "cached req count[%u] for entity[%s] is up to max[%u].", cachedReqCount_, entityDesc_.c_str(),
745 : maxCachedReqCount_);
746 8 : return false;
747 : }
748 8 : ++cachedReqCount_;
749 8 : cachedReqCountLock.Unlock();
750 8 : DGW_LOG_DEBUG(
751 : "Success to add cached req count for entity[%s], current count:[%u].", entityDesc_.c_str(), cachedReqCount_);
752 8 : return true;
753 : }
754 :
755 4 : bool ChannelEntity::ReduceCachedReqCount()
756 : {
757 4 : cachedReqCountLock.Lock();
758 4 : if (cachedReqCount_ == 0U) {
759 2 : cachedReqCountLock.Unlock();
760 2 : DGW_LOG_ERROR("Entity[%s] has no cached req!", entityDesc_.c_str());
761 2 : return false;
762 : }
763 2 : --cachedReqCount_;
764 2 : cachedReqCountLock.Unlock();
765 2 : DGW_LOG_DEBUG(
766 : "Success to reduce cached req count for entity[%s], current count:[%u].", entityDesc_.c_str(), cachedReqCount_);
767 2 : return true;
768 : }
769 :
770 1 : const CommChannel* ChannelEntity::GetCommChannel() const { return channelPtr_; }
771 :
772 67 : uint32_t ChannelEntity::GetQueueId() const { return compReqQueueId_; }
773 :
774 5 : bool ChannelEntity::CheckRecvReqEventContinue()
775 : {
776 5 : if (cachedEnvelopeQueue_.IsEmpty()) {
777 2 : return false;
778 : }
779 3 : bool flag = false;
780 3 : cachedReqCountLock.Lock();
781 3 : flag = (cachedReqCount_ != maxCachedReqCount_) ? true : false;
782 3 : cachedReqCountLock.Unlock();
783 3 : DGW_LOG_DEBUG(
784 : "Check entity[%s] to supply receive request event, flag:[%d].", entityDesc_.c_str(),
785 : static_cast<int32_t>(flag));
786 3 : return flag;
787 : }
788 :
789 4 : FsmStatus ChannelEntity::ProcessSendCompletion(Mbuf* mbuf)
790 : {
791 4 : uint64_t dataLen = 0UL;
792 4 : auto drvRet = halMbufGetBuffSize(mbuf, &dataLen);
793 4 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
794 1 : DGW_LOG_ERROR(
795 : "Unhandled error!! Fail to get buff size for mbuf, entity:[%s], ret=[%d]", entityDesc_.c_str(), drvRet);
796 : }
797 :
798 4 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
799 4 : MbufTypeInfo typeInfo = {};
800 4 : uint32_t outLen = sizeof(typeInfo);
801 4 : drvRet = halBuffGetInfo(
802 : BUFF_GET_MBUF_TYPE_INFO, PtrToPtr<Mbuf*, void>(&mbuf), static_cast<uint32_t>(sizeof(mbuf)),
803 : PtrToPtr<MbufTypeInfo, void>(&typeInfo), &outLen);
804 4 : if ((drvRet == static_cast<int32_t>(DRV_ERROR_NONE)) &&
805 3 : (typeInfo.type == static_cast<uint32_t>(MBUF_CREATE_BY_BUILD))) {
806 2 : void* buff = nullptr;
807 2 : uint64_t len = 0U;
808 2 : drvRet = halMbufUnBuild(mbuf, &buff, &len);
809 2 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
810 1 : DGW_LOG_ERROR("halMbufUnBuild fail, ret: %d", drvRet);
811 : } else {
812 1 : halBuffPut(nullptr, buff);
813 1 : DGW_LOG_INFO("Free head success");
814 : }
815 2 : } else {
816 2 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
817 1 : DGW_LOG_ERROR("halBuffGetInfo fail, ret: %d", drvRet);
818 : }
819 2 : (void)halMbufFree(mbuf);
820 2 : DGW_LOG_INFO("Free mbuf.");
821 : }
822 :
823 8 : bqs::ProfileManager::GetInstance(resIndex_).AddMbufFreeCost(
824 4 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - begin);
825 4 : statInfo_.freeMbufTimes++;
826 4 : bqs::StatisticManager::GetInstance().MbufFreeStat(dataLen);
827 4 : DGW_LOG_INFO("Success to free mbuf for entity[%s] when processing send completion event.", entityDesc_.c_str());
828 :
829 4 : return FsmStatus::FSM_SUCCESS;
830 : }
831 :
832 4 : FsmStatus ChannelEntity::ProcessReceiveCompletion(Mbuf* const mbuf)
833 : {
834 4 : bqs::ProfInfo reportData = {};
835 4 : if (bqs::BqsMsprofManager::GetInstance().IsStartProfling()) {
836 2 : reportData.type = static_cast<uint32_t>(bqs::DgwProfInfoType::ENQUEUE_DATA);
837 2 : reportData.itemId = transId_;
838 2 : reportData.timeStamp = bqs::GetTimeStamp();
839 : }
840 4 : DGW_LOG_INFO("Tag[%u] recv completion", channelPtr_->GetPeerTagId());
841 :
842 4 : if (ScheduleConfig::GetInstance().IsStopped(schedCfgKey_)) {
843 1 : (void)halMbufFree(mbuf);
844 1 : DGW_LOG_INFO("Entity[%s] discard mbuf for schedule_stopped", entityDesc_.c_str());
845 1 : return FsmStatus::FSM_SUCCESS;
846 : }
847 : // recv completion
848 3 : const uint64_t begin = bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick();
849 3 : const auto drvRet = halQueueEnQueue(deviceId_, compReqQueueId_, PtrToPtr<void, Mbuf>(mbuf));
850 :
851 3 : DGW_LOG_INFO(
852 : "%s halQueueEnQueue queue id:[%u] device id:[%u] result:[%d].", entityDesc_.c_str(), compReqQueueId_, deviceId_,
853 : static_cast<int32_t>(drvRet));
854 3 : bqs::BqsMsprofManager::GetInstance().ReportApiPerf(reportData);
855 6 : bqs::ProfileManager::GetInstance(resIndex_).AddHcclEnqueueCost(
856 3 : bqs::ProfileManager::GetInstance(resIndex_).GetCpuTick() - begin);
857 3 : if (drvRet != DRV_ERROR_NONE) {
858 1 : statInfo_.hcclEnqueueFailTimes++;
859 1 : DGW_LOG_ERROR(
860 : "Drop mbuf! Failed to enqueue completed req mbuf, entity:[%s], ret:[%d].", entityDesc_.c_str(),
861 : static_cast<int32_t>(drvRet));
862 1 : (void)halMbufFree(mbuf);
863 1 : return FsmStatus::FSM_FAILED;
864 : }
865 2 : statInfo_.hcclEnqueueSuccTimes++;
866 2 : return FsmStatus::FSM_SUCCESS;
867 : }
868 :
869 5 : FsmStatus ChannelEntity::ProcessLinkRequest(const HcclRequest& req, const float64_t reqProcCost)
870 : {
871 : (void)req;
872 : (void)reqProcCost;
873 5 : DGW_LOG_RUN_INFO(
874 : "Time cost to complete link request is %.2fus, entity:[%s], isSrc[%d].", reqProcCost, entityDesc_.c_str(),
875 : (direction_ == EntityDirection::DIRECTION_SEND));
876 :
877 : // pop request: pop failed, unhandled error
878 5 : const int32_t count = uncompReqQueue_.Pop();
879 5 : if (count == 0) {
880 1 : DGW_LOG_ERROR("Failed to pop link request from uncompleted req queue, entity:[%s].", entityDesc_.c_str());
881 1 : return FsmStatus::FSM_FAILED;
882 : }
883 :
884 4 : DGW_LOG_INFO("Success to pop link request from uncompleted req queue, entity:[%s].", entityDesc_.c_str());
885 :
886 4 : linkStatus_ = dgw::ChannelLinkStatus::CONNECTED;
887 4 : const uint32_t unlinkTagCount = bqs::StatisticManager::GetInstance().ReduceUnlinkCount();
888 4 : DGW_LOG_RUN_INFO(
889 : "Success to establish a link for entity:[%s], current unlink tag count is [%u]", entityDesc_.c_str(),
890 : unlinkTagCount);
891 4 : return FsmStatus::FSM_SUCCESS;
892 : }
893 :
894 32 : void ChannelEntity::Dump() const
895 : {
896 32 : const std::string desc = (direction_ == EntityDirection::DIRECTION_SEND) ? "Src" : "Dst";
897 : const auto maxCompletionGapForBody =
898 32 : bqs::ProfileManager::GetInstance(resIndex_).GetTimeCost(statInfo_.maxCompletionGapTickForBody);
899 : const auto minCompletionGapForBody =
900 32 : bqs::ProfileManager::GetInstance(resIndex_).GetTimeCost(statInfo_.minCompletionGapTickForBody);
901 : const auto avgCompletionGapForBody =
902 32 : (statInfo_.totalCompletionCountForBody == 0U) ?
903 : 0.0 :
904 4 : bqs::ProfileManager::GetInstance(resIndex_).GetTimeCost(statInfo_.totalCompletionGapTickForBody) /
905 4 : statInfo_.totalCompletionCountForBody;
906 : const auto maxCompletionGapForHead =
907 32 : bqs::ProfileManager::GetInstance(resIndex_).GetTimeCost(statInfo_.maxCompletionGapTickForHead);
908 : const auto minCompletionGapForHead =
909 32 : bqs::ProfileManager::GetInstance(resIndex_).GetTimeCost(statInfo_.minCompletionGapTickForHead);
910 : const auto avgCompletionGapForHead =
911 32 : (statInfo_.totalCompletionCountForHead == 0U) ?
912 : 0.0 :
913 3 : bqs::ProfileManager::GetInstance(resIndex_).GetTimeCost(statInfo_.totalCompletionGapTickForHead) /
914 3 : statInfo_.totalCompletionCountForHead;
915 32 : DGW_LOG_RUN_INFO(
916 : "%s entity statistic info: desc=[%s], HcclImprobe=[succ:%lu, fail:%lu, total:%lu], "
917 : "alloc mbuf=[%lu], HcclImrecv=[succ:%lu, fail:%lu], HcclTestSome=[succ:%lu], "
918 : "uncompReqQueue=[push:%lu, pop:%lu], bodyCostUs=[max: %.2f, avg: %.2f, min: %.2f], "
919 : "headCostUs=[max: %.2f, avg: %.2f, min: %.2f], "
920 : "HcclIsend=[succ:%lu, full:%lu, fail:%lu], "
921 : "free mbuf=[%lu], hccl enqueue=[succ:%lu, fail:%lu], dequeue=[succ:%lu, fail:%lu], "
922 : "cached envelope=[%u], link status=[%d].",
923 : desc.c_str(), entityDesc_.c_str(), statInfo_.hcclImprobeSuccTimes, statInfo_.hcclImprobeFailTimes,
924 : statInfo_.hcclImprobeTotalTimes, statInfo_.allocMbufTimes, statInfo_.hcclImrecvSuccTimes,
925 : statInfo_.hcclImrecvFailTimes, statInfo_.hcclTestSomeSuccTimes, statInfo_.uncompReqQueuePushTimes,
926 : statInfo_.uncompReqQueuePopTimes, maxCompletionGapForBody, avgCompletionGapForBody, minCompletionGapForBody,
927 : maxCompletionGapForHead, avgCompletionGapForHead, minCompletionGapForHead, statInfo_.hcclIsendSuccTimes,
928 : statInfo_.hcclIsendFullTimes, statInfo_.hcclIsendFailTimes, statInfo_.freeMbufTimes,
929 : statInfo_.hcclEnqueueSuccTimes, statInfo_.hcclEnqueueFailTimes, statInfo_.dequeueSuccTimes,
930 : statInfo_.dequeueFailTimes, cachedEnvelopeQueue_.Size(), static_cast<int32_t>(linkStatus_));
931 32 : }
932 :
933 1 : FsmStatus ChannelEntity::MakeSureOutputCompletion()
934 : {
935 1 : DGW_LOG_INFO("Entity[%s] start to wait send completion", entityDesc_.c_str());
936 1 : FsmStatus ret = FsmStatus::FSM_SUCCESS;
937 1 : uint32_t totalWaitUs = 0U;
938 1001 : while (!uncompReqQueue_.IsEmpty()) {
939 1001 : if (totalWaitUs >= CHECK_SEND_COMPLETION_LIMIT_US) {
940 1 : DGW_LOG_RUN_INFO(
941 : "Entity[%s] fail to finish sending in [%u] us", entityDesc_.c_str(), CHECK_SEND_COMPLETION_LIMIT_US);
942 1 : ret = FsmStatus::FSM_FAILED;
943 1 : break;
944 : }
945 1000 : usleep(CHECK_SEND_COMPLETION_INTERVAL_US);
946 1000 : totalWaitUs += CHECK_SEND_COMPLETION_INTERVAL_US;
947 : }
948 :
949 1 : DGW_LOG_INFO(
950 : "Entity[%s] Finish to wait send completion, cost [%u] us, left [%u] requests", entityDesc_.c_str(), totalWaitUs,
951 : uncompReqQueue_.Size());
952 3 : while (!uncompReqQueue_.IsEmpty()) {
953 2 : uncompReqQueue_.Pop();
954 : }
955 1 : return ret;
956 : }
957 :
958 1 : void ChannelEntity::PostDeque()
959 : {
960 1 : const bool firstRet = ReduceCachedReqCount();
961 1 : const bool secondRet = ReduceCachedReqCount();
962 1 : if ((!firstRet) || (!secondRet)) {
963 1 : DGW_LOG_ERROR(
964 : "Unhandled error! Reduce cached req count failed! first ret:[%d], second ret:[%d].",
965 : static_cast<int32_t>(firstRet), static_cast<int32_t>(secondRet));
966 : }
967 1 : }
968 :
969 : } // namespace dgw
|
