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-07-16 09:44:54 +08:00
parent 8bb155029b
commit fb57d51ee5
6 changed files with 795 additions and 839 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/Data_Source/Data_Source_Handler.cpp
+36 -22
View File
@@ -2,6 +2,7 @@
#include "Data_Source.h"
#include "Local_Server/server/Global.h"
#include <string_view>
#include "Local_Server/server/io_coro.h"
using namespace Psc;
std::shared_ptr<Data_Source> ds(Data_Source_Handler* dsh) {
return dynamic_cast<Data_Source*>(dsh)->that();
@@ -130,28 +131,41 @@ size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data)
std::cout << source->key + " read:[mode_s_serial]:" << mem2hex(mode_data)
<< std::endl;
}
SSR::Binary_Format_handle_buffer(
source->buffer, mode_data, [this, source, &ret](std::string& packet) {
ret++;
auto msg = create_msg(packet);
if (!msg) return;
source->push_to_feed(msg);
auto mt = msg->type;
bool mode_s = mt == SSR::Msg::S7 || mt == SSR::Msg::S14;
if (mt == SSR::Msg::HULC_Status) {
source->handle_HULC(packet);
}
else if (mt == SSR::Msg::Radarcape_status) {
auto radarcape_msg = SSR::create_Radarcape_STATUS_Message(packet);
std::cout << radarcape_msg.toJson().to_json_string() << std::endl;
SSR::mode_s_logger->debug("Radarcape_status/radarcape", {},
radarcape_msg.toJson().to_json_string());
}
else if (mode_s) {
// 拓展点
handle_mode_s(std::dynamic_pointer_cast<SSR::Mode_S_Msg>(msg));
}
});
auto handle_packet = [this, source, &ret](std::string& packet) {
ret++;
auto msg = create_msg(packet);
if (!msg) return;
source->push_to_feed(msg);
auto mt = msg->type;
bool mode_s = mt == SSR::Msg::S7 || mt == SSR::Msg::S14;
if (mt == SSR::Msg::HULC_Status) {
source->handle_HULC(packet);
}
else if (mt == SSR::Msg::Radarcape_status) {
auto radarcape_msg = SSR::create_Radarcape_STATUS_Message(packet);
std::cout << radarcape_msg.toJson().to_json_string() << std::endl;
SSR::mode_s_logger->debug("Radarcape_status/radarcape", {},
radarcape_msg.toJson().to_json_string());
}
else if (mode_s) {
// 拓展点
handle_mode_s(std::dynamic_pointer_cast<SSR::Mode_S_Msg>(msg));
}
};
auto f = [this, source, handle_packet](std::string& packet) {
auto& wait = Global::instance()->mode_acs.wait_process_msg;
if (wait) {
handle_packet(packet);
}
else {
auto co = Coro::instance();
auto executor = co->process_data->get_executor();
asio::post(executor, [handle_packet, packet = std::move(packet)]() mutable {
handle_packet(packet);
});
}
};
SSR::Binary_Format_handle_buffer(source->buffer, mode_data, f);
return ret;
}
void Data_Source_Handler::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> mode_s_msg) {
module/Local_Server/server/Config.h
+1
View File
@@ -47,6 +47,7 @@ struct Mode_ACS_Config_Base_Data {
PSC_USE_JSON
};
struct Mode_ACS_Config_Data {
Psc::Copyable_Atomic<bool> wait_process_msg;
Psc::Copyable_Atomic<bool> time_space_filter;
Psc::Copyable_Atomic<bool> speed_filter;
Psc::Copyable_Atomic<std::size_t> max_track_point_size{};
module/Local_Server/server/io_coro.cpp
+270 -276
View File
@@ -14,299 +14,293 @@
#include <format>
#include <sstream>
#include <thread>
static std::string thread_id_str() {
std::ostringstream oss;
oss << std::this_thread::get_id();
return oss.str();
std::ostringstream oss;
oss << std::this_thread::get_id();
return oss.str();
}
std::vector<std::shared_ptr<With_Loop_Coro> > get_all() {
auto g = Global::instance();
std::vector<std::shared_ptr<With_Loop_Coro> > ret;
auto sources = g->mode_acs.data_source_config.map.list();
for (auto &source: sources) {
ret.emplace_back(source);
}
auto feeds = g->mode_acs.data_feed_config.map.list();
for (auto &feed: feeds) {
ret.emplace_back(feed);
}
return ret;
std::vector<std::shared_ptr<With_Loop_Coro>> get_all() {
auto g = Global::instance();
std::vector<std::shared_ptr<With_Loop_Coro>> ret;
auto sources = g->mode_acs.data_source_config.map.list();
for (auto& source : sources) {
ret.emplace_back(source);
}
auto feeds = g->mode_acs.data_feed_config.map.list();
for (auto& feed : feeds) {
ret.emplace_back(feed);
}
return ret;
}
bool Coro::has_running_loop_tasks() {
auto list = get_all();
for (auto &li: list) {
if (li->running()) {
return true;
}
}
return false;
auto list = get_all();
for (auto& li : list) {
if (li->running()) {
return true;
}
}
return false;
}
bool data_feed_debug = false;
bool data_source_debug = false;
// 不开启wait 性能巨差 数据积压
bool wait = true;
void Coro::start() {
std::cout << "1================ start_io_coro" << std::endl;
running.store(true, std::memory_order_release);
io.restart();
io_work = std::make_unique<asio::executor_work_guard<asio::io_context::executor_type> >(io.get_executor());
io_thread = std::thread([this] {
try {
std::cout << std::format("协程io:{} 协程启动！\n", thread_id_str());
io.run();
std::cout << std::format("协程io:{} 销毁！\n", thread_id_str());
} catch (const std::exception &e) {
std::cerr << std::format("协程io异常: {}\n", e.what());
std::terminate();
} catch (...) {
std::cerr << "协程io未知异常\n";
std::terminate();
}
});
auto n = std::max<unsigned int>(1, std::thread::hardware_concurrency());
auto other_need = std::max<unsigned int>(2, n / 2);
if (n > other_need) {
n -= other_need;
} else {
n = n / 2;
}
n = std::max<unsigned int>(1, n);
process_data = std::make_unique<asio::thread_pool>(n);
std::cout << std::format("协程已经启动 process_data concurrency: 在n个线程上{}\n", n);
auto promise = std::make_shared<std::promise<void> >();
loop_task = std::make_unique<std::future<void> >(promise->get_future());
asio::co_spawn(io, coro_thread(), [promise](std::exception_ptr ep) {
std::cout << "发生了异常" << std::endl;
if (ep) {
promise->set_exception(ep);
std::rethrow_exception(ep);
}
promise->set_value();
});
std::cout << std::flush;
std::cout << "1================ start_io_coro" << std::endl;
running.store(true, std::memory_order_release);
io.restart();
io_work = std::make_unique<asio::executor_work_guard<asio::io_context::executor_type>>(io.get_executor());
io_thread = std::thread([this] {
try {
std::cout << std::format("协程io:{} 协程启动！\n", thread_id_str());
io.run();
std::cout << std::format("协程io:{} 销毁！\n", thread_id_str());
}
catch (const std::exception& e) {
std::cerr << std::format("协程io异常: {}\n", e.what());
std::terminate();
}
catch (...) {
std::cerr << "协程io未知异常\n";
std::terminate();
}
});
auto n = std::max<unsigned int>(1, std::thread::hardware_concurrency());
auto other_need = std::max<unsigned int>(2, n / 2);
if (n > other_need) {
n -= other_need;
}
else {
n = n / 2;
}
n = std::max<unsigned int>(1, n);
process_data = std::make_unique<asio::thread_pool>(n);
std::cout << std::format("协程已经启动 process_data concurrency: 在n个线程上{}\n", n);
auto promise = std::make_shared<std::promise<void>>();
loop_task = std::make_unique<std::future<void>>(promise->get_future());
asio::co_spawn(io, coro_thread(), [promise](std::exception_ptr ep) {
std::cout << "发生了异常" << std::endl;
if (ep) {
promise->set_exception(ep);
std::rethrow_exception(ep);
}
promise->set_value();
});
std::cout << std::flush;
}
asio::awaitable<void> Coro::sleep_for(std::chrono::milliseconds ms) {
auto executor = co_await asio::this_coro::executor;
asio::steady_timer timer(executor, ms);
co_await timer.async_wait(asio::use_awaitable);
co_return;
auto executor = co_await asio::this_coro::executor;
asio::steady_timer timer(executor, ms);
co_await timer.async_wait(asio::use_awaitable);
co_return;
}
void Coro::stop() {
std::cout << "stop_io_coro begin" << std::endl;
auto list = get_all();
for (auto &li: list) {
if (li->running()) {
std::cout << std::format("请求停止 {}:{} 当前状态为{}\n", li->type, li->key, Psc::to_string(li->state));
li->async_stop();
}
}
// 同步等待
for (auto &li: list) {
li->sync_wait();
}
// 停止状态更新状态机
running = false;
asio::post(io, [] {
});
if (loop_task) {
std::cout << "等待 coro_thread 退出" << std::endl;
loop_task->get();
loop_task.reset();
std::cout << "coro_thread 已退出" << std::endl;
}
// 清理资源
for (auto &li: list) {
li->loop_task.reset();
}
if (process_data) {
process_data->join();
process_data.reset();
}
io_work.reset();
io.stop();
if (io_thread.joinable()) {
io_thread.join();
}
std::cout << "stop_io_coro end" << std::endl;
std::cout << "stop_io_coro begin" << std::endl;
auto list = get_all();
for (auto& li : list) {
if (li->running()) {
std::cout << std::format("请求停止 {}:{} 当前状态为{}\n", li->type, li->key, Psc::to_string(li->state));
li->async_stop();
}
}
// 同步等待
for (auto& li : list) {
li->sync_wait();
}
// 停止状态更新状态机
running = false;
asio::post(io, [] {});
if (loop_task) {
std::cout << "等待 coro_thread 退出" << std::endl;
loop_task->get();
loop_task.reset();
std::cout << "coro_thread 已退出" << std::endl;
}
// 清理资源
for (auto& li : list) {
li->loop_task.reset();
}
if (process_data) {
process_data->join();
process_data.reset();
}
io_work.reset();
io.stop();
if (io_thread.joinable()) {
io_thread.join();
}
std::cout << "stop_io_coro end" << std::endl;
}
asio::awaitable<void> Coro::coro_thread() {
while (running.load(std::memory_order_acquire)) {
auto list = get_all();
for (auto &li: list) {
co_await li->tick(); //状态机驱动函数
}
co_await sleep_for(std::chrono::milliseconds(10));
}
co_return;
while (running.load(std::memory_order_acquire)) {
auto list = get_all();
for (auto& li : list) {
co_await li->tick(); //状态机驱动函数
}
co_await sleep_for(std::chrono::milliseconds(10));
}
co_return;
}
asio::awaitable<void> Data_Feed::loop_coro() {
auto feed = this;
auto co = Coro::instance();
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 (feed->running()) {
if (data_feed_debug) {
std::cout << std::format("{} feed loop begin {}\n", key, thread_id_str());
}
std::vector<std::string> messages;
{
if (!feed->running() || !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<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 && 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 co->sleep_for(std::chrono::milliseconds(10));
continue;
}
if (!feed->running() || !feed->registered()) {
break;
}
for (auto &msg: messages) {
if (!feed->enable) {
break;
}
if (data_feed_debug) {
std::cout << std::format("{} before send {}\n", key, thread_id_str());
}
co_await feed->send_coro(msg);
if (data_feed_debug) {
std::cout << std::format("{} after send {}\n", key, thread_id_str());
}
}
co_await resume_on(co->io.get_executor());
}
std::cout << std::format("{} feed loop exit {}\n", key, thread_id_str());
co_return;
auto feed = this;
auto co = Coro::instance();
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 (feed->running()) {
if (data_feed_debug) {
std::cout << std::format("{} feed loop begin {}\n", key, thread_id_str());
}
std::vector<std::string> messages;
{
if (!feed->running() || !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<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 && 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 co->sleep_for(std::chrono::milliseconds(10));
continue;
}
if (!feed->running() || !feed->registered()) {
break;
}
for (auto& msg : messages) {
if (!feed->enable) {
break;
}
if (data_feed_debug) {
std::cout << std::format("{} before send {}\n", key, thread_id_str());
}
co_await feed->send_coro(msg);
if (data_feed_debug) {
std::cout << std::format("{} after send {}\n", key, thread_id_str());
}
}
co_await resume_on(co->io.get_executor());
}
std::cout << std::format("{} feed loop exit {}\n", key, thread_id_str());
co_return;
}
asio::awaitable<void> Data_Source::loop_coro() {
auto source = this;
auto co = Coro::instance();
auto process_strand = std::make_shared<asio::strand<asio::thread_pool::executor_type> >(
co->process_data->get_executor());
while (source->running()) {
if (data_source_debug) {
std::cout << std::format("[1] ds:{} source loop begin {}\n", key, thread_id_str()) << std::flush;;
}
if (!source->running() || !source->registered()) {
break;
}
if (data_source_debug) {
std::cout << std::format("[2] ds:{} before handle thread:{}\n", key, thread_id_str()) << std::flush;;
}
co_await source->handle_in_loop_coro();
if (!enable || !source->running() || !source->registered()) {
break;
}
if (data_source_debug) {
std::cout << std::format("[3] ds:{} before read thread:{}\n", key, thread_id_str()) << std::flush;;
}
auto mode_data = co_await source->read_coro();
if (data_source_debug) {
std::cout << std::format("[4] ds:{} after read thread:{}\n", key, thread_id_str()) << std::flush;;
}
if (wait) {
co_await resume_on(co->process_data->get_executor());
if (data_source_debug) {
std::cout << std::format("wait [5] ds:{} before process {} thread:{} \n", key, thread_id_str(),
mode_data.size()) <<
std::flush;;
}
auto num = source->process_mode_acs_data(mode_data);
if (data_source_debug) {
std::cout << std::format("[6] ds:{} after process {} {}\n", key, num, thread_id_str()) << std::flush;;
}
co_await resume_on(co->io.get_executor());
static Frequency_Limit_Multi mt(0.1);
auto name = type + ":" + key;
if (mt.test(name)) {
if (num == 0) {
if (data_source_debug)
std::cout << std::format("[7] {} 没解析到数据睡眠10ms {} {}\n", name, num,
thread_id_str()) << std::flush;;
ask_sleep = true;
} else {
if (data_source_debug)
std::cout << std::format("[7] {} 解析到数据 {} {}\n", name, num,
thread_id_str()) << std::flush;;
}
}
if (num == 0) {
co_await co->sleep_for(std::chrono::milliseconds(10));
ask_sleep = false;
}
} else {
asio::post(*process_strand, [this, source, mode_data = std::move(mode_data)]() mutable {
try {
if (data_source_debug) {
std::cout << std::format("no wait [5] ds:{} before process {} size:{}\n", source->key,
thread_id_str(),
mode_data.size());
}
auto num = source->process_mode_acs_data(mode_data);
if (data_source_debug) {
std::cout << std::format("[6] ds:{} after process {} {}\n", source->key, num, thread_id_str());
}
static Frequency_Limit_Multi mt(0.1);
auto name = source->type + ":" + source->key;
if (mt.test(name)) {
if (num == 0) {
std::cout << std::format("[7] {} 没解析到数据 {} {}\n", name, num, thread_id_str());
ask_sleep = true;
} else {
std::cout << std::format("[8] {} 解析到数据 {} {}\n", name, num, thread_id_str());
}
}
} catch (const std::exception &e) {
std::cout << std::format("[9] {} process exception: {}\n", source->key, e.what());
} catch (...) {
std::cout << std::format("[9] {} process unknown exception\n", source->key);
}
});
co_await asio::post(co->io, asio::use_awaitable);
if (ask_sleep) {
co_await co->sleep_for(std::chrono::milliseconds(10));
ask_sleep = false;
}
}
std::cout << std::flush;
}
std::cout << std::format("{} source loop exit {}\n", key, thread_id_str());
co_return;
auto co = Coro::instance();
auto process_strand = std::make_shared<asio::strand<asio::thread_pool::executor_type>>(co->process_data->get_executor());
while (running()) {
if (data_source_debug) {
std::cout << std::format("[1] ds:{} source loop begin {}\n", key, thread_id_str()) << std::flush;;
}
if (!running() || !registered()) {
break;
}
if (data_source_debug) {
std::cout << std::format("[2] ds:{} before handle thread:{}\n", key, thread_id_str()) << std::flush;;
}
co_await handle_in_loop_coro();
if (!enable || !running() || !registered()) {
break;
}
if (data_source_debug) {
std::cout << std::format("[3] ds:{} before read thread:{}\n", key, thread_id_str()) << std::flush;;
}
auto mode_data = co_await read_coro();
if (data_source_debug) {
std::cout << std::format("[4] ds:{} after read thread:{}\n", key, thread_id_str()) << std::flush;;
}
auto& wait = Global::instance()->mode_acs.wait_process_msg;
if (wait) {
// 调度到别的线程
co_await resume_on(co->process_data->get_executor());
if (data_source_debug) {
std::cout << std::format("wait [5] ds:{} before process {} thread:{} \n", key, thread_id_str(),
mode_data.size()) <<
std::flush;;
}
auto num = process_mode_acs_data(mode_data);
if (data_source_debug) {
std::cout << std::format("[6] ds:{} after process {} {}\n", key, num, thread_id_str()) << std::flush;;
}
// 调度回io主线程
co_await resume_on(co->io.get_executor());
static Frequency_Limit_Multi mt(0.1);
auto name = type + ":" + key;
if (mt.test(name)) {
if (num == 0) {
if (data_source_debug)
std::cout << std::format("[7] {} 没解析到数据睡眠10ms {} {}\n", name, num,
thread_id_str()) << std::flush;;
ask_sleep = true;
}
else {
if (data_source_debug)
std::cout << std::format("[7] {} 解析到数据 {} {}\n", name, num,
thread_id_str()) << std::flush;;
}
}
if (num == 0) {
co_await co->sleep_for(std::chrono::milliseconds(10));
ask_sleep = false;
}
}
else {
asio::post(*process_strand, [this, mode_data = std::move(mode_data)]() mutable {
try {
if (data_source_debug) {
std::cout << std::format("no wait [5] ds:{} before process {} size:{}\n", key,
thread_id_str(),
mode_data.size());
}
auto num = process_mode_acs_data(mode_data);
if (data_source_debug) {
std::cout << std::format("[6] ds:{} after process {} {}\n", key, num, thread_id_str());
}
static Frequency_Limit_Multi mt(0.1);
auto name = type + ":" + key;
if (mt.test(name)) {
if (num == 0) {
std::cout << std::format("[7] {} 没解析到数据 {} {}\n", name, num, thread_id_str());
ask_sleep = true;
}
else {
std::cout << std::format("[8] {} 解析到数据 {} {}\n", name, num, thread_id_str());
}
}
}
catch (const std::exception& e) {
std::cout << std::format("[9] {} process exception: {}\n", key, e.what());
}
catch (...) {
std::cout << std::format("[9] {} process unknown exception\n", key);
}
});
co_await asio::post(co->io, asio::use_awaitable);
if (ask_sleep) {
co_await co->sleep_for(std::chrono::milliseconds(10));
ask_sleep = false;
}
}
std::cout << std::flush;
}
std::cout << std::format("{} source loop exit {}\n", key, thread_id_str());
co_return;
}
module/Local_Server/server/server.cpp
+481 -541
View File
File diff suppressed because it is too large Load Diff