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-23 18:06:44 +08:00
parent a9293932cc
commit 05dec135ed
5 changed files with 528 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/Data_Feed/data_feed_thread.cpp
-146
View File
@@ -1,146 +0,0 @@
#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include <future>
#include <iostream>
long long get_current_milliseconds() {
auto now = std::chrono::system_clock::now();
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch());
return duration.count();
}
Frequency_Limit too_many_msg_limit;
Frequency_Limit flush_limit;
ucoro::awaitable<void> data_feed_thread_coro(std::atomic<bool>& 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<std::shared_ptr<Data_Feed>> 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<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);
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<int>(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<std::shared_ptr<Msg>>& all = feed->msg_buffer.get_all();
const std::vector<std::string*>& 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<bool>& 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("");
}
}
module/Local_Server/Data_Source/Data_Source.cpp
+28 -8
View File
@@ -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<SSR::Mode_Msg>& 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(&params);
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<int>("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();
module/Local_Server/Data_Source/Data_Source.h
+8 -1
View File
@@ -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<bool> data_source_loop_requested_ = false;
std::atomic<bool> data_source_loop_running_ = false;
std::atomic<std::uint64_t> data_source_loop_generation_ = 0;
virtual bool _open() = 0;
virtual void _close() = 0;
};
module/Local_Server/server/io_coro.cpp
+490
View File
@@ -0,0 +1,490 @@
#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <future>
#include <iostream>
#include <map>
#include <mutex>
#include <queue>
#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&) {
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<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();
}
);
});
}
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});
}
);
}
}
);
}
);
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;
}
}
}
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->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<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);
}
}
);
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<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;
}
}
}
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) {
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("");
}
}
module/Local_Server_main.cpp
+2 -2
View File
@@ -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);
}