system_executor.cpp 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341
  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)
  18. {
  19. m_thread_pool.thread_pool_init(threads_size);
  20. m_system_executor_status = SYSTEM_EXECUTOR_READY;
  21. return Error_code::SUCCESS;
  22. }
  23. //反初始化
  24. Error_manager System_executor::system_executor_uninit()
  25. {
  26. m_thread_pool.thread_pool_uninit();
  27. m_system_executor_status = SYSTEM_EXECUTOR_UNKNOW;
  28. return Error_code::SUCCESS;
  29. }
  30. //检查执行者的状态, 判断能否处理这条消息,
  31. Error_manager System_executor::check_executer(Communication_message* p_msg)
  32. {
  33. if ( p_msg == NULL )
  34. {
  35. return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
  36. " POINTER IS NULL ");
  37. }
  38. Error_manager t_error;
  39. //通过 p_msg->get_message_type() 和 p_msg->get_receiver() 找到处理模块的实例对象, 查询执行人是否可以处理这条消息
  40. switch ( p_msg->get_message_type() )
  41. {
  42. case Communication_message::Message_type::eLocate_request_msg:
  43. {
  44. Error_manager t_laser_result = Laser_manager::get_instance_references().check_status();
  45. Error_manager t_executor_result = System_executor::get_instance_references().check_status();
  46. Error_manager t_locate_result = Locate_manager::get_instance_references().check_status();
  47. if (t_laser_result == SUCCESS
  48. && t_executor_result == SUCCESS
  49. && t_locate_result == SUCCESS)
  50. {
  51. return Error_code::SUCCESS;
  52. }
  53. else
  54. {
  55. //整合所有的错误码
  56. t_error.compare_and_cover_error(t_laser_result);
  57. t_error.compare_and_cover_error(t_executor_result);
  58. t_error.compare_and_cover_error(t_locate_result);
  59. if (t_error.get_error_level() == NEGLIGIBLE_ERROR)//一级故障,轻微故障,
  60. {
  61. std::cout << "executer_is_busy , " << std::endl;
  62. //返回繁忙之后, 通信模块1秒后再次调用check
  63. return Error_code::COMMUNICATION_EXCUTER_IS_BUSY;
  64. }
  65. else//返回二级故障,可以封装一条答复信息, 返回错误码
  66. {
  67. message::Measure_request_msg t_measure_request_msg;
  68. //针对消息类型, 对消息进行二次解析
  69. if (t_measure_request_msg.ParseFromString(p_msg->get_message_buf()))
  70. {
  71. //创建一条答复消息
  72. message::Measure_response_msg t_measure_response_msg;
  73. t_measure_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eLocate_response_msg);
  74. t_measure_response_msg.mutable_base_info()->set_timeout_ms(5000);
  75. t_measure_response_msg.mutable_base_info()->set_sender(message::Communicator::eMeasurer);
  76. t_measure_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  77. t_measure_response_msg.set_command_id(t_measure_request_msg.command_id());
  78. t_measure_response_msg.set_terminal_id(t_measure_request_msg.terminal_id());
  79. t_measure_response_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  80. t_measure_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  81. t_measure_response_msg.mutable_error_manager()->set_error_description(t_error.get_error_description());
  82. string t_msg = t_measure_response_msg.SerializeAsString();
  83. System_communication::get_instance_references().encapsulate_msg(t_msg);
  84. LOG(INFO) << " System_executor::check_executer executer status error "<< this;
  85. return t_error;
  86. }
  87. else
  88. {
  89. LOG(INFO) << " System_executor::check_executer second PARSE_ERROR "<< this;
  90. return Error_manager(Error_code::SYSTEM_EXECUTOR_PARSE_ERROR, Error_level::MINOR_ERROR,
  91. " message::Measure_request_msg ParseFromString error ");
  92. }
  93. }
  94. }
  95. break;
  96. }
  97. default :
  98. ;
  99. break;
  100. }
  101. return t_error;
  102. }
  103. //处理消息的执行函数
  104. Error_manager System_executor::execute_msg(Communication_message* p_msg)
  105. {
  106. if ( p_msg == NULL )
  107. {
  108. return Error_manager(Error_code::POINTER_IS_NULL, Error_level::MINOR_ERROR,
  109. " POINTER IS NULL ");
  110. }
  111. switch ( p_msg->get_message_type() )
  112. {
  113. case Communication_message::eLocate_request_msg:
  114. message::Measure_request_msg t_measure_request_msg;
  115. //针对消息类型, 对消息进行二次解析
  116. if( t_measure_request_msg.ParseFromString(p_msg->get_message_buf()) )
  117. {
  118. //往线程池添加执行任务, 之后会唤醒一个线程去执行他.
  119. m_thread_pool.enqueue(&System_executor::execute_for_measure, this,
  120. t_measure_request_msg.command_id(), t_measure_request_msg.terminal_id());
  121. }
  122. else
  123. {
  124. return Error_manager(Error_code::SYSTEM_EXECUTOR_PARSE_ERROR, Error_level::MINOR_ERROR,
  125. " message::Measure_request_msg ParseFromString error ");
  126. }
  127. break;
  128. }
  129. return Error_code::SUCCESS;
  130. }
  131. //检查状态
  132. Error_manager System_executor::check_status()
  133. {
  134. if ( m_system_executor_status == SYSTEM_EXECUTOR_READY )
  135. {
  136. if ( m_thread_pool.thread_is_full_load() == false )
  137. {
  138. return Error_code::SUCCESS;
  139. }
  140. else
  141. {
  142. return Error_manager(Error_code::SYSTEM_EXECUTOR_STATUS_BUSY, Error_level::NEGLIGIBLE_ERROR,
  143. " System_executor::check_status error ");
  144. }
  145. }
  146. else
  147. {
  148. return Error_manager(Error_code::SYSTEM_EXECUTOR_STATUS_ERROR, Error_level::MINOR_ERROR,
  149. " System_executor::check_status error ");
  150. }
  151. }
  152. //判断是否为待机,如果已经准备好,则可以执行任务。
  153. bool System_executor::is_ready()
  154. {
  155. if ( m_system_executor_status == SYSTEM_EXECUTOR_READY && m_thread_pool.thread_is_full_load() == false )
  156. {
  157. return true;
  158. }
  159. else
  160. {
  161. return false;
  162. }
  163. }
  164. System_executor::System_executor_status System_executor::get_system_executor_status()
  165. {
  166. return m_system_executor_status;
  167. }
  168. //雷达感测定位 的处理函数
  169. //input::command_id, 消息指令id, 由主控制系统生成的唯一码
  170. //input::command_id, 终端id, 对应具体的某个车位
  171. //return::void, 没有返回, 执行结果直接生成一条答复消息, 然后通过通信返回
  172. void System_executor::execute_for_measure(int command_id, int terminal_id)
  173. {
  174. Error_manager t_error;
  175. LOG(INFO) << " System_executor::execute_for_measure run "<< this;
  176. //第一步, 创建点云缓存, 和 指定雷达,目前就一个
  177. std::map<int,pcl::PointCloud<pcl::PointXYZ>::Ptr> point_cloud_map;
  178. std::mutex cloud_lock;
  179. std::vector<int> select_laser_id_vector;
  180. select_laser_id_vector.push_back(0);
  181. select_laser_id_vector.push_back(1);
  182. select_laser_id_vector.push_back(2);
  183. select_laser_id_vector.push_back(3);
  184. //第二步, 按照时间生成中间文件的保存路径
  185. time_t nowTime;
  186. nowTime = time(NULL);
  187. struct tm* sysTime = localtime(&nowTime);
  188. char t_save_path[256] = { 0 };
  189. sprintf(t_save_path, "../data/%04d%02d%02d_%02d%02d%02d",
  190. sysTime->tm_year + 1900, sysTime->tm_mon + 1, sysTime->tm_mday, sysTime->tm_hour, sysTime->tm_min, sysTime->tm_sec);
  191. //第三步, 创建雷达管理模块的任务单, 并发送到 laser_manager
  192. Laser_manager_task laser_manager_task ;
  193. laser_manager_task.task_init(TASK_CREATED,"",NULL,std::chrono::milliseconds(15000),
  194. true,t_save_path,&cloud_lock,&point_cloud_map,false,
  195. 1000,false,select_laser_id_vector );
  196. t_error = Task_command_manager::get_instance_references().execute_task(&laser_manager_task);
  197. // tp_laser_manager->execute_task(laser_manager_task);
  198. //第四步, 等待任务单完成
  199. if ( t_error != Error_code::SUCCESS )
  200. {
  201. LOG(INFO) << " Laser_manager execute_task error "<< this;
  202. }
  203. else
  204. {
  205. //判断任务是否结束, 阻塞 等待, 直到任务完成或者超时
  206. while ( laser_manager_task.is_task_end() == false)
  207. {
  208. if ( laser_manager_task.is_over_time() )
  209. {
  210. //超时处理。取消任务。
  211. Laser_manager::get_instance_pointer()->cancel_task();
  212. laser_manager_task.set_task_statu(TASK_DEAD);
  213. t_error.error_manager_reset(Error_code::LASER_MANAGER_TASK_OVER_TIME, Error_level::MINOR_ERROR,
  214. " laser_manager_task is_over_time ");
  215. laser_manager_task.set_task_error_manager(t_error);
  216. }
  217. else
  218. {
  219. //继续等待
  220. std::this_thread::sleep_for(std::chrono::microseconds(1));
  221. std::this_thread::yield();
  222. }
  223. }
  224. //提取任务单 的错误码
  225. t_error = laser_manager_task.get_task_error_manager();
  226. std::cout << "huli laser_manager_task error :::::::" << laser_manager_task.get_task_error_manager().to_string() << std::endl;
  227. }
  228. Locate_manager_task locate_manager_task ;
  229. //第五步, 创建定位模块的任务单, 并发送到 Locate_manager
  230. if ( t_error != Error_code::SUCCESS )
  231. {
  232. LOG(INFO) << " laser_manager_task error "<< this;
  233. }
  234. else
  235. {
  236. locate_manager_task.task_init(TASK_CREATED,"",NULL,std::chrono::milliseconds(15000),
  237. true,t_save_path,&cloud_lock,&point_cloud_map, false );
  238. t_error = Task_command_manager::get_instance_references().execute_task(&locate_manager_task);
  239. //第六步, 等待任务单完成
  240. if ( t_error != Error_code::SUCCESS )
  241. {
  242. LOG(INFO) << " Locate_manager execute_task error "<< this;
  243. }
  244. else
  245. {
  246. //判断任务是否结束, 阻塞 等待, 直到任务完成或者超时
  247. while ( locate_manager_task.is_task_end() == false)
  248. {
  249. if ( locate_manager_task.is_over_time() )
  250. {
  251. //超时处理。取消任务。
  252. Locate_manager::get_instance_pointer()->cancel_task();
  253. locate_manager_task.set_task_statu(TASK_DEAD);
  254. t_error.error_manager_reset(Error_code::LOCATE_MANAGER_TASK_OVER_TIME, Error_level::MINOR_ERROR,
  255. " locate_manager_task is_over_time ");
  256. locate_manager_task.set_task_error_manager(t_error);
  257. }
  258. else
  259. {
  260. //继续等待
  261. std::this_thread::sleep_for(std::chrono::microseconds(1));
  262. std::this_thread::yield();
  263. }
  264. }
  265. //提取任务单 的错误码
  266. t_error = locate_manager_task.get_task_error_manager();
  267. std::cout << "huli locate_manager_task error :::::::" << locate_manager_task.get_task_error_manager().to_string() << std::endl;
  268. }
  269. }
  270. //第七步, 创建一条答复消息
  271. message::Measure_response_msg t_measure_response_msg;
  272. t_measure_response_msg.mutable_base_info()->set_msg_type(message::Message_type::eLocate_response_msg);
  273. t_measure_response_msg.mutable_base_info()->set_timeout_ms(5000);
  274. t_measure_response_msg.mutable_base_info()->set_sender(message::Communicator::eMeasurer);
  275. t_measure_response_msg.mutable_base_info()->set_receiver(message::Communicator::eMain);
  276. t_measure_response_msg.set_command_id(command_id);
  277. t_measure_response_msg.set_terminal_id(terminal_id);
  278. t_measure_response_msg.mutable_error_manager()->set_error_code(t_error.get_error_code());
  279. t_measure_response_msg.mutable_error_manager()->set_error_level((message::Error_level)t_error.get_error_level());
  280. t_measure_response_msg.mutable_error_manager()->set_error_description(t_error.get_error_description(), t_error.get_description_length());
  281. if ( t_error == Error_code::SUCCESS )
  282. {
  283. t_measure_response_msg.mutable_locate_information()->set_locate_x(locate_manager_task.get_task_locate_information_ex()->locate_x);
  284. t_measure_response_msg.mutable_locate_information()->set_locate_y(locate_manager_task.get_task_locate_information_ex()->locate_y);
  285. t_measure_response_msg.mutable_locate_information()->set_locate_angle(locate_manager_task.get_task_locate_information_ex()->locate_angle);
  286. t_measure_response_msg.mutable_locate_information()->set_locate_length(locate_manager_task.get_task_locate_information_ex()->locate_length);
  287. t_measure_response_msg.mutable_locate_information()->set_locate_width(locate_manager_task.get_task_locate_information_ex()->locate_width);
  288. t_measure_response_msg.mutable_locate_information()->set_locate_height(locate_manager_task.get_task_locate_information_ex()->locate_height);
  289. t_measure_response_msg.mutable_locate_information()->set_locate_wheel_base(locate_manager_task.get_task_locate_information_ex()->locate_wheel_base);
  290. t_measure_response_msg.mutable_locate_information()->set_locate_wheel_width(locate_manager_task.get_task_locate_information_ex()->locate_wheel_width);
  291. t_measure_response_msg.mutable_locate_information()->set_locate_correct(locate_manager_task.get_task_locate_information_ex()->locate_correct);
  292. }
  293. string t_msg = t_measure_response_msg.SerializeAsString();
  294. System_communication::get_instance_references().encapsulate_msg(t_msg);
  295. std::cout << "huli t_measure_response_msg = " << t_measure_response_msg.DebugString() << std::endl;
  296. return ;
  297. }