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-24 09:54:17 +08:00
parent 05dec135ed
commit 351fc5c453
7 changed files with 252 additions and 483 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/server/Global.h
+1
View File
@@ -291,6 +291,7 @@ public:
DELETE_COPY(Global)
};
void wake_data_feed_thread();
void io_coro(std::atomic<bool>& running);
module/Local_Server/server/io_coro.cpp
+220 -395
View File
@@ -1,6 +1,5 @@
#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include <atomic>
#include <chrono>
#include <condition_variable>
@@ -13,477 +12,303 @@
#include <sstream>
#include <thread>
#include <vector>
#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&) {
}
catch (const ucoro::operation_cancelled&) {
return true;
} catch (...) {
}
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<void> data_feed_thread_yield_coro() {
co_await ucoro::callback_awaitable<void>([](auto done) mutable {
data_feed_thread_scheduler().post(
[done = std::move(done)]() mutable {
done();
ucoro::Single_Thread_Scheduler& data_feed_thread_scheduler() {
static ucoro::Single_Thread_Scheduler scheduler;
return scheduler;
}
ucoro::awaitable<void> data_feed_thread_yield_coro() {
co_await ucoro::callback_awaitable<void>([](auto done) mutable {
data_feed_thread_scheduler().post(
[done = std::move(done)]() mutable {
done();
}
);
});
}
ucoro::awaitable<void> data_feed_thread_wait_event_coro() {
co_await ucoro::callback_awaitable<void>([](auto done) mutable {
data_feed_thread_scheduler().async_wait(
[done = std::move(done)]() mutable {
done();
}
);
});
}
ucoro::awaitable<void> data_source_loop_coro(
std::atomic<bool>& running,
std::shared_ptr<Data_Source> source
) {
while (running.load(std::memory_order_acquire) &&
source &&
source->enable &&
source->registered()) {
if (!source->is_open() && !source->open()) {
co_await data_feed_thread_wait_event_coro();
continue;
}
);
});
}
ucoro::awaitable<void> data_feed_thread_wait_event_coro() {
co_await ucoro::callback_awaitable<void>([](auto done) mutable {
data_feed_thread_scheduler().async_wait(
[done = std::move(done)]() mutable {
done();
co_await source->handle_in_loop_coro();
auto mode_data = co_await source->read_coro();
if (!running.load(std::memory_order_acquire) ||
!source->enable ||
!source->registered()) {
break;
}
);
});
}
struct Data_Source_Process_Result {
std::exception_ptr exception;
};
ucoro::awaitable<Data_Source_Process_Result> process_data_source_data_coro(
std::shared_ptr<Data_Source> source,
std::string mode_data
) {
auto result = co_await ucoro::callback_awaitable<Data_Source_Process_Result>(
[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});
}
);
source->process_mode_acs_data(mode_data);
co_await data_feed_thread_yield_coro();
}
co_return;
}
ucoro::awaitable<void> data_feed_loop_coro(
std::atomic<bool>& running,
std::shared_ptr<Data_Feed> 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 &&
feed->registered()) {
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<std::string*>& 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<Data_Source*, ucoro::awaitable<void>>;
using Data_Feed_Task_Map = std::map<Data_Feed*, ucoro::awaitable<void>>;
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<bool>& 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 || !source->enable) {
continue;
}
if (tasks.find(source.get()) != tasks.end()) {
continue;
}
if (!source->is_open() && !source->open()) {
continue;
}
auto task = data_source_loop_coro(running, source).detach_with_callback(
[source](std::exception_ptr exception) mutable {
if (!is_ucoro_operation_cancelled(exception)) {
data_feed_thread_scheduler().set_exception(exception);
}
}
);
}
);
co_return result;
}
ucoro::awaitable<void> data_source_loop_coro(
std::atomic<bool>& running,
std::shared_ptr<Data_Source> 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<Data_Feed>& 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<void> data_feed_loop_coro(
std::atomic<bool>& running,
std::shared_ptr<Data_Feed> 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<std::string*>& 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;
}
task.start();
if (task.valid()) {
tasks.emplace(source.get(), std::move(task));
}
}
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;
}
void sync_data_feed_loop_tasks(
std::atomic<bool>& 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;
}
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<Data_Source*, ucoro::awaitable<void>>;
using Data_Feed_Task_Map = std::map<Data_Feed*, ucoro::awaitable<void>>;
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<bool>& 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);
}
if (tasks.find(feed.get()) != tasks.end()) {
continue;
}
);
task.start();
if (task.valid()) {
tasks.emplace(source.get(), std::move(task));
}
}
}
void sync_data_feed_loop_tasks(
std::atomic<bool>& 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);
}
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));
}
);
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);
void stop_all_data_source_loop_tasks(Data_Source_Task_Map& tasks) {
auto& scheduler = data_feed_thread_scheduler();
scheduler.wake();
scheduler.cleanup_abandoned_tasks();
if (tasks.empty()) {
break;
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));
}
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<void> data_feed_thread_coro(std::atomic<bool>& 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;
<< 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<bool>& running) {
try {
ucoro::sync_await(data_feed_thread_coro(running));
} catch (const std::exception& e) {
}
catch (const std::exception& e) {
std::cerr << "data_feed_thread_coro exception: "
<< e.what()
<< std::endl;
<< e.what()
<< std::endl;
Psc::fail_fast_core_dump("");
} catch (...) {
}
catch (...) {
std::cerr << "data_feed_thread_coro unknown exception" << std::endl;
Psc::fail_fast_core_dump("");
}