From 03bfac8d0cfc25f2b6a18f915d408f4a92592b69 Mon Sep 17 00:00:00 2001 From: wyc <1104749580@qq.com> Date: Thu, 25 Jun 2026 11:45:41 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BE=AE=E8=B0=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- module/Local_Server/server/io_coro.cpp | 224 ++++++++++++------------- module/Local_Server/server/io_coro.h | 31 ++++ module/Local_Server_main.cpp | 11 +- 3 files changed, 146 insertions(+), 120 deletions(-) create mode 100644 module/Local_Server/server/io_coro.h diff --git a/module/Local_Server/server/io_coro.cpp b/module/Local_Server/server/io_coro.cpp index cfd0bbe..2270890 100644 --- a/module/Local_Server/server/io_coro.cpp +++ b/module/Local_Server/server/io_coro.cpp @@ -1,31 +1,72 @@ #include "../server/Global.h" #include "../server/Mode_Msg_Buffer.h" -#include -#include -#include -#include -#include -#include -#include -#include -#include -Frequency_Limit too_many_msg_limit; -Frequency_Limit flush_limit; -concurrencpp::runtime runtime_; -std::shared_ptr executor_; -std::unique_ptr> data_feed_thread_task; -std::atomic running = false; -concurrencpp::result coro_thread(std::atomic& running); -void start_io_coro() { +#include "io_coro.h" + +concurrencpp::result Coro::coro_thread() { + co_await concurrencpp::resume_on(executor_); + auto g = Global::instance(); + for (auto& feed : g->mode_acs.data_feed_config.map.list()) { + if (feed->enable) { + auto ret = feed->check_and_open(); + if (!ret) { + std::cout << "Data feed first open failed: " << feed->key << std::endl; + } + } + } + while (running.load(std::memory_order_acquire)) { + auto sources = Global::instance()->mode_acs.data_source_config.map.list(); + for (auto& source : sources) { + if (!source || !source->enable) { + continue; + } + if (task_is_running(source->loop_task)) { + continue; + } + if (!source->is_open() && !source->open()) { + continue; + } + source->loop_task = std::make_unique>(data_source_loop_coro(source)); + } + auto feeds = Global::instance()->mode_acs.data_feed_config.map.list(); + for (auto& feed : feeds) { + if (!feed || !feed->enable) { + continue; + } + if (task_is_running(feed->loop_task)) { + continue; + } + auto ret = feed->check_and_open(); + if (!ret) { + continue; + } + feed->loop_task = std::make_unique>(data_feed_loop_coro(feed)); + } + co_await concurrencpp::resume_on(executor_); + } + auto start = std::chrono::steady_clock::now(); + while (has_running_loop_tasks()) { + if (std::chrono::steady_clock::now() - start > std::chrono::seconds(10)) { + std::cerr << "data loop task stop timeout" << std::endl; + break; + } + co_await concurrencpp::resume_on(executor_); + } + co_return; +} + +void Coro::start() { if (data_feed_thread_task) { return; } std::cout << "start_io_coro" << std::endl; running.store(true, std::memory_order_release); executor_ = runtime_.make_worker_thread_executor(); - data_feed_thread_task = std::make_unique>(coro_thread(running)); + data_feed_thread_task = std::make_unique>(coro_thread()); } -void stop_io_coro() { + + +void Coro::stop() { running.store(false, std::memory_order_release); if (!data_feed_thread_task) { return; @@ -44,7 +85,8 @@ void stop_io_coro() { data_feed_thread_task.reset(); executor_.reset(); } -bool task_is_running(std::unique_ptr>& task) { + +bool Coro::task_is_running(std::unique_ptr>& task) { if (!task) { return false; } @@ -55,7 +97,7 @@ bool task_is_running(std::unique_ptr>& task) { task.reset(); return false; } -concurrencpp::result data_source_loop_coro(std::atomic& running, std::shared_ptr source) { +concurrencpp::result Coro::data_source_loop_coro(std::shared_ptr source) { co_await concurrencpp::resume_on(executor_); while (running.load(std::memory_order_acquire)) { if (!source->enable || !source->registered()) { @@ -75,56 +117,57 @@ concurrencpp::result data_source_loop_coro(std::atomic& running, std } co_return; } -concurrencpp::result data_feed_loop_coro(std::atomic& running, std::shared_ptr feed) { - co_await concurrencpp::resume_on(executor_); - auto& mode_acs = Global::instance()->mode_acs; - auto& cfg = mode_acs.data_feed_config; - auto& pool = cfg.pool_; - auto& report_data_feed_msg_mum = mode_acs.report_data_feed_msg_mum; - while (running.load(std::memory_order_acquire)) { - std::vector messages; - { +concurrencpp::result Coro::data_feed_loop_coro(std::shared_ptr feed) { + co_await concurrencpp::resume_on(executor_); + auto& mode_acs = Global::instance()->mode_acs; + auto& cfg = mode_acs.data_feed_config; + auto& pool = cfg.pool_; + auto& report_data_feed_msg_mum = mode_acs.report_data_feed_msg_mum; + while (running.load(std::memory_order_acquire)) { + std::vector messages; + { + if (!feed->enable || !feed->registered()) { + break; + } + co_await feed->handle_in_loop_coro(); + auto s_num = feed->msg_buffer.mode_s_msg_num.load(); + auto other_num = feed->msg_buffer.mode_other_msg_num.load(); + const std::vector& all = feed->msg_buffer.get_all(); + Pool_Guard pg(&pool, all); + auto num = report_data_feed_msg_mum.load(); + auto monitor_msg_live = mode_acs.monitor_msg_live.load(); + if (monitor_msg_live && num != 0 && all.size() > num && too_many_msg_limit.test()) { + std::ostringstream oss; + oss << "feed_key:" << feed->key << " "; + oss << "recv_s:" << s_num << " "; + oss << "recv_other:" << other_num << " "; + std::cout << oss.str() << std::endl; + } + messages.reserve(all.size()); + for (const auto* str : all) { + if (!str || str->empty()) { + std::cout << "empty feed message:" << feed->key << std::endl; + continue; + } + messages.emplace_back(*str); + } + } + if (messages.empty()) { + co_await concurrencpp::resume_on(executor_); + continue; + } if (!feed->enable || !feed->registered()) { break; } - co_await feed->handle_in_loop_coro(); - auto s_num = feed->msg_buffer.mode_s_msg_num.load(); - auto other_num = feed->msg_buffer.mode_other_msg_num.load(); - const std::vector& all = feed->msg_buffer.get_all(); - Pool_Guard pg(&pool, all); - auto num = report_data_feed_msg_mum.load(); - auto monitor_msg_live = mode_acs.monitor_msg_live.load(); - if (monitor_msg_live && num != 0 && all.size() > num && too_many_msg_limit.test()) { - std::ostringstream oss; - oss << "feed_key:" << feed->key << " "; - oss << "recv_s:" << s_num << " "; - oss << "recv_other:" << other_num << " "; - std::cout << oss.str() << std::endl; + for (auto& msg : messages) { + co_await feed->send_coro(msg); } - messages.reserve(all.size()); - for (const auto* str : all) { - if (!str || str->empty()) { - std::cout << "empty feed message:" << feed->key << std::endl; - continue; - } - messages.emplace_back(*str); - } - } - if (messages.empty()) { co_await concurrencpp::resume_on(executor_); - continue; } - if (!feed->enable || !feed->registered()) { - break; - } - for (auto& msg : messages) { - co_await feed->send_coro(msg); - } - co_await concurrencpp::resume_on(executor_); + co_return; } - co_return; -} -bool has_running_loop_tasks() { + +bool Coro::has_running_loop_tasks() { auto sources = Global::instance()->mode_acs.data_source_config.map.list(); for (auto& source : sources) { if (source && task_is_running(source->loop_task)) { @@ -139,54 +182,5 @@ bool has_running_loop_tasks() { } return false; } -concurrencpp::result coro_thread(std::atomic& running) { - co_await concurrencpp::resume_on(executor_); - auto g = Global::instance(); - for (auto& feed : g->mode_acs.data_feed_config.map.list()) { - if (feed->enable) { - auto ret = feed->check_and_open(); - if (!ret) { - std::cout << "Data feed first open failed: " << feed->key << std::endl; - } - } - } - while (running.load(std::memory_order_acquire)) { - auto sources = Global::instance()->mode_acs.data_source_config.map.list(); - for (auto& source : sources) { - if (!source || !source->enable) { - continue; - } - if (task_is_running(source->loop_task)) { - continue; - } - if (!source->is_open() && !source->open()) { - continue; - } - source->loop_task = std::make_unique>(data_source_loop_coro(running, source)); - } - auto feeds = Global::instance()->mode_acs.data_feed_config.map.list(); - for (auto& feed : feeds) { - if (!feed || !feed->enable) { - continue; - } - if (task_is_running(feed->loop_task)) { - continue; - } - auto ret = feed->check_and_open(); - if (!ret) { - continue; - } - feed->loop_task = std::make_unique>(data_feed_loop_coro(running, feed)); - } - co_await concurrencpp::resume_on(executor_); - } - auto start = std::chrono::steady_clock::now(); - while (has_running_loop_tasks()) { - if (std::chrono::steady_clock::now() - start > std::chrono::seconds(10)) { - std::cerr << "data loop task stop timeout" << std::endl; - break; - } - co_await concurrencpp::resume_on(executor_); - } - co_return; -} \ No newline at end of file + + diff --git a/module/Local_Server/server/io_coro.h b/module/Local_Server/server/io_coro.h new file mode 100644 index 0000000..e654764 --- /dev/null +++ b/module/Local_Server/server/io_coro.h @@ -0,0 +1,31 @@ +#pragma once +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "Core/Statistics/Frequency_Limit.h" +#include "Local_Server/Data_Feed/Data_Feed.h" +#include "Local_Server/Data_Source/Data_Source.h" +class Coro { +public: + void start(); + void stop(); +private: + bool task_is_running(std::unique_ptr>& task); + concurrencpp::result data_source_loop_coro(std::shared_ptr source); + concurrencpp::result data_feed_loop_coro(std::shared_ptr feed); + concurrencpp::result coro_thread(); + bool has_running_loop_tasks(); + Frequency_Limit too_many_msg_limit; + Frequency_Limit flush_limit; + concurrencpp::runtime runtime_; + std::shared_ptr executor_; + std::unique_ptr> data_feed_thread_task; + std::atomic running = false; +}; diff --git a/module/Local_Server_main.cpp b/module/Local_Server_main.cpp index e2cfd6b..f3735fc 100644 --- a/module/Local_Server_main.cpp +++ b/module/Local_Server_main.cpp @@ -11,6 +11,8 @@ #include #include #include + +#include "Local_Server/server/io_coro.h" // ./server ./config.json #ifdef __linux__ @@ -182,8 +184,9 @@ int psc_main(int argc, char *argv[]) { bool enable = g->mlat.enable; - void start_io_coro(); - start_io_coro(); + Coro coro; + + coro.start(); // Catch_Memory cm; @@ -208,9 +211,7 @@ int psc_main(int argc, char *argv[]) { } } - void stop_io_coro(); - stop_io_coro(); - + coro.stop(); if (stop_program == SIGINT || stop_program == SIGTERM) { clear();