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 11:00:00 +08:00
parent 3a4a3abe3f
commit 8ca05115ec
5 changed files with 153 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/Data_Feed/Data_Feed.cpp
+11 -1
View File
@@ -1,7 +1,17 @@
#include "Local_Server/Data_Feed/Data_Feed.h"
#include "../server/Global.h"
void Data_Feed_UDP_Server::handle_in_loop() {
ucoro::sync_await(svr.tick_coro());
ucoro::sync_await(handle_in_loop_coro());
}
ucoro::awaitable<void> Data_Feed_UDP_Server::handle_in_loop_coro() {
co_await svr.tick_coro();
co_return;
}
ucoro::awaitable<void> Data_Feed_UDP_Server::send_coro(const std::string& data) {
co_await svr.write_to_all_clients_coro(data);
co_return;
}
bool Data_Feed_UDP_Server::_open() {
svr.set_bind_address("0.0.0.0", port);
module/Local_Server/Data_Feed/Data_Feed.h
+38 -5
View File
@@ -55,6 +55,13 @@ public:
Frequency_Limit_Multi sbs_flm{};
virtual void handle_in_loop() {
}
virtual ucoro::awaitable<void> handle_in_loop_coro() {
handle_in_loop();
co_return;
}
virtual ucoro::awaitable<void> send_coro(const std::string&) {
co_return;
}
virtual Psc::JSON to_json() {
Psc::JSON ret = Psc::JSON::object();
@@ -126,8 +133,16 @@ public:
~Data_Feed_TCP_Server() override {}
void handle_in_loop() override {
//std::cout << socket.to_string() << "flush_clients" << std::endl;
ucoro::sync_await(svr.flush_clients_coro());
ucoro::sync_await(svr.tick_coro());
ucoro::sync_await(handle_in_loop_coro());
}
ucoro::awaitable<void> handle_in_loop_coro() override {
co_await svr.flush_clients_coro();
co_await svr.tick_coro();
co_return;
}
ucoro::awaitable<void> send_coro(const std::string& data) override {
co_await svr.write_to_all_clients_coro(data);
co_return;
}
void from_json(const Psc::JSON* that_json) override {
Data_Feed::from_json(that_json);
@@ -185,7 +200,15 @@ public:
class Data_Feed_TCP_Client : public Data_Feed {
public:
void handle_in_loop() override {
ucoro::sync_await(cli.tick_coro());
ucoro::sync_await(handle_in_loop_coro());
}
ucoro::awaitable<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
co_return;
}
ucoro::awaitable<void> send_coro(const std::string& data) override {
co_await cli.send_coro(data);
co_return;
}
std::string url;
Psc::JSON get_custom_state_json() override {
@@ -229,9 +252,11 @@ public:
Data_Feed_UDP_Server()= default;
Data_Feed_UDP_Server()= default;
~Data_Feed_UDP_Server() override = default;
void handle_in_loop() override;
ucoro::awaitable<void> handle_in_loop_coro() override;
ucoro::awaitable<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) {
@@ -268,7 +293,15 @@ public:
return VAR_JSON_1(state);
}
void handle_in_loop() override {
ucoro::sync_await(cli.tick_coro());
ucoro::sync_await(handle_in_loop_coro());
}
ucoro::awaitable<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
co_return;
}
ucoro::awaitable<void> send_coro(const std::string& data) override {
co_await cli.send_coro(data);
co_return;
}
Psc::asio_socket::UDP_Client_Coro cli;
std::string url;
module/Local_Server/Data_Feed/data_feed_thread.cpp
+20 -19
View File
@@ -133,8 +133,13 @@ std::optional<std::string> convert_to_send_format(Data_Source* ds, const std::sh
return send_msg;
}
void data_feed_thread(std::atomic<bool>& running) {
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();
@@ -150,8 +155,14 @@ void data_feed_thread(std::atomic<bool>& running) {
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();
bool all_empty = true;
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->source_step_coro();
}
// 这两个每次循环都要重新计算 保证实时性
// if (flush_limit.test()) {
auto all_feed = cfg.map.list();
@@ -160,7 +171,7 @@ void data_feed_thread(std::atomic<bool>& running) {
if (!feed->enable) continue;
enable_feeds.push_back(feed);
// 执行特定的刷新逻辑 比如 tcp_server 检查自身的连接
feed->handle_in_loop();
co_await feed->handle_in_loop_coro();
}
@@ -200,22 +211,7 @@ void data_feed_thread(std::atomic<bool>& running) {
std::cout << "警告:未知原因:" << feed->key << "内有空的消息体" << std::endl;
continue;
}
if (feed->type == "Data_Feed_TCP_Server") {
auto it = dynamic_cast<Data_Feed_TCP_Server*>(feed.get());
ucoro::sync_await(it->svr.write_to_all_clients_coro(*str));
}
else if (feed->type == "Data_Feed_UDP_Server") {
auto it = dynamic_cast<Data_Feed_UDP_Server*>(feed.get());
ucoro::sync_await(it->svr.write_to_all_clients_coro(*str));
}
else if (feed->type == "Data_Feed_TCP_Client") {
auto it = dynamic_cast<Data_Feed_TCP_Client*>(feed.get());
ucoro::sync_await(it->cli.send_coro(*str));
}
else if (feed->type == "Data_Feed_UDP_Client") {
auto it = dynamic_cast<Data_Feed_UDP_Client*>(feed.get());
ucoro::sync_await(it->cli.send_coro(*str));
}
co_await feed->send_coro(*str);
}
}
if (t != 0) {
@@ -226,5 +222,10 @@ void data_feed_thread(std::atomic<bool>& running) {
}
}
}
co_return;
}
void data_feed_thread(std::atomic<bool>& running) {
ucoro::sync_await(data_feed_thread_coro(running));
}
module/Local_Server/Data_Source/Data_Source.cpp
+67 -11
View File
@@ -1,10 +1,23 @@
#include "Data_Source.h"
#include <algorithm>
#include <asio/post.hpp>
#include <asio/thread_pool.hpp>
#include <codecvt>
#include <thread>
#include "../server/Global.h"
#include "Database.h"
namespace {
asio::thread_pool& data_source_process_pool() {
const auto hardware_threads = std::thread::hardware_concurrency();
const auto worker_count = std::max(2u, hardware_threads > 2 ? hardware_threads - 2 : hardware_threads);
static asio::thread_pool pool(worker_count);
return pool;
}
}
Psc::serial::Serial* create_serial(const std::string& serial_name, Baud_Rate_Type baud_rate) {
auto serial = new Psc::serial::Serial;
serial->set_serial_name(serial_name);
@@ -55,17 +68,9 @@ std::string Data_Source::thread_key() const {
}
void Data_Source::test_and_attach_thread() {
//static int t = Global::instance()->mode_acs.read_milliseconds;
Global::instance()->thread_manager.test_and_start_thread(thread_key(), [this](std::atomic<bool>& running) {
if (enable) {
open();
}
while (running.load(std::memory_order_acquire) == true)
{
handle_in_loop();
handle_mode_acs_source();
}
});
if (enable) {
open();
}
}
@@ -174,7 +179,10 @@ void Data_Source::handle_mode_acs_source() {
return;
}
process_mode_acs_data(mode_data);
}
void Data_Source::process_mode_acs_data(const std::string& mode_data) {
auto size = mode_data.size();
read_speed.update(size);
value_statistics.update(size);
@@ -204,6 +212,49 @@ void Data_Source::handle_mode_acs_source() {
});
}
ucoro::awaitable<void> Data_Source::handle_in_loop_coro() {
handle_in_loop();
co_return;
}
ucoro::awaitable<std::string> Data_Source::recv_coro() {
co_return read();
}
ucoro::awaitable<void> Data_Source::source_step_coro() {
co_await handle_in_loop_coro();
auto mode_data = co_await recv_coro();
if (mode_data.empty()) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
co_return;
}
struct Process_Result {
std::exception_ptr exception;
};
auto self = that();
auto result = co_await ucoro::callback_awaitable<Process_Result>(
[this, self = std::move(self), mode_data = std::move(mode_data)](auto done) mutable {
asio::post(
data_source_process_pool(),
[this, self = std::move(self), mode_data = std::move(mode_data), done = std::move(done)]() mutable {
try {
process_mode_acs_data(mode_data);
done(Process_Result{});
} catch (...) {
done(Process_Result{std::current_exception()});
}
});
});
if (result.exception) {
std::rethrow_exception(result.exception);
}
co_return;
}
void Data_Source::test_and_stop_thread() const {
Global::instance()->thread_manager.test_and_stop_thread(thread_key());
@@ -675,6 +726,11 @@ void File_Data_Source::handle_mode_acs_source() {
}
}
ucoro::awaitable<void> File_Data_Source::source_step_coro() {
handle_mode_acs_source();
co_return;
}
std::string Dll_Data_Source::read() {
auto start = std::chrono::steady_clock::now();
module/Local_Server/Data_Source/Data_Source.h
+17 -2
View File
@@ -107,6 +107,12 @@ public:
virtual std::shared_ptr<SSR::Msg> create_msg(const std::shared_ptr<SSR::Data_Source_Interface>& source, const std::string& packet);
virtual void update_msg_day_time(std::shared_ptr<SSR::Msg> msg);
virtual void handle_mode_acs_source();
virtual ucoro::awaitable<void> handle_in_loop_coro();
virtual ucoro::awaitable<std::string> recv_coro();
virtual ucoro::awaitable<void> source_step_coro();
void process_mode_acs_data(const std::string& mode_data);
void refresh_data_feed_key_list();
@@ -196,6 +202,7 @@ public:
}
return true;
}
bool is_open() const { return _open_; }
void close() {
if (_open_) {
_open_ = false;
@@ -224,7 +231,11 @@ protected:
struct TCP_Client_Data_Source : Data_Source {
void handle_in_loop() override {
ucoro::sync_await(cli.tick_coro());
ucoro::sync_await(handle_in_loop_coro());
}
ucoro::awaitable<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
co_return;
}
Psc::JSON get_custom_state_json() override {
auto& state = cli.state;
@@ -263,7 +274,10 @@ struct TCP_Client_Data_Source : Data_Source {
}
std::string read() override {
return ucoro::sync_await(cli.read_coro());
return ucoro::sync_await(recv_coro());
}
ucoro::awaitable<std::string> recv_coro() override {
co_return co_await cli.read_coro();
}
};
struct Serial_Data_Source : Data_Source {
@@ -377,6 +391,7 @@ struct File_Data_Source : public Data_Source {
}
std::string read() override;
void handle_mode_acs_source() override;
ucoro::awaitable<void> source_step_coro() override;
protected:
std::optional<std::string> get_raw_line(int &ret_index);