ae/ECAP_Server
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

常规更新

Browse Source
This commit is contained in:
种太阳的热带美短
2026-06-26 18:08:41 +08:00
parent 64a4218c00
commit c683038d3a
1 changed files with 62 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/server/io_coro.cpp
+62 -51
View File
@@ -8,13 +8,12 @@
#include <format> #include <format>
#include <sstream> #include <sstream>
#include <thread> #include <thread>
#include <exception>
static std::string thread_id_str() { static std::string thread_id_str() {
std::ostringstream oss; std::ostringstream oss;
oss << std::this_thread::get_id(); oss << std::this_thread::get_id();
return oss.str(); return oss.str();
} }
std::vector<std::shared_ptr<With_Loop_Coro>> get_all() { std::vector<std::shared_ptr<With_Loop_Coro>> get_all() {
auto g = Global::instance(); auto g = Global::instance();
std::vector<std::shared_ptr<With_Loop_Coro>> ret; std::vector<std::shared_ptr<With_Loop_Coro>> ret;
@@ -28,7 +27,6 @@ std::vector<std::shared_ptr<With_Loop_Coro>> get_all() {
} }
return ret; return ret;
} }
bool Coro::has_running_loop_tasks() { bool Coro::has_running_loop_tasks() {
auto list = get_all(); auto list = get_all();
for (auto& li : list) { for (auto& li : list) {
@@ -38,54 +36,70 @@ bool Coro::has_running_loop_tasks() {
} }
return false; return false;
} }
void Coro::start() { void Coro::start() {
std::cout << "start_io_coro" << std::endl; std::cout << "start_io_coro" << std::endl;
running.store(true, std::memory_order_release); running.store(true, std::memory_order_release);
io = runtime_.make_executor<concurrencpp::worker_thread_executor>( io = runtime_.make_executor<concurrencpp::worker_thread_executor>(
[](std::string_view thread_name) { [](std::string_view thread_name) {
std::cout << std::format("io:{} 协程启动！\n", thread_name); std::cout << std::format("协程io:{} 协程启动！\n", thread_name);
}, },
[](std::string_view thread_name) { [](std::string_view thread_name) {
std::cout << std::format("io:{} 协程销毁！\n", thread_name); std::cout << std::format("协程io:{} 销毁！\n", thread_name);
}); });
// process_data = runtime_.make_executor<concurrencpp::worker_thread_executor>(
// [](std::string_view thread_name) {
// std::cout << std::format("process_data 协程:{} 协程启动！\n", thread_name);
// },
// [](std::string_view thread_name) {
// std::cout << std::format("process_data 协程:{} 销毁！\n", thread_name);
// });
auto n = std::thread::hardware_concurrency();
auto other_need = std::max<size_t>(2, std::thread::hardware_concurrency() / 2);
if (n > other_need) {
n -= other_need;
}
else {
n = n / 2;
}
process_data = runtime_.make_executor<concurrencpp::thread_pool_executor>( process_data = runtime_.make_executor<concurrencpp::thread_pool_executor>(
"process_data", "process_data",
4, n,
std::chrono::seconds(5), std::chrono::seconds(5),
[](std::string_view thread_name) { [](std::string_view thread_name) {
std::cout << std::format("{} 协程启动！\n", thread_name); std::cout << std::format("多线程协程{} 启动！\n", thread_name);
}, },
[](std::string_view thread_name) { [](std::string_view thread_name) {
std::cout << std::format("{} 协程销毁！\n", thread_name); std::cout << std::format("多线程协程{} 销毁！\n", thread_name);
}); });
std::cout << std::format("process_data concurrency: {}\n", process_data->max_concurrency_level()); std::cout << std::format("协程已经启动 process_data concurrency: 在n个线程上{}\n", process_data->max_concurrency_level());
data_feed_thread_task = std::make_unique<concurrencpp::result<void>>(coro_thread()); data_feed_thread_task = std::make_unique<concurrencpp::result<void>>(coro_thread());
} }
concurrencpp::result<void> Coro::sleep_for(std::chrono::milliseconds ms) {
co_await runtime_.timer_queue()->make_delay_object(ms, io);
co_return;
}
void Coro::stop() { void Coro::stop() {
if (!running.exchange(false, std::memory_order_acq_rel)) {
return;
}
std::cout << "stop_io_coro begin" << std::endl; std::cout << "stop_io_coro begin" << std::endl;
auto list = get_all(); auto list = get_all();
for (auto& li : list) { for (auto& li : list) {
std::cout << std::format("请求停止 {}:{}\n", li->type, li->key); if (li->running()) {
li->async_stop(); std::cout << std::format("请求停止 {}:{} 当前状态为{}\n", li->type, li->key, Psc::to_string(li->state));
li->async_stop();
}
} }
// 同步等待
for (auto& li : list) { for (auto& li : list) {
std::cout << std::format("强制关闭 {}:{}\n", li->type, li->key); li->sync_wait();
li->force_close();
} }
// 停止状态更新状态机
running = false;
if (data_feed_thread_task) { if (data_feed_thread_task) {
std::cout << "等待 coro_thread 退出" << std::endl; std::cout << "等待 coro_thread 退出" << std::endl;
data_feed_thread_task->get(); data_feed_thread_task->get();
data_feed_thread_task.reset(); data_feed_thread_task.reset();
std::cout << "coro_thread 已退出" << std::endl; std::cout << "coro_thread 已退出" << std::endl;
} }
for (auto& li : list) { // 清理资源
li->sync_wait();
}
for (auto& li : list) { for (auto& li : list) {
li->loop_task.reset(); li->loop_task.reset();
} }
@@ -93,33 +107,18 @@ void Coro::stop() {
io.reset(); io.reset();
std::cout << "stop_io_coro end" << std::endl; std::cout << "stop_io_coro end" << std::endl;
} }
concurrencpp::result<void> Coro::sleep_for(std::chrono::milliseconds ms) {
co_await runtime_.timer_queue()->make_delay_object(ms, io);
co_return;
}
concurrencpp::result<void> Coro::coro_thread() { concurrencpp::result<void> Coro::coro_thread() {
co_await concurrencpp::resume_on(io); co_await concurrencpp::resume_on(io);
while (running.load(std::memory_order_acquire)) { while (running.load(std::memory_order_acquire)) {
auto list = get_all(); auto list = get_all();
for (auto& li : list) { for (auto& li : list) {
co_await li->sync_coro_loop_and_enable(); co_await li->tick(); //状态机驱动函数
} }
co_await sleep_for(std::chrono::milliseconds(10)); co_await sleep_for(std::chrono::milliseconds(10));
} }
std::cout << "coro_thread exit begin" << std::endl;
auto list = get_all();
for (auto& li : list) {
li->async_stop();
li->force_close();
}
std::cout << "coro_thread exit end" << std::endl;
co_return; co_return;
} }
bool data_feed_debug = false;
bool debug = false;
concurrencpp::result<void> Data_Feed::loop_coro() { concurrencpp::result<void> Data_Feed::loop_coro() {
auto feed = this; auto feed = this;
auto co = Coro::instance(); auto co = Coro::instance();
@@ -128,8 +127,8 @@ concurrencpp::result<void> Data_Feed::loop_coro() {
auto& cfg = mode_acs.data_feed_config; auto& cfg = mode_acs.data_feed_config;
auto& pool = cfg.pool_; auto& pool = cfg.pool_;
auto& report_data_feed_msg_mum = mode_acs.report_data_feed_msg_mum; auto& report_data_feed_msg_mum = mode_acs.report_data_feed_msg_mum;
while (feed->enable) { while (feed->running()) {
if (debug) { if (data_feed_debug) {
std::cout << std::format("{} feed loop begin {}\n", key, thread_id_str()); std::cout << std::format("{} feed loop begin {}\n", key, thread_id_str());
} }
std::vector<std::string> messages; std::vector<std::string> messages;
@@ -152,7 +151,7 @@ concurrencpp::result<void> Data_Feed::loop_coro() {
std::cout << oss.str() << std::endl; std::cout << oss.str() << std::endl;
} }
messages.reserve(all.size()); messages.reserve(all.size());
for (const auto* str : all) { for (const auto *str : all) {
if (!str || str->empty()) { if (!str || str->empty()) {
std::cout << "empty feed message:" << feed->key << std::endl; std::cout << "empty feed message:" << feed->key << std::endl;
continue; continue;
@@ -171,11 +170,11 @@ concurrencpp::result<void> Data_Feed::loop_coro() {
if (!feed->enable) { if (!feed->enable) {
break; break;
} }
if (debug) { if (data_feed_debug) {
std::cout << std::format("{} before send {}\n", key, thread_id_str()); std::cout << std::format("{} before send {}\n", key, thread_id_str());
} }
co_await feed->send_coro(msg); co_await feed->send_coro(msg);
if (debug) { if (data_feed_debug) {
std::cout << std::format("{} after send {}\n", key, thread_id_str()); std::cout << std::format("{} after send {}\n", key, thread_id_str());
} }
} }
@@ -184,43 +183,55 @@ concurrencpp::result<void> Data_Feed::loop_coro() {
std::cout << std::format("{} feed loop exit {}\n", key, thread_id_str()); std::cout << std::format("{} feed loop exit {}\n", key, thread_id_str());
co_return; co_return;
} }
bool data_source_debug = false;
concurrencpp::result<void> Data_Source::loop_coro() { concurrencpp::result<void> Data_Source::loop_coro() {
auto source = this; auto source = this;
auto co = Coro::instance(); auto co = Coro::instance();
co_await concurrencpp::resume_on(co->io); co_await concurrencpp::resume_on(co->io);
while (enable) { while (source->running()) {
if (debug) { if (data_source_debug) {
std::cout << std::format("{} source loop begin {}\n", key, thread_id_str()); std::cout << std::format("{} source loop begin {}\n", key, thread_id_str());
} }
if (!source->running() || !source->registered()) { if (!source->running() || !source->registered()) {
break; break;
} }
if (debug) { if (data_source_debug) {
std::cout << std::format("{} before handle {}\n", key, thread_id_str()); std::cout << std::format("{} before handle {}\n", key, thread_id_str());
} }
co_await source->handle_in_loop_coro(); co_await source->handle_in_loop_coro();
if (debug) { if (data_source_debug) {
std::cout << std::format("{} before read {}\n", key, thread_id_str()); std::cout << std::format("{} before read {}\n", key, thread_id_str());
} }
auto mode_data = co_await source->read_coro(); auto mode_data = co_await source->read_coro();
if (debug) { if (data_source_debug) {
std::cout << std::format("{} after read {}\n", key, thread_id_str()); std::cout << std::format("{} after read {}\n", key, thread_id_str());
} }
if (!enable || !source->running() || !source->registered()) { if (!enable || !source->running() || !source->registered()) {
break; break;
} }
co_await concurrencpp::resume_on(co->process_data); co_await concurrencpp::resume_on(co->process_data);
if (debug) { if (data_source_debug) {
std::cout << std::format("{} before process {}\n", key, thread_id_str()); std::cout << std::format("{} before process {}\n", key, thread_id_str());
} }
auto num = source->process_mode_acs_data(mode_data); auto num = source->process_mode_acs_data(mode_data);
if (debug) { if (data_source_debug) {
std::cout << std::format("{} after process {} {}\n", key, num, thread_id_str()); std::cout << std::format("{} after process {} {}\n", key, num, thread_id_str());
} }
if (num == 0) { if (num == 0) {
static Frequency_Limit_Multi mt(0.1);
auto name = type + ":" + key;
if (mt.test(name)) {
std::cout << std::format("{} 没解析到数据睡眠10ms {} {}\n", name, num, thread_id_str());
}
co_await co->sleep_for(std::chrono::milliseconds(10)); co_await co->sleep_for(std::chrono::milliseconds(10));
} }
else {
static Frequency_Limit_Multi mt(0.1);
auto name = type + ":" + key;
if (mt.test(name)) {
std::cout << std::format("{} 解析到数据 {} {}\n", name, num, thread_id_str());
}
}
co_await concurrencpp::resume_on(co->io); co_await concurrencpp::resume_on(co->io);
} }
std::cout << std::format("{} source loop exit {}\n", key, thread_id_str()); std::cout << std::format("{} source loop exit {}\n", key, thread_id_str());