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