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