// // Created by huli on 2020/7/20. // #include "dispatch_manager.h" #include Dispatch_manager::Dispatch_manager() { m_dispatch_manager_status = E_DISPATCH_MANAGER_UNKNOW; m_dispatch_id = -1; m_catcher_size = 0; m_carrier_size = 0; m_passageway_size = 0; m_process_thread_queue = nullptr; m_dispatch_manager_thread = nullptr; } Dispatch_manager::~Dispatch_manager() { dispatch_manager_uninit(); } //调度管理 初始化 Error_manager Dispatch_manager::dispatch_manager_init(int dispatch_id) { LOG(INFO) << " ----Dispatch_manager::dispatch_manager_init----- "<< this; m_dispatch_id = dispatch_id; m_catcher_size = 2; for (int i = 1; i < m_catcher_size; ++i) { std::shared_ptr tp_dispatch_device_base(new Catcher); Catcher* tp_catcher = (Catcher*)tp_dispatch_device_base.get(); tp_catcher->dispatch_device_base_init(i); m_catcher_map[i] = tp_dispatch_device_base; } m_carrier_size = 3; for (int i = 0; i < m_carrier_size; ++i) { std::shared_ptr tp_dispatch_device_base(new Carrier); Carrier* tp_carrier = (Carrier*)tp_dispatch_device_base.get(); tp_carrier->dispatch_device_base_init(i); m_carrier_map[i] = tp_dispatch_device_base; } // m_passageway_size = 8; // for (int i = 0; i < m_passageway_size; ++i) // { // std::shared_ptr tp_dispatch_device_base(new Passageway); // Passageway* tp_passageway = (Passageway*)tp_dispatch_device_base.get(); // tp_passageway->dispatch_device_base_init(i); // m_passageway_map[i] = tp_dispatch_device_base; // } //创建线程池 if (m_process_thread_queue == nullptr) { m_process_thread_queue = tq::TQFactory::CreateDefaultQueue(); m_process_thread_queue->Start(48); } m_dispatch_manager_status = E_DISPATCH_MANAGER_READY; return Error_code::SUCCESS; } //调度管理 反初始化 Error_manager Dispatch_manager::dispatch_manager_uninit() { if (m_dispatch_manager_thread) { m_dispatch_manager_condition.kill_all(); } if (m_dispatch_manager_thread) { m_dispatch_manager_thread->join(); delete m_dispatch_manager_thread; m_dispatch_manager_thread = NULL; } //等待线程池完成 if(m_process_thread_queue!=nullptr) { m_process_thread_queue->WaitForFinish(); m_process_thread_queue->Stop(); m_process_thread_queue = nullptr; } m_dispatch_manager_status = E_DISPATCH_MANAGER_UNKNOW; m_dispatch_id = -1; // for (int i = 0; i < m_catcher_size; ++i) // { // m_catcher_vector[i]->dispatch_device_base_uninit(); // delete(m_catcher_vector[i]); // } // m_catcher_size = 0; // m_catcher_vector.clear(); // // for (int i = 0; i < m_carrier_size; ++i) // { // m_carrier_vector[i]->dispatch_device_base_uninit(); // delete(m_carrier_vector[i]); // } // m_carrier_size = 0; // m_carrier_vector.clear(); // // for (int i = 0; i < m_passageway_size; ++i) // { // m_passageway_vector[i]->dispatch_device_base_uninit(); // delete(m_passageway_vector[i]); // } // m_passageway_size = 0; // m_passageway_vector.clear(); return Error_code::SUCCESS; } //对外的接口函数,负责接受并处理任务单, Error_manager Dispatch_manager::execute_task(Dispatch_manager::Dispatch_motion_direction dispatch_motion_direction) { return Error_code::SUCCESS; // std::this_thread::sleep_for(std::chrono::seconds(rand()%3+3)); if ( dispatch_motion_direction == E_PICKUP_CAR ) { return Error_code::SUCCESS; } // return Error_code::SUCCESS; // srand(0); unsigned int t_probability = rand(); if ( t_probability%100 >=20 ) { return Error_code::SUCCESS; } else { return Error_manager(Error_code::ERROR, Error_level::MINOR_ERROR, " Dispatch_manager::execute_task() error "); } } //检查能否执行消息指令 Error_manager Dispatch_manager::check_execute_msg(Communication_message* p_msg) { Error_manager t_error = Dispatch_manager::get_instance_references().check_status(); if ( t_error == Error_code::SUCCESS ) { return Error_code::SUCCESS; } else if (t_error.get_error_level() == NEGLIGIBLE_ERROR)//一级故障,轻微故障, { std::cout << "Dispatch_manager _is_busy , " << std::endl; //返回繁忙之后, 通信模块1秒后再次调用check return Error_code::COMMUNICATION_EXCUTER_IS_BUSY; } switch ( p_msg->get_message_type() ) { case Communication_message::Message_type::eDispatch_request_msg: { message::Dispatch_request_msg t_dispatch_request_msg; //针对消息类型, 对消息进行二次解析 if (t_dispatch_request_msg.ParseFromString(p_msg->get_message_buf())) { //创建一条答复消息 message::Dispatch_response_msg t_dispatch_response_msg; t_dispatch_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eDispatch_response_msg); t_dispatch_response_msg.mutable_base_info()->set_timeout_ms(5000); t_dispatch_response_msg.mutable_base_info()->set_sender(message::Communicator::eDispatch_manager); t_dispatch_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain); t_dispatch_response_msg.set_command_key(t_dispatch_request_msg.command_key()); t_dispatch_response_msg.mutable_error_manager()->set_error_code(t_error.get_error_code()); t_dispatch_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level()); t_dispatch_response_msg.mutable_error_manager()->set_error_description(t_error.get_error_description()); std::string t_msg = t_dispatch_response_msg.SerializeAsString(); System_communication::get_instance_references().encapsulate_msg(t_msg); LOG(INFO) << " System_executor::check_execute_msg executer status error "<< this; return t_error; } else { LOG(INFO) << " System_executor::check_executer Second analysis ERROR "<< this; return Error_manager(Error_code::SYSTEM_EXECUTOR_PARSE_ERROR, Error_level::MINOR_ERROR, " message::Measure_request_msg ParseFromString error "); } } case Communication_message::Message_type::eDispatch_plan_response_msg: { return Error_code::SUCCESS; break; } case Communication_message::Message_type::eDispatch_control_request_msg: { return Error_code::SUCCESS; break; } default : //无效的消息, return Error_manager(Error_code::INVALID_MESSAGE, Error_level::NEGLIGIBLE_ERROR, " INVALID_MESSAGE error "); break; } } //检查状态 Error_manager Dispatch_manager::check_status() { if ( m_dispatch_manager_status == E_DISPATCH_MANAGER_READY ) { return Error_code::SUCCESS; } else if ( m_dispatch_manager_status == E_DISPATCH_MANAGER_STORE || m_dispatch_manager_status == E_DISPATCH_MANAGER_STORE ) { return Error_manager(Error_code::DISPATCH_MANAGER_STATUS_BUSY, Error_level::NEGLIGIBLE_ERROR, " Dispatch_manager::check_status() error "); } else { return Error_manager(Error_code::DISPATCH_MANAGER_STATUS_ERROR, Error_level::MINOR_ERROR, " Dispatch_manager::check_status() error "); } return Error_code::SUCCESS; } //调度模块 //执行搬运请求(主控->调度管理) Error_manager Dispatch_manager::execute_for_dispatch_request_msg(message::Dispatch_request_msg& dispatch_request_msg) { LOG(INFO) << " ---Dispatch_manager::execute_for_dispatch_request_msg--- "<< this; LOG(INFO) << " dispatch_request_msg->m_command_key = "<Dispatch_process_init(dispatch_request_msg); if ( t_error != Error_code::SUCCESS ) { delete(tp_dispatch_process); return t_error; } std::unique_lock t_lock(m_lock); //流程缓存到map if ( tp_dispatch_process->m_dispatch_process_type == Dispatch_process::Dispatch_process_type::DISPATCH_PROCESS_STORE ) { m_key_to_dispatch_process_store_map[tp_dispatch_process->m_command_key] = tp_dispatch_process; } else if( tp_dispatch_process->m_dispatch_process_type == Dispatch_process::Dispatch_process_type::DISPATCH_PROCESS_PICKUP ) { m_key_to_dispatch_process_pickup_map[tp_dispatch_process->m_command_key] = tp_dispatch_process; } else { delete(tp_dispatch_process); return Error_manager(Error_code::DISPATCH_PROCESS_TASK_STATUS_ERROR, Error_level::MINOR_ERROR, " tp_dispatch_process->m_dispatch_process_type PARAMRTER ERROR "); } //把流程添加到 调度线程池里面, 里面的线程池会自动回收调度流程的内存 m_process_thread_queue->AddTask(tp_dispatch_process); return Error_code::SUCCESS; } //调度模块 //调度总规划的答复(调度算法->调度管理) Error_manager Dispatch_manager::execute_for_dispatch_plan_response_msg(message::Dispatch_plan_response_msg &dispatch_plan_response_msg) { LOG(INFO) << " ---Dispatch_manager::execute_for_dispatch_plan_response_msg--- "<< this; LOG(INFO) << " dispatch_plan_response_msg->m_command_key = "< t_lock(m_lock); //使用 command_key 找到对应的流程 std::string t_command_key = dispatch_plan_response_msg.command_key(); for (auto iter = m_key_to_dispatch_process_store_map.begin(); iter != m_key_to_dispatch_process_store_map.end(); ++iter) { if ( 0 == iter->first.compare(0, COMMAND_KEY_DEFAULT_LENGTH, t_command_key, 0, COMMAND_KEY_DEFAULT_LENGTH) ) { t_error = iter->second->check_process_ready() ; if ( t_error != Error_code::SUCCESS ) { return t_error; } //把消息 写入流程 std::unique_lock t_lock(iter->second->m_lock); iter->second->m_dispatch_plan_response_msg = dispatch_plan_response_msg; return Error_code::SUCCESS; } } for (auto iter = m_key_to_dispatch_process_pickup_map.begin(); iter != m_key_to_dispatch_process_pickup_map.end(); ++iter) { if ( 0 == iter->first.compare(0, COMMAND_KEY_DEFAULT_LENGTH, t_command_key, 0, COMMAND_KEY_DEFAULT_LENGTH) ) { t_error = iter->second->check_process_ready() ; if ( t_error != Error_code::SUCCESS ) { return t_error; } //把消息 写入流程 std::unique_lock t_lock(iter->second->m_lock); iter->second->m_dispatch_plan_response_msg = dispatch_plan_response_msg; return Error_code::SUCCESS; } } return Error_code::SUCCESS; } //调度模块 //调度控制的任务请求(调度算法->调度管理) Error_manager Dispatch_manager::execute_for_dispatch_control_request_msg(message::Dispatch_control_request_msg &dispatch_control_request_msg) { LOG(INFO) << " ---Dispatch_manager::execute_for_dispatch_control_request_msg--- "<< this; LOG(INFO) << " dispatch_control_request_msg->m_command_key = "< t_lock(m_lock); //使用 command_key 找到对应的流程 std::string t_command_key = dispatch_control_request_msg.command_key(); for (auto iter = m_key_to_dispatch_process_store_map.begin(); iter != m_key_to_dispatch_process_store_map.end(); ++iter) { if ( 0 == iter->first.compare(0, COMMAND_KEY_DEFAULT_LENGTH, t_command_key, 0, COMMAND_KEY_DEFAULT_LENGTH) ) { t_error = iter->second->check_process_ready() ; if ( t_error != Error_code::SUCCESS ) { return t_error; } //把消息 写入流程 std::unique_lock t_lock(iter->second->m_lock); int t_dispatch_device_type = dispatch_control_request_msg.dispatch_device_type(); if ( iter->second->m_dispatch_control_node_map.find(t_dispatch_device_type) == iter->second->m_dispatch_control_node_map.end() ) { Dispatch_process::Dispatch_control_node t_dispatch_control_node; iter->second->m_dispatch_control_node_map[t_dispatch_device_type] = t_dispatch_control_node; iter->second->m_dispatch_control_node_map[t_dispatch_device_type].m_dispatch_control_status = Dispatch_process::Dispatch_control_status::DISPATCH_CONTROL_READY; } iter->second->m_dispatch_control_node_map[t_dispatch_device_type].m_dispatch_control_request_msg = dispatch_control_request_msg; // iter->second->m_dispatch_control_node_map[t_dispatch_device_type].m_dispatch_control_status = Dispatch_process::Dispatch_control_status::DISPATCH_CONTROL_READY; return Error_code::SUCCESS; } } for (auto iter = m_key_to_dispatch_process_pickup_map.begin(); iter != m_key_to_dispatch_process_pickup_map.end(); ++iter) { if ( 0 == iter->first.compare(0, COMMAND_KEY_DEFAULT_LENGTH, t_command_key, 0, COMMAND_KEY_DEFAULT_LENGTH) ) { t_error = iter->second->check_process_ready() ; if ( t_error != Error_code::SUCCESS ) { return t_error; } //把消息 写入流程 std::unique_lock t_lock(iter->second->m_lock); int t_dispatch_device_type = dispatch_control_request_msg.dispatch_device_type(); if ( iter->second->m_dispatch_control_node_map.find(t_dispatch_device_type) == iter->second->m_dispatch_control_node_map.end() ) { Dispatch_process::Dispatch_control_node t_dispatch_control_node; iter->second->m_dispatch_control_node_map[t_dispatch_device_type] = t_dispatch_control_node; iter->second->m_dispatch_control_node_map[t_dispatch_device_type].m_dispatch_control_status = Dispatch_process::Dispatch_control_status::DISPATCH_CONTROL_READY; } iter->second->m_dispatch_control_node_map[t_dispatch_device_type].m_dispatch_control_request_msg = dispatch_control_request_msg; // iter->second->m_dispatch_control_node_map[t_dispatch_device_type].m_dispatch_control_status = Dispatch_process::Dispatch_control_status::DISPATCH_CONTROL_READY; return Error_code::SUCCESS; } } return Error_code::SUCCESS; } //在流程的map 里面释放指定的流程 Error_manager Dispatch_manager::release_dispatch_process(std::string command_key) { std::unique_lock t_lock(m_lock); auto iter1 = m_key_to_dispatch_process_store_map.find(command_key); if ( iter1 != m_key_to_dispatch_process_store_map.end()) { m_key_to_dispatch_process_store_map.erase(iter1); } auto iter2 = m_key_to_dispatch_process_pickup_map.find(command_key); if ( iter2 != m_key_to_dispatch_process_pickup_map.end()) { m_key_to_dispatch_process_pickup_map.erase(iter2); } return Error_code::SUCCESS; } Dispatch_manager::Dispatch_manager_status Dispatch_manager::get_dispatch_manager_status() { return m_dispatch_manager_status; } int Dispatch_manager::get_dispatch_id() { return m_dispatch_id; } void Dispatch_manager::set_dispatch_id(int dispatch_id) { m_dispatch_id = dispatch_id; }