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-25 11:45:41 +08:00
parent f0510f5438
commit 03bfac8d0c
3 changed files with 146 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/server/io_coro.cpp
+109 -115
View File
@@ -1,31 +1,72 @@
#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include <concurrencpp/concurrencpp.h>
#include <atomic>
#include <chrono>
#include <exception>
#include <iostream>
#include <memory>
#include <sstream>
#include <utility>
#include <vector>
Frequency_Limit too_many_msg_limit;
Frequency_Limit flush_limit;
concurrencpp::runtime runtime_;
std::shared_ptr<concurrencpp::worker_thread_executor> executor_;
std::unique_ptr<concurrencpp::result<void>> data_feed_thread_task;
std::atomic<bool> running = false;
concurrencpp::result<void> coro_thread(std::atomic<bool>& running);
void start_io_coro() {
#include "io_coro.h"
concurrencpp::result<void> 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<concurrencpp::result<
void>>(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<concurrencpp::result<void>>(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<concurrencpp::result<void>>(coro_thread(running));
data_feed_thread_task = std::make_unique<concurrencpp::result<void>>(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<concurrencpp::result<void>>& task) {
bool Coro::task_is_running(std::unique_ptr<concurrencpp::result<void>>& task) {
if (!task) {
return false;
}
@@ -55,7 +97,7 @@ bool task_is_running(std::unique_ptr<concurrencpp::result<void>>& task) {
task.reset();
return false;
}
concurrencpp::result<void> data_source_loop_coro(std::atomic<bool>& running, std::shared_ptr<Data_Source> source) {
concurrencpp::result<void> Coro::data_source_loop_coro(std::shared_ptr<Data_Source> 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<void> data_source_loop_coro(std::atomic<bool>& running, std
}
co_return;
}
concurrencpp::result<void> data_feed_loop_coro(std::atomic<bool>& running, std::shared_ptr<Data_Feed> 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<std::string> messages;
{
concurrencpp::result<void> Coro::data_feed_loop_coro(std::shared_ptr<Data_Feed> 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<std::string> 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<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 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<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;
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<void> coro_thread(std::atomic<bool>& 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<concurrencpp::result<void>>(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<concurrencpp::result<void>>(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;
}
module/Local_Server/server/io_coro.h
+31
View File
@@ -0,0 +1,31 @@
#pragma once
#include <concurrencpp/concurrencpp.h>
#include <atomic>
#include <chrono>
#include <exception>
#include <iostream>
#include <memory>
#include <sstream>
#include <utility>
#include <vector>
#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<concurrencpp::result<void>>& task);
concurrencpp::result<void> data_source_loop_coro(std::shared_ptr<Data_Source> source);
concurrencpp::result<void> data_feed_loop_coro(std::shared_ptr<Data_Feed> feed);
concurrencpp::result<void> coro_thread();
bool has_running_loop_tasks();
Frequency_Limit too_many_msg_limit;
Frequency_Limit flush_limit;
concurrencpp::runtime runtime_;
std::shared_ptr<concurrencpp::worker_thread_executor> executor_;
std::unique_ptr<concurrencpp::result<void>> data_feed_thread_task;
std::atomic<bool> running = false;
};
module/Local_Server_main.cpp
+6 -5
View File
@@ -11,6 +11,8 @@
#include <codecvt>
#include <csignal>
#include <future>
#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();