diff --git a/module/Local_Server/Data_Feed/data_feed_thread.cpp b/module/Local_Server/Data_Feed/data_feed_thread.cpp index acfd701..4c23d30 100644 --- a/module/Local_Server/Data_Feed/data_feed_thread.cpp +++ b/module/Local_Server/Data_Feed/data_feed_thread.cpp @@ -39,7 +39,33 @@ ucoro::awaitable data_feed_thread_coro(std::atomic& running) { for (auto& source : mode_acs.data_source_config.map.list()) { if (!source->enable) continue; if (!source->is_open() && !source->open()) continue; - co_await source->source_step_coro(); + + + co_await source->handle_in_loop_coro(); + auto mode_data = co_await source->read_coro(); + + struct Process_Result { + std::exception_ptr exception; + }; + + auto result = co_await ucoro::callback_awaitable( + [source, mode_data = std::move(mode_data)](auto done) mutable { + asio::post( + data_source_process_pool(), + [source, mode_data = std::move(mode_data), done = std::move(done)]() mutable { + try { + source->process_mode_acs_data(mode_data); + done(Process_Result{}); + } catch (...) { + done(Process_Result{std::current_exception()}); + } + }); + }); + + if (result.exception) { + std::rethrow_exception(result.exception); + } + } // 这两个每次循环都要重新计算 保证实时性 diff --git a/module/Local_Server/Data_Source/Data_Source.cpp b/module/Local_Server/Data_Source/Data_Source.cpp index ae36a17..288ca04 100644 --- a/module/Local_Server/Data_Source/Data_Source.cpp +++ b/module/Local_Server/Data_Source/Data_Source.cpp @@ -9,15 +9,12 @@ #include "../server/Global.h" #include "Database.h" -namespace { asio::thread_pool& data_source_process_pool() { const auto hardware_threads = std::thread::hardware_concurrency(); const auto worker_count = std::max(2u, hardware_threads > 2 ? hardware_threads - 2 : hardware_threads); static asio::thread_pool pool(worker_count); return pool; } -} - Psc::serial::Serial* create_serial(const std::string& serial_name, Baud_Rate_Type baud_rate) { auto serial = new Psc::serial::Serial; serial->set_serial_name(serial_name); @@ -107,33 +104,6 @@ void File_Data_Source::before_handle_msg(std::shared_ptr& msg) { -ucoro::awaitable Data_Source::source_step_coro() { - co_await handle_in_loop_coro(); - auto mode_data = co_await read_coro(); - - struct Process_Result { - std::exception_ptr exception; - }; - auto self = that(); - auto result = co_await ucoro::callback_awaitable( - [this, self = std::move(self), mode_data = std::move(mode_data)](auto done) mutable { - asio::post( - data_source_process_pool(), - [this, self = std::move(self), mode_data = std::move(mode_data), done = std::move(done)]() mutable { - try { - process_mode_acs_data(mode_data); - done(Process_Result{}); - } catch (...) { - done(Process_Result{std::current_exception()}); - } - }); - }); - - if (result.exception) { - std::rethrow_exception(result.exception); - } - co_return; -} diff --git a/module/Local_Server/Data_Source/Data_Source.h b/module/Local_Server/Data_Source/Data_Source.h index a825a13..7ca98f7 100644 --- a/module/Local_Server/Data_Source/Data_Source.h +++ b/module/Local_Server/Data_Source/Data_Source.h @@ -10,7 +10,7 @@ class SM_RingBuffer; // 统一封装数据输出的模式 class Data_Source; std::shared_ptr create_from_json(const Psc::JSON *that_json); - +asio::thread_pool& data_source_process_pool(); class Input_Format { public: @@ -79,7 +79,7 @@ public: Data_Source(); std::shared_ptr last_prase_msg = nullptr; virtual ucoro::awaitable handle_in_loop_coro(){co_return;} - virtual ucoro::awaitable source_step_coro(); + std::atomic enable{}; std::string type; std::atomic base_station_show{};