From 8ca05115ec2fdfb096c92f3c375cc938d6e77b42 Mon Sep 17 00:00:00 2001 From: wyc <1104749580@qq.com> Date: Tue, 23 Jun 2026 11:00:00 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=B2=E5=8F=A3=E6=94=B9=E5=8D=8F=E7=A8=8B?= =?UTF-8?q?=E5=89=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- module/Local_Server/Data_Feed/Data_Feed.cpp | 12 ++- module/Local_Server/Data_Feed/Data_Feed.h | 43 ++++++++-- .../Data_Feed/data_feed_thread.cpp | 39 +++++----- .../Local_Server/Data_Source/Data_Source.cpp | 78 ++++++++++++++++--- module/Local_Server/Data_Source/Data_Source.h | 19 ++++- 5 files changed, 153 insertions(+), 38 deletions(-) diff --git a/module/Local_Server/Data_Feed/Data_Feed.cpp b/module/Local_Server/Data_Feed/Data_Feed.cpp index 950db64..6965dbe 100644 --- a/module/Local_Server/Data_Feed/Data_Feed.cpp +++ b/module/Local_Server/Data_Feed/Data_Feed.cpp @@ -1,7 +1,17 @@ #include "Local_Server/Data_Feed/Data_Feed.h" #include "../server/Global.h" void Data_Feed_UDP_Server::handle_in_loop() { - ucoro::sync_await(svr.tick_coro()); + ucoro::sync_await(handle_in_loop_coro()); +} + +ucoro::awaitable Data_Feed_UDP_Server::handle_in_loop_coro() { + co_await svr.tick_coro(); + co_return; +} + +ucoro::awaitable Data_Feed_UDP_Server::send_coro(const std::string& data) { + co_await svr.write_to_all_clients_coro(data); + co_return; } bool Data_Feed_UDP_Server::_open() { svr.set_bind_address("0.0.0.0", port); diff --git a/module/Local_Server/Data_Feed/Data_Feed.h b/module/Local_Server/Data_Feed/Data_Feed.h index aedf2cc..62b6c1a 100644 --- a/module/Local_Server/Data_Feed/Data_Feed.h +++ b/module/Local_Server/Data_Feed/Data_Feed.h @@ -55,6 +55,13 @@ public: Frequency_Limit_Multi sbs_flm{}; virtual void handle_in_loop() { + } + virtual ucoro::awaitable handle_in_loop_coro() { + handle_in_loop(); + co_return; + } + virtual ucoro::awaitable send_coro(const std::string&) { + co_return; } virtual Psc::JSON to_json() { Psc::JSON ret = Psc::JSON::object(); @@ -126,8 +133,16 @@ public: ~Data_Feed_TCP_Server() override {} void handle_in_loop() override { //std::cout << socket.to_string() << "flush_clients" << std::endl; - ucoro::sync_await(svr.flush_clients_coro()); - ucoro::sync_await(svr.tick_coro()); + ucoro::sync_await(handle_in_loop_coro()); + } + ucoro::awaitable handle_in_loop_coro() override { + co_await svr.flush_clients_coro(); + co_await svr.tick_coro(); + co_return; + } + ucoro::awaitable send_coro(const std::string& data) override { + co_await svr.write_to_all_clients_coro(data); + co_return; } void from_json(const Psc::JSON* that_json) override { Data_Feed::from_json(that_json); @@ -185,7 +200,15 @@ public: class Data_Feed_TCP_Client : public Data_Feed { public: void handle_in_loop() override { - ucoro::sync_await(cli.tick_coro()); + ucoro::sync_await(handle_in_loop_coro()); + } + ucoro::awaitable handle_in_loop_coro() override { + co_await cli.tick_coro(); + co_return; + } + ucoro::awaitable send_coro(const std::string& data) override { + co_await cli.send_coro(data); + co_return; } std::string url; Psc::JSON get_custom_state_json() override { @@ -229,9 +252,11 @@ public: - Data_Feed_UDP_Server()= default; + Data_Feed_UDP_Server()= default; ~Data_Feed_UDP_Server() override = default; void handle_in_loop() override; + ucoro::awaitable handle_in_loop_coro() override; + ucoro::awaitable send_coro(const std::string& data) override; [[nodiscard]] Psc::JSON get_clients_json() const { Psc::JSON ret = Psc::JSON::array(); for (const auto& i : svr.clients) { @@ -268,7 +293,15 @@ public: return VAR_JSON_1(state); } void handle_in_loop() override { - ucoro::sync_await(cli.tick_coro()); + ucoro::sync_await(handle_in_loop_coro()); + } + ucoro::awaitable handle_in_loop_coro() override { + co_await cli.tick_coro(); + co_return; + } + ucoro::awaitable send_coro(const std::string& data) override { + co_await cli.send_coro(data); + co_return; } Psc::asio_socket::UDP_Client_Coro cli; std::string url; diff --git a/module/Local_Server/Data_Feed/data_feed_thread.cpp b/module/Local_Server/Data_Feed/data_feed_thread.cpp index 29189f1..553805a 100644 --- a/module/Local_Server/Data_Feed/data_feed_thread.cpp +++ b/module/Local_Server/Data_Feed/data_feed_thread.cpp @@ -133,8 +133,13 @@ std::optional convert_to_send_format(Data_Source* ds, const std::sh return send_msg; } -void data_feed_thread(std::atomic& running) { +ucoro::awaitable data_feed_thread_coro(std::atomic& running) { auto g = Global::instance(); + for (auto& source : g->mode_acs.data_source_config.map.list()) { + if (source->enable) { + source->open(); + } + } for (auto& t : g->mode_acs.data_feed_config.map.list()) { if (t->enable) { auto ret = t->check_and_open(); @@ -150,8 +155,14 @@ void data_feed_thread(std::atomic& running) { auto& report_data_feed_msg_mum = mode_acs.report_data_feed_msg_mum; std::vector> enable_feeds; // all_feed 保证声明周期仍然存在 auto monitor_msg_live = mode_acs.monitor_msg_live.load(); - bool all_empty = true; while (running.load(std::memory_order_acquire) == true) { + bool all_empty = true; + 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(); + } + // 这两个每次循环都要重新计算 保证实时性 // if (flush_limit.test()) { auto all_feed = cfg.map.list(); @@ -160,7 +171,7 @@ void data_feed_thread(std::atomic& running) { if (!feed->enable) continue; enable_feeds.push_back(feed); // 执行特定的刷新逻辑 比如 tcp_server 检查自身的连接 - feed->handle_in_loop(); + co_await feed->handle_in_loop_coro(); } @@ -200,22 +211,7 @@ void data_feed_thread(std::atomic& running) { std::cout << "警告:未知原因:" << feed->key << "内有空的消息体" << std::endl; continue; } - if (feed->type == "Data_Feed_TCP_Server") { - auto it = dynamic_cast(feed.get()); - ucoro::sync_await(it->svr.write_to_all_clients_coro(*str)); - } - else if (feed->type == "Data_Feed_UDP_Server") { - auto it = dynamic_cast(feed.get()); - ucoro::sync_await(it->svr.write_to_all_clients_coro(*str)); - } - else if (feed->type == "Data_Feed_TCP_Client") { - auto it = dynamic_cast(feed.get()); - ucoro::sync_await(it->cli.send_coro(*str)); - } - else if (feed->type == "Data_Feed_UDP_Client") { - auto it = dynamic_cast(feed.get()); - ucoro::sync_await(it->cli.send_coro(*str)); - } + co_await feed->send_coro(*str); } } if (t != 0) { @@ -226,5 +222,10 @@ void data_feed_thread(std::atomic& running) { } } } + co_return; +} + +void data_feed_thread(std::atomic& running) { + ucoro::sync_await(data_feed_thread_coro(running)); } diff --git a/module/Local_Server/Data_Source/Data_Source.cpp b/module/Local_Server/Data_Source/Data_Source.cpp index 525c32c..205b3e6 100644 --- a/module/Local_Server/Data_Source/Data_Source.cpp +++ b/module/Local_Server/Data_Source/Data_Source.cpp @@ -1,10 +1,23 @@ #include "Data_Source.h" +#include +#include +#include #include +#include #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); @@ -55,17 +68,9 @@ std::string Data_Source::thread_key() const { } void Data_Source::test_and_attach_thread() { - //static int t = Global::instance()->mode_acs.read_milliseconds; - Global::instance()->thread_manager.test_and_start_thread(thread_key(), [this](std::atomic& running) { - if (enable) { - open(); - } - while (running.load(std::memory_order_acquire) == true) - { - handle_in_loop(); - handle_mode_acs_source(); - } - }); + if (enable) { + open(); + } } @@ -174,7 +179,10 @@ void Data_Source::handle_mode_acs_source() { return; } + process_mode_acs_data(mode_data); +} +void Data_Source::process_mode_acs_data(const std::string& mode_data) { auto size = mode_data.size(); read_speed.update(size); value_statistics.update(size); @@ -204,6 +212,49 @@ void Data_Source::handle_mode_acs_source() { }); } +ucoro::awaitable Data_Source::handle_in_loop_coro() { + handle_in_loop(); + co_return; +} + +ucoro::awaitable Data_Source::recv_coro() { + co_return read(); +} + +ucoro::awaitable Data_Source::source_step_coro() { + co_await handle_in_loop_coro(); + auto mode_data = co_await recv_coro(); + if (mode_data.empty()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + co_return; + } + 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; +} + + + + void Data_Source::test_and_stop_thread() const { Global::instance()->thread_manager.test_and_stop_thread(thread_key()); @@ -675,6 +726,11 @@ void File_Data_Source::handle_mode_acs_source() { } } +ucoro::awaitable File_Data_Source::source_step_coro() { + handle_mode_acs_source(); + co_return; +} + std::string Dll_Data_Source::read() { auto start = std::chrono::steady_clock::now(); diff --git a/module/Local_Server/Data_Source/Data_Source.h b/module/Local_Server/Data_Source/Data_Source.h index eb2bcc4..c54bbd0 100644 --- a/module/Local_Server/Data_Source/Data_Source.h +++ b/module/Local_Server/Data_Source/Data_Source.h @@ -107,6 +107,12 @@ public: virtual std::shared_ptr create_msg(const std::shared_ptr& source, const std::string& packet); virtual void update_msg_day_time(std::shared_ptr msg); virtual void handle_mode_acs_source(); + virtual ucoro::awaitable handle_in_loop_coro(); + virtual ucoro::awaitable recv_coro(); + virtual ucoro::awaitable source_step_coro(); + + + void process_mode_acs_data(const std::string& mode_data); void refresh_data_feed_key_list(); @@ -196,6 +202,7 @@ public: } return true; } + bool is_open() const { return _open_; } void close() { if (_open_) { _open_ = false; @@ -224,7 +231,11 @@ protected: struct TCP_Client_Data_Source : Data_Source { void handle_in_loop() override { - ucoro::sync_await(cli.tick_coro()); + ucoro::sync_await(handle_in_loop_coro()); + } + ucoro::awaitable handle_in_loop_coro() override { + co_await cli.tick_coro(); + co_return; } Psc::JSON get_custom_state_json() override { auto& state = cli.state; @@ -263,7 +274,10 @@ struct TCP_Client_Data_Source : Data_Source { } std::string read() override { - return ucoro::sync_await(cli.read_coro()); + return ucoro::sync_await(recv_coro()); + } + ucoro::awaitable recv_coro() override { + co_return co_await cli.read_coro(); } }; struct Serial_Data_Source : Data_Source { @@ -377,6 +391,7 @@ struct File_Data_Source : public Data_Source { } std::string read() override; void handle_mode_acs_source() override; + ucoro::awaitable source_step_coro() override; protected: std::optional get_raw_line(int &ret_index);