dispatch_manager.cpp 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380
  1. //
  2. // Created by huli on 2020/7/20.
  3. //
  4. #include "dispatch_manager.h"
  5. #include <cstdlib>
  6. Dispatch_manager::Dispatch_manager()
  7. {
  8. m_dispatch_manager_status = E_DISPATCH_MANAGER_UNKNOW;
  9. m_dispatch_id = -1;
  10. m_catcher_size = 0;
  11. m_carrier_size = 0;
  12. m_passageway_size = 0;
  13. m_process_thread_queue = nullptr;
  14. m_dispatch_manager_thread = nullptr;
  15. }
  16. Dispatch_manager::~Dispatch_manager()
  17. {
  18. dispatch_manager_uninit();
  19. }
  20. //调度管理 初始化
  21. Error_manager Dispatch_manager::dispatch_manager_init(int dispatch_id)
  22. {
  23. m_dispatch_id = dispatch_id;
  24. m_catcher_size = 2;
  25. for (int i = 0; i < m_catcher_size; ++i)
  26. {
  27. std::shared_ptr<Dispatch_device_base> tp_dispatch_device_base(new Catcher);
  28. Catcher* tp_catcher = (Catcher*)tp_dispatch_device_base.get();
  29. tp_catcher->dispatch_device_base_init(i);
  30. m_catcher_map[i] = tp_dispatch_device_base;
  31. }
  32. m_carrier_size = 3;
  33. for (int i = 0; i < m_carrier_size; ++i)
  34. {
  35. std::shared_ptr<Dispatch_device_base> tp_dispatch_device_base(new Carrier);
  36. Carrier* tp_carrier = (Carrier*)tp_dispatch_device_base.get();
  37. tp_carrier->dispatch_device_base_init(i);
  38. m_carrier_map[i] = tp_dispatch_device_base;
  39. }
  40. m_passageway_size = 8;
  41. for (int i = 0; i < m_passageway_size; ++i)
  42. {
  43. std::shared_ptr<Dispatch_device_base> tp_dispatch_device_base(new Passageway);
  44. Passageway* tp_passageway = (Passageway*)tp_dispatch_device_base.get();
  45. tp_passageway->dispatch_device_base_init(i);
  46. m_passageway_map[i] = tp_dispatch_device_base;
  47. }
  48. //创建线程池
  49. if (m_process_thread_queue == nullptr) {
  50. m_process_thread_queue = tq::TQFactory::CreateDefaultQueue();
  51. m_process_thread_queue->Start(48);
  52. }
  53. m_dispatch_manager_status = E_DISPATCH_MANAGER_READY;
  54. return Error_code::SUCCESS;
  55. }
  56. //调度管理 反初始化
  57. Error_manager Dispatch_manager::dispatch_manager_uninit()
  58. {
  59. if (m_dispatch_manager_thread)
  60. {
  61. m_dispatch_manager_condition.kill_all();
  62. }
  63. if (m_dispatch_manager_thread)
  64. {
  65. m_dispatch_manager_thread->join();
  66. delete m_dispatch_manager_thread;
  67. m_dispatch_manager_thread = NULL;
  68. }
  69. //等待线程池完成
  70. if(m_process_thread_queue!=nullptr) {
  71. m_process_thread_queue->WaitForFinish();
  72. m_process_thread_queue->Stop();
  73. m_process_thread_queue = nullptr;
  74. }
  75. m_dispatch_manager_status = E_DISPATCH_MANAGER_UNKNOW;
  76. m_dispatch_id = -1;
  77. // for (int i = 0; i < m_catcher_size; ++i)
  78. // {
  79. // m_catcher_vector[i]->dispatch_device_base_uninit();
  80. // delete(m_catcher_vector[i]);
  81. // }
  82. // m_catcher_size = 0;
  83. // m_catcher_vector.clear();
  84. //
  85. // for (int i = 0; i < m_carrier_size; ++i)
  86. // {
  87. // m_carrier_vector[i]->dispatch_device_base_uninit();
  88. // delete(m_carrier_vector[i]);
  89. // }
  90. // m_carrier_size = 0;
  91. // m_carrier_vector.clear();
  92. //
  93. // for (int i = 0; i < m_passageway_size; ++i)
  94. // {
  95. // m_passageway_vector[i]->dispatch_device_base_uninit();
  96. // delete(m_passageway_vector[i]);
  97. // }
  98. // m_passageway_size = 0;
  99. // m_passageway_vector.clear();
  100. return Error_code::SUCCESS;
  101. }
  102. //对外的接口函数,负责接受并处理任务单,
  103. Error_manager Dispatch_manager::execute_task(Dispatch_manager::Dispatch_motion_direction dispatch_motion_direction)
  104. {
  105. return Error_code::SUCCESS;
  106. // std::this_thread::sleep_for(std::chrono::seconds(rand()%3+3));
  107. if ( dispatch_motion_direction == E_PICKUP_CAR )
  108. {
  109. return Error_code::SUCCESS;
  110. }
  111. // return Error_code::SUCCESS;
  112. // srand(0);
  113. unsigned int t_probability = rand();
  114. if ( t_probability%100 >=20 )
  115. {
  116. return Error_code::SUCCESS;
  117. }
  118. else
  119. {
  120. return Error_manager(Error_code::ERROR, Error_level::MINOR_ERROR,
  121. " Dispatch_manager::execute_task() error ");
  122. }
  123. }
  124. //检查能否执行消息指令
  125. Error_manager Dispatch_manager::check_execute_msg(Communication_message* p_msg)
  126. {
  127. Error_manager t_error = Dispatch_manager::get_instance_references().check_status();
  128. if ( t_error == Error_code::SUCCESS )
  129. {
  130. return Error_code::SUCCESS;
  131. }
  132. else if (t_error.get_error_level() == NEGLIGIBLE_ERROR)//一级故障,轻微故障,
  133. {
  134. std::cout << "Dispatch_manager _is_busy , " << std::endl;
  135. //返回繁忙之后, 通信模块1秒后再次调用check
  136. return Error_code::COMMUNICATION_EXCUTER_IS_BUSY;
  137. }
  138. switch ( p_msg->get_message_type() )
  139. {
  140. case Communication_message::Message_type::eDispatch_request_msg:
  141. {
  142. message::Dispatch_request_msg t_dispatch_request_msg;
  143. //针对消息类型, 对消息进行二次解析
  144. if (t_dispatch_request_msg.ParseFromString(p_msg->get_message_buf()))
  145. {
  146. //创建一条答复消息
  147. message::Dispatch_response_msg t_dispatch_response_msg;
  148. t_dispatch_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eDispatch_response_msg);
  149. t_dispatch_response_msg.mutable_base_info()->set_timeout_ms(5000);
  150. t_dispatch_response_msg.mutable_base_info()->set_sender(message::Communicator::eDispatch_mamager);
  151. t_dispatch_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  152. t_dispatch_response_msg.set_command_key(t_dispatch_request_msg.command_key());
  153. t_dispatch_response_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  154. t_dispatch_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  155. t_dispatch_response_msg.mutable_error_manager()->set_error_description(t_error.get_error_description());
  156. std::string t_msg = t_dispatch_response_msg.SerializeAsString();
  157. System_communication::get_instance_references().encapsulate_msg(t_msg);
  158. LOG(INFO) << " System_executor::check_execute_msg executer status error "<< this;
  159. return t_error;
  160. }
  161. else
  162. {
  163. LOG(INFO) << " System_executor::check_executer Second analysis ERROR "<< this;
  164. return Error_manager(Error_code::SYSTEM_EXECUTOR_PARSE_ERROR, Error_level::MINOR_ERROR,
  165. " message::Measure_request_msg ParseFromString error ");
  166. }
  167. }
  168. case Communication_message::Message_type::eDispatch_plan_response_msg:
  169. {
  170. return Error_code::SUCCESS;
  171. break;
  172. }
  173. case Communication_message::Message_type::eDispatch_control_request_msg:
  174. {
  175. return Error_code::SUCCESS;
  176. break;
  177. }
  178. default :
  179. //无效的消息,
  180. return Error_manager(Error_code::INVALID_MESSAGE, Error_level::NEGLIGIBLE_ERROR,
  181. " INVALID_MESSAGE error ");
  182. break;
  183. }
  184. }
  185. //检查状态
  186. Error_manager Dispatch_manager::check_status()
  187. {
  188. if ( m_dispatch_manager_status == E_DISPATCH_MANAGER_READY )
  189. {
  190. return Error_code::SUCCESS;
  191. }
  192. else if ( m_dispatch_manager_status == E_DISPATCH_MANAGER_STORE || m_dispatch_manager_status == E_DISPATCH_MANAGER_STORE )
  193. {
  194. return Error_manager(Error_code::DISPATCH_MANAGER_STATUS_BUSY, Error_level::NEGLIGIBLE_ERROR,
  195. " Dispatch_manager::check_status() error ");
  196. }
  197. else
  198. {
  199. return Error_manager(Error_code::DISPATCH_MANAGER_STATUS_ERROR, Error_level::MINOR_ERROR,
  200. " Dispatch_manager::check_status() error ");
  201. }
  202. return Error_code::SUCCESS;
  203. }
  204. //调度模块 //执行搬运请求(主控->调度管理)
  205. Error_manager Dispatch_manager::execute_for_dispatch_request_msg(message::Dispatch_request_msg& dispatch_request_msg)
  206. {
  207. Error_manager t_error;
  208. //创建调度流程
  209. Dispatch_process * tp_dispatch_process(new Dispatch_process);
  210. t_error = tp_dispatch_process->Dispatch_process_init(dispatch_request_msg);
  211. if ( t_error != Error_code::SUCCESS )
  212. {
  213. return t_error;
  214. }
  215. //流程缓存到map
  216. if ( tp_dispatch_process->m_dispatch_process_type == Dispatch_process::Dispatch_process_type::DISPATCH_PROCESS_STORE )
  217. {
  218. m_key_to_dispatch_process_store_map[tp_dispatch_process->m_command_key] = tp_dispatch_process;
  219. }
  220. else if( tp_dispatch_process->m_dispatch_process_type == Dispatch_process::Dispatch_process_type::DISPATCH_PROCESS_PICKUP )
  221. {
  222. m_key_to_dispatch_process_pickup_map[tp_dispatch_process->m_command_key] = tp_dispatch_process;
  223. }
  224. else
  225. {
  226. return Error_manager(Error_code::DISPATCH_PROCESS_TASK_STATUS_ERROR, Error_level::MINOR_ERROR,
  227. " tp_dispatch_process->m_dispatch_process_type PARAMRTER ERROR ");
  228. }
  229. //把流程添加到 调度线程池里面
  230. m_process_thread_queue->AddTask(tp_dispatch_process);
  231. return Error_code::SUCCESS;
  232. }
  233. //调度模块 //调度总规划的答复(调度算法->调度管理)
  234. Error_manager Dispatch_manager::execute_for_dispatch_plan_response_msg(message::Dispatch_plan_response_msg &dispatch_plan_response_msg)
  235. {
  236. Error_manager t_error;
  237. //使用 command_key 找到对应的流程
  238. std::string t_command_key = dispatch_plan_response_msg.command_key();
  239. for (auto iter = m_key_to_dispatch_process_store_map.begin(); iter != m_key_to_dispatch_process_store_map.end(); ++iter)
  240. {
  241. if ( 0 == iter->first.compare(0, COMMAND_KEY_DEFAULT_LENGTH, t_command_key, 0, COMMAND_KEY_DEFAULT_LENGTH) )
  242. {
  243. t_error = iter->second->check_process_ready() ;
  244. if ( t_error != Error_code::SUCCESS )
  245. {
  246. return t_error;
  247. }
  248. //把消息 写入流程
  249. iter->second->m_dispatch_plan_response_msg = dispatch_plan_response_msg;
  250. return Error_code::SUCCESS;
  251. }
  252. }
  253. for (auto iter = m_key_to_dispatch_process_pickup_map.begin(); iter != m_key_to_dispatch_process_pickup_map.end(); ++iter)
  254. {
  255. if ( 0 == iter->first.compare(0, COMMAND_KEY_DEFAULT_LENGTH, t_command_key, 0, COMMAND_KEY_DEFAULT_LENGTH) )
  256. {
  257. t_error = iter->second->check_process_ready() ;
  258. if ( t_error != Error_code::SUCCESS )
  259. {
  260. return t_error;
  261. }
  262. //把消息 写入流程
  263. iter->second->m_dispatch_plan_response_msg = dispatch_plan_response_msg;
  264. return Error_code::SUCCESS;
  265. }
  266. }
  267. return Error_code::SUCCESS;
  268. }
  269. //调度模块 //调度控制的任务请求(调度算法->调度管理)
  270. Error_manager Dispatch_manager::execute_for_dispatch_control_request_msg(message::Dispatch_control_request_msg &dispatch_control_request_msg)
  271. {
  272. Error_manager t_error;
  273. //使用 command_key 找到对应的流程
  274. std::string t_command_key = dispatch_control_request_msg.command_key();
  275. for (auto iter = m_key_to_dispatch_process_store_map.begin(); iter != m_key_to_dispatch_process_store_map.end(); ++iter)
  276. {
  277. if ( 0 == iter->first.compare(0, COMMAND_KEY_DEFAULT_LENGTH, t_command_key, 0, COMMAND_KEY_DEFAULT_LENGTH) )
  278. {
  279. t_error = iter->second->check_process_ready() ;
  280. if ( t_error != Error_code::SUCCESS )
  281. {
  282. return t_error;
  283. }
  284. //把消息 写入流程
  285. iter->second->m_dispatch_control_request_msg = dispatch_control_request_msg;
  286. return Error_code::SUCCESS;
  287. }
  288. }
  289. for (auto iter = m_key_to_dispatch_process_pickup_map.begin(); iter != m_key_to_dispatch_process_pickup_map.end(); ++iter)
  290. {
  291. if ( 0 == iter->first.compare(0, COMMAND_KEY_DEFAULT_LENGTH, t_command_key, 0, COMMAND_KEY_DEFAULT_LENGTH) )
  292. {
  293. t_error = iter->second->check_process_ready() ;
  294. if ( t_error != Error_code::SUCCESS )
  295. {
  296. return t_error;
  297. }
  298. //把消息 写入流程
  299. iter->second->m_dispatch_control_request_msg = dispatch_control_request_msg;
  300. return Error_code::SUCCESS;
  301. }
  302. }
  303. return Error_code::SUCCESS;
  304. }
  305. //在流程的map 里面释放指定的流程
  306. Error_manager Dispatch_manager::release_dispatch_process(std::string command_key)
  307. {
  308. std::unique_lock<std::mutex> t_lock(m_lock);
  309. auto iter1 = m_key_to_dispatch_process_store_map.find(command_key);
  310. if ( iter1 != m_key_to_dispatch_process_store_map.end())
  311. {
  312. m_key_to_dispatch_process_store_map.erase(iter1);
  313. }
  314. auto iter2 = m_key_to_dispatch_process_pickup_map.find(command_key);
  315. if ( iter2 != m_key_to_dispatch_process_pickup_map.end())
  316. {
  317. m_key_to_dispatch_process_pickup_map.erase(iter2);
  318. }
  319. return Error_code::SUCCESS;
  320. }
  321. Dispatch_manager::Dispatch_manager_status Dispatch_manager::get_dispatch_manager_status()
  322. {
  323. return m_dispatch_manager_status;
  324. }
  325. int Dispatch_manager::get_dispatch_id()
  326. {
  327. return m_dispatch_id;
  328. }
  329. void Dispatch_manager::set_dispatch_id(int dispatch_id)
  330. {
  331. m_dispatch_id = dispatch_id;
  332. }