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:17:52 +08:00
parent 2aad10ccf4
commit f0510f5438
11 changed files with 370 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/Data_Feed/Data_Feed.cpp
+3 -3
View File
@@ -10,15 +10,15 @@ bool Data_Feed::registered() {
return false;
}
void Data_Feed_UDP_Server::handle_in_loop() {
psco::sync_await(handle_in_loop_coro());
(handle_in_loop_coro()).get();
}
psco::awaitable<void> Data_Feed_UDP_Server::handle_in_loop_coro() {
concurrencpp::result<void> Data_Feed_UDP_Server::handle_in_loop_coro() {
co_await svr.tick_coro();
co_return;
}
psco::awaitable<void> Data_Feed_UDP_Server::send_coro(const std::string& data) {
concurrencpp::result<void> Data_Feed_UDP_Server::send_coro(const std::string& data) {
co_await svr.write_to_all_clients_coro(data);
co_return;
}
module/Local_Server/Data_Feed/Data_Feed.h
+21 -21
View File
@@ -47,7 +47,7 @@ struct Output_Format {
class Data_Feed {
public:
virtual ~Data_Feed() = default;
std::unique_ptr<psco::awaitable<void>> loop_task;
std::unique_ptr<concurrencpp::result<void>> loop_task;
BIN_Msg_Buffer msg_buffer{};
std::string key;
bool enable = false;
@@ -57,11 +57,11 @@ public:
virtual void handle_in_loop() {
}
virtual psco::awaitable<void> handle_in_loop_coro() {
virtual concurrencpp::result<void> handle_in_loop_coro() {
handle_in_loop();
co_return;
}
virtual psco::awaitable<void> send_coro(const std::string&) {
virtual concurrencpp::result<void> send_coro(const std::string&) {
co_return;
}
virtual Psc::JSON to_json() {
@@ -99,7 +99,7 @@ public:
}
std::string to_string() {
return "Data_Feed【" + VAR_STR_3(key, enable, type) + "】";
return "Data_Feed{" + VAR_STR_3(key, enable, type) + "}";
}
bool registered();
@@ -136,14 +136,14 @@ public:
~Data_Feed_TCP_Server() override {}
void handle_in_loop() override {
//std::cout << socket.to_string() << "flush_clients" << std::endl;
psco::sync_await(handle_in_loop_coro());
(handle_in_loop_coro()).get();
}
psco::awaitable<void> handle_in_loop_coro() override {
concurrencpp::result<void> handle_in_loop_coro() override {
co_await svr.flush_clients_coro();
co_await svr.tick_coro();
co_return;
}
psco::awaitable<void> send_coro(const std::string& data) override {
concurrencpp::result<void> send_coro(const std::string& data) override {
co_await svr.write_to_all_clients_coro(data);
co_return;
}
@@ -166,7 +166,7 @@ public:
svr.set_connect_system_buffer_size(connect_system_buffer_size);
svr.set_tcp_no_delay(false);
svr.create();
return psco::sync_await(svr.listen_coro("0.0.0.0", port));
return (svr.listen_coro("0.0.0.0", port)).get();
}
Psc::JSON get_clients_json() {
Psc::JSON ret = Psc::JSON::array();
@@ -191,7 +191,7 @@ public:
{
auto& ins = conn->send_num.instant;
auto& avr =conn->send_num.average;
cur.append({"send_num(byte/次)", VAR_STR_2(ins, avr)});
cur.append({"send_num(byte/count)", VAR_STR_2(ins, avr)});
}
ret.append(cur);
}
@@ -203,13 +203,13 @@ public:
class Data_Feed_TCP_Client : public Data_Feed {
public:
void handle_in_loop() override {
psco::sync_await(handle_in_loop_coro());
(handle_in_loop_coro()).get();
}
psco::awaitable<void> handle_in_loop_coro() override {
concurrencpp::result<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
co_return;
}
psco::awaitable<void> send_coro(const std::string& data) override {
concurrencpp::result<void> send_coro(const std::string& data) override {
co_await cli.send_coro(data);
co_return;
}
@@ -228,7 +228,7 @@ public:
sockaddr_in.port = port;
cli.set_dest_address(sockaddr_in);
cli.create();
psco::sync_await(cli.connect_coro());
(cli.connect_coro()).get();
return true;
}
void from_json(const Psc::JSON* that_json) override {
@@ -258,8 +258,8 @@ public:
Data_Feed_UDP_Server()= default;
~Data_Feed_UDP_Server() override = default;
void handle_in_loop() override;
psco::awaitable<void> handle_in_loop_coro() override;
psco::awaitable<void> send_coro(const std::string& data) override;
concurrencpp::result<void> handle_in_loop_coro() override;
concurrencpp::result<void> send_coro(const std::string& data) override;
[[nodiscard]] Psc::JSON get_clients_json() const {
Psc::JSON ret = Psc::JSON::array();
for (const auto& i : svr.clients) {
@@ -296,13 +296,13 @@ public:
return VAR_JSON_1(state);
}
void handle_in_loop() override {
psco::sync_await(handle_in_loop_coro());
(handle_in_loop_coro()).get();
}
psco::awaitable<void> handle_in_loop_coro() override {
concurrencpp::result<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
co_return;
}
psco::awaitable<void> send_coro(const std::string& data) override {
concurrencpp::result<void> send_coro(const std::string& data) override {
co_await cli.send_coro(data);
co_return;
}
@@ -321,13 +321,13 @@ public:
return ret;
}
void _close() override {
std::cout << "udp客户端 目标地址：" << cli.dest_address.to_string() << "关闭！" << std::endl;
std::cout << "udp client target address:" << cli.dest_address.to_string() << " closed" << std::endl;
cli.close();
}
bool _open() override {
cli.create();
cli.set_dest_address(url, port);
psco::sync_await(cli.connect_coro());
(cli.connect_coro()).get();
return true;
}
};
@@ -340,7 +340,7 @@ inline std::shared_ptr<Data_Feed> create_data_feed_from_type(const std::string&
if (t == "Data_Feed_UDP_Server") ret = std::make_shared<Data_Feed_UDP_Server>(); else
if (t == "Data_Feed_TCP_Client") ret = std::make_shared<Data_Feed_TCP_Client>(); else
if (t == "Data_Feed_UDP_Client") ret = std::make_shared<Data_Feed_UDP_Client>(); else {
std::cout << "未知的 Data_Feed type类型! t:[" << t << "]"<< std::endl;
std::cout << "鏈煡鐨? Data_Feed type绫诲瀷! t:[" << t << "]"<< std::endl;
throw std::invalid_argument("unknown Data_Feed type: " + t);
}
return ret;
module/Local_Server/Data_Source/Data_Source.cpp
+3 -3
View File
@@ -593,7 +593,7 @@ void Data_Source_Config::server(Global* g) {
ds->enable = false;
ds->close();
ds->from_json(&params);
wake_data_feed_thread();
g->save();
g->mode_acs.source_feed_relation_config.set_need_refresh();
});
@@ -610,7 +610,7 @@ void Data_Source_Config::server(Global* g) {
(void)enable;
HTTP_REQUIRE_TRUE(map.insert(index, ds), "index")
wake_data_feed_thread();
//std::cout << params.to_json_string() << std::endl;
g->mode_acs.source_feed_relation_config.set_need_refresh();
Config::save();
@@ -625,7 +625,7 @@ void Data_Source_Config::server(Global* g) {
ds->close();
HTTP_REQUIRE_TRUE(map.remove(index), "index")
wake_data_feed_thread();
g->save();
res->setBody(warp(to_json()).to_json_string());
g->mode_acs.source_feed_relation_config.set_need_refresh();
module/Local_Server/Data_Source/Data_Source.h
+14 -14
View File
@@ -7,7 +7,7 @@
namespace Psc {
class SM_RingBuffer;
}
// 统一封装数据输出的模式
class Data_Source;
std::shared_ptr<Data_Source> create_from_json(const Psc::JSON *that_json);
@@ -68,17 +68,17 @@ protected:
class Data_Source : public std::enable_shared_from_this<Data_Source>, public Data_Source_Handler {
public:
~Data_Source() override = default;
std::unique_ptr<psco::awaitable<void>> loop_task;
std::unique_ptr<concurrencpp::result<void>> loop_task;
std::shared_ptr<Data_Source> that();
virtual psco::awaitable<std::string> read_coro() {co_return "";};
virtual concurrencpp::result<std::string> read_coro() {co_return "";};
bool registered() const;
virtual Psc::JSON get_custom_state_json() = 0;
Psc::JSON get_state();
std::string last_char; // 用于处理奇数字节的情况
std::string last_char; // 用于处理奇数字节的情�?
// std::shared_ptr<Input_Format> parse_format;
Data_Source();
std::shared_ptr<SSR::Mode_Msg> last_prase_msg = nullptr;
virtual psco::awaitable<void> handle_in_loop_coro(){co_return;}
virtual concurrencpp::result<void> handle_in_loop_coro(){co_return;}
std::atomic<bool> enable{};
std::string type;
std::atomic<bool> base_station_show{};
@@ -99,8 +99,8 @@ public:
Psc::JSON ret = Psc::JSON::object();
Ret_J(enable);
Ret_J(type);
ret.append({"飞机总数", get_aircraft_num()});
ret.append({"解出位置的数量", have_pos_aircraft_num});
ret.append({"aircraft_total", get_aircraft_num()});
ret.append({"aircraft_with_position", have_pos_aircraft_num});
ret.append({"mode_ac_statistic", mode_ac_statistic.to_json()});
ret.append({"mode_s_statistic", mode_s_statistic.to_json()});
ret.append({"all_connect_feed", get_all_connect_feed_status()});
@@ -167,7 +167,7 @@ protected:
struct TCP_Client_Data_Source : Data_Source {
psco::awaitable<void> handle_in_loop_coro() override {
concurrencpp::result<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
co_return;
}
@@ -190,7 +190,7 @@ struct TCP_Client_Data_Source : Data_Source {
cli.set_dest_address(addr);
cli.create();
psco::sync_await(cli.connect_coro());
(cli.connect_coro()).get();
return true;
}
void _close() override { cli.close(); }
@@ -208,7 +208,7 @@ struct TCP_Client_Data_Source : Data_Source {
}
psco::awaitable<std::string> read_coro() override {
concurrencpp::result<std::string> read_coro() override {
co_return co_await cli.read_coro();
}
};
@@ -238,13 +238,13 @@ struct Serial_Data_Source : Data_Source {
}
void _close() override { if (serial) serial->close(); }
psco::awaitable<void> handle_in_loop_coro() override {
concurrencpp::result<void> handle_in_loop_coro() override {
if (serial) {
co_await serial->tick_coro();
}
co_return;
}
psco::awaitable<std::string> read_coro() override {
concurrencpp::result<std::string> read_coro() override {
if (!serial) {
co_return "";
}
@@ -270,7 +270,7 @@ struct Serial_Data_Source : Data_Source {
enum class File_Data_Type {
SIMPLE_BIN_Blank, // 没有1a转义的
SIMPLE_BIN_Blank, // 没有1a转义�?
BIN_Blank_Text,
AVR,
BIN_Text,
@@ -452,7 +452,7 @@ inline std::shared_ptr<Data_Source> create_data_source_from_type(const std::stri
else if (t == "Shared_Memory_Data_Source")
ret = std::make_shared<Shared_Memory_Data_Source>();
else {
std::cout << "未知的 Data_Source type类型!" << std::endl;
std::cout << "未知�?Data_Source type类型!" << std::endl;
throw std::invalid_argument("unknown Data_Source type: " + t);
}
return ret;
module/Local_Server/Data_Source/Data_Source_Handler.cpp
+8
View File
@@ -497,6 +497,14 @@ void Data_Source_Handler::push_to_feed(const std::shared_ptr<SSR::Msg>& msg) {
if (msg->type == SSR::Mode_Msg::T::S7 || msg->type == SSR::Mode_Msg::T::S14) {
// static Frequency_Limit fl;
// if (!fl.test()) {
// std::ostringstream oss;
// oss << "收到消息 ===" << VAR_STR_3(msg->type ,feed->msg_buffer.mode_s_msg_num, s_size) << std::endl;
// std::cout << oss.str() << std::endl;
// }
if (feed->msg_buffer.mode_s_msg_num > s_size) {
continue;
} else {
module/Local_Server/External_Database/External_Database.cpp
+42 -31
View File
@@ -4,6 +4,7 @@
#include "Resources.h"
#include "../server/Global.h"
#include "../server/Performance_Monitor.h"
#include "Core/Base/Coro_Result.h"
#include <SQLiteCpp/Statement.h>
#include <SQLiteCpp/Transaction.h>
@@ -67,30 +68,22 @@ namespace
}
template <typename T, typename Starter>
psco::awaitable<T> await_external_database_callback(Starter starter)
concurrencpp::result<T> await_external_database_callback(Starter starter)
{
using Result = psco::traits::exception_with_result_t<T>;
auto result = co_await psco::callback_awaitable<Result>(
[starter = std::move(starter)](auto handler) mutable
auto result = co_await Psc::coro::callback_result<T>(
[starter = std::move(starter)](auto done) mutable
{
starter([handler = std::move(handler)](std::exception_ptr exception,
T value) mutable
starter([done = std::move(done)](std::exception_ptr exception, T value) mutable
{
if (exception)
{
handler(Result{exception});
done.set_exception(exception);
return;
}
handler(Result{std::move(value)});
done(std::move(value));
});
});
if (std::holds_alternative<std::exception_ptr>(result))
{
std::rethrow_exception(std::get<std::exception_ptr>(result));
}
co_return std::move(std::get<T>(result));
co_return std::move(result);
}
} // 命名空间
@@ -494,7 +487,7 @@ void External_Resources_Manager::async_external_database_status(
std::move(callback));
}
psco::awaitable<std::vector<External_Resource_Status>>
concurrencpp::result<std::vector<External_Resource_Status>>
External_Resources_Manager::refresh_external_databases_coro(
const bool force_download)
{
@@ -506,7 +499,7 @@ External_Resources_Manager::refresh_external_databases_coro(
});
}
psco::awaitable<External_Resource_Status>
concurrencpp::result<External_Resource_Status>
External_Resources_Manager::refresh_external_database_coro(
std::string resource_name,
const bool force_download)
@@ -521,7 +514,7 @@ External_Resources_Manager::refresh_external_database_coro(
});
}
psco::awaitable<External_Resource_Status>
concurrencpp::result<External_Resource_Status>
External_Resources_Manager::import_external_database_coro(
std::string resource_name,
std::filesystem::path source_file)
@@ -536,7 +529,7 @@ External_Resources_Manager::import_external_database_coro(
});
}
psco::awaitable<External_Resource_Status>
concurrencpp::result<External_Resource_Status>
External_Resources_Manager::clear_external_database_table_coro(
std::string resource_name)
{
@@ -548,7 +541,7 @@ External_Resources_Manager::clear_external_database_table_coro(
});
}
psco::awaitable<std::optional<External_Database_Row>>
concurrencpp::result<std::optional<External_Database_Row>>
External_Resources_Manager::query_external_database_coro(
std::string resource_name,
std::vector<std::string> primary_key_values) const
@@ -566,33 +559,51 @@ External_Resources_Manager::query_external_database_coro(
});
}
psco::awaitable<std::map<std::string, std::shared_ptr<External_Database_Row>>>
concurrencpp::result<std::shared_ptr<External_Database_Row_Map>>
External_Resources_Manager::query_aircraft_external_databases_coro(
std::string icao24) const
{
co_return co_await await_external_database_callback<
std::map<std::string, std::shared_ptr<External_Database_Row>>>(
[this, icao24 = std::move(icao24)](Row_Map_Callback callback) mutable
auto result = co_await Psc::coro::callback_result<std::shared_ptr<External_Database_Row_Map>>(
[this, icao24 = std::move(icao24)](auto done) mutable
{
async_query_aircraft_external_databases(
std::move(icao24), std::move(callback));
std::move(icao24),
[done = std::move(done)](std::exception_ptr exception, External_Database_Row_Map value) mutable
{
if (exception)
{
done.set_exception(exception);
return;
}
done(std::make_shared<External_Database_Row_Map>(std::move(value)));
});
});
co_return result;
}
psco::awaitable<std::map<std::string, std::shared_ptr<External_Database_Row>>>
concurrencpp::result<std::shared_ptr<External_Database_Row_Map>>
External_Resources_Manager::query_callsign_external_databases_coro(
std::optional<std::string> callsign) const
{
co_return co_await await_external_database_callback<
std::map<std::string, std::shared_ptr<External_Database_Row>>>(
[this, callsign = std::move(callsign)](Row_Map_Callback callback) mutable
auto result = co_await Psc::coro::callback_result<std::shared_ptr<External_Database_Row_Map>>(
[this, callsign = std::move(callsign)](auto done) mutable
{
async_query_callsign_external_databases(
std::move(callsign), std::move(callback));
std::move(callsign),
[done = std::move(done)](std::exception_ptr exception, External_Database_Row_Map value) mutable
{
if (exception)
{
done.set_exception(exception);
return;
}
done(std::make_shared<External_Database_Row_Map>(std::move(value)));
});
});
co_return result;
}
psco::awaitable<std::vector<External_Resource_Status>>
concurrencpp::result<std::vector<External_Resource_Status>>
External_Resources_Manager::external_database_status_coro() const
{
co_return co_await await_external_database_callback<std::vector<External_Resource_Status>>
module/Local_Server/External_Database/global.h
+16 -12
View File
@@ -12,7 +12,7 @@
#include <optional>
#include <string>
#include <vector>
#include <psco/awaitable.hpp>
#include <concurrencpp/concurrencpp.h>
#include "Core/Base/JSON.h"
@@ -20,11 +20,17 @@ class Global;
struct External_Database_Row {
std::map<std::string, std::optional<std::string>> columns;
External_Database_Row() = default;
External_Database_Row(const External_Database_Row&) = default;
External_Database_Row& operator=(const External_Database_Row&) = default;
External_Database_Row(External_Database_Row&&) noexcept = default;
External_Database_Row& operator=(External_Database_Row&&) noexcept = default;
[[nodiscard]] std::optional<std::string> get(const std::string& column) const;
[[nodiscard]] Psc::JSON to_json() const;
};
using External_Database_Row_Map = std::map<std::string, std::shared_ptr<External_Database_Row>>;
struct External_Resource_Status {
std::string name;
std::size_t row_count{};
@@ -33,8 +39,6 @@ struct External_Resource_Status {
std::string message;
};
// 每个数据源模块只管理一张 SQLite 表。主键由数据源按业务含义选择：
// ICAO24、航班呼号、机场 ICAO 代码、机型代码或其他业务键。
class External_Resources {
public:
External_Resources(std::string name,
@@ -105,7 +109,7 @@ public:
using Row_Callback = std::function<void(std::exception_ptr, std::optional<External_Database_Row>)>;
using Row_Map_Callback = std::function<void(
std::exception_ptr,
std::map<std::string, std::shared_ptr<External_Database_Row>>)>;
External_Database_Row_Map)>;
External_Resources_Manager();
@@ -149,24 +153,24 @@ public:
Row_Map_Callback callback) const;
void async_external_database_status(Status_List_Callback callback) const;
[[nodiscard]] psco::awaitable<std::vector<External_Resource_Status>>
[[nodiscard]] concurrencpp::result<std::vector<External_Resource_Status>>
refresh_external_databases_coro(bool force_download = true);
[[nodiscard]] psco::awaitable<External_Resource_Status>
[[nodiscard]] concurrencpp::result<External_Resource_Status>
refresh_external_database_coro(std::string resource_name,
bool force_download = true);
[[nodiscard]] psco::awaitable<External_Resource_Status>
[[nodiscard]] concurrencpp::result<External_Resource_Status>
import_external_database_coro(std::string resource_name,
std::filesystem::path source_file);
[[nodiscard]] psco::awaitable<External_Resource_Status>
[[nodiscard]] concurrencpp::result<External_Resource_Status>
clear_external_database_table_coro(std::string resource_name);
[[nodiscard]] psco::awaitable<std::optional<External_Database_Row>>
[[nodiscard]] concurrencpp::result<std::optional<External_Database_Row>>
query_external_database_coro(std::string resource_name,
std::vector<std::string> primary_key_values) const;
[[nodiscard]] psco::awaitable<std::map<std::string, std::shared_ptr<External_Database_Row>>>
[[nodiscard]] concurrencpp::result<std::shared_ptr<External_Database_Row_Map>>
query_aircraft_external_databases_coro(std::string icao24) const;
[[nodiscard]] psco::awaitable<std::map<std::string, std::shared_ptr<External_Database_Row>>>
[[nodiscard]] concurrencpp::result<std::shared_ptr<External_Database_Row_Map>>
query_callsign_external_databases_coro(std::optional<std::string> callsign) const;
[[nodiscard]] psco::awaitable<std::vector<External_Resource_Status>>
[[nodiscard]] concurrencpp::result<std::vector<External_Resource_Status>>
external_database_status_coro() const;
void register_external_resource(std::unique_ptr<External_Resources> external_res);
module/Local_Server/server/Global.h
-2
View File
@@ -291,8 +291,6 @@ public:
DELETE_COPY(Global)
};
void wake_data_feed_thread();
void io_coro(std::atomic<bool>& running);
void handle_buffer_muti_start(std::string& buffer, const std::string& data, const std::set<std::string>& prefix_list,
module/Local_Server/server/Ucoro_Drogon_Glue.h
+83 -67
View File
@@ -1,103 +1,95 @@
#pragma once
#include <concurrencpp/concurrencpp.h>
#include <drogon/drogon.h>
#include <trantor/net/EventLoop.h>
#include <psco/awaitable.hpp>
#include <coroutine>
#include <exception>
#include <memory>
#include <optional>
#include <type_traits>
#include <utility>
#include <variant>
namespace Ecap_Coro {
template <typename T>
class Ucoro_Drogon_Awaiter {
class Concurrencpp_Drogon_Awaiter {
public:
explicit Ucoro_Drogon_Awaiter(psco::awaitable<T>&& task)
: state_(std::make_shared<State>()), task_(std::move(task)) {
explicit Concurrencpp_Drogon_Awaiter(concurrencpp::result<T>&& task)
: state_(std::make_shared<State>()), task_(std::move(task))
{
}
bool await_ready() const noexcept {
bool await_ready() const noexcept
{
return false;
}
void await_suspend(std::coroutine_handle<> continuation) {
void await_suspend(std::coroutine_handle<> continuation)
{
state_->continuation = continuation;
state_->loop = trantor::EventLoop::getEventLoopOfCurrentThread();
if (state_->loop == nullptr) {
state_->loop = drogon::app().getLoop();
}
auto state = state_;
state_->running_task.emplace(
psco::with_callback(
std::move(task_),
[state](psco::traits::exception_with_result_t<T> result) mutable {
state->result = std::move(result);
auto resume = [state]() {
state->continuation.resume();
};
if (state->loop != nullptr) {
state->loop->queueInLoop(std::move(resume));
} else {
resume();
}
}
)
);
state_->running_task->start();
state_->running_task.emplace(run(state_, std::move(task_)));
}
T await_resume() {
if (std::holds_alternative<std::exception_ptr>(state_->result)) {
auto exception = std::get<std::exception_ptr>(state_->result);
if (exception) {
std::rethrow_exception(exception);
}
T await_resume()
{
if (state_->exception) {
std::rethrow_exception(state_->exception);
}
return std::move(std::get<T>(state_->result));
return std::move(*state_->value);
}
private:
struct State {
trantor::EventLoop* loop{};
std::coroutine_handle<> continuation{};
psco::traits::exception_with_result_t<T> result{};
std::optional<psco::awaitable<void>> running_task{};
std::optional<T> value{};
std::exception_ptr exception{};
std::optional<concurrencpp::result<void>> running_task{};
};
static concurrencpp::result<void> run(std::shared_ptr<State> state, concurrencpp::result<T> task)
{
try {
state->value.emplace(co_await task);
} catch (...) {
state->exception = std::current_exception();
}
resume(std::move(state));
co_return;
}
static void resume(std::shared_ptr<State> state)
{
auto resume_fn = [state]() {
state->continuation.resume();
};
if (state->loop != nullptr) {
state->loop->queueInLoop(std::move(resume_fn));
} else {
resume_fn();
}
}
std::shared_ptr<State> state_;
psco::awaitable<T> task_;
concurrencpp::result<T> task_;
};
template <>
class Ucoro_Drogon_Awaiter<void> {
class Concurrencpp_Drogon_Awaiter<void> {
public:
explicit Ucoro_Drogon_Awaiter(psco::awaitable<void>&& task)
: state_(std::make_shared<State>()), task_(std::move(task)) {
explicit Concurrencpp_Drogon_Awaiter(concurrencpp::result<void>&& task)
: state_(std::make_shared<State>()), task_(std::move(task))
{
}
bool await_ready() const noexcept {
bool await_ready() const noexcept
{
return false;
}
void await_suspend(std::coroutine_handle<> continuation) {
void await_suspend(std::coroutine_handle<> continuation)
{
state_->continuation = continuation;
state_->loop = trantor::EventLoop::getEventLoopOfCurrentThread();
if (state_->loop == nullptr) {
state_->loop = drogon::app().getLoop();
}
auto state = state_;
state_->running_task.emplace(
psco::with_callback(
std::move(task_),
[state](std::exception_ptr exception) mutable {
state->exception = exception;
auto resume = [state]() {
state->continuation.resume();
};
if (state->loop != nullptr) {
state->loop->queueInLoop(std::move(resume));
} else {
resume();
}
}
)
);
state_->running_task->start();
state_->running_task.emplace(run(state_, std::move(task_)));
}
void await_resume() {
void await_resume()
{
if (state_->exception) {
std::rethrow_exception(state_->exception);
}
@@ -107,21 +99,45 @@ private:
trantor::EventLoop* loop{};
std::coroutine_handle<> continuation{};
std::exception_ptr exception{};
std::optional<psco::awaitable<void>> running_task{};
std::optional<concurrencpp::result<void>> running_task{};
};
static concurrencpp::result<void> run(std::shared_ptr<State> state, concurrencpp::result<void> task)
{
try {
co_await task;
} catch (...) {
state->exception = std::current_exception();
}
resume(std::move(state));
co_return;
}
static void resume(std::shared_ptr<State> state)
{
auto resume_fn = [state]() {
state->continuation.resume();
};
if (state->loop != nullptr) {
state->loop->queueInLoop(std::move(resume_fn));
} else {
resume_fn();
}
}
std::shared_ptr<State> state_;
psco::awaitable<void> task_;
concurrencpp::result<void> task_;
};
template <typename T>
auto to_drogon(psco::awaitable<T>&& task) {
return Ucoro_Drogon_Awaiter<T>{std::move(task)};
auto to_drogon(concurrencpp::result<T>&& task)
{
return Concurrencpp_Drogon_Awaiter<T>{std::move(task)};
}
template <typename T>
drogon::Task<T> to_drogon_task(psco::awaitable<T>&& task) {
drogon::Task<T> to_drogon_task(concurrencpp::result<T>&& task)
{
co_return co_await to_drogon(std::move(task));
}
inline drogon::Task<> to_drogon_task(psco::awaitable<void>&& task) {
inline drogon::Task<> to_drogon_task(concurrencpp::result<void>&& task)
{
co_await to_drogon(std::move(task));
co_return;
}
}
}
module/Local_Server/server/io_coro.cpp
+166 -231
View File
@@ -1,6 +1,6 @@
#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include "psco/single_thread.h"
#include <concurrencpp/concurrencpp.h>
#include <atomic>
#include <chrono>
#include <exception>
@@ -9,242 +9,29 @@
#include <sstream>
#include <utility>
#include <vector>
Frequency_Limit too_many_msg_limit;
Frequency_Limit flush_limit;
namespace {
psco::Single_Thread_Scheduler& data_feed_thread_scheduler() {
static psco::Single_Thread_Scheduler scheduler;
return scheduler;
}
psco::awaitable<void> data_feed_thread_yield_coro() {
co_await psco::callback_awaitable<void>([](auto done) mutable {
data_feed_thread_scheduler().post([done = std::move(done)]() mutable {
done();
});
});
co_return;
}
psco::awaitable<void> data_feed_thread_wait_event_coro() {
co_await psco::callback_awaitable<void>([](auto done) mutable {
data_feed_thread_scheduler().async_wait([done = std::move(done)]() mutable {
done();
});
});
co_return;
}
psco::awaitable<void> data_source_loop_coro(std::atomic<bool>& running, std::weak_ptr<Data_Source> source_ref) {
while (running.load(std::memory_order_acquire)) {
auto source = source_ref.lock();
if (!source || !source->enable || !source->registered()) {
break;
}
if (!source->is_open() && !source->open()) {
source.reset();
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->enable || !source->registered()) {
break;
}
source->process_mode_acs_data(mode_data);
source.reset();
co_await data_feed_thread_yield_coro();
}
co_return;
}
psco::awaitable<void> data_feed_loop_coro(std::atomic<bool>& running, std::weak_ptr<Data_Feed> feed_ref) {
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;
{
auto feed = feed_ref.lock();
if (!feed || !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;
}
if (all.empty()) {
feed.reset();
} else {
messages.reserve(all.size());
for (const auto* str : all) {
if (!str || str->empty()) {
std::cout << "警告:未知原因:" << feed->key << "内有空的消息体" << std::endl;
continue;
}
messages.emplace_back(*str);
}
}
}
if (messages.empty()) {
co_await data_feed_thread_wait_event_coro();
continue;
}
auto feed = feed_ref.lock();
if (!feed || !feed->enable || !feed->registered()) {
break;
}
for (auto& msg : messages) {
co_await feed->send_coro(msg);
}
feed.reset();
co_await data_feed_thread_yield_coro();
}
co_return;
}
void handle_loop_exception(std::exception_ptr exception) {
if (!exception) {
data_feed_thread_scheduler().set_exception(exception);
}
try {
std::rethrow_exception(exception);
}
catch (const psco::operation_cancelled&) {
}
catch (...) {
data_feed_thread_scheduler().set_exception(exception);
}
}
void sync_data_source_loop_tasks(std::atomic<bool>& running) {
auto sources = Global::instance()->mode_acs.data_source_config.map.list();
for (auto& source : sources) {
if (!source || !source->enable) {
continue;
}
if (source->loop_task && source->loop_task->valid()) {
continue;
}
if (!source->is_open() && !source->open()) {
continue;
}
source->loop_task = std::make_unique<psco::awaitable<void>>(psco::with_callback(data_source_loop_coro(running, std::weak_ptr<Data_Source>(source)), [](std::exception_ptr exception) mutable {
handle_loop_exception(exception);
}));
source->loop_task->start();
}
}
void sync_data_feed_loop_tasks(std::atomic<bool>& running) {
auto feeds = Global::instance()->mode_acs.data_feed_config.map.list();
for (auto& feed : feeds) {
if (!feed || !feed->enable) {
continue;
}
if (feed->loop_task && feed->loop_task->valid()) {
continue;
}
auto ret = feed->check_and_open();
if (!ret) {
continue;
}
feed->loop_task = std::make_unique<psco::awaitable<void>>(psco::with_callback(data_feed_loop_coro(running, std::weak_ptr<Data_Feed>(feed)), [](std::exception_ptr exception) mutable {
handle_loop_exception(exception);
}));
feed->loop_task->start();
}
}
bool has_running_loop_tasks() {
auto sources = Global::instance()->mode_acs.data_source_config.map.list();
for (auto& source : sources) {
if (source && source->loop_task && source->loop_task->valid()) {
return true;
}
}
auto feeds = Global::instance()->mode_acs.data_feed_config.map.list();
for (auto& feed : feeds) {
if (feed && feed->loop_task && feed->loop_task->valid()) {
return true;
}
}
return false;
}
void cancel_loop_tasks() {
auto sources = Global::instance()->mode_acs.data_source_config.map.list();
for (auto& source : sources) {
if (source && source->loop_task && source->loop_task->valid()) {
source->loop_task->cancel();
}
}
auto feeds = Global::instance()->mode_acs.data_feed_config.map.list();
for (auto& feed : feeds) {
if (feed && feed->loop_task && feed->loop_task->valid()) {
feed->loop_task->cancel();
}
}
}
void stop_loop_tasks() {
auto& scheduler = data_feed_thread_scheduler();
scheduler.wake();
auto start = std::chrono::steady_clock::now();
while (has_running_loop_tasks()) {
scheduler.rethrow_if_exception();
scheduler.drain();
if (!has_running_loop_tasks()) {
break;
}
if (std::chrono::steady_clock::now() - start > std::chrono::seconds(10)) {
std::cerr << "data loop task stop timeout, cancel remaining tasks" << std::endl;
cancel_loop_tasks();
break;
}
scheduler.wait_for_callback_for(std::chrono::milliseconds(1));
}
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() {
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));
}
void wake_data_feed_thread() {
data_feed_thread_scheduler().wake();
}
void stop_data_feed_thread_scheduler() {
data_feed_thread_scheduler().stop();
}
psco::awaitable<void> data_feed_thread_coro(std::atomic<bool>& running) {
data_feed_thread_scheduler().reset();
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_Config] [" << feed->key << "] 第一次打开失败！ 程序继续运行，等待后续重试。" << std::endl;
}
}
void stop_io_coro() {
running.store(false, std::memory_order_release);
if (!data_feed_thread_task) {
return;
}
while (running.load(std::memory_order_acquire)) {
auto& scheduler = data_feed_thread_scheduler();
scheduler.rethrow_if_exception();
sync_data_source_loop_tasks(running);
sync_data_feed_loop_tasks(running);
const auto resumed = scheduler.drain();
if (resumed == 0) {
scheduler.wait_for_work([&running] {
return !running.load(std::memory_order_acquire);
});
}
}
data_feed_thread_scheduler().stop();
stop_loop_tasks();
co_return;
}
void io_coro(std::atomic<bool>& running) {
try {
psco::sync_await(data_feed_thread_coro(running));
data_feed_thread_task->get();
}
catch (const std::exception& e) {
std::cerr << "data_feed_thread_coro exception: " << e.what() << std::endl;
@@ -254,4 +41,152 @@ void io_coro(std::atomic<bool>& running) {
std::cerr << "data_feed_thread_coro unknown exception" << std::endl;
Psc::fail_fast_core_dump("");
}
data_feed_thread_task.reset();
executor_.reset();
}
bool task_is_running(std::unique_ptr<concurrencpp::result<void>>& task) {
if (!task) {
return false;
}
if (task->status() == concurrencpp::result_status::idle) {
return true;
}
task->get();
task.reset();
return false;
}
concurrencpp::result<void> data_source_loop_coro(std::atomic<bool>& running, std::shared_ptr<Data_Source> source) {
co_await concurrencpp::resume_on(executor_);
while (running.load(std::memory_order_acquire)) {
if (!source->enable || !source->registered()) {
break;
}
if (!source->is_open() && !source->open()) {
co_await concurrencpp::resume_on(executor_);
continue;
}
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;
}
source->process_mode_acs_data(mode_data);
co_await concurrencpp::resume_on(executor_);
}
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;
{
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;
}
for (auto& msg : messages) {
co_await feed->send_coro(msg);
}
co_await concurrencpp::resume_on(executor_);
}
co_return;
}
bool 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)) {
return true;
}
}
auto feeds = Global::instance()->mode_acs.data_feed_config.map.list();
for (auto& feed : feeds) {
if (feed && task_is_running(feed->loop_task)) {
return true;
}
}
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_main.cpp
+14 -2
View File
@@ -55,6 +55,10 @@ struct Catch_Memory {
std::int64_t sm;
};
int psc_main(int argc, char *argv[]) {
std::cout << "wyc_main" << std::endl;
@@ -171,13 +175,17 @@ int psc_main(int argc, char *argv[]) {
}
}
});
manager.test_and_start_thread("io_coro 协程线程", io_coro);
// manager.test_and_start_thread("io_coro 协程线程", io_coro);
static int t = Global::instance()->mode_acs.read_milliseconds;
wake_data_feed_thread();
g->dsp_config.init_env();
bool enable = g->mlat.enable;
void start_io_coro();
start_io_coro();
// Catch_Memory cm;
while (stop_program == 0) {
for (auto &ds : g->mode_acs.data_source_config.map.list()) {
@@ -200,6 +208,10 @@ int psc_main(int argc, char *argv[]) {
}
}
void stop_io_coro();
stop_io_coro();
if (stop_program == SIGINT || stop_program == SIGTERM) {
clear();
}