From 05dec135ed498a83aa53c6be0bdf1fb6ac4b8b11 Mon Sep 17 00:00:00 2001 From: wyc <1104749580@qq.com> Date: Tue, 23 Jun 2026 18:06:44 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8D=8F=E7=A8=8B=E8=B0=83=E5=BA=A6=E5=99=A8?= =?UTF-8?q?=EF=BC=8C=E9=87=8D=E5=A4=A7=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Data_Feed/data_feed_thread.cpp | 146 ------ .../Local_Server/Data_Source/Data_Source.cpp | 36 +- module/Local_Server/Data_Source/Data_Source.h | 9 +- module/Local_Server/server/io_coro.cpp | 490 ++++++++++++++++++ module/Local_Server_main.cpp | 4 +- 5 files changed, 528 insertions(+), 157 deletions(-) delete mode 100644 module/Local_Server/Data_Feed/data_feed_thread.cpp create mode 100644 module/Local_Server/server/io_coro.cpp diff --git a/module/Local_Server/Data_Feed/data_feed_thread.cpp b/module/Local_Server/Data_Feed/data_feed_thread.cpp deleted file mode 100644 index 4c23d30..0000000 --- a/module/Local_Server/Data_Feed/data_feed_thread.cpp +++ /dev/null @@ -1,146 +0,0 @@ -#include "../server/Global.h" -#include "../server/Mode_Msg_Buffer.h" -#include -#include -long long get_current_milliseconds() { - auto now = std::chrono::system_clock::now(); - auto duration = std::chrono::duration_cast(now.time_since_epoch()); - return duration.count(); -} -Frequency_Limit too_many_msg_limit; -Frequency_Limit flush_limit; - - - -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(); - if (!ret) { - std::cout << "[Data_feed_Config] [" << t->key << "] 第一次打开失败！ 程序退出! " << std::endl; - } - } - } - auto& mode_acs = Global::instance()->mode_acs; - auto& cfg = mode_acs.data_feed_config; - auto& pool = cfg.pool_; - auto t = cfg.empty_wait_milliseconds.load(); - 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(); - 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->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); - } - - } - - // 这两个每次循环都要重新计算 保证实时性 - // if (flush_limit.test()) { - auto all_feed = cfg.map.list(); - enable_feeds.clear(); - for (auto& feed : all_feed) { - if (!feed->enable) continue; - enable_feeds.push_back(feed); - // 执行特定的刷新逻辑 比如 tcp_server 检查自身的连接 - co_await feed->handle_in_loop_coro(); - } - - - int m = static_cast(enable_feeds.size()); - for (int j = 0; j < m; j++) { - auto& feed = enable_feeds[j]; - 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(); - const std::vector& all = feed->msg_buffer.get_all(); - // 释放内存池 - - - Pool_Guard pg(&pool, all); - - auto num = report_data_feed_msg_mum.load(); - if (monitor_msg_live) { - if (num != 0 && all.size() > num) { - if (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; - } - } - } - if (all.empty()) continue; - all_empty = false; - auto n = all.size(); - - - - for (auto i = 0; i < n; i++) { - const auto& str = all[i]; - if (str->empty()) { - std::cout << "警告:未知原因:" << feed->key << "内有空的消息体" << std::endl; - continue; - } - co_await feed->send_coro(*str); - } - } - if (t != 0) { - if (all_empty) { - std::this_thread::sleep_for(std::chrono::milliseconds(t * 2)); - } else { - std::this_thread::sleep_for(std::chrono::milliseconds(t)); - } - } - } - co_return; -} - -void io_coro(std::atomic& running) { - - try { - ucoro::sync_await(data_feed_thread_coro(running)); - } catch (const std::exception& e) { - std::cerr << "data_feed_thread_coro exception: " << e.what() << std::endl; - Psc::fail_fast_core_dump(""); - } catch (...) { - std::cerr << "data_feed_thread_coro unknown exception" << std::endl; - Psc::fail_fast_core_dump(""); - } - -} - diff --git a/module/Local_Server/Data_Source/Data_Source.cpp b/module/Local_Server/Data_Source/Data_Source.cpp index 288ca04..df7f4e4 100644 --- a/module/Local_Server/Data_Source/Data_Source.cpp +++ b/module/Local_Server/Data_Source/Data_Source.cpp @@ -79,8 +79,8 @@ std::string Data_Source::thread_key() const { } void Data_Source::test_and_attach_thread() { - if (enable) { - open(); + if (enable && open()) { + data_source_loop_requested_.store(true, std::memory_order_release); } } @@ -109,8 +109,30 @@ void File_Data_Source::before_handle_msg(std::shared_ptr& msg) { -void Data_Source::test_and_stop_thread() const { - Global::instance()->thread_manager.test_and_stop_thread(thread_key()); +void Data_Source::test_and_stop_thread() { + data_source_loop_requested_.store(false, std::memory_order_release); + data_source_loop_generation_.fetch_add(1, std::memory_order_acq_rel); +} + +bool Data_Source::data_source_loop_should_run() const { + return data_source_loop_requested_.load(std::memory_order_acquire); +} + +std::uint64_t Data_Source::data_source_loop_generation() const { + return data_source_loop_generation_.load(std::memory_order_acquire); +} + +bool Data_Source::try_mark_data_source_loop_running() { + bool expected = false; + return data_source_loop_running_.compare_exchange_strong( + expected, + true, + std::memory_order_acq_rel, + std::memory_order_acquire); +} + +void Data_Source::mark_data_source_loop_stopped() { + data_source_loop_running_.store(false, std::memory_order_release); } @@ -603,14 +625,12 @@ void Data_Source_Config::server(Global* g) { HTTP_REQUIRE_VALUE(key, params.try_get_string("key")) HTTP_REQUIRE_VALUE(ds, map.get(key)) // 直接关闭 + ds->test_and_stop_thread(); ds->close(); ds->from_json(¶ms); if (ds->enable) { ds->open(); ds->test_and_attach_thread(); - } else - { - ds->test_and_stop_thread(); } g->save(); g->mode_acs.source_feed_relation_config.set_need_refresh(); @@ -642,8 +662,8 @@ void Data_Source_Config::server(Global* g) { CHECK_JSON_PARAM HTTP_REQUIRE_VALUE(index, params.try_get_number("index")) HTTP_REQUIRE_VALUE(ds, map.try_get(index)) - ds->close(); ds->test_and_stop_thread(); + ds->close(); HTTP_REQUIRE_TRUE(map.remove(index), "index") g->save(); diff --git a/module/Local_Server/Data_Source/Data_Source.h b/module/Local_Server/Data_Source/Data_Source.h index 7ca98f7..d762a75 100644 --- a/module/Local_Server/Data_Source/Data_Source.h +++ b/module/Local_Server/Data_Source/Data_Source.h @@ -98,7 +98,11 @@ public: Frequency_Limit statistic_fl; std::string thread_key() const; void test_and_attach_thread(); - void test_and_stop_thread() const; + void test_and_stop_thread(); + bool data_source_loop_should_run() const; + std::uint64_t data_source_loop_generation() const; + bool try_mark_data_source_loop_running(); + void mark_data_source_loop_stopped(); Psc::JSON statistic_json() { Psc::JSON ret = Psc::JSON::object(); Ret_J(enable); @@ -173,6 +177,9 @@ protected: bool _open_ = false; + std::atomic data_source_loop_requested_ = false; + std::atomic data_source_loop_running_ = false; + std::atomic data_source_loop_generation_ = 0; virtual bool _open() = 0; virtual void _close() = 0; }; diff --git a/module/Local_Server/server/io_coro.cpp b/module/Local_Server/server/io_coro.cpp new file mode 100644 index 0000000..65dbd94 --- /dev/null +++ b/module/Local_Server/server/io_coro.cpp @@ -0,0 +1,490 @@ +#include "../server/Global.h" +#include "../server/Mode_Msg_Buffer.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "ucoro/single_thread.h" + +bool is_ucoro_operation_cancelled(std::exception_ptr exception) noexcept { + if (!exception) { + return false; + } + + try { + std::rethrow_exception(exception); + } catch (const ucoro::operation_cancelled&) { + return true; + } catch (...) { + return false; + } +} + + + +Frequency_Limit too_many_msg_limit; +Frequency_Limit flush_limit; + +namespace { + + + + +ucoro::Single_Thread_Scheduler& data_feed_thread_scheduler() { + static ucoro::Single_Thread_Scheduler scheduler; + return scheduler; +} + +ucoro::awaitable data_feed_thread_yield_coro() { + co_await ucoro::callback_awaitable([](auto done) mutable { + data_feed_thread_scheduler().post( + [done = std::move(done)]() mutable { + done(); + } + ); + }); +} + +ucoro::awaitable data_feed_thread_wait_event_coro() { + co_await ucoro::callback_awaitable([](auto done) mutable { + data_feed_thread_scheduler().async_wait( + [done = std::move(done)]() mutable { + done(); + } + ); + }); +} + +struct Data_Source_Process_Result { + std::exception_ptr exception; +}; + +ucoro::awaitable process_data_source_data_coro( + std::shared_ptr source, + std::string mode_data +) { + 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); + + data_feed_thread_scheduler().post( + [done = std::move(done)]() mutable { + done(Data_Source_Process_Result{}); + } + ); + } catch (...) { + auto exception = std::current_exception(); + + data_feed_thread_scheduler().post( + [done = std::move(done), exception]() mutable { + done(Data_Source_Process_Result{exception}); + } + ); + } + } + ); + } + ); + + co_return result; +} + +ucoro::awaitable data_source_loop_coro( + std::atomic& running, + std::shared_ptr source +) { + const auto generation = source->data_source_loop_generation(); + + while (running.load(std::memory_order_acquire) && + source->data_source_loop_generation() == generation && + source->data_source_loop_should_run()) { + if (!source->enable) { + break; + } + + if (!source->is_open() && !source->open()) { + co_await data_feed_thread_wait_event_coro(); + continue; + } + + co_await source->handle_in_loop_coro(); + + auto mode_data = co_await source->read_coro(); + + if (!running.load(std::memory_order_acquire) || + source->data_source_loop_generation() != generation || + !source->data_source_loop_should_run()) { + break; + } + + auto result = co_await process_data_source_data_coro( + source, + std::move(mode_data) + ); + + if (result.exception) { + std::rethrow_exception(result.exception); + } + + co_await data_feed_thread_yield_coro(); + } + + co_return; +} + +bool data_feed_still_registered(const std::shared_ptr& feed) { + if (!feed) { + return false; + } + + auto all_feed = Global::instance()->mode_acs.data_feed_config.map.list(); + + for (auto& item : all_feed) { + if (item.get() == feed.get()) { + return true; + } + } + + return false; +} + +ucoro::awaitable data_feed_loop_coro( + std::atomic& running, + std::shared_ptr feed +) { + 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) && + feed && + feed->enable && + data_feed_still_registered(feed)) { + 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) { + if (num != 0 && all.size() > num) { + if (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; + } + } + } + + if (all.empty()) { + co_await data_feed_thread_wait_event_coro(); + continue; + } + + const auto n = all.size(); + + for (size_t i = 0; i < n; ++i) { + const auto* str = all[i]; + + if (!str || str->empty()) { + std::cout << "警告:未知原因:" + << feed->key + << "内有空的消息体" + << std::endl; + continue; + } + + std::string msg = *str; + co_await feed->send_coro(msg); + } + + co_await data_feed_thread_yield_coro(); + } + + co_return; +} + +using Data_Source_Task_Map = std::map>; +using Data_Feed_Task_Map = std::map>; + +void cleanup_data_source_loop_tasks(Data_Source_Task_Map& tasks) { + for (auto it = tasks.begin(); it != tasks.end();) { + if (!it->second.valid()) { + it = tasks.erase(it); + } else { + ++it; + } + } +} + +void cleanup_data_feed_loop_tasks(Data_Feed_Task_Map& tasks) { + for (auto it = tasks.begin(); it != tasks.end();) { + if (!it->second.valid()) { + it = tasks.erase(it); + } else { + ++it; + } + } +} + +void sync_data_source_loop_tasks( + std::atomic& running, + Data_Source_Task_Map& tasks +) { + cleanup_data_source_loop_tasks(tasks); + + auto sources = Global::instance()->mode_acs.data_source_config.map.list(); + + for (auto& source : sources) { + if (!source->enable || !source->data_source_loop_should_run()) { + continue; + } + + if (tasks.find(source.get()) != tasks.end()) { + continue; + } + + source->test_and_attach_thread(); + + if (!source->is_open() && !source->open()) { + continue; + } + + if (!source->try_mark_data_source_loop_running()) { + continue; + } + + auto task = data_source_loop_coro(running, source).detach_with_callback( + [source](std::exception_ptr exception) mutable { + source->mark_data_source_loop_stopped(); + + if (!is_ucoro_operation_cancelled(exception)) { + data_feed_thread_scheduler().set_exception(exception); + } + } + ); + + task.start(); + + if (task.valid()) { + tasks.emplace(source.get(), std::move(task)); + } + } +} + +void sync_data_feed_loop_tasks( + std::atomic& running, + Data_Feed_Task_Map& tasks +) { + cleanup_data_feed_loop_tasks(tasks); + + auto feeds = Global::instance()->mode_acs.data_feed_config.map.list(); + + for (auto& feed : feeds) { + if (!feed || !feed->enable) { + continue; + } + + if (tasks.find(feed.get()) != tasks.end()) { + continue; + } + + auto ret = feed->check_and_open(); + + if (!ret) { + continue; + } + + auto task = data_feed_loop_coro(running, feed).detach_with_callback( + [feed](std::exception_ptr exception) mutable { + if (!is_ucoro_operation_cancelled(exception)) { + data_feed_thread_scheduler().set_exception(exception); + } + } + ); + + task.start(); + + if (task.valid()) { + tasks.emplace(feed.get(), std::move(task)); + } + } +} + +void stop_all_data_source_loop_tasks(Data_Source_Task_Map& tasks) { + auto sources = Global::instance()->mode_acs.data_source_config.map.list(); + + for (auto& source : sources) { + source->test_and_stop_thread(); + } + + auto& scheduler = data_feed_thread_scheduler(); + scheduler.wake(); + scheduler.cleanup_abandoned_tasks(); + + auto start = std::chrono::steady_clock::now(); + + while (!tasks.empty()) { + scheduler.rethrow_if_exception(); + + scheduler.drain(); + + cleanup_data_source_loop_tasks(tasks); + scheduler.cleanup_abandoned_tasks(); + + if (tasks.empty()) { + break; + } + + if (std::chrono::steady_clock::now() - start > std::chrono::seconds(10)) { + std::cerr << "data_source_loop_coro stop timeout, abandon remaining tasks" + << std::endl; + + scheduler.abandon_remaining_tasks(tasks); + break; + } + + scheduler.wait_for_callback_for(std::chrono::milliseconds(1)); + } + + scheduler.cleanup_abandoned_tasks(); +} + +void stop_all_data_feed_loop_tasks(Data_Feed_Task_Map& tasks) { + auto& scheduler = data_feed_thread_scheduler(); + scheduler.wake(); + scheduler.cleanup_abandoned_tasks(); + + auto start = std::chrono::steady_clock::now(); + + while (!tasks.empty()) { + scheduler.rethrow_if_exception(); + + scheduler.drain(); + + cleanup_data_feed_loop_tasks(tasks); + scheduler.cleanup_abandoned_tasks(); + + if (tasks.empty()) { + break; + } + + if (std::chrono::steady_clock::now() - start > std::chrono::seconds(10)) { + std::cerr << "data_feed_loop_coro stop timeout, abandon remaining tasks" + << std::endl; + + scheduler.abandon_remaining_tasks(tasks); + break; + } + + scheduler.wait_for_callback_for(std::chrono::milliseconds(1)); + } + + scheduler.cleanup_abandoned_tasks(); +} + +} // namespace + +void wake_data_feed_thread() { + data_feed_thread_scheduler().wake(); +} + +void stop_data_feed_thread_scheduler() { + data_feed_thread_scheduler().stop(); +} + +ucoro::awaitable data_feed_thread_coro(std::atomic& running) { + data_feed_thread_scheduler().reset(); + + auto g = Global::instance(); + + for (auto& source : g->mode_acs.data_source_config.map.list()) { + if (source->enable) { + source->test_and_attach_thread(); + } + } + + 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_Config] [" + << feed->key + << "] 第一次打开失败！ 程序继续运行，等待后续重试。" + << std::endl; + } + } + } + + Data_Source_Task_Map data_source_tasks; + Data_Feed_Task_Map data_feed_tasks; + + while (running.load(std::memory_order_acquire)) { + auto& scheduler = data_feed_thread_scheduler(); + + scheduler.rethrow_if_exception(); + scheduler.cleanup_abandoned_tasks(); + + sync_data_source_loop_tasks(running, data_source_tasks); + sync_data_feed_loop_tasks(running, data_feed_tasks); + + const auto resumed = scheduler.drain(); + + cleanup_data_source_loop_tasks(data_source_tasks); + cleanup_data_feed_loop_tasks(data_feed_tasks); + scheduler.cleanup_abandoned_tasks(); + + if (resumed == 0) { + scheduler.wait_for_work([&running] { + return !running.load(std::memory_order_acquire); + }); + } + } + + data_feed_thread_scheduler().stop(); + + stop_all_data_source_loop_tasks(data_source_tasks); + stop_all_data_feed_loop_tasks(data_feed_tasks); + + co_return; +} + +void io_coro(std::atomic& running) { + try { + ucoro::sync_await(data_feed_thread_coro(running)); + } catch (const std::exception& e) { + std::cerr << "data_feed_thread_coro exception: " + << e.what() + << std::endl; + Psc::fail_fast_core_dump(""); + } catch (...) { + std::cerr << "data_feed_thread_coro unknown exception" << std::endl; + Psc::fail_fast_core_dump(""); + } +} diff --git a/module/Local_Server_main.cpp b/module/Local_Server_main.cpp index 0dcfa0b..a0ab003 100644 --- a/module/Local_Server_main.cpp +++ b/module/Local_Server_main.cpp @@ -55,7 +55,7 @@ struct Catch_Memory { std::int64_t sm; }; -int wyc_main(int argc, char *argv[]) { +int psc_main(int argc, char *argv[]) { std::cout << "wyc_main" << std::endl; if (argc >= 2) { @@ -217,7 +217,7 @@ int wyc_main(int argc, char *argv[]) { int main(int argc, char* argv[]) { - return redict_main_with_gtest(argc, argv, wyc_main); + return redict_main_with_gtest(argc, argv, psc_main); }