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 "deploy/abnormal_status_handler/abnormal_status_handler.h"
12 : #include <algorithm>
13 : #include <future>
14 : #include <sys/inotify.h>
15 : #include "mmpa/mmpa_api.h"
16 : #include "framework/common/util.h"
17 : #include "deploy/deployer/heterogeneous_model_deployer.h"
18 : #include "deploy/resource/resource_manager.h"
19 : #include "securec.h"
20 : #include "common/compile_profiling/ge_call_wrapper.h"
21 :
22 : namespace ge {
23 : namespace {
24 : constexpr const char_t *const kRedeployFileName = "redeploy";
25 : constexpr const char_t *const kRedeployDoneFileName = "redeploy.done";
26 : constexpr const char_t *const kRedeployErrorFileName = "redeploy.error";
27 : constexpr size_t kMaxPathLen = 1024UL;
28 : constexpr int32_t kMaxReadMonitorFileStrLen = 1024;
29 : constexpr int32_t kFileMonitorInterval = 500;
30 : constexpr int32_t kCheckReployFileInterval = 50;
31 : constexpr int32_t kCheckReployFileTimes = 10;
32 : constexpr int32_t kNotSupportDefault = 0;
33 : constexpr int32_t kNotSupportDynamicSched = 1;
34 : constexpr int32_t kNotSupportRedeploy = 2;
35 : } // namespace
36 :
37 : void AbnormalStatusHandler::FindOldDevice(DeployPlan::DeviceStateList &device_state_list, NodeConfig node_new,
38 : NodeConfig node_old) const {
39 : for (auto &iter_device_old : node_old.device_list) { // 新devices里面找老的device，找不到说明老的device损坏
40 : bool find_old_in_new = false;
41 : for (auto &iter_device_new : node_new.device_list) {
42 : if (iter_device_old.device_type == iter_device_new.device_type &&
43 : iter_device_old.device_id == iter_device_new.device_id) {
44 : DeployPlan::DeviceInfo device_info =
45 : DeployPlan::DeviceInfo(iter_device_old.device_type, node_old.node_id, iter_device_old.device_id);
46 : device_state_list.emplace(device_info, true);
47 : GELOGI("AbnormalStatusMonitor, device is normal, node_id=%d, device_id=%u, device_type=%u",
48 : device_info.GetNodeId(), device_info.GetDeviceId(), device_info.GetType());
49 : find_old_in_new = true;
50 : break;
51 : }
52 : }
53 : if (!find_old_in_new) {
54 : DeployPlan::DeviceInfo device_info =
55 : DeployPlan::DeviceInfo(iter_device_old.device_type, node_old.node_id, iter_device_old.device_id);
56 : device_state_list.emplace(device_info, false);
57 : GEEVENT("AbnormalStatusMonitor, device is abnormal, node_id=%d, device_id=%u, device_type=%u",
58 : device_info.GetNodeId(), device_info.GetDeviceId(), device_info.GetType());
59 : }
60 : }
61 : }
62 :
63 : Status AbnormalStatusHandler::FindAbnormalDeviceOnServer(DeployPlan::DeviceStateList &device_state_list,
64 : DeployerConfig information_new,
65 : DeployerConfig information_old) const {
66 : // 把所有node_config信息汇总到node_config_list
67 : std::vector<NodeConfig> node_config_list_new;
68 : node_config_list_new.assign(information_new.remote_node_config_list.begin(),
69 : information_new.remote_node_config_list.end());
70 : node_config_list_new.insert(node_config_list_new.begin(), information_new.node_config);
71 : std::vector<NodeConfig> node_config_list_old;
72 : node_config_list_old.assign(information_old.remote_node_config_list.begin(),
73 : information_old.remote_node_config_list.end());
74 : node_config_list_old.insert(node_config_list_old.begin(), information_old.node_config);
75 :
76 : for (auto &iter_old : node_config_list_old) {
77 : bool is_node_abnormal = true;
78 : for (auto &iter_new : node_config_list_new) {
79 : if (iter_old.ipaddr.compare(iter_new.ipaddr) != 0) {
80 : continue;
81 : }
82 : is_node_abnormal = false;
83 : FindOldDevice(device_state_list, iter_new, iter_old);
84 : }
85 : if (is_node_abnormal) {
86 : for (auto &iter_device_old : iter_old.device_list) {
87 : DeployPlan::DeviceInfo device_info =
88 : DeployPlan::DeviceInfo(iter_device_old.device_type, iter_old.node_id, iter_device_old.device_id);
89 : device_state_list.emplace(device_info, false);
90 : GEEVENT("AbnormalStatusMonitor, node is abnormal, node_id=%d, device_id=%u, device_type=%u",
91 : device_info.GetNodeId(), device_info.GetDeviceId(), device_info.GetType());
92 : }
93 : }
94 : }
95 : return SUCCESS;
96 : }
97 :
98 : Status AbnormalStatusHandler::ParseDeviceStateList(const std::string &file_path,
99 : DeployPlan::DeviceStateList &device_state_list) {
100 : // 解析异常设备信息
101 : auto information_old = Configurations::GetInstance().GetHostInformation();
102 : GELOGI("AbnormalStatusMonitor, show old node info");
103 : ShowNodeInfo(information_old);
104 : DeployerConfig information_new;
105 : GE_CHK_STATUS_RET_NOLOG(ConfigParser::ParseServerInfo(file_path, information_new));
106 : GELOGI("AbnormalStatusMonitor, show new node info on server");
107 : ShowNodeInfo(information_new);
108 : GE_CHK_STATUS_RET(FindAbnormalDeviceOnServer(device_state_list, information_new, information_old),
109 : "AbnormalStatusMonitor, failed to do FindAbnormalDevice if is on server");
110 : return SUCCESS;
111 : }
112 :
113 : bool AbnormalStatusHandler::IsHeartbeatNormal() const {
114 : auto &deploy_context = DeployContext::LocalContext();
115 : std::lock_guard<std::mutex> lk(deploy_context.GetAbnormalHeartbeatInfoMu());
116 : return deploy_context.GetAbnormalSubmodelInstanceName().empty() && deploy_context.GetAbnormalNodeConfig().empty() &&
117 : deploy_context.GetAbnormalDeviceInfo().empty();
118 : }
119 :
120 : void AbnormalStatusHandler::ParseAbnormalNodeConfig(DeployPlan::DeviceStateList &device_state_list) const {
121 : auto &deploy_context = DeployContext::LocalContext();
122 : for (auto &iter : deploy_context.GetAbnormalNodeConfig()) {
123 : for (auto &iter_device : iter.first.device_list) {
124 : DeployPlan::DeviceInfo device_info =
125 : DeployPlan::DeviceInfo(iter_device.device_type, iter.first.node_id, iter_device.device_id);
126 : device_state_list.emplace(device_info, false);
127 : GELOGI("AbnormalStatusMonitor, node abnormal(process OnServer), node_id=%d, device_id=%d, device_type=%d",
128 : device_info.GetNodeId(), device_info.GetDeviceId(), device_info.GetType());
129 : }
130 : }
131 : }
132 :
133 : void AbnormalStatusHandler::ParseAbnormalDeviceInfo(DeployPlan::DeviceStateList &device_state_list) const {
134 : auto &deploy_context = DeployContext::LocalContext();
135 : for (auto &iter : deploy_context.GetAbnormalDeviceInfo()) {
136 : GELOGI("AbnormalStatusMonitor, ParseAbnormalDeviceInfo: node_id=%d, device_id=%d, device_type=%d",
137 : iter.first.GetNodeId(), iter.first.GetDeviceId(), iter.first.GetType());
138 : device_state_list.emplace(iter.first, false);
139 : }
140 : }
141 :
142 : void AbnormalStatusHandler::ParseAbnormalModelInstances(bool &is_new_abnormal) {
143 : auto &deploy_context = DeployContext::LocalContext();
144 : for (auto &iter : deploy_context.GetAbnormalSubmodelInstanceName()) {
145 : for (auto &submodel_instance_name : iter.second) {
146 : auto instances_model_iter = abnormal_submodel_instances_name_[iter.first].find(submodel_instance_name.first);
147 : if (instances_model_iter == abnormal_submodel_instances_name_[iter.first].end()) {
148 : abnormal_submodel_instances_name_[iter.first][submodel_instance_name.first] = true;
149 : GELOGI(
150 : "AbnormalStatusMonitor, ParseAbnormalModelInstances: root model id=%u,"
151 : " submodel[%s] is on abnormal process",
152 : iter.first, submodel_instance_name.first.c_str());
153 : is_new_abnormal = true;
154 : }
155 : }
156 : }
157 : }
158 :
159 : void AbnormalStatusHandler::ParseHeartbeatAbnormalInfo(bool &is_new_abnormal,
160 : DeployPlan::DeviceStateList &device_state_list) {
161 : GELOGI("AbnormalStatusMonitor, ParseHeartbeatAbnormalInfo start");
162 : auto information_old = Configurations::GetInstance().GetHostInformation();
163 : ShowNodeInfo(information_old);
164 : auto &deploy_context = DeployContext::LocalContext();
165 : std::lock_guard<std::mutex> lk(deploy_context.GetAbnormalHeartbeatInfoMu());
166 : if (!deploy_context.GetAbnormalNodeConfig().empty()) { // tsd进程异常或心跳监测失败异常
167 : GELOGI("AbnormalStatusMonitor, tsd process abnormal, abnormal node info size=%zu",
168 : deploy_context.GetAbnormalNodeConfig().size());
169 : ParseAbnormalNodeConfig(device_state_list);
170 : AbnormalDevices2ModelInstances(device_state_list, is_new_abnormal); // node异常转为model instance异常
171 : deploy_context.ClearAbnormalNodeConfig();
172 : }
173 : if (!deploy_context.GetAbnormalDeviceInfo().empty()) { // flowgw进程异常
174 : GELOGI("AbnormalStatusMonitor, flowgw process abnormal, abnormal device info size=%zu",
175 : deploy_context.GetAbnormalDeviceInfo().size());
176 : ParseAbnormalDeviceInfo(device_state_list);
177 : AbnormalDevices2ModelInstances(device_state_list, is_new_abnormal); // device异常转为model instance异常
178 : deploy_context.ClearAbnormalDeviceInfo();
179 : }
180 : if (!deploy_context.GetAbnormalSubmodelInstanceName().empty()) { // 执行进程异常
181 : GELOGI("AbnormalStatusMonitor, executor process abnormal, abnormal submodel instances size=%zu",
182 : deploy_context.GetAbnormalSubmodelInstanceName().size());
183 : ParseAbnormalModelInstances(is_new_abnormal);
184 : deploy_context.ClearAbnormalSubmodelInstanceName();
185 : }
186 : GELOGI("AbnormalStatusMonitor, ParseHeartbeatAbnormalInfo end");
187 : }
188 :
189 : bool AbnormalStatusHandler::IsModelMulInstance(std::map<const std::string, bool> &abnormal_submodel_instances_name,
190 : DeployPlan::ModelDeployInfo model_deploy_infos) const {
191 : for (auto model_deploy_info = model_deploy_infos.begin(); model_deploy_info != model_deploy_infos.end();
192 : model_deploy_info++) {
193 : bool is_model_mul_model_instance = false;
194 : for (auto &model_instance_info : model_deploy_info->second) {
195 : auto abnormal_submodel_instances = abnormal_submodel_instances_name.find(model_instance_info.first);
196 : if (abnormal_submodel_instances == abnormal_submodel_instances_name.end()) {
197 : GELOGI("AbnormalStatusMonitor, the model[%s]'s instance[%s] deployed on normal process",
198 : model_deploy_info->first.c_str(), model_instance_info.first.c_str());
199 : is_model_mul_model_instance = true;
200 : }
201 : }
202 : if (!is_model_mul_model_instance) {
203 : GEEVENT(
204 : "AbnormalStatusMonitor, it doesn't support dynamic sched,"
205 : "the model[%s] only deployed on abnormal process",
206 : model_deploy_info->first.c_str());
207 : return false;
208 : }
209 : }
210 : return true;
211 : }
212 :
213 : bool AbnormalStatusHandler::IsSupportDynamicSchedRecover(const uint32_t &root_model_id) {
214 : if (!is_dynamic_sched_) {
215 : GELOGI("AbnormalStatusMonitor, is_dynamic_sched_ is unenable");
216 : return false;
217 : }
218 :
219 : // 获取root_model_id对应的submodel_instance_name 的映射
220 : DeployPlan::ModelDeployInfo model_deploy_info;
221 : {
222 : std::lock_guard<std::mutex> lk(mu_);
223 : model_deploy_info = deployed_models_[root_model_id].model_deploy_infos;
224 : }
225 :
226 : // 检查root_model_id对应的异常模型是否多实例
227 : if (!IsModelMulInstance(abnormal_submodel_instances_name_[root_model_id], model_deploy_info)) {
228 : return false;
229 : }
230 : return true;
231 : }
232 :
233 : Status AbnormalStatusHandler::GenerateFile(const std::string &file_path, const char_t *const file_name) const {
234 : auto pos = file_path.find_last_of('/');
235 : GE_CHK_BOOL_RET_STATUS(pos != std::string::npos, FAILED, "AbnormalStatusMonitor, failed to handle path[%s]",
236 : file_path.c_str());
237 : std::string new_file_path = file_path.substr(0, pos + 1) + file_name;
238 : std::ofstream file(new_file_path);
239 : GE_CHK_BOOL_RET_STATUS(file.is_open(), FAILED, "AbnormalStatusMonitor, failed generate path[%s]",
240 : new_file_path.c_str());
241 : file.close();
242 : GEEVENT("AbnormalStatusMonitor, the path[%s] has generated", new_file_path.c_str());
243 : return SUCCESS;
244 : }
245 :
246 : Status AbnormalStatusHandler::AfterHandleAbnormalInfo(const std::string &file_path, const char_t *const file_name) {
247 : GE_CHK_STATUS_RET(GenerateFile(file_path, file_name),
248 : "AbnormalStatusMonitor, failed to do GenerateFile, file_name[%s]", file_name);
249 : return SUCCESS;
250 : }
251 :
252 : Status AbnormalStatusHandler::FailedHandleAbnormal(uint32_t root_model_id) {
253 : GELOGI("AbnormalStatusMonitor, abnormal status doesn't support redeploy, callback exec, model_id=%u", root_model_id);
254 : std::lock_guard<std::mutex> lk(abnormal_status_callback_info_.mu);
255 : if (abnormal_status_callback_info_.callback_list[root_model_id] != nullptr) {
256 : GE_CHK_STATUS_RET(abnormal_status_callback_info_.callback_list[root_model_id](kCallbackFailedRedeploy,
257 : abnormal_submodel_instances_name_),
258 : "AbnormalStatusMonitor, callback exec failed, root_model_id=%u", root_model_id);
259 : }
260 : return SUCCESS;
261 : }
262 :
263 : void AbnormalStatusHandler::GetDeviceListDiff(const DeployPlan::DeviceStateList &device_state_list_new,
264 : DeployPlan::DeviceStateList &device_state_list_old,
265 : DeployPlan::DeviceStateList &device_state_list_diff) const {
266 : for (auto &iter_new : device_state_list_new) {
267 : if (iter_new.second) {
268 : continue;
269 : }
270 : DeployPlan::DeviceInfo device_info =
271 : DeployPlan::DeviceInfo(iter_new.first.GetType(), iter_new.first.GetNodeId(), iter_new.first.GetDeviceId());
272 : auto iter_old = device_state_list_old.find(device_info);
273 : if (iter_old != device_state_list_old.end() && !(iter_old->second)) {
274 : continue;
275 : } else {
276 : GELOGI("AbnormalStatusMonitor, GetDeviceListDiff: node_id=%d, device_id=%u, device_type=%u",
277 : device_info.GetNodeId(), device_info.GetDeviceId(), device_info.GetType());
278 : device_state_list_diff[device_info] = false;
279 : device_state_list_old[device_info] = false;
280 : }
281 : }
282 : }
283 :
284 : bool AbnormalStatusHandler::IsInDeviceList(std::set<DeployPlan::DeviceInfo> &instance_device_infos,
285 : DeployPlan::DeviceStateList &device_state_list_diff) const {
286 : for (const auto &instance_device_info : instance_device_infos) {
287 : DeployPlan::DeviceInfo device_info = DeployPlan::DeviceInfo(
288 : instance_device_info.GetType(), instance_device_info.GetNodeId(), instance_device_info.GetDeviceId());
289 : auto new_abnormal_device = device_state_list_diff.find(device_info);
290 : if (new_abnormal_device != device_state_list_diff.end() && !(new_abnormal_device->second)) {
291 : return true;
292 : }
293 : }
294 : return false;
295 : }
296 :
297 : bool AbnormalStatusHandler::IsInModelInstanceList(uint32_t root_model_id, const std::string &model_instance_name,
298 : RootModelId2SubmodelName &abnormal_submodel_instances_name) const {
299 : auto iter = abnormal_submodel_instances_name[root_model_id].find(model_instance_name);
300 : if (iter != abnormal_submodel_instances_name[root_model_id].end() && !(iter->second)) {
301 : return true;
302 : }
303 : return false;
304 : }
305 :
306 : void AbnormalStatusHandler::Add2ModelInstanceList(uint32_t root_model_id, const std::string &model_instance_name,
307 : RootModelId2SubmodelName &abnormal_submodel_instances_name) const {
308 : abnormal_submodel_instances_name[root_model_id][model_instance_name] = false;
309 : }
310 :
311 : void AbnormalStatusHandler::AbnormalDiffDevices2ModelInstances(
312 : uint32_t root_model_id, std::map<std::string, std::set<DeployPlan::DeviceInfo>> &model_deploy_info,
313 : DeployPlan::DeviceStateList &device_state_list_diff, bool &is_new_abnormal) {
314 : for (auto &model_instance_info : model_deploy_info) {
315 : if (!IsInDeviceList(model_instance_info.second, device_state_list_diff)) {
316 : GELOGI("AbnormalStatusMonitor, model instance[%s] is not on abnormal device list",
317 : model_instance_info.first.c_str());
318 : continue;
319 : }
320 : if (IsInModelInstanceList(root_model_id, model_instance_info.first, abnormal_submodel_instances_name_)) {
321 : GELOGI("AbnormalStatusMonitor, model instance[%s] is already in abnormal model instance list",
322 : model_instance_info.first.c_str());
323 : continue;
324 : } else {
325 : GELOGI("AbnormalStatusMonitor, model instance[%s] is add to abnormal list", model_instance_info.first.c_str());
326 : Add2ModelInstanceList(root_model_id, model_instance_info.first, abnormal_submodel_instances_name_);
327 : is_new_abnormal = true;
328 : }
329 : }
330 : }
331 :
332 : void AbnormalStatusHandler::AbnormalDevices2ModelInstances(DeployPlan::DeviceStateList &device_state_list,
333 : bool &is_new_abnormal) {
334 : DeployPlan::DeviceStateList device_state_list_diff;
335 : GetDeviceListDiff(device_state_list, device_state_list_, device_state_list_diff);
336 : if (device_state_list_diff.empty()) {
337 : GELOGI("AbnormalStatusMonitor, device_state_list has no new abnormal devices");
338 : return;
339 : }
340 : std::lock_guard<std::mutex> lk(mu_);
341 : for (auto &deployed_model : deployed_models_) {
342 : for (auto &model_deploy_info : deployed_model.second.model_deploy_infos) {
343 : AbnormalDiffDevices2ModelInstances(deployed_model.first, model_deploy_info.second, device_state_list_diff,
344 : is_new_abnormal);
345 : }
346 : }
347 : }
348 :
349 : Status AbnormalStatusHandler::RedeployStart(const uint32_t &root_model_id) {
350 : GELOGI("AbnormalStatusMonitor, redeploy start: pre callback exec, root_model_id=%u", root_model_id);
351 : std::lock_guard<std::mutex> lk(abnormal_status_callback_info_.mu);
352 : if (abnormal_status_callback_info_.callback_list[root_model_id] != nullptr) {
353 : GE_CHK_STATUS_RET(abnormal_status_callback_info_.callback_list[root_model_id](kCallbackStartRedeploy,
354 : abnormal_submodel_instances_name_),
355 : "AbnormalStatusMonitor, failed to do exec callback, root_model_id=%u", root_model_id);
356 : }
357 : return SUCCESS;
358 : }
359 :
360 : void AbnormalStatusHandler::PreHandleAbnormalInfo() {
361 : {
362 : std::lock_guard<std::mutex> lk(abnormal_status_callback_info_.mu);
363 : GELOGI("AbnormalStatusMonitor, wait callback init, callback size=%zu, deployed models size=%zu",
364 : abnormal_status_callback_info_.callback_list.size(), deployed_models_.size());
365 : }
366 : while (!IsDeployingRootModel() &&
367 : !IsAllCallbackInitFinished()) { // 无模型正在部署并且所有执行器把callbacks初始化后可开始重部署
368 : std::this_thread::sleep_for(std::chrono::milliseconds(kFileMonitorInterval)); // 等待0.5秒进行下一次检查
369 : }
370 : GELOGI("AbnormalStatusMonitor, callback init success");
371 : return;
372 : }
373 :
374 : void AbnormalStatusHandler::ShowNodeInfo(DeployerConfig &information) const {
375 : GELOGI("AbnormalStatusMonitor, ShowNodeInfo start");
376 : std::vector<NodeConfig> node_config_list;
377 : node_config_list.assign(information.remote_node_config_list.begin(), information.remote_node_config_list.end());
378 : node_config_list.insert(node_config_list.begin(), information.node_config);
379 : for (auto &iter : node_config_list) {
380 : GELOGI("AbnormalStatusMonitor, device info: node_id=%d", iter.node_id);
381 : for (auto &iter_device : iter.device_list) {
382 : GELOGI("AbnormalStatusMonitor, device info: device_type=%u, node_id=%d, device_id=%u", iter_device.device_type,
383 : iter.node_id, iter_device.device_id);
384 : }
385 : }
386 : GELOGI("AbnormalStatusMonitor, ShowNodeInfo end");
387 : }
388 :
389 : Status AbnormalStatusHandler::ParallelAbnormalStatusHandle(uint32_t check_devices_flag) {
390 : auto root_model_num = abnormal_submodel_instances_name_.size();
391 : if (root_model_num > 1) {
392 : ThreadPool pool("ge_dpl_rd", static_cast<uint32_t>(root_model_num), false);
393 : std::vector<std::future<Status>> fut_rets;
394 : for (const auto &it : abnormal_submodel_instances_name_) {
395 : const auto &root_model_id = it.first;
396 : auto fut = pool.commit([this, &root_model_id, &check_devices_flag]() -> Status {
397 : GE_CHK_STATUS_RET(RedeployProc(root_model_id, check_devices_flag),
398 : "AbnormalStatusMonitor, failed to do RedeployProc, root_model_id=%u", root_model_id);
399 : return SUCCESS;
400 : });
401 : fut_rets.emplace_back(std::move(fut));
402 : }
403 : bool has_root_model_redeploy_failed = false;
404 : for (auto &fut : fut_rets) {
405 : if (fut.get() != SUCCESS) { // fut.get()是同步操作
406 : has_root_model_redeploy_failed = true;
407 : }
408 : }
409 : if (has_root_model_redeploy_failed) {
410 : GELOGE(FAILED, "AbnormalStatusMonitor, there is a model redeployed failed");
411 : return FAILED;
412 : }
413 : return SUCCESS;
414 : }
415 : for (const auto &it : abnormal_submodel_instances_name_) {
416 : const auto &root_model_id = it.first;
417 : GE_CHK_STATUS_RET(RedeployProc(root_model_id, check_devices_flag),
418 : "AbnormalStatusMonitor, failed to do RedeployProc, root_model_id=%u", root_model_id);
419 : }
420 : return SUCCESS;
421 : }
422 :
423 : Status AbnormalStatusHandler::FileMonitorProc(const std::string &file_path) {
424 : GELOGI("AbnormalStatusMonitor, FileMonitorProc start, file_path=%s", file_path.c_str());
425 : DeployPlan::DeviceStateList device_state_list;
426 : GE_CHK_STATUS_RET(ParseDeviceStateList(file_path, device_state_list),
427 : "AbnormalStatusMonitor, failed to do ParseDeviceStateList");
428 : bool is_new_abnormal = false;
429 : AbnormalDevices2ModelInstances(device_state_list,
430 : is_new_abnormal); // device异常转换为model_instance异常(即执行进程异常)
431 : if (is_new_abnormal) {
432 : PreHandleAbnormalInfo();
433 : auto check_devices_flag = CheckAbnormalDevices(device_state_list);
434 : if (check_devices_flag == kNotSupportRedeploy) {
435 : // host异常，无法恢复业务, 写redeploy.error文件
436 : GE_CHK_STATUS_RET(AfterHandleAbnormalInfo(file_path, kRedeployErrorFileName),
437 : "AbnormalStatusMonitor, failed to do AfterHandleAbnormalInfo, kRedeployErrorFileName");
438 : GELOGE(FAILED, "AbnormalStatusMonitor, it(cause by abnormal device) can't recover by redeploying");
439 : }
440 : if (ParallelAbnormalStatusHandle(check_devices_flag) == SUCCESS) {
441 : GE_CHK_STATUS_RET(AfterHandleAbnormalInfo(file_path, kRedeployDoneFileName),
442 : "AbnormalStatusMonitor, failed to do AfterHandleAbnormalInfo, kRedeployDoneFileName");
443 : return SUCCESS;
444 : } else {
445 : GE_CHK_STATUS_RET(AfterHandleAbnormalInfo(file_path, kRedeployErrorFileName),
446 : "AbnormalStatusMonitor, failed to do AfterHandleAbnormalInfo, kRedeployErrorFileName");
447 : return FAILED;
448 : }
449 : }
450 : GELOGI(
451 : "AbnormalStatusMonitor, FileMonitorProc end, no new abnormal model instance"); // 无新异常或新异常上无模型实例部署
452 : return SUCCESS;
453 : }
454 :
455 : Status AbnormalStatusHandler::WaitReployFileGenerate(const std::string &file_path) {
456 : uint32_t times = 0;
457 : while (times++ < kCheckReployFileTimes) {
458 : GELOGI("AbnormalStatusMonitor, Wait for redeploy file generated, times=%u", times);
459 : if (IsReployFileGeneratedThenRemove(file_path)) {
460 : return SUCCESS;
461 : }
462 : std::this_thread::sleep_for(std::chrono::milliseconds(kCheckReployFileInterval)); // 0.05秒检查一次，检查10次
463 : }
464 : return FAILED;
465 : }
466 :
467 : bool AbnormalStatusHandler::IsReployFileGeneratedThenRemove(const std::string &file_path) const {
468 : auto pos = file_path.find_last_of('/');
469 : if (pos == std::string::npos) {
470 : GELOGE(FAILED, "AbnormalStatusMonitor, file path[%s] is illegal", file_path.c_str());
471 : return false;
472 : }
473 : std::string redeploy_file_path = file_path.substr(0, pos + 1) + kRedeployFileName;
474 : std::ifstream ifs(redeploy_file_path);
475 : if (!ifs.is_open()) {
476 : GELOGI("AbnormalStatusMonitor, The path[%s] hasn't generated", redeploy_file_path.c_str());
477 : return false;
478 : }
479 : ifs.close();
480 : (void)std::remove(redeploy_file_path.c_str());
481 : GELOGI("AbnormalStatusMonitor, The path[%s] has generated, now remove it", redeploy_file_path.c_str());
482 : return true;
483 : }
484 :
485 : Status AbnormalStatusHandler::GetFilePath(std::string &config_dir, const char_t *const path_env) const {
486 : char_t file_path[kMaxPathLen]{};
487 : const int32_t ret = mmGetEnv(path_env, file_path, kMaxPathLen);
488 : if (ret == EN_OK) {
489 : const std::string real_path = RealPath(file_path);
490 : if (!real_path.empty()) {
491 : config_dir = file_path;
492 : GELOGI("AbnormalStatusMonitor, Get file_path[%s] success from env", file_path);
493 : return SUCCESS;
494 : }
495 : }
496 : return FAILED;
497 : }
498 :
499 : Status AbnormalStatusHandler::GetMonitorFilePath(std::string &file_path) {
500 : GELOGI("AbnormalStatusMonitor, try get resource config path, on server");
501 : GE_CHK_STATUS_RET(Configurations::GetResourceConfigPath(file_path), "Failed to get resource file");
502 : GELOGI("AbnormalStatusMonitor, get resource config path[%s] success", file_path.c_str());
503 : return SUCCESS;
504 : }
505 :
506 : void AbnormalStatusHandler::MonitorFileAndHeartbeatProc(const std::string &file_path, const int32_t &fd) {
507 : char_t buf[kMaxReadMonitorFileStrLen];
508 : ssize_t len = 0;
509 : GELOGI("AbnormalStatusMonitor, monitoring path[%s]", file_path.c_str());
510 : while (file_monitor_flag_.load()) {
511 : len = read(fd, buf, sizeof(buf));
512 : if (len <= 0) { // 优先文件监测无异常，再心跳监测
513 : if (IsHeartbeatNormal()) {
514 : std::this_thread::sleep_for(std::chrono::milliseconds(kFileMonitorInterval)); // 等待0.5秒进行下一次检查
515 : } else if (HeartbeatMonitorProc() != SUCCESS) {
516 : GELOGE(FAILED, "AbnormalStatusMonitor, failed to do HeartbeatMonitorProc");
517 : break;
518 : }
519 : continue;
520 : }
521 : GELOGI("AbnormalStatusMonitor, The path[%s] is different, parser the different type", file_path.c_str());
522 : char_t *event_buf = buf;
523 : bool resource_config_modify = false;
524 : while (event_buf < buf + len) {
525 : struct inotify_event *event = static_cast<struct inotify_event *>(static_cast<void *>(event_buf));
526 : if ((event->mask & IN_MODIFY) != 0) {
527 : resource_config_modify = true;
528 : break;
529 : }
530 : event_buf += sizeof(struct inotify_event) + event->len;
531 : }
532 : if (resource_config_modify) {
533 : GELOGI("AbnormalStatusMonitor, The path[%s] has modify", file_path.c_str());
534 : if (WaitReployFileGenerate(file_path) == SUCCESS && FileMonitorProc(file_path) != SUCCESS) {
535 : GELOGE(FAILED, "AbnormalStatusMonitor, failed to FileMonitorProc");
536 : break;
537 : }
538 : }
539 : }
540 : }
541 :
542 : void AbnormalStatusHandler::AbnormalStatusMonitorRun() {
543 : GEEVENT("AbnormalStatusMonitor, abnormal status monitor thread start");
544 : std::string file_path;
545 : if (GetMonitorFilePath(file_path) != SUCCESS) {
546 : GELOGW("AbnormalStatusMonitor, get no monitor file path");
547 : return;
548 : }
549 : int32_t fd = inotify_init1(IN_NONBLOCK); // 设置非阻塞模式，inotify_init没有非阻塞模式
550 : if (fd < 0) {
551 : GELOGW("AbnormalStatusMonitor, inotify1 init get fd = %d", fd);
552 : return;
553 : }
554 : int32_t wd = inotify_add_watch(fd, file_path.c_str(), IN_MODIFY);
555 : if (wd < 0) {
556 : GELOGW("AbnormalStatusMonitor, inotify and watch get wd = %d", wd);
557 : close(fd);
558 : return;
559 : }
560 : MonitorFileAndHeartbeatProc(file_path, fd);
561 : inotify_rm_watch(fd, wd);
562 : close(fd);
563 : return;
564 : }
565 :
566 : void AbnormalStatusHandler::Initialize() {
567 : GELOGI("AbnormalStatusMonitor, Initialize");
568 : file_monitor_flag_.store(true);
569 : run_context_ = GetThreadLocalContext();
570 : file_monitor_thread_ = std::thread([this]() {
571 : SET_THREAD_NAME(pthread_self(), "ge_dpl_fmon");
572 : GetThreadLocalContext() = run_context_;
573 : AbnormalStatusMonitorRun();
574 : });
575 : return;
576 : }
577 :
578 : void AbnormalStatusHandler::Finalize() {
579 : GELOGI("AbnormalStatusMonitor, Finalize");
580 : file_monitor_flag_.store(false);
581 : if (file_monitor_thread_.joinable()) {
582 : file_monitor_thread_.join();
583 : }
584 : deployed_models_.clear();
585 : {
586 : std::lock_guard<std::mutex> lk(abnormal_status_callback_info_.mu);
587 : abnormal_status_callback_info_.callback_list.clear();
588 : }
589 : device_state_list_.clear();
590 : abnormal_submodel_instances_name_.clear();
591 : return;
592 : }
593 :
594 : Status AbnormalStatusHandler::DynamicSchedRecoverProc(uint32_t root_model_id) {
595 : std::vector<DeployPlan::DeviceInfo> device_infos;
596 : for (auto &iter : device_state_list_) {
597 : if (!iter.second) {
598 : device_infos.push_back(iter.first);
599 : }
600 : }
601 : GE_CHK_STATUS_RET(
602 : ClearModelExceptionData(root_model_id, device_infos), // root_model_id用于数据清理，device_infos用于删除路由
603 : "AbnormalStatusMonitor, failed to do ClearModelExceptionData");
604 : GELOGI("AbnormalStatusMonitor, dynamic sched callback exec, model_id=%u", root_model_id);
605 : std::lock_guard<std::mutex> lk(abnormal_status_callback_info_.mu);
606 : if (abnormal_status_callback_info_.callback_list[root_model_id] != nullptr) {
607 : GE_CHK_STATUS_RET(abnormal_status_callback_info_.callback_list[root_model_id](kCallbackDynamicSched,
608 : abnormal_submodel_instances_name_),
609 : "AbnormalStatusMonitor, callback exec failed, root_model_id=%u", root_model_id);
610 : }
611 : return SUCCESS;
612 : }
613 :
614 : uint32_t AbnormalStatusHandler::CheckAbnormalDevices(DeployPlan::DeviceStateList &device_state_list) const {
615 : // 异常的device_info为主server所在的host(数据清理需要通过host），或者主server所在的device0异常（调度请求接收在device0）则不支持动态调度
616 : for (auto iter = device_state_list.begin(); iter != device_state_list.end(); iter++) {
617 : if (!iter->second && (iter->first.GetNodeId() == 0)) { // 主server异常
618 : if (iter->first.GetType() == static_cast<int32_t>(CPU)) { // host异常
619 : GELOGE(FAILED, "AbnormalStatusMonitor, it doesn't support redeploy, host is abnormal");
620 : return kNotSupportRedeploy;
621 : }
622 : if (is_dynamic_sched_ && (iter->first.GetDeviceId() == 0)) { // 动态调度场景device0异常
623 : GELOGI("AbnormalStatusMonitor, it doesn't support dynamic sched redeploy, device0 is abnormal");
624 : return kNotSupportDynamicSched;
625 : }
626 : }
627 : }
628 : return kNotSupportDefault;
629 : }
630 :
631 : Status AbnormalStatusHandler::RedeployProc(uint32_t root_model_id, uint32_t check_devices_flag) {
632 : GELOGI("AbnormalStatusMonitor, RedeployProc start, root_model_id=%u", root_model_id);
633 : GE_DISMISSABLE_GUARD(
634 : guard, ([this, &root_model_id]() {
635 : if (FailedHandleAbnormal(root_model_id) != SUCCESS) { // 无法恢复业务, 做相应处理:ModelIO调用返回失败
636 : GELOGE(FAILED, "AbnormalStatusMonitor, failed to do FailedHandleAbnormal");
637 : }
638 : }));
639 : GE_CHK_STATUS_RET(RedeployStart(root_model_id), "AbnormalStatusMonitor, failed to do RedeployStart");
640 : if (check_devices_flag != kNotSupportRedeploy &&
641 : IsSupportDynamicSchedRecover(root_model_id)) { // 1、动态调度降级服务
642 : GE_CHK_STATUS_RET(DynamicSchedRecoverProc(root_model_id),
643 : "AbnormalStatusMonitor, failed to do dynamic sched recover proc");
644 : GELOGI("AbnormalStatusMonitor, redeploy success");
645 : GE_DISMISS_GUARD(guard);
646 : return SUCCESS;
647 : }
648 : // 2、重部署恢复业务
649 : // 3、无法恢复业务, 做相应处理:ModelIO调用返回失败:见FailedHandleAbnormal
650 : GELOGE(FAILED, "AbnormalStatusMonitor, root_model_id=%u can't recover by redeploying", root_model_id);
651 : return FAILED;
652 : }
653 :
654 : Status AbnormalStatusHandler::HeartbeatMonitorProc() {
655 : GELOGI("AbnormalStatusMonitor, HeartbeatMonitorProc start");
656 : bool is_new_abnormal = false;
657 : DeployPlan::DeviceStateList device_state_list;
658 : ParseHeartbeatAbnormalInfo(is_new_abnormal, device_state_list);
659 : if (is_new_abnormal) {
660 : PreHandleAbnormalInfo();
661 : auto check_devices_flag = CheckAbnormalDevices(device_state_list);
662 : if (check_devices_flag == kNotSupportRedeploy) {
663 : GELOGE(FAILED, "AbnormalStatusMonitor, it(cause by abnormal process) can't recover by redeploying");
664 : }
665 : GE_CHK_STATUS_RET(ParallelAbnormalStatusHandle(check_devices_flag),
666 : "AbnormalStatusMonitor, failed to do ParallelAbnormalStatusHandle");
667 : return SUCCESS;
668 : }
669 : GELOGI("AbnormalStatusMonitor, HeartbeatMonitorProc end, no new abnormal process");
670 : return SUCCESS;
671 : }
672 :
673 : AbnormalStatusHandler::AbnormalStatusHandler(std::mutex &mu) : mu_(mu) {
674 : GELOGI("AbnormalStatusHandler, start.");
675 : }
676 :
677 : AbnormalStatusHandler::~AbnormalStatusHandler() {
678 : if (file_monitor_flag_) {
679 : GELOGW("AbnormalStatusHandler, file_monitor_thread_ is not stopped");
680 : Finalize();
681 : }
682 : }
683 :
684 : void AbnormalStatusHandler::IncDeployingRootModelNum() {
685 : deploying_root_model_cnt_++;
686 : }
687 :
688 : void AbnormalStatusHandler::DecreaseDeployingRootModelNum() {
689 : deploying_root_model_cnt_--;
690 : }
691 :
692 10 : bool AbnormalStatusHandler::IsDeployingRootModel() const {
693 : return deploying_root_model_cnt_.load() != 0U;
694 : }
695 :
696 10 : bool AbnormalStatusHandler::IsAllCallbackInitFinished() const {
697 : std::lock_guard<std::mutex> lk(abnormal_status_callback_info_.mu);
698 : return abnormal_status_callback_info_.callback_list.size() == deployed_models_.size();
699 : }
700 :
701 : void AbnormalStatusHandler::DelCallback(const uint32_t root_model_id) {
702 : std::lock_guard<std::mutex> lk(abnormal_status_callback_info_.mu);
703 : abnormal_status_callback_info_.callback_list.erase(root_model_id);
704 : }
705 :
706 : DeployPlan::AbnormalStatusCallbackInfo *AbnormalStatusHandler::GetAbnormalStatusCallbackInfo() {
707 : return &abnormal_status_callback_info_;
708 : }
709 :
710 : void AbnormalStatusHandler::SetDynamicSchedFlag(bool flag) {
711 : is_dynamic_sched_ = flag;
712 : }
713 :
714 : void AbnormalStatusHandler::AddDeployedModelInfo(uint32_t model_id,
715 : const DeployPlan::ModelDeployInfo &model_deploy_infos,
716 : const std::set<int32_t> &deployed_remote_nodes) {
717 : deployed_models_[model_id].model_id = model_id;
718 : deployed_models_[model_id].model_deploy_infos = model_deploy_infos;
719 : deployed_models_[model_id].deployed_remote_nodes = deployed_remote_nodes;
720 : }
721 :
722 : void AbnormalStatusHandler::DelDeployedModelInfo(uint32_t model_id) {
723 : deployed_models_.erase(model_id);
724 : }
725 :
726 : Status AbnormalStatusHandler::ParallelClearData(const std::pair<uint32_t, std::set<uint32_t>> &need_clear_root_models,
727 : const std::vector<DeployPlan::DeviceInfo> &device_infos,
728 : const int32_t type) const {
729 : auto clear_num = need_clear_root_models.second.size();
730 : if (clear_num > 1U) {
731 : ThreadPool pool("ge_dpl_rdc", static_cast<uint32_t>(clear_num), false);
732 : std::vector<std::future<Status>> fut_rets;
733 : for (const auto &node_id : need_clear_root_models.second) {
734 : const auto &model_id = need_clear_root_models.first;
735 : auto fut = pool.commit([&node_id, &model_id, &device_infos, type]() -> Status {
736 : GE_CHK_STATUS_RET_NOLOG(
737 : HeterogeneousModelDeployer::ClearNodelExceptionData(node_id, model_id, device_infos, type));
738 : return SUCCESS;
739 : });
740 : fut_rets.emplace_back(std::move(fut));
741 : }
742 : for (auto &fut : fut_rets) {
743 : GE_CHK_STATUS_RET_NOLOG(fut.get());
744 : }
745 : } else {
746 : for (const auto &node_id : need_clear_root_models.second) {
747 : GE_CHK_STATUS_RET_NOLOG(HeterogeneousModelDeployer::ClearNodelExceptionData(node_id, need_clear_root_models.first,
748 : device_infos, type));
749 : }
750 : }
751 : return SUCCESS;
752 : }
753 :
754 : // 如果node整个坏了或者host坏了就不要执行清理动作了
755 : Status AbnormalStatusHandler::ClearModelExceptionData(uint32_t root_model_id,
756 : const std::vector<DeployPlan::DeviceInfo> &device_infos) {
757 : for (const auto &device_info : device_infos) {
758 : GELOGI("AbnormalStatusMonitor, Exception root_model_id=%u, device info: device = %s.", root_model_id,
759 : device_info.GetDesc().c_str());
760 : }
761 : std::pair<uint32_t, std::set<uint32_t>> need_clear_root_models; // model_id, node_ids
762 : need_clear_root_models.first = root_model_id;
763 : const int32_t local_node_id = ResourceManager::GetInstance().GetLocalNodeId(); // local node也要一同清理
764 : {
765 : std::lock_guard<std::mutex> lk(mu_);
766 : auto &model_info_iter = deployed_models_[root_model_id];
767 : auto clear_nodes = model_info_iter.deployed_remote_nodes;
768 : clear_nodes.emplace(local_node_id); // local也要算上
769 : for (auto clear_node : clear_nodes) {
770 : need_clear_root_models.second.emplace(clear_node);
771 : }
772 : }
773 : for (auto &node_id : need_clear_root_models.second) {
774 : GELOGI("AbnormalStatusMonitor, Exception model info: root_model_id = %u, node_id = %u.",
775 : need_clear_root_models.first, node_id);
776 : }
777 :
778 : GE_CHK_STATUS_RET_NOLOG(ParallelClearData(need_clear_root_models, device_infos, EXCEPTION_HANDLE_STOP));
779 : GE_CHK_STATUS_RET_NOLOG(ParallelClearData(need_clear_root_models, device_infos, EXCEPTION_HANDLE_CLEAR));
780 : return SUCCESS;
781 : }
782 : } // namespace ge
|
