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 "dynamic_sched_mgr.hpp"
12 : #include <securec.h>
13 : #include "driver/ascend_hal.h"
14 : #include "bqs_feature_ctrl.h"
15 :
16 : namespace dgw {
17 : namespace {
18 : constexpr int32_t kMicrosecondToNanosecond = 1000;
19 : constexpr int32_t kDynamicSchedDuration = 2000 * kMicrosecondToNanosecond; // 2ms
20 : constexpr int32_t kRequestCacheNum = 3;
21 : constexpr uint32_t KCoLocateNum = 2;
22 : }
23 :
24 : // 考虑合设flowgw场景，一个flowgw可能服务两个device，默认device0
25 69 : DynamicSchedMgr &DynamicSchedMgr::GetInstance(uint32_t deviceId)
26 : {
27 69 : uint32_t index = deviceId >= KCoLocateNum ? 0U : deviceId;
28 71 : static DynamicSchedMgr mgr[KCoLocateNum];
29 69 : return mgr[index];
30 : }
31 :
32 11 : FsmStatus DynamicSchedMgr::AddRootModelInfo(const RootModelInfo &rootModelInfo)
33 : {
34 11 : const uint32_t rootModelId = rootModelInfo.rootModelId;
35 11 : const auto iter = rootModelInfos_.find(rootModelId);
36 11 : if (iter != rootModelInfos_.end()) {
37 1 : DGW_LOG_ERROR("Root model info has been added, rootModelId=%u.", rootModelId);
38 1 : return FsmStatus::FSM_FAILED;
39 : }
40 10 : (void)rootModelInfos_.emplace(rootModelId, rootModelInfo);
41 10 : return FsmStatus::FSM_SUCCESS;
42 : }
43 :
44 9 : void DynamicSchedMgr::DeleteQueue(const uint32_t globalLogicId, const uint32_t rootModelId)
45 : {
46 9 : DynamicSchedDurationPrint();
47 9 : for (auto iter = rootModelInfos_.begin(); iter != rootModelInfos_.end(); iter++) {
48 8 : if ((iter->first == rootModelId) &&
49 4 : (iter->second.responseQue.globalLogicId == globalLogicId)) {
50 4 : (void)rootModelInfos_.erase(iter);
51 4 : return;
52 : }
53 : }
54 : }
55 :
56 :
57 9 : void DynamicSchedMgr::UpdateNodeId(const int32_t nodeId)
58 : {
59 9 : nodeId_ = nodeId;
60 9 : }
61 :
62 6 : void DynamicSchedMgr::GenerateRequest(const std::vector<RequestInfo> &requests,
63 : const int32_t centerResponseQueIdx,
64 : dynamic::FlowgwRequest &flowgwRequest) const
65 : {
66 6 : flowgwRequest.set_node_id(nodeId_);
67 6 : flowgwRequest.set_input_index(centerResponseQueIdx);
68 11 : for (const auto &request : requests) {
69 10 : for (const auto &decision : request.decisions) {
70 10 : for (const auto &dst : request.dsts) {
71 5 : auto queueInfo = flowgwRequest.add_queue_infos();
72 5 : queueInfo->set_logic_group_id(dst.logicGroupId);
73 5 : queueInfo->set_model_uuid(request.src.modelUuid);
74 5 : queueInfo->set_root_model_id(request.src.rootModelId);
75 5 : queueInfo->set_trans_id(decision.transId);
76 5 : queueInfo->set_route_label(decision.routeLabel);
77 : // Compatible for old version
78 5 : queueInfo->set_trans_id_old(static_cast<int32_t>(decision.transId));
79 5 : queueInfo->set_route_label_old(static_cast<int32_t>(decision.routeLabel));
80 5 : auto queueAttr = queueInfo->mutable_queue_attrs();
81 5 : queueAttr->set_queue_id(request.src.queueId);
82 5 : queueAttr->set_device_id(request.src.deviceId);
83 5 : queueAttr->set_logic_id(request.src.queueLogicId);
84 : }
85 : }
86 : }
87 6 : }
88 :
89 8 : FsmStatus DynamicSchedMgr::SendRequest(const uint32_t rootModelId,
90 : const std::vector<RequestInfo> &requests)
91 : {
92 : // get root model info
93 8 : const auto iter = rootModelInfos_.find(rootModelId);
94 8 : if (iter == rootModelInfos_.end()) {
95 1 : DGW_LOG_ERROR("Root model info has not been added, rootModelId=%u.", rootModelId);
96 1 : return FsmStatus::FSM_FAILED;
97 : }
98 : // some requests will not be enqueue because result has been cached
99 7 : std::vector<RequestInfo> requestsAfterCache;
100 7 : SendRequestToCacheResult(requests, requestsAfterCache);
101 :
102 7 : if (requestSentNum_ > kRequestCacheNum) {
103 4 : for (auto &request : requestsAfterCache) {
104 2 : iter->second.requestCache.push_back(request);
105 : }
106 2 : return FsmStatus::FSM_SUCCESS;
107 : }
108 : // construct FlowgwRequest protobuf object
109 5 : auto cacheSize = iter->second.requestCache.size();
110 5 : dynamic::FlowgwRequest flowgwRequest;
111 5 : if (cacheSize == 0) {
112 4 : GenerateRequest(requestsAfterCache, iter->second.responseQue.globalLogicId, flowgwRequest);
113 : } else {
114 1 : GenerateRequest(iter->second.requestCache, iter->second.responseQue.globalLogicId, flowgwRequest);
115 1 : GenerateRequest(requestsAfterCache, iter->second.responseQue.globalLogicId, flowgwRequest);
116 : }
117 5 : if (flowgwRequest.queue_infos_size() == 0U) {
118 0 : return FsmStatus::FSM_SUCCESS;
119 : }
120 :
121 5 : const auto enqueueRet = EnqueueRequest(flowgwRequest, iter->second.requestQue.deviceId,
122 5 : iter->second.requestQue.queueId);
123 5 : if (enqueueRet != FsmStatus::FSM_SUCCESS) {
124 1 : return enqueueRet;
125 : }
126 4 : iter->second.requestCache.clear();
127 4 : requestSentNum_++;
128 4 : return FsmStatus::FSM_SUCCESS;
129 7 : }
130 :
131 5 : FsmStatus DynamicSchedMgr::EnqueueRequest(const dynamic::FlowgwRequest &flowgwRequest,
132 : const uint32_t deviceId, const uint32_t queueId) const
133 : {
134 5 : const auto reqSize = flowgwRequest.ByteSizeLong();
135 5 : Mbuf *mbuf = nullptr;
136 5 : auto drvRet = halMbufAlloc(reqSize, &mbuf);
137 5 : if (drvRet != DRV_ERROR_NONE) {
138 0 : DGW_LOG_ERROR("halMbufAlloc failed, drvRet=%d, dataSize=%lu.", drvRet, reqSize);
139 0 : return FsmStatus::FSM_FAILED;
140 : }
141 0 : auto mbufDeleter = [](Mbuf *buf) { (void)halMbufFree(buf); };
142 5 : std::unique_ptr<Mbuf, decltype(mbufDeleter)> mbufGuard(mbuf, mbufDeleter);
143 5 : drvRet = halMbufSetDataLen(mbuf, reqSize);
144 5 : if (drvRet != DRV_ERROR_NONE) {
145 0 : DGW_LOG_ERROR("halMbufSetDataLen failed, drvRet=%d, dataSize=%lu.", drvRet, reqSize);
146 0 : return FsmStatus::FSM_FAILED;
147 : }
148 : // write data
149 5 : void *buffAddr = nullptr;
150 5 : drvRet = halMbufGetBuffAddr(mbuf, &buffAddr);
151 5 : if (drvRet != DRV_ERROR_NONE || buffAddr == nullptr) {
152 1 : DGW_LOG_ERROR("Failed to get buff addr, ret[%d].", drvRet);
153 1 : return FsmStatus::FSM_FAILED;
154 : }
155 4 : flowgwRequest.SerializeToArray(buffAddr, static_cast<int32_t>(reqSize));
156 : // enqueue
157 4 : drvRet = halQueueEnQueue(deviceId, queueId, mbuf);
158 4 : if (drvRet == DRV_ERROR_QUEUE_FULL) {
159 0 : return FsmStatus::FSM_DEST_FULL;
160 4 : } else if (drvRet != DRV_ERROR_NONE) {
161 0 : DGW_LOG_ERROR("Failed to enqueue mbuf, ret[%d].", drvRet);
162 0 : return FsmStatus::FSM_FAILED;
163 : }
164 4 : PrintRequestLog(flowgwRequest);
165 4 : mbufGuard.release();
166 4 : return FsmStatus::FSM_SUCCESS;
167 5 : }
168 :
169 4 : void DynamicSchedMgr::PrintRequestLog(const dynamic::FlowgwRequest &flowgwRequest) const
170 : {
171 4 : if (!bqs::HostQsLog::GetInstance().CheckLogLevel(static_cast<int32_t>(AICPU), DLOG_INFO)) {
172 0 : return;
173 : }
174 4 : const int32_t queue_infos_size = flowgwRequest.queue_infos_size();
175 8 : for (int32_t queue_infos_index = 0; queue_infos_index < queue_infos_size; queue_infos_index++) {
176 4 : const auto &queue_info = flowgwRequest.queue_infos(queue_infos_index);
177 4 : DGW_LOG_INFO("Dynamic sched send request, node_id=%d, input_index=%d, queue_id=%u, device_type=%d, "
178 : "device_id=%d, logic_id=%u, logic_group_id=%u, model_uuid=%u, trans_id=%lu, route_label=%u, "
179 : "root_model_id=%u, queue_infos_index=%d.",
180 : flowgwRequest.node_id(), flowgwRequest.input_index(),
181 : queue_info.queue_attrs().queue_id(), queue_info.queue_attrs().device_type(),
182 : queue_info.queue_attrs().device_id(), queue_info.queue_attrs().logic_id(),
183 : queue_info.logic_group_id(), queue_info.model_uuid(), queue_info.trans_id(),
184 : queue_info.route_label(), queue_info.root_model_id(), queue_infos_index);
185 : }
186 : }
187 :
188 0 : void DynamicSchedMgr::PrintResponseLog(const dynamic::FlowgwResponse &flowgwResponse) const
189 : {
190 0 : if (!bqs::HostQsLog::GetInstance().CheckLogLevel(static_cast<int32_t>(AICPU), DLOG_INFO)) {
191 0 : return;
192 : }
193 0 : const int32_t queue_infos_size = flowgwResponse.queue_infos_size();
194 0 : for (int32_t queue_infos_index = 0; queue_infos_index < queue_infos_size; queue_infos_index++) {
195 0 : const auto &queue_info = flowgwResponse.queue_infos(queue_infos_index);
196 0 : DGW_LOG_INFO("Dynamic sched get response, queue_id=%u, device_type=%d, "
197 : "device_id=%d, logic_id=%u, logic_group_id=%u, model_uuid=%u, trans_id=%lu, route_label=%u, "
198 : "choose_logic_id=%u, root_model_id=%u, queue_infos_index=%d, need_cache=%d.",
199 : queue_info.queue_attrs().queue_id(), queue_info.queue_attrs().device_type(),
200 : queue_info.queue_attrs().device_id(), queue_info.queue_attrs().logic_id(),
201 : queue_info.logic_group_id(), queue_info.model_uuid(), queue_info.trans_id(),
202 : queue_info.route_label(), queue_info.choose_logic_id(),
203 : queue_info.root_model_id(), queue_infos_index, static_cast<int32_t>(queue_info.need_cache()));
204 : }
205 : }
206 :
207 :
208 5 : FsmStatus DynamicSchedMgr::GetResponse(const uint32_t rootModelId,
209 : std::vector<ResponseInfo> &responses)
210 : {
211 : // get root model info
212 5 : const auto iter = rootModelInfos_.find(rootModelId);
213 5 : if (iter == rootModelInfos_.end()) {
214 1 : return FsmStatus::FSM_SUCCESS;
215 : }
216 4 : GetResponseFromCacheResult(responses);
217 :
218 4 : auto cacheSize = iter->second.requestCache.size();
219 4 : if (requestSentNum_ == 0) {
220 2 : if (cacheSize != 0) {
221 1 : SendRequest(rootModelId, {});
222 : }
223 2 : return FsmStatus::FSM_SUCCESS;
224 : }
225 : // dequeue
226 2 : void *mbuf = nullptr;
227 2 : const auto ret = halQueueDeQueue(iter->second.responseQue.deviceId, iter->second.responseQue.queueId, &mbuf);
228 2 : if (ret == DRV_ERROR_NONE) {
229 1 : requestSentNum_--;
230 : // parse mbuf to FlowgwResponse
231 0 : auto mbufDeleter = [](Mbuf *buf) { (void)halMbufFree(buf); };
232 : std::unique_ptr<Mbuf, decltype(mbufDeleter)> mbufGuard(
233 1 : PtrToPtr<void, Mbuf>(mbuf) , mbufDeleter);
234 1 : dynamic::FlowgwResponse flowgwResponse;
235 1 : void *buffer_addr = nullptr;
236 1 : uint64_t buffer_size = 0U;
237 1 : if (halMbufGetBuffAddr(PtrToPtr<void, Mbuf>(mbuf), &buffer_addr) != DRV_ERROR_NONE) {
238 1 : DGW_LOG_ERROR("halMbufGetBuffAddr failed");
239 1 : return FsmStatus::FSM_FAILED;
240 : };
241 0 : if (halMbufGetBuffSize(PtrToPtr<void, Mbuf>(mbuf), &buffer_size) != DRV_ERROR_NONE) {
242 0 : DGW_LOG_ERROR("halMbufGetBuffAddr failed");
243 0 : return FsmStatus::FSM_FAILED;
244 : };
245 0 : google::protobuf::io::ArrayInputStream stream(buffer_addr, static_cast<int32_t>(buffer_size));
246 0 : if (!flowgwResponse.ParseFromZeroCopyStream(&stream)) {
247 0 : DGW_LOG_ERROR("Response ParseFromZeroCopyStream failed");
248 0 : return FsmStatus::FSM_FAILED;
249 : }
250 0 : PrintResponseLog(flowgwResponse);
251 : // write response data
252 0 : for (const auto &queueInfo : flowgwResponse.queue_infos()) {
253 0 : ResponseInfo responseInfo;
254 0 : responseInfo.src.queueId = queueInfo.queue_attrs().queue_id();
255 0 : responseInfo.src.queueLogicId = queueInfo.queue_attrs().logic_id();
256 0 : responseInfo.src.modelUuid = queueInfo.model_uuid();
257 0 : responseInfo.src.rootModelId = queueInfo.root_model_id();
258 : GroupResult groupResult;
259 0 : groupResult.logicGroupId = queueInfo.logic_group_id();
260 0 : groupResult.index = queueInfo.choose_logic_id();
261 0 : responseInfo.groupResults.emplace_back(std::move(groupResult));
262 0 : if (queueInfo.need_cache()) {
263 0 : UpdateCacheResult(responseInfo);
264 0 : continue;
265 : }
266 0 : responses.emplace_back(std::move(responseInfo));
267 0 : }
268 :
269 0 : if (requestSentNum_ == 0) {
270 0 : if (cacheSize != 0) {
271 0 : SendRequest(rootModelId, {});
272 : }
273 : }
274 3 : } else if (ret != DRV_ERROR_QUEUE_EMPTY) {
275 1 : DGW_LOG_ERROR("failed to dequeue, device_id = %u, queue_id = %u, ret = %d",
276 : iter->second.responseQue.deviceId, iter->second.responseQue.queueId, ret);
277 1 : return FsmStatus::FSM_FAILED;
278 : }
279 0 : GetResponseFromCacheResult(responses);
280 0 : return FsmStatus::FSM_SUCCESS;
281 : }
282 :
283 1 : FsmStatus DynamicSchedMgr::ClearCacheRouteResult()
284 : {
285 1 : validCacheInfos_.clear();
286 1 : invalidCacheInfos_.clear();
287 1 : return FsmStatus::FSM_SUCCESS;
288 : }
289 :
290 7 : void DynamicSchedMgr::SendRequestToCacheResult(const std::vector<RequestInfo> &requests,
291 : std::vector<RequestInfo> &requestsAfterCache)
292 : {
293 13 : for (const auto &request : requests) {
294 6 : RequestInfo requestAfterCache;
295 6 : requestAfterCache.src = request.src;
296 6 : requestAfterCache.decisions = request.decisions;
297 12 : for (const auto &dst : request.dsts) {
298 6 : CacheRouteKey key = {request.src, dst};
299 6 : const auto iterValidCache = validCacheInfos_.find(key);
300 6 : if (iterValidCache != validCacheInfos_.end()) {
301 0 : iterValidCache->second.num++;
302 0 : continue;
303 : }
304 6 : invalidCacheInfos_[key]++;
305 6 : requestAfterCache.dsts.emplace_back(dst);
306 : }
307 6 : if (requestAfterCache.dsts.empty()) {
308 0 : continue;
309 : }
310 6 : requestsAfterCache.emplace_back(std::move(requestAfterCache));
311 6 : }
312 7 : }
313 :
314 0 : void DynamicSchedMgr::UpdateCacheResult(const ResponseInfo &getResponseInfo)
315 : {
316 0 : for (const auto &result : getResponseInfo.groupResults) {
317 0 : DstGroupInfo dstGroupInfo = {result.logicGroupId};
318 0 : CacheRouteKey key = {getResponseInfo.src, dstGroupInfo};
319 0 : const auto iterValidCache = validCacheInfos_.find(key);
320 0 : if (iterValidCache != validCacheInfos_.end()) {
321 0 : continue;
322 : }
323 0 : const auto iterInvalidCache = invalidCacheInfos_.find(key);
324 0 : if (iterInvalidCache != invalidCacheInfos_.end()) {
325 0 : CacheRouteValue cacheRouteValue = {result, iterInvalidCache->second};
326 0 : validCacheInfos_.emplace(key, std::move(cacheRouteValue));
327 0 : invalidCacheInfos_.erase(iterInvalidCache);
328 : }
329 : }
330 0 : }
331 :
332 4 : void DynamicSchedMgr::GetResponseFromCacheResult(std::vector<ResponseInfo> &responses)
333 : {
334 4 : for (auto &cacheInfo : validCacheInfos_) {
335 0 : for (uint32_t index = 0U; index < cacheInfo.second.num; index++) {
336 0 : std::vector<GroupResult> groupResults;
337 0 : groupResults.emplace_back(cacheInfo.second.result);
338 0 : ResponseInfo response = {cacheInfo.first.srcQueueInfo, groupResults};
339 0 : responses.emplace_back(std::move(response));
340 0 : DGW_LOG_INFO("get response from cache result, root_model_id=%u, src queue_id=%u, " \
341 : "logic_id=%u, logic_group_id=%u, result_index=%u.",
342 : cacheInfo.first.srcQueueInfo.rootModelId, cacheInfo.first.srcQueueInfo.queueId,
343 : cacheInfo.first.srcQueueInfo.queueLogicId,
344 : cacheInfo.second.result.logicGroupId, cacheInfo.second.result.index);
345 0 : }
346 0 : cacheInfo.second.num = 0U;
347 : }
348 4 : }
349 :
350 4 : void DynamicSchedMgr::DynamicSchedDurationEnd(uint64_t begin)
351 : {
352 4 : uint64_t duration = DynamicSchedNow() - begin;
353 4 : durationTotal_ += duration;
354 4 : if (duration > kDynamicSchedDuration) {
355 0 : durationSize_++;
356 : }
357 4 : if (duration > durationMax_) {
358 4 : durationMax_ = duration;
359 : }
360 4 : cntTotal_++;
361 4 : }
362 :
363 9 : void DynamicSchedMgr::DynamicSchedDurationPrint()
364 : {
365 9 : BQS_LOG_RUN_INFO("DynamicSched, flowgw data: Total(us)=%lu, Cnt=%lu, Per duration(ns)=%lu, Max duration(ns)=%lu,"
366 : " Greater 2ms cnt=%lu", durationTotal_ / kMicrosecondToNanosecond, cntTotal_,
367 : (durationTotal_ / (cntTotal_ != 0ULL ? cntTotal_ : 1UL)), durationMax_, durationSize_);
368 9 : durationTotal_ = 0ULL;
369 9 : cntTotal_ = 0ULL;
370 9 : durationMax_ = 0ULL;
371 9 : durationSize_ = 0ULL;
372 9 : call_ = 0ULL;
373 9 : }
374 : }
|
