system_executor.cpp 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468
  1. //
  2. // Created by huli on 2020/7/2.
  3. //
  4. #include "system_executor.h"
  5. #include "../message/measure_message.pb.h"
  6. #include "../laser/laser_manager.h"
  7. #include "../locate/locate_manager.h"
  8. #include "../system/system_communication.h"
  9. System_executor::System_executor()
  10. {
  11. }
  12. System_executor::~System_executor()
  13. {
  14. system_executor_uninit();
  15. }
  16. //初始化
  17. Error_manager System_executor::system_executor_init(int threads_size, int terminal_id)
  18. {
  19. m_thread_pool.thread_pool_init(threads_size);
  20. m_system_executor_status = SYSTEM_EXECUTOR_READY;
  21. m_terminal_id = terminal_id;
  22. return Error_code::SUCCESS;
  23. }
  24. //反初始化
  25. Error_manager System_executor::system_executor_uninit()
  26. {
  27. m_thread_pool.thread_pool_uninit();
  28. m_system_executor_status = SYSTEM_EXECUTOR_UNKNOW;
  29. return Error_code::SUCCESS;
  30. }
  31. //检查消息是否有效, 主要检查消息类型和接受者, 判断这条消息是不是给我的.
  32. Error_manager System_executor::check_msg(Communication_message* p_msg)
  33. {
  34. if ( p_msg == NULL )
  35. {
  36. return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
  37. " POINTER IS NULL ");
  38. }
  39. //检查消息类型
  40. switch ( p_msg->get_message_type() )
  41. {
  42. case Communication_message::Message_type::eLocate_request_msg:
  43. {
  44. //检查接受人
  45. if ( p_msg->get_receiver() == Communication_message::Communicator::eMeasurer )
  46. {
  47. message::Measure_request_msg t_measure_request_msg;
  48. //针对消息类型, 对消息进行二次解析
  49. if (t_measure_request_msg.ParseFromString(p_msg->get_message_buf()))
  50. {
  51. //检查终端id
  52. if ( t_measure_request_msg.terminal_id() == m_terminal_id )
  53. {
  54. return Error_code::SUCCESS;
  55. }
  56. }
  57. }
  58. break;
  59. }
  60. default :
  61. ;
  62. break;
  63. }
  64. //无效的消息,
  65. return Error_manager(Error_code::INVALID_MESSAGE, Error_level::NEGLIGIBLE_ERROR,
  66. " INVALID_MESSAGE error ");
  67. }
  68. //检查执行者的状态, 判断能否处理这条消息,
  69. Error_manager System_executor::check_executer(Communication_message* p_msg)
  70. {
  71. //huli DDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD 记得删除
  72. return Error_code::SUCCESS;
  73. if ( p_msg == NULL )
  74. {
  75. return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
  76. " POINTER IS NULL ");
  77. }
  78. Error_manager t_error;
  79. //通过 p_msg->get_message_type() 和 p_msg->get_receiver() 找到处理模块的实例对象, 查询执行人是否可以处理这条消息
  80. switch ( p_msg->get_message_type() )
  81. {
  82. case Communication_message::Message_type::eLocate_request_msg:
  83. {
  84. Error_manager t_laser_result = Laser_manager::get_instance_references().check_status();
  85. Error_manager t_executor_result = System_executor::get_instance_references().check_status();
  86. Error_manager t_locate_result = Locate_manager::get_instance_references().check_status();
  87. if (t_laser_result == SUCCESS
  88. && t_executor_result == SUCCESS
  89. && t_locate_result == SUCCESS)
  90. {
  91. return Error_code::SUCCESS;
  92. }
  93. else
  94. {
  95. //整合所有的错误码
  96. t_error.compare_and_cover_error(t_laser_result);
  97. t_error.compare_and_cover_error(t_executor_result);
  98. t_error.compare_and_cover_error(t_locate_result);
  99. if (t_error.get_error_level() == NEGLIGIBLE_ERROR)//一级故障,轻微故障,
  100. {
  101. std::cout << "executer_is_busy , " << std::endl;
  102. //返回繁忙之后, 通信模块1秒后再次调用check
  103. return Error_code::COMMUNICATION_EXCUTER_IS_BUSY;
  104. }
  105. else//返回二级故障,可以封装一条答复信息, 返回错误码
  106. {
  107. message::Measure_request_msg t_measure_request_msg;
  108. //针对消息类型, 对消息进行二次解析
  109. if (t_measure_request_msg.ParseFromString(p_msg->get_message_buf()))
  110. {
  111. //创建一条答复消息
  112. message::Measure_response_msg t_measure_response_msg;
  113. t_measure_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eLocate_response_msg);
  114. t_measure_response_msg.mutable_base_info()->set_timeout_ms(5000);
  115. t_measure_response_msg.mutable_base_info()->set_sender(message::Communicator::eMeasurer);
  116. t_measure_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  117. t_measure_response_msg.set_command_key(t_measure_request_msg.command_key());
  118. t_measure_response_msg.set_terminal_id(t_measure_request_msg.terminal_id());
  119. t_measure_response_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  120. t_measure_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  121. t_measure_response_msg.mutable_error_manager()->set_error_description(t_error.get_error_description());
  122. string t_msg = t_measure_response_msg.SerializeAsString();
  123. System_communication::get_instance_references().encapsulate_msg(t_msg);
  124. LOG(INFO) << " System_executor::check_executer executer status error "<< this;
  125. return t_error;
  126. }
  127. else
  128. {
  129. LOG(INFO) << " System_executor::check_executer second PARSE_ERROR "<< this;
  130. return Error_manager(Error_code::SYSTEM_EXECUTOR_PARSE_ERROR, Error_level::MINOR_ERROR,
  131. " message::Measure_request_msg ParseFromString error ");
  132. }
  133. }
  134. }
  135. break;
  136. }
  137. default :
  138. ;
  139. break;
  140. }
  141. return t_error;
  142. }
  143. //处理消息的执行函数
  144. Error_manager System_executor::execute_msg(Communication_message* p_msg)
  145. {
  146. if ( p_msg == NULL )
  147. {
  148. return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
  149. " POINTER IS NULL ");
  150. }
  151. switch ( p_msg->get_message_type() )
  152. {
  153. case Communication_message::eLocate_request_msg:
  154. message::Measure_request_msg t_measure_request_msg;
  155. //针对消息类型, 对消息进行二次解析
  156. if( t_measure_request_msg.ParseFromString(p_msg->get_message_buf()) )
  157. {
  158. //往线程池添加执行任务, 之后会唤醒一个线程去执行他.
  159. m_thread_pool.enqueue(&System_executor::execute_for_measure, this,
  160. t_measure_request_msg.command_key(), t_measure_request_msg.terminal_id());
  161. }
  162. else
  163. {
  164. return Error_manager(Error_code::SYSTEM_EXECUTOR_PARSE_ERROR, Error_level::MINOR_ERROR,
  165. " message::Measure_request_msg ParseFromString error ");
  166. }
  167. break;
  168. }
  169. return Error_code::SUCCESS;
  170. }
  171. //检查状态
  172. Error_manager System_executor::check_status()
  173. {
  174. if ( m_system_executor_status == SYSTEM_EXECUTOR_READY )
  175. {
  176. if ( m_thread_pool.thread_is_full_load() == false )
  177. {
  178. return Error_code::SUCCESS;
  179. }
  180. else
  181. {
  182. return Error_manager(Error_code::SYSTEM_EXECUTOR_STATUS_BUSY, Error_level::NEGLIGIBLE_ERROR,
  183. " System_executor::check_status error ");
  184. }
  185. }
  186. else
  187. {
  188. return Error_manager(Error_code::SYSTEM_EXECUTOR_STATUS_ERROR, Error_level::MINOR_ERROR,
  189. " System_executor::check_status error ");
  190. }
  191. }
  192. //定时发送状态信息
  193. Error_manager System_executor::encapsulate_send_status()
  194. {
  195. Error_manager t_error;
  196. //创建一条状态消息
  197. message::Measure_status_msg t_measure_status_msg;
  198. t_measure_status_msg.mutable_base_info()->set_msg_type(message::Message_type::eLocate_status_msg);
  199. t_measure_status_msg.mutable_base_info()->set_timeout_ms(5000);
  200. t_measure_status_msg.mutable_base_info()->set_sender(message::Communicator::eMeasurer);
  201. t_measure_status_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  202. t_measure_status_msg.set_terminal_id(m_terminal_id);
  203. Laser_manager::Laser_manager_status t_laser_manager_status = Laser_manager::get_instance_references().get_laser_manager_status();
  204. t_measure_status_msg.set_laser_manager_status((message::Laser_manager_status)t_laser_manager_status);
  205. std::vector<Laser_base*> & t_laser_vector = Laser_manager::get_instance_references().get_laser_vector();
  206. for (auto iter = t_laser_vector.begin(); iter != t_laser_vector.end(); ++iter)
  207. {
  208. Laser_statu t_laser_statu = (*iter)->get_laser_statu();
  209. t_measure_status_msg.add_laser_statu_vector((message::Laser_statu)t_laser_statu);
  210. }
  211. Locate_manager::Locate_manager_status t_locate_manager_status = Locate_manager::get_instance_references().get_locate_manager_status();
  212. t_measure_status_msg.set_locate_manager_status((message::Locate_manager_status)t_locate_manager_status);
  213. t_measure_status_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  214. t_measure_status_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  215. t_measure_status_msg.mutable_error_manager()->set_error_description(t_error.get_error_description(), t_error.get_description_length());
  216. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_x(0);
  217. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_y(0);
  218. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_angle(0);
  219. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_length(0);
  220. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_width(0);
  221. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_height(0);
  222. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_wheel_base(0);
  223. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_wheel_width(0);
  224. t_measure_status_msg.mutable_locate_information_realtime()->set_locate_correct(0);
  225. string t_msg = t_measure_status_msg.SerializeAsString();
  226. System_communication::get_instance_references().encapsulate_msg(t_msg);
  227. return Error_code::SUCCESS;
  228. }
  229. //判断是否为待机,如果已经准备好,则可以执行任务。
  230. bool System_executor::is_ready()
  231. {
  232. if ( m_system_executor_status == SYSTEM_EXECUTOR_READY && m_thread_pool.thread_is_full_load() == false )
  233. {
  234. return true;
  235. }
  236. else
  237. {
  238. return false;
  239. }
  240. }
  241. System_executor::System_executor_status System_executor::get_system_executor_status()
  242. {
  243. return m_system_executor_status;
  244. }
  245. //雷达感测定位 的处理函数
  246. //input::command_id, 消息指令id, 由主控制系统生成的唯一码
  247. //input::command_id, 终端id, 对应具体的某个车位
  248. //return::void, 没有返回, 执行结果直接生成一条答复消息, 然后通过通信返回
  249. void System_executor::execute_for_measure(std::string command_info, int terminal_id)
  250. {
  251. Error_manager t_error;
  252. LOG(INFO) << " System_executor::execute_for_measure run "<< this;
  253. /*
  254. //第一步, 创建点云缓存, 和 指定雷达,目前就一个
  255. std::map<int,pcl::PointCloud<pcl::PointXYZ>::Ptr> point_cloud_map;
  256. std::mutex cloud_lock;
  257. std::vector<int> select_laser_id_vector;
  258. select_laser_id_vector.push_back(0);
  259. select_laser_id_vector.push_back(1);
  260. select_laser_id_vector.push_back(2);
  261. select_laser_id_vector.push_back(3);
  262. //第二步, 按照时间生成中间文件的保存路径
  263. time_t nowTime;
  264. nowTime = time(NULL);
  265. struct tm* sysTime = localtime(&nowTime);
  266. char t_save_path[256] = { 0 };
  267. sprintf(t_save_path, "../data/%04d%02d%02d_%02d%02d%02d",
  268. sysTime->tm_year + 1900, sysTime->tm_mon + 1, sysTime->tm_mday, sysTime->tm_hour, sysTime->tm_min, sysTime->tm_sec);
  269. //第三步, 创建雷达管理模块的任务单, 并发送到 laser_manager
  270. Laser_manager_task laser_manager_task ;
  271. laser_manager_task.task_init(TASK_CREATED,"",NULL,std::chrono::milliseconds(15000),
  272. true,t_save_path,&cloud_lock,&point_cloud_map,false,
  273. 1000,false,select_laser_id_vector );
  274. t_error = Task_command_manager::get_instance_references().execute_task(&laser_manager_task);
  275. // tp_laser_manager->execute_task(laser_manager_task);
  276. //第四步, 等待任务单完成
  277. if ( t_error != Error_code::SUCCESS )
  278. {
  279. LOG(INFO) << " Laser_manager execute_task error "<< this;
  280. }
  281. else
  282. {
  283. //判断任务是否结束, 阻塞 等待, 直到任务完成或者超时
  284. while ( laser_manager_task.is_task_end() == false)
  285. {
  286. if ( laser_manager_task.is_over_time() )
  287. {
  288. //超时处理。取消任务。
  289. Laser_manager::get_instance_pointer()->cancel_task();
  290. laser_manager_task.set_task_statu(TASK_DEAD);
  291. t_error.error_manager_reset(Error_code::LASER_MANAGER_TASK_OVER_TIME, Error_level::MINOR_ERROR,
  292. " laser_manager_task is_over_time ");
  293. laser_manager_task.set_task_error_manager(t_error);
  294. }
  295. else
  296. {
  297. //继续等待
  298. std::this_thread::sleep_for(std::chrono::microseconds(1));
  299. std::this_thread::yield();
  300. }
  301. }
  302. //提取任务单 的错误码
  303. t_error = laser_manager_task.get_task_error_manager();
  304. std::cout << "huli laser_manager_task error :::::::" << laser_manager_task.get_task_error_manager().to_string() << std::endl;
  305. }
  306. Locate_manager_task locate_manager_task ;
  307. //第五步, 创建定位模块的任务单, 并发送到 Locate_manager
  308. if ( t_error != Error_code::SUCCESS )
  309. {
  310. LOG(INFO) << " laser_manager_task error "<< this;
  311. }
  312. else
  313. {
  314. locate_manager_task.task_init(TASK_CREATED,"",NULL,std::chrono::milliseconds(15000),
  315. true,t_save_path,&cloud_lock,&point_cloud_map, false );
  316. t_error = Task_command_manager::get_instance_references().execute_task(&locate_manager_task);
  317. //第六步, 等待任务单完成
  318. if ( t_error != Error_code::SUCCESS )
  319. {
  320. LOG(INFO) << " Locate_manager execute_task error "<< this;
  321. }
  322. else
  323. {
  324. //判断任务是否结束, 阻塞 等待, 直到任务完成或者超时
  325. while ( locate_manager_task.is_task_end() == false)
  326. {
  327. if ( locate_manager_task.is_over_time() )
  328. {
  329. //超时处理。取消任务。
  330. Locate_manager::get_instance_pointer()->cancel_task();
  331. locate_manager_task.set_task_statu(TASK_DEAD);
  332. t_error.error_manager_reset(Error_code::LOCATE_MANAGER_TASK_OVER_TIME, Error_level::MINOR_ERROR,
  333. " locate_manager_task is_over_time ");
  334. locate_manager_task.set_task_error_manager(t_error);
  335. }
  336. else
  337. {
  338. //继续等待
  339. std::this_thread::sleep_for(std::chrono::microseconds(1));
  340. std::this_thread::yield();
  341. }
  342. }
  343. //提取任务单 的错误码
  344. t_error = locate_manager_task.get_task_error_manager();
  345. std::cout << "huli locate_manager_task error :::::::" << locate_manager_task.get_task_error_manager().to_string() << std::endl;
  346. }
  347. }
  348. //第七步, 创建一条答复消息
  349. message::Measure_response_msg t_measure_response_msg;
  350. t_measure_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eLocate_response_msg);
  351. t_measure_response_msg.mutable_base_info()->set_timeout_ms(5000);
  352. t_measure_response_msg.mutable_base_info()->set_sender(message::Communicator::eMeasurer);
  353. t_measure_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  354. t_measure_response_msg.set_command_id(command_id);
  355. t_measure_response_msg.set_terminal_id(terminal_id);
  356. t_measure_response_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  357. t_measure_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  358. t_measure_response_msg.mutable_error_manager()->set_error_description(t_error.get_error_description(), t_error.get_description_length());
  359. if ( t_error == Error_code::SUCCESS )
  360. {
  361. t_measure_response_msg.mutable_locate_information()->set_locate_x(locate_manager_task.get_task_locate_information_ex()->locate_x);
  362. t_measure_response_msg.mutable_locate_information()->set_locate_y(locate_manager_task.get_task_locate_information_ex()->locate_y);
  363. t_measure_response_msg.mutable_locate_information()->set_locate_angle(locate_manager_task.get_task_locate_information_ex()->locate_angle);
  364. t_measure_response_msg.mutable_locate_information()->set_locate_length(locate_manager_task.get_task_locate_information_ex()->locate_length);
  365. t_measure_response_msg.mutable_locate_information()->set_locate_width(locate_manager_task.get_task_locate_information_ex()->locate_width);
  366. t_measure_response_msg.mutable_locate_information()->set_locate_height(locate_manager_task.get_task_locate_information_ex()->locate_height);
  367. t_measure_response_msg.mutable_locate_information()->set_locate_wheel_base(locate_manager_task.get_task_locate_information_ex()->locate_wheel_base);
  368. t_measure_response_msg.mutable_locate_information()->set_locate_wheel_width(locate_manager_task.get_task_locate_information_ex()->locate_wheel_width);
  369. t_measure_response_msg.mutable_locate_information()->set_locate_correct(locate_manager_task.get_task_locate_information_ex()->locate_correct);
  370. }
  371. string t_msg = t_measure_response_msg.SerializeAsString();
  372. System_communication::get_instance_references().encapsulate_msg(t_msg);
  373. std::cout << "huli t_measure_response_msg = " << t_measure_response_msg.DebugString() << std::endl;
  374. return ;
  375. */
  376. message::Measure_response_msg t_measure_response_msg;
  377. t_measure_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eLocate_response_msg);
  378. t_measure_response_msg.mutable_base_info()->set_timeout_ms(5000);
  379. t_measure_response_msg.mutable_base_info()->set_sender(message::Communicator::eMeasurer);
  380. t_measure_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  381. t_measure_response_msg.set_command_key(command_info);
  382. t_measure_response_msg.set_terminal_id(terminal_id);
  383. t_measure_response_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  384. t_measure_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  385. t_measure_response_msg.mutable_error_manager()->set_error_description(t_error.get_error_description(), t_error.get_description_length());
  386. if ( t_error == Error_code::SUCCESS )
  387. {
  388. t_measure_response_msg.mutable_locate_information()->set_locate_x(0);
  389. t_measure_response_msg.mutable_locate_information()->set_locate_y(0);
  390. t_measure_response_msg.mutable_locate_information()->set_locate_angle(0);
  391. t_measure_response_msg.mutable_locate_information()->set_locate_length(0);
  392. t_measure_response_msg.mutable_locate_information()->set_locate_width(0);
  393. t_measure_response_msg.mutable_locate_information()->set_locate_height(0);
  394. t_measure_response_msg.mutable_locate_information()->set_locate_wheel_base(0);
  395. t_measure_response_msg.mutable_locate_information()->set_locate_wheel_width(0);
  396. t_measure_response_msg.mutable_locate_information()->set_locate_correct(0);
  397. }
  398. string t_msg = t_measure_response_msg.SerializeAsString();
  399. System_communication::get_instance_references().encapsulate_msg(t_msg);
  400. std::cout << "huli t_measure_response_msg = " << t_measure_response_msg.DebugString() << std::endl;
  401. return ;
  402. }