ae/ECAP_Server
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修完bug

Browse Source
This commit is contained in:
种太阳的热带美短
2026-08-10 16:36:12 +08:00
parent 18e92d3631
commit d28edab2d7
12 changed files with 389 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/Data_Feed/Data_Feed.cpp
+25 -3
View File
@@ -33,6 +33,7 @@ asio::awaitable<void> Data_Feed_UDP_Server::_open() {
const auto value = specific.read([](const auto& value) { return value; });
svr.set_bind_address("0.0.0.0", value.port);
svr.create();
co_await svr.bind_coro();
server_logger->c_debug({}, {}, to_string() + "开启！");
co_return;
}
@@ -74,11 +75,16 @@ void BIN_Msg_Buffer::push(std::string_view msg) {
else {
msg_list_cache.back()->append(msg);
}
cache_bytes += msg.size();
high_water_messages = std::max(high_water_messages, msg_list_cache.size());
high_water_bytes = std::max(high_water_bytes, cache_bytes);
}
const std::vector<std::string*>& BIN_Msg_Buffer::get_all() {
std::lock_guard<std::mutex> g(mtx);
std::swap(msg_list_cache, msg_list);
msg_list_cache.clear();
dispatch_bytes = cache_bytes;
cache_bytes = 0;
mode_s_msg_num = 0;
mode_other_msg_num = 0;
return msg_list;
@@ -89,15 +95,31 @@ BIN_Msg_Buffer::BIN_Msg_Buffer() {
}
Psc::JSON BIN_Msg_Buffer::state_json() {
auto output_packet_size = Global::instance()->mode_acs.data_feed_config.settings.member<&Data_feed_Config_Data::packet_byte_size>().read([](const auto& value) { return value; });
size_t size, cache_size, pool_capacity, pool_free_count;
size_t size, cache_size, current_cache_bytes, current_dispatch_bytes, current_high_water_messages, current_high_water_bytes, pool_capacity, pool_free_count;
{
std::lock_guard<std::mutex> g(mtx);
auto& pool = Global::instance()->mode_acs.data_feed_config.pool_;
size = msg_list.size();
cache_size = msg_list_cache.size();
current_cache_bytes = cache_bytes;
current_dispatch_bytes = dispatch_bytes;
current_high_water_messages = high_water_messages;
current_high_water_bytes = high_water_bytes;
pool_capacity = pool.capacity();
pool_free_count = pool.free_count();
}
return VAR_JSON_7(mode_s_msg_num, mode_other_msg_num, output_packet_size, cache_size, size, pool_capacity,
pool_free_count);
Psc::JSON ret = Psc::JSON::object();
ret.append({"mode_s_pending", mode_s_msg_num.load()});
ret.append({"mode_other_pending", mode_other_msg_num.load()});
ret.append({"batch_target_bytes", output_packet_size});
ret.append({"pending_batches", cache_size});
ret.append({"pending_bytes", current_cache_bytes});
ret.append({"last_dispatch_batches", size});
ret.append({"last_dispatch_bytes", current_dispatch_bytes});
ret.append({"high_water_batches", current_high_water_messages});
ret.append({"high_water_bytes", current_high_water_bytes});
ret.append({"pool_capacity", pool_capacity});
ret.append({"pool_free_count", pool_free_count});
ret.append({"pool_utilization_percent", pool_capacity == 0 ? 0.0 : static_cast<double>(pool_capacity - pool_free_count) * 100.0 / static_cast<double>(pool_capacity)});
return ret;
}
module/Local_Server/Data_Feed/Data_Feed.h
+64 -27
View File
@@ -33,6 +33,12 @@ public:
BIN_Msg_Buffer msg_buffer{};
adminive::Managed_Value<Output_Format> output_format;
Frequency_Limit_Multi sbs_flm{};
Psc::Throughput_Statistics queued_statistics;
Psc::Throughput_Statistics dispatched_statistics;
Psc::Throughput_Statistics dropped_statistics;
Psc::Throughput_Statistics filtered_statistics;
std::atomic<std::uint64_t> mode_s_limit_drop_total{};
std::atomic<std::uint64_t> mode_other_limit_drop_total{};
asio::awaitable<void> loop_coro() final;
virtual void handle_in_loop() {
}
@@ -60,7 +66,17 @@ public:
return Psc::JSON::object();
}
virtual Psc::JSON get_state_json() {
Psc::JSON ret = msg_buffer.state_json();
Psc::JSON pipeline = Psc::JSON::object();
pipeline.append({"queued", queued_statistics.to_json()});
pipeline.append({"dispatched", dispatched_statistics.to_json()});
pipeline.append({"dropped", dropped_statistics.to_json()});
pipeline.append({"filtered", filtered_statistics.to_json()});
pipeline.append({"mode_s_limit_drop_total", mode_s_limit_drop_total.load()});
pipeline.append({"mode_other_limit_drop_total", mode_other_limit_drop_total.load()});
Psc::JSON ret = Psc::JSON::object();
ret.append({"lifecycle", lifecycle_state_json()});
ret.append({"queue", msg_buffer.state_json()});
ret.append({"pipeline", pipeline});
ret += get_custom_state_json();
return ret;
}
@@ -68,6 +84,22 @@ public:
return "Data_Feed{" + key + "," + (enabled() ? "true" : "false") + "," + type + "}";
}
bool registered();
void record_queued(std::size_t bytes) {
queued_statistics.update(bytes);
}
void record_dispatched(std::size_t bytes) {
dispatched_statistics.update(bytes);
}
void record_dropped(std::size_t bytes, bool mode_s) {
dropped_statistics.update(bytes);
if (mode_s)
++mode_s_limit_drop_total;
else
++mode_other_limit_drop_total;
}
void record_filtered(std::size_t bytes) {
filtered_statistics.update(bytes);
}
protected:
Data_Feed() = default;
Frequency_Limit too_many_msg_limit;
@@ -96,11 +128,13 @@ public:
adminive::Managed_Value<Data_Feed_TCP_Server_Data> specific;
Psc::asio_socket::TCP_Server_Coro svr;
Psc::JSON get_custom_state_json() override {
auto& state = svr.state;
auto value = specific.read([](const auto& value) { return value; });
auto ret = value.to_base_json();
ret.key = "fixed";
auto result = VAR_JSON_1(state);
Psc::JSON result = Psc::JSON::object();
result.append({"transport_state", Psc::to_string(svr.state.load())});
result.append({"transport", svr.get_statistics_json()});
result.append({"client_count", svr.get_all_clients().size()});
result.append(ret);
return result;
}
@@ -146,21 +180,7 @@ public:
auto& cur_port = info.sockaddr.port;
cur.append({"base", VAR_STR_3(cur_ip, cur_port, fd)});
cur.append({"state", conn->send_buffer.state_str()});
{
auto& ins = conn->push_speed.instant_speed;
auto& avr = conn->push_speed.average_speed;
cur.append({"send_speed(KB/s)", VAR_STR_2(ins, avr)});
}
{
auto& ins = conn->lose_speed.instant_speed;
auto& avr = conn->lose_speed.average_speed;
cur.append({"lose_speed(KB/s)", VAR_STR_2(ins, avr)});
}
{
auto& ins = conn->send_num.instant;
auto& avr = conn->send_num.average;
cur.append({"send_num(byte/count)", VAR_STR_2(ins, avr)});
}
cur.append({"statistics", conn->statistics.to_json()});
ret.append(cur);
}
return ret;
@@ -195,8 +215,10 @@ public:
co_return;
}
Psc::JSON get_custom_state_json() override {
auto& state = cli.state;
return VAR_JSON_1(state);
Psc::JSON ret = Psc::JSON::object();
ret.append({"transport_state", Psc::to_string(cli.state.load())});
ret.append({"transport", cli.statistics.to_json()});
return ret;
}
Psc::asio_socket::TCP_Client_Coro cli;
~Data_Feed_TCP_Client() override = default;
@@ -243,8 +265,20 @@ class Data_Feed_UDP_Server : public Data_Feed {
public:
adminive::Managed_Value<Data_Feed_UDP_Server_Data> specific;
Psc::JSON get_custom_state_json() override {
auto& state = svr.state;
return VAR_JSON_1(state);
const auto clients = svr.get_all_clients();
const auto now = Psc::get_current_millisecond_timestamp();
std::size_t active_client_count = 0;
for (const auto& client : clients) {
const auto last_activity = client.statistics->last_activity_unix_ms();
if (last_activity != 0 && now - last_activity <= 30000)
++active_client_count;
}
Psc::JSON ret = Psc::JSON::object();
ret.append({"transport_state", Psc::to_string(svr.state.load())});
ret.append({"transport", svr.statistics.to_json()});
ret.append({"client_count", clients.size()});
ret.append({"active_client_count", active_client_count});
return ret;
}
Data_Feed_UDP_Server() = default;
~Data_Feed_UDP_Server() override = default;
@@ -252,10 +286,11 @@ public:
asio::awaitable<void> send_coro(std::string_view data) override;
[[nodiscard]] Psc::JSON get_clients_json() const {
Psc::JSON ret = Psc::JSON::array();
for (const auto& i : svr.clients) {
for (const auto& client : svr.get_all_clients()) {
Psc::JSON cur = Psc::JSON::object();
cur.append({"ip", i.ip});
cur.append({"port", i.port});
cur.append({"ip", client.address.ip});
cur.append({"port", client.address.port});
cur.append({"statistics", client.statistics->to_json()});
ret.append(cur);
}
return ret;
@@ -297,8 +332,10 @@ class Data_Feed_UDP_Client : public Data_Feed {
public:
adminive::Managed_Value<Data_Feed_UDP_Client_Data> specific;
Psc::JSON get_custom_state_json() override {
auto& state = cli.state;
return VAR_JSON_1(state);
Psc::JSON ret = Psc::JSON::object();
ret.append({"transport_state", Psc::to_string(cli.state.load())});
ret.append({"transport", cli.statistics.to_json()});
return ret;
}
asio::awaitable<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
module/Local_Server/Data_Source/Data_Source.cpp
+22 -5
View File
@@ -23,10 +23,17 @@ bool Data_Source::registered() const {
}
Psc::JSON Data_Source::get_state() {
Psc::JSON ret = Psc::JSON::object();
ret.append({"state", Psc::to_string(state)});
ret.append({"lifecycle", lifecycle_state_json()});
ret.append_list(get_custom_state_json().children);
ret.append({"ds: read_speed（byte）", read_speed});
ret.append({"ds: value_statistics（byte）", value_statistics});
ret.append({"input", input_statistics.to_json()});
ret.append({"mode_ac", mode_ac_statistic.to_json()});
ret.append({"mode_s", mode_s_statistic.to_json()});
Psc::JSON aircraft = Psc::JSON::object();
aircraft.append({"total", get_aircraft_num()});
aircraft.append({"with_position", have_pos_aircraft_num.load()});
ret.append({"aircraft", aircraft});
ret.append({"base_station", base_station.to_Json()});
ret.append({"connected_feeds", get_all_connect_feed_status()});
return ret;
}
Data_Source::Data_Source() {
@@ -158,6 +165,7 @@ std::optional<std::string> File_Data_Source::get_raw_line(int& ret_index) {
const auto config = specific.read([](const auto& value) { return value; });
if (config.play_mode == Play_Mode::loop) {
index = 0;
++loop_total;
}
else {
if (!have_report_play_back_all_success) {
@@ -173,7 +181,9 @@ std::optional<std::string> File_Data_Source::get_raw_line(int& ret_index) {
}
}
ret_index = index + 1;
return part_infos[index++];
const auto& record = part_infos[index++];
raw_read_statistics.update(record.size());
return record;
}
// 1A 33 1A 1A F1 FB 87 73 7E 7F a8001d81a87543b0a80000
std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, int index, std::string_view log_info) {
@@ -340,6 +350,8 @@ void File_Data_Source::origin_data_transform_mode_data(std::string& mode_data) {
if (config.play_mode != Play_Mode::analysis) {
std::optional<std::string> log_info = get_raw_line(ret_index);
mode_data = log_info.has_value() ? get_true_from_raw_line(this, config.data_type, ret_index, log_info.value()) : "";
if (log_info.has_value() && !log_info->empty() && mode_data.empty())
conversion_dropped_statistics.update(log_info->size());
}
else {
int num = 0;
@@ -349,6 +361,8 @@ void File_Data_Source::origin_data_transform_mode_data(std::string& mode_data) {
num++;
auto data = log_info.value();
auto cur = get_true_from_raw_line(this, config.data_type, ret_index, data);
if (!data.empty() && cur.empty())
conversion_dropped_statistics.update(data.size());
ret += cur;
// 一定要在这个位置，否则会导致遗漏报文
if (num == 1000) {
@@ -398,7 +412,7 @@ void Dll_Data_Source::origin_data_transform_mode_data(std::string& mode_data) {
const auto config = specific.read([](const auto& value) { return value; });
auto buf = reinterpret_cast<char*>(buffer.data());
auto len = read_func_ptr(buf, config.buffer_size);
vs.update(len);
read_statistics.update(len);
std::string data(buf, len);
if (data.empty()) {
auto end = std::chrono::steady_clock::now();
@@ -416,10 +430,12 @@ asio::awaitable<void> Shared_Memory_Data_Source::_open() {
const auto config = specific.read([](const auto& value) { return value; });
sm = std::make_unique<Psc::SM_RingBuffer>();
sm->init(config.shared_memory_name, config.shared_memory_size);
shared_memory_open = true;
co_return;
}
asio::awaitable<void> Shared_Memory_Data_Source::_close() {
sm.reset();
shared_memory_open = false;
co_return;
}
void Shared_Memory_Data_Source::origin_data_transform_mode_data(std::string& mode_data) {
@@ -430,6 +446,7 @@ void Shared_Memory_Data_Source::origin_data_transform_mode_data(std::string& mod
bool ok = sm->read((uint8_t*)data.data(), length);
if (!ok) {
mode_data = "";
return;
}
if (length != size) {
data.resize(length);
module/Local_Server/Data_Source/Data_Source.h
+51 -14
View File
@@ -221,8 +221,10 @@ public:
co_return;
}
Psc::JSON get_custom_state_json() override {
auto& state = cli.state;
return VAR_JSON_1(state);
Psc::JSON ret = Psc::JSON::object();
ret.append({"transport_state", Psc::to_string(cli.state.load())});
ret.append({"transport", cli.statistics.to_json()});
return ret;
}
Psc::asio_socket::TCP_Client_Coro cli;
~TCP_Client_Data_Source() override = default;
@@ -277,7 +279,11 @@ class Serial_Data_Source : public Data_Source {
public:
adminive::Managed_Value<Serial_Data_Source_Data> specific;
Psc::JSON get_custom_state_json() override {
return Psc::JSON::object();
Psc::JSON ret = Psc::JSON::object();
ret.append({"serial_open", serial_open.load()});
if (serial)
ret.append({"transport", serial->statistics.to_json()});
return ret;
}
Serial_Data_Source() {
type = "Serial_Data_Source";
@@ -292,13 +298,15 @@ public:
serial->set_stop_bits(Psc::serial::StopBits::OneStop);
serial->set_flow_control(Psc::serial::FlowControl::HardwareControl);
serial->set_buffer_byte_size(10 * 1024);
if (!serial->open())
serial_open = serial->open();
if (!serial_open)
std::cerr << "createSerial " + value.port_name + ":" + std::to_string(value.baud_rate) + " 打开串口失败！\n";
co_return;
}
asio::awaitable<void> _close() override {
if (serial)
serial->close();
serial_open = false;
co_return;
}
asio::awaitable<void> handle_in_loop_coro() override {
@@ -321,6 +329,7 @@ public:
});
}
std::unique_ptr<Psc::serial::Serial_Coro> serial{};
std::atomic_bool serial_open{};
~Serial_Data_Source() override {
if (serial)
serial->close();
@@ -363,9 +372,23 @@ public:
bool have_report_play_back_all_success = false;
std::string state = "null";
Psc::JSON get_custom_state_json() override {
auto cur_virtual_time = player_clock.get_cur_time_point();
auto playback_speed_rate = player_clock.playback_speed_rate;
return VAR_JSON_5(cur_virtual_time, playback_speed_rate, state, index, part_infos.size());
std::lock_guard lock(mtx);
const auto total_records = part_infos.size();
const auto current_index = static_cast<std::size_t>(std::max<long long>(index, 0));
Psc::JSON playback = Psc::JSON::object();
playback.append({"state", state});
playback.append({"virtual_time", player_clock.get_cur_time_point().to_string()});
playback.append({"speed_rate", player_clock.playback_speed_rate});
playback.append({"current_index", current_index});
playback.append({"total_records", total_records});
playback.append({"remaining_records", total_records > current_index ? total_records - current_index : 0});
playback.append({"progress_percent", total_records == 0 ? 0.0 : static_cast<double>(current_index) * 100.0 / static_cast<double>(total_records)});
playback.append({"loop_total", loop_total.load()});
playback.append({"raw_read", raw_read_statistics.to_json()});
playback.append({"conversion_dropped", conversion_dropped_statistics.to_json()});
Psc::JSON ret = Psc::JSON::object();
ret.append({"playback", playback});
return ret;
}
File_Data_Source() {
type = "File_Data_Source";
@@ -382,6 +405,9 @@ public:
std::shared_ptr<SSR::Mode_S_Msg> msg;
};
Player_Clock player_clock;
Psc::Throughput_Statistics raw_read_statistics;
Psc::Throughput_Statistics conversion_dropped_statistics;
std::atomic<std::uint64_t> loop_total{};
std::shared_ptr<Cache_Msg> pre_cache = nullptr;
std::deque<std::shared_ptr<Cache_Msg>> cache_list;
std::string get_true_file_path() const {
@@ -432,11 +458,13 @@ class Dll_Data_Source : public Data_Source {
public:
adminive::Managed_Value<Dll_Data_Source_Data> specific;
Psc::JSON get_custom_state_json() override {
auto ret = VAR_JSON_1(state);
ret.append({"read_vaild_len", vs.to_string()});
std::lock_guard lock(state_mtx);
Psc::JSON ret = Psc::JSON::object();
ret.append({"state", state});
ret.append({"library_read", read_statistics.to_json()});
return ret;
}
Psc::Value_Statistics vs;
Psc::Throughput_Statistics read_statistics;
Dll_Data_Source() {
type = "Dll_Data_Source";
}
@@ -459,6 +487,7 @@ public:
using Call_Back = void (*)(char* buf, std::size_t len);
using set_Call_back = void (*)(Call_Back);
std::string state;
mutable std::mutex state_mtx;
asio::awaitable<void> _close() override {
Psc::free_library(lib);
co_return;
@@ -468,7 +497,8 @@ public:
{
auto r = Psc::try_load_library(Psc::get_abs_path(value.library_path));
if (!r) {
state = r.error().message();
std::lock_guard lock(state_mtx);
state = Psc::platform_2_utf8(r.error().message());
co_return;
}
lib = r.value();
@@ -476,13 +506,17 @@ public:
{
auto r = Psc::try_load_function(lib, value.function_name);
if (!r) {
state = r.error().message();
std::lock_guard lock(state_mtx);
state = Psc::platform_2_utf8(r.error().message());
std::cout << LOG_POS << " [" << value.function_name << "] 函数指针加载失败!" << std::endl;
co_return;
}
read_func_ptr = (Func_Type)r.value();
}
state = "加载成功";
{
std::lock_guard lock(state_mtx);
state = "加载成功";
}
co_return;
}
void origin_data_transform_mode_data(std::string& data) override;
@@ -510,7 +544,9 @@ class Shared_Memory_Data_Source : public Data_Source {
public:
adminive::Managed_Value<Shared_Memory_Data_Source_Data> specific;
Psc::JSON get_custom_state_json() override {
return Psc::JSON::object();
Psc::JSON ret = Psc::JSON::object();
ret.append({"shared_memory_open", shared_memory_open.load()});
return ret;
}
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Source::from_json(that_json, not_exist_use_default_value);
@@ -526,6 +562,7 @@ public:
void origin_data_transform_mode_data(std::string& data) override;
protected:
std::unique_ptr<Psc::SM_RingBuffer> sm;
std::atomic_bool shared_memory_open{};
};
inline std::shared_ptr<Data_Source> create_data_source_from_type(std::string_view t) {
std::shared_ptr<Data_Source> ret{};
module/Local_Server/Data_Source/Data_Source_Handler.cpp
+24 -18
View File
@@ -108,8 +108,7 @@ size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data)
}
auto source = ds(this);
auto size = mode_data.size();
read_speed.update(size);
value_statistics.update(size);
input_statistics.update(size);
// 预处理数据
static bool mode_s_console = Global::instance()->console_config.mode_s_console;
static bool record_playback = Global::instance()->console_config.record_playback;
@@ -410,9 +409,14 @@ void Data_Source_Handler::push_to_feed(const std::shared_ptr<SSR::Msg>& msg) {
const auto [other_size, s_size] = Global::instance()->mode_acs.settings.read([](const auto& value) {
return std::pair{value.mode_other_max_num, value.mode_s_max_num};
});
std::vector<std::string> list;
for (const auto& item : cached_data_feed_key_list) {
auto& key = item.key;
std::vector<std::string> feed_keys;
{
std::lock_guard lock(cdf_mtx);
feed_keys.reserve(cached_data_feed_key_list.size());
for (const auto& item : cached_data_feed_key_list)
feed_keys.push_back(item.key);
}
for (const auto& key : feed_keys) {
auto opt_feed = Global::instance()->mode_acs.data_feed_config.map.get(key);
if (!opt_feed.has_value()) {
std::cout << "wrong data feed: " << key << std::endl;
@@ -421,7 +425,13 @@ void Data_Source_Handler::push_to_feed(const std::shared_ptr<SSR::Msg>& msg) {
std::shared_ptr<Data_Feed>& feed = opt_feed.value();
if (!feed->enabled())
continue;
if (msg->type == SSR::Mode_Msg::T::S7 || msg->type == SSR::Mode_Msg::T::S14) {
const bool mode_s = msg->type == SSR::Mode_Msg::T::S7 || msg->type == SSR::Mode_Msg::T::S14;
std::optional<std::string> binary = convert_to_send_format(this, feed, msg);
if (!binary.has_value()) {
feed->record_filtered(msg->packet.size());
continue;
}
if (mode_s) {
// static Frequency_Limit fl;
// if (!fl.test()) {
// std::ostringstream oss;
@@ -429,7 +439,8 @@ void Data_Source_Handler::push_to_feed(const std::shared_ptr<SSR::Msg>& msg) {
// ,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) {
if (feed->msg_buffer.mode_s_msg_num >= s_size) {
feed->record_dropped(binary->size(), true);
continue;
}
else {
@@ -437,27 +448,22 @@ void Data_Source_Handler::push_to_feed(const std::shared_ptr<SSR::Msg>& msg) {
}
}
else {
if (feed->msg_buffer.mode_other_msg_num > other_size) {
if (feed->msg_buffer.mode_other_msg_num >= other_size) {
feed->record_dropped(binary->size(), false);
continue;
}
else {
++feed->msg_buffer.mode_other_msg_num;
}
}
std::optional<std::string> binary = convert_to_send_format(this, feed, msg);
if (binary.has_value()) {
feed->msg_buffer.push(binary.value());
}
feed->msg_buffer.push(binary.value());
feed->record_queued(binary->size());
}
}
Psc::JSON Data_Source_Handler::get_all_connect_feed_status() {
Psc::JSON ret = Psc::JSON::object();
std::vector<Cached_Source_Info> list;
{
std::lock_guard g(cdf_mtx);
list = cached_data_feed_key_list;
}
for (auto& i : list) {
std::lock_guard lock(cdf_mtx);
for (auto& i : cached_data_feed_key_list) {
Psc::JSON cur = Psc::JSON::object();
auto odf = Global::instance()->mode_acs.data_feed_config.map.get(i.key);
if (odf.has_value()) {
module/Local_Server/Data_Source/Data_Source_Handler.h
+1 -2
View File
@@ -21,8 +21,7 @@ public:
size_t process_mode_acs_data(std::string_view mode_data);
Mode_AC_Statistic_Data mode_ac_statistic;
Mode_S_Statistic_Data mode_s_statistic;
Psc::Speed_Statistics read_speed;
Psc::Value_Statistics value_statistics;
Psc::Throughput_Statistics input_statistics;
~Data_Source_Handler() override = default;
// 推送到 data_feed 相关代码
void push_to_feed(const std::shared_ptr<SSR::Msg>& msg);
module/Local_Server/global_include.h
+71 -57
View File
@@ -174,16 +174,15 @@ public:
Psc::JSON to_json() {
std::lock_guard g(mtx);
Psc::JSON ret = Psc::JSON::object();
Ret_J(pre_second_num);
Ret_J(total_num);
Ret_J(length_error_num);
// Ret_J(speed);
const auto messages_per_second = utc == std::time(nullptr) ? pre_second_num : 0;
ret.append({"messages_per_second", messages_per_second});
ret.append({"messages_total", total_num});
ret.append({"length_errors_total", length_error_num.load()});
return ret;
}
void add() {
std::lock_guard g(mtx);
add_inter();
// speed.update(1);
}
void add_length_error() {
std::lock_guard g(mtx);
@@ -205,7 +204,6 @@ protected:
size_t pre_second_num{};
size_t total_num{};
time_t utc{};
// Psc::Speed_Statistics speed;
};
class Statistic_Data {
public:
@@ -214,8 +212,9 @@ public:
size_t total_num{};
[[nodiscard]] Psc::JSON to_json() const {
Psc::JSON ret = Psc::JSON::object();
Ret_J(pre_second_num);
Ret_J(total_num);
const auto messages_per_second = utc == std::time(nullptr) ? pre_second_num : 0;
ret.append({"messages_per_second", messages_per_second});
ret.append({"messages_total", total_num});
return ret;
}
Statistic_Data() = default;
@@ -227,32 +226,41 @@ public:
size_t total_num{};
size_t crc_error_num{};
Psc::JSON to_json() {
Psc::JSON ret = Psc::JSON::object();
Ret_J(pre_second_num);
Ret_J(total_num);
Ret_J(crc_error_num);
// Ret_J(speed);
std::string t = "null";
if (total_num != 0) {
t = std::to_string(static_cast<double>(crc_error_num) / static_cast<double>(total_num)) + "%";
}
ret.append({"crc误码率", t});
for (auto& cur : statistic_map) {
ret.append({cur.first, cur.second.to_json()});
}
return ret;
std::lock_guard lock(mtx);
return to_json_locked();
}
[[nodiscard]] std::tuple<size_t, size_t, size_t> counters() const {
std::lock_guard lock(mtx);
return {utc == std::time(nullptr) ? pre_second_num : 0, total_num, crc_error_num};
}
void add() {
std::lock_guard lock(mtx);
add_inter();
}
void add_crc_error() {
std::lock_guard lock(mtx);
crc_error_num++;
}
void add_sub_part(std::string_view key) {
std::lock_guard lock(mtx);
add_sub_part_inter(key);
}
// Psc::Speed_Statistics speed;
protected:
Psc::JSON to_json_locked() const {
Psc::JSON ret = Psc::JSON::object();
const auto messages_per_second = utc == std::time(nullptr) ? pre_second_num : 0;
const auto crc_error_percent = total_num == 0 ? 0.0 : static_cast<double>(crc_error_num) * 100.0 / static_cast<double>(total_num);
ret.append({"messages_per_second", messages_per_second});
ret.append({"messages_total", total_num});
ret.append({"crc_errors_total", crc_error_num});
ret.append({"crc_error_percent", crc_error_percent});
Psc::JSON subtypes = Psc::JSON::object();
for (const auto& cur : statistic_map) {
subtypes.append({cur.first, cur.second.to_json()});
}
ret.append({"subtypes", subtypes});
return ret;
}
void add_inter() {
total_num++;
auto cur_time = std::time(nullptr);
@@ -261,7 +269,6 @@ protected:
pre_second_num = 0;
}
pre_second_num++;
// speed.update(1);
}
void add_sub_part_inter(std::string_view key) {
auto t = get_create_statistic_data(key);
@@ -285,10 +292,12 @@ protected:
return t;
}
std::map<std::string, Statistic_Data> statistic_map;
mutable std::mutex mtx;
};
class CPR_Percentage_Statistic_Data {
public:
void add(SSR::CPR_Type cpr_type, SSR::CPR_Ret_Type result_type) {
std::lock_guard lock(mtx);
switch (result_type) {
case SSR::CPR_Ret_Type::Speed_Error:
case SSR::CPR_Ret_Type::Parse_OK:
@@ -305,18 +314,22 @@ public:
++statistic.result_num[result_type];
}
Psc::JSON to_json() const {
std::lock_guard lock(mtx);
Psc::JSON result = Psc::JSON::object();
for (const auto& [cpr_type, statistic] : statistic_map) {
Psc::JSON type_result = Psc::JSON::object();
type_result.append({"总计数", statistic.total_num});
type_result.append({"total", statistic.total_num});
for (auto result_type : statistic_types) {
size_t num = 0;
auto iter = statistic.result_num.find(result_type);
if (iter != statistic.result_num.end()) {
num = iter->second;
}
double rate = static_cast<double>(num) / static_cast<double>(statistic.total_num) * 100.0;
type_result.append({Psc::to_string(result_type), std::format("{} {:.2f}%", num, rate)});
double rate = statistic.total_num == 0 ? 0.0 : static_cast<double>(num) / static_cast<double>(statistic.total_num) * 100.0;
Psc::JSON value = Psc::JSON::object();
value.append({"count", num});
value.append({"percent", rate});
type_result.append({Psc::to_string(result_type), value});
}
result.append({Psc::to_string(cpr_type), type_result});
}
@@ -331,29 +344,35 @@ protected:
SSR::CPR_Ret_Type::Speed_Error, SSR::CPR_Ret_Type::Parse_OK, SSR::CPR_Ret_Type::Inter_Error,
SSR::CPR_Ret_Type::Time_Space_Too_Long, SSR::CPR_Ret_Type::Out_of_Maximum_Detection_Range};
std::map<SSR::CPR_Type, Statistic_Data> statistic_map;
mutable std::mutex mtx;
};
class Mode_S_Statistic_Data {
public:
Psc::JSON to_json() {
size_t total_num = get_total_num();
size_t crc_error_num = get_crc_error_num();
Psc::JSON ret = Psc::JSON::object();
ret.append({"每秒数量", get_total_pre_second_num()});
Ret_J(length_error_num);
Ret_J(total_num);
std::string t = "null";
if (total_num != 0) {
t = std::format("{}%", static_cast<double>(crc_error_num) / static_cast<double>(total_num));
std::lock_guard lock(mtx);
size_t total_pre_second_num = 0;
size_t total_num = 0;
size_t crc_error_num = 0;
for (auto& [_, statistic] : DF_Statistic_Data_map) {
const auto [pre_second, total, crc_error] = statistic.counters();
total_pre_second_num += pre_second;
total_num += total;
crc_error_num += crc_error;
}
Ret_J(crc_error_num);
ret.append({"crc_error_rate", t});
Psc::JSON ret = Psc::JSON::object();
const auto crc_error_percent = total_num == 0 ? 0.0 : static_cast<double>(crc_error_num) * 100.0 / static_cast<double>(total_num);
ret.append({"messages_per_second", total_pre_second_num});
ret.append({"messages_total", total_num});
ret.append({"length_errors_total", length_error_num.load()});
ret.append({"crc_errors_total", crc_error_num});
ret.append({"crc_error_percent", crc_error_percent});
Psc::JSON df_sub_type = Psc::JSON::object();
for (auto& cur : DF_Statistic_Data_map) {
df_sub_type.append({Psc::to_string(cur.first) + " [" + std::to_string(static_cast<int>(cur.first)) + "]",
cur.second.to_json()});
}
ret.append({"DF子类型", df_sub_type});
ret.append({"CPR统计信息", cpr_statistic_data.to_json()});
ret.append({"downlink_formats", df_sub_type});
ret.append({"cpr", cpr_statistic_data.to_json()});
return ret;
}
void add_length_error(std::string_view key) {
@@ -363,40 +382,35 @@ public:
cpr_statistic_data.add(cpr_type, result_type);
}
DF_Statistic_Data* get_create_df_statistic_data(const SSR::Downlink_Format& key) {
DF_Statistic_Data* t;
auto iter = DF_Statistic_Data_map.find(key);
if (iter == DF_Statistic_Data_map.end()) {
DF_Statistic_Data cur{};
DF_Statistic_Data_map.insert(std::make_pair(key, cur));
}
t = &DF_Statistic_Data_map[key];
return t;
std::lock_guard lock(mtx);
return &DF_Statistic_Data_map.try_emplace(key).first->second;
}
size_t get_total_pre_second_num() {
std::lock_guard lock(mtx);
size_t ret = 0;
for (auto& cur : DF_Statistic_Data_map) {
ret += cur.second.pre_second_num;
}
for (auto& cur : DF_Statistic_Data_map)
ret += std::get<0>(cur.second.counters());
return ret;
}
size_t get_total_num() {
std::lock_guard lock(mtx);
size_t ret = 0;
for (auto& cur : DF_Statistic_Data_map) {
ret += cur.second.total_num;
}
for (auto& cur : DF_Statistic_Data_map)
ret += std::get<1>(cur.second.counters());
return ret;
}
size_t get_crc_error_num() {
std::lock_guard lock(mtx);
size_t ret = 0;
for (auto& cur : DF_Statistic_Data_map) {
ret += cur.second.crc_error_num;
}
for (auto& cur : DF_Statistic_Data_map)
ret += std::get<2>(cur.second.counters());
return ret;
}
protected:
std::atomic<size_t> length_error_num{};
std::map<SSR::Downlink_Format, DF_Statistic_Data> DF_Statistic_Data_map;
CPR_Percentage_Statistic_Data cpr_statistic_data;
std::mutex mtx;
};
template <typename T>
class Immutable : public T {};
module/Local_Server/server/Adminive_Config.cpp
+45 -12
View File
@@ -453,7 +453,7 @@ Data_Source_Row source_row(const std::shared_ptr<Data_Source>& source) {
source->settings.read([&](const auto& value) {
row.config = value;
});
row.state = Psc::to_string(source->state);
row.state = Psc::to_string(source->state.load());
if(const auto source_value = dynamic_cast<TCP_Client_Data_Source*>(source.get())) {
source_value->specific.read([&](const auto& value) {
row.ip = value.ip;
@@ -533,7 +533,7 @@ Data_Feed_Row feed_row(const std::shared_ptr<Data_Feed>& feed) {
row.output_format = value;
row.format = value.type;
});
row.state = Psc::to_string(feed->state);
row.state = Psc::to_string(feed->state.load());
if(const auto feed_value = dynamic_cast<Data_Feed_TCP_Server*>(feed.get())) {
feed_value->specific.read([&](const auto& value) {
row.port = value.port;
@@ -641,13 +641,13 @@ Json data_topology_view(Global* global) {
Json data_topology_connection_status(Global* global, std::uint64_t id) {
const auto feed = item_by_id(global->mode_acs.data_feed_config.map.list(), id);
const auto source = global->mode_acs.data_source_config.map.get(feed->source_key);
Json result{{"连接", feed->source_key + " → " + feed->key}, {"数据馈送状态", Psc::to_string(feed->state)}, {"数据馈送启用", feed->enabled()}};
Json result{{"连接", feed->source_key + " → " + feed->key}, {"数据馈送状态", Psc::to_string(feed->state.load())}, {"数据馈送启用", feed->enabled()}};
if(!source) {
result["连接状态"] = "数据源不存在";
result["已在数据源注册"] = false;
return result;
}
result["数据源状态"] = Psc::to_string(source.value()->state);
result["数据源状态"] = Psc::to_string(source.value()->state.load());
result["数据源启用"] = source.value()->enabled();
const auto all_status = adminive::parse_json<Json>(source.value()->get_all_connect_feed_status().to_json_string());
const auto status = all_status.find(feed->key);
@@ -655,8 +655,30 @@ Json data_topology_connection_status(Global* global, std::uint64_t id) {
result["连接状态"] = status != all_status.end() && source.value()->enabled() && feed->enabled() ? "运行中" : "未运行";
if(status != all_status.end())
result["缓存统计"] = *status;
result["数据馈送观测"] = adminive::parse_json<Json>(feed->get_state_json().to_json_string());
return result;
}
Json data_topology_runtime_client_status(Global* global, std::uint64_t feed_id, std::uint64_t client_id) {
const auto feed = item_by_id(global->mode_acs.data_feed_config.map.list(), feed_id);
if(const auto tcp = dynamic_cast<Data_Feed_TCP_Server*>(feed.get())) {
const auto client = tcp->svr.get_client(static_cast<Psc::asio_socket::Socket_FD>(client_id));
if(!client)
throw std::out_of_range("TCP client is no longer connected");
return Json{{"connection_state", client->closing.load() ? "closing" : "connected"},
{"remote", client->info.sockaddr.to_string()}, {"fd", client->info.fd},
{"statistics", adminive::parse_json<Json>(client->statistics.to_json().to_json_string())}};
}
if(const auto udp = dynamic_cast<Data_Feed_UDP_Server*>(feed.get())) {
const auto client = udp->svr.get_client(static_cast<std::size_t>(client_id));
if(!client)
throw std::out_of_range("UDP client is no longer available");
const auto last_activity = client->statistics->last_activity_unix_ms();
const auto idle_ms = last_activity == 0 ? 0 : Psc::get_current_millisecond_timestamp() - last_activity;
return Json{{"connection_state", last_activity != 0 && idle_ms <= 30000 ? "active" : "idle"}, {"remote", client->address.to_string()}, {"idle_ms", idle_ms},
{"statistics", adminive::parse_json<Json>(client->statistics->to_json().to_json_string())}};
}
throw std::invalid_argument("data feed does not expose runtime clients");
}
Json data_topology(Global* global) {
const auto sources = global->mode_acs.data_source_config.map.list();
const auto feeds = global->mode_acs.data_feed_config.map.list();
@@ -670,7 +692,7 @@ Json data_topology(Global* global) {
auto config = adminive::to_frontend_json<Json>(source_row(source));
config["id"] = index + 1;
Json details = Json::array();
details.push_back(Json{{"label", "运行状态"}, {"value", Psc::to_string(source->state)}});
details.push_back(Json{{"label", "运行状态"}, {"value", Psc::to_string(source->state.load())}});
if(const auto value = dynamic_cast<TCP_Client_Data_Source*>(source.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "目标"}, {"value", config.ip + ":" + std::to_string(config.port)}}); });
} else if(const auto value = dynamic_cast<Serial_Data_Source*>(source.get())) {
@@ -691,7 +713,7 @@ Json data_topology(Global* global) {
const auto id = "data_feed:" + feed->key;
auto config = adminive::to_frontend_json<Json>(feed_row(feed));
config["id"] = index + 1;
Json details = Json::array({Json{{"label", "运行状态"}, {"value", Psc::to_string(feed->state)}}});
Json details = Json::array({Json{{"label", "运行状态"}, {"value", Psc::to_string(feed->state.load())}}});
feed->output_format.read([&](const auto& value) { details.push_back(Json{{"label", "数据格式"}, {"value", std::string(magic_enum::enum_name(value.type))}}); });
if(const auto value = dynamic_cast<Data_Feed_TCP_Server*>(feed.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "监听端口"}, {"value", config.port}}); });
@@ -716,15 +738,21 @@ Json data_topology(Global* global) {
for(const auto& client : value->svr.get_all_clients()) {
const auto client_id = "client:tcp:" + feed->key + ":" + std::to_string(client->info.fd);
const auto endpoint = client->info.sockaddr.to_string();
nodes.push_back(Json{{"id", client_id}, {"kind", "client"}, {"entity_id", 0}, {"key", endpoint}, {"label", endpoint}, {"type", "TCP 客户端"}, {"enabled", true}, {"details", Json::array({Json{{"label", "文件描述符"}, {"value", client->info.fd}}, Json{{"label", "发送缓冲区"}, {"value", client->send_buffer.state_str()}}})}});
edges.push_back(Json{{"id", "feed-client:" + feed->key + ":" + std::to_string(client->info.fd)}, {"kind", "runtime"}, {"source", id}, {"target", client_id}, {"active", true}, {"details", Json::array({Json{{"label", "连接状态"}, {"value", "已连接"}}, Json{{"label", "远端"}, {"value", endpoint}}, Json{{"label", "发送缓冲区"}, {"value", client->send_buffer.state_str()}}})}});
const auto status_api = data_topology_path + "/status/runtime_client/" + std::to_string(index + 1) + "/" + std::to_string(client->info.fd);
nodes.push_back(Json{{"id", client_id}, {"kind", "client"}, {"entity_id", 0}, {"key", endpoint}, {"label", endpoint}, {"type", "TCP 客户端"}, {"enabled", true}, {"status_api", status_api}, {"details", Json::array({Json{{"label", "文件描述符"}, {"value", client->info.fd}}, Json{{"label", "远端"}, {"value", endpoint}}})}});
edges.push_back(Json{{"id", "feed-client:" + feed->key + ":" + std::to_string(client->info.fd)}, {"kind", "runtime"}, {"source", id}, {"target", client_id}, {"active", !client->closing.load()}, {"status_api", status_api}, {"details", Json::array({Json{{"label", "连接状态"}, {"value", "已连接"}}, Json{{"label", "远端"}, {"value", endpoint}}})}});
}
} else if(const auto value = dynamic_cast<Data_Feed_UDP_Server*>(feed.get())) {
for(const auto& client : value->svr.clients) {
const auto endpoint = client.to_string();
const auto clients = value->svr.get_all_clients();
for(std::size_t client_index = 0; client_index < clients.size(); ++client_index) {
const auto& client = clients[client_index];
const auto endpoint = client.address.to_string();
const auto last_activity = client.statistics->last_activity_unix_ms();
const auto active = last_activity != 0 && Psc::get_current_millisecond_timestamp() - last_activity <= 30000;
const auto client_id = "client:udp:" + feed->key + ":" + endpoint;
nodes.push_back(Json{{"id", client_id}, {"kind", "client"}, {"entity_id", 0}, {"key", endpoint}, {"label", endpoint}, {"type", "UDP 客户端"}, {"enabled", true}, {"details", Json::array()}});
edges.push_back(Json{{"id", "feed-client:" + feed->key + ":" + endpoint}, {"kind", "runtime"}, {"source", id}, {"target", client_id}, {"active", true}, {"details", Json::array({Json{{"label", "连接状态"}, {"value", "已连接"}}, Json{{"label", "远端"}, {"value", endpoint}}})}});
const auto status_api = data_topology_path + "/status/runtime_client/" + std::to_string(index + 1) + "/" + std::to_string(client_index);
nodes.push_back(Json{{"id", client_id}, {"kind", "client"}, {"entity_id", 0}, {"key", endpoint}, {"label", endpoint}, {"type", "UDP 客户端"}, {"enabled", true}, {"status_api", status_api}, {"details", Json::array({Json{{"label", "远端"}, {"value", endpoint}}})}});
edges.push_back(Json{{"id", "feed-client:" + feed->key + ":" + endpoint}, {"kind", "runtime"}, {"source", id}, {"target", client_id}, {"active", active}, {"status_api", status_api}, {"details", Json::array({Json{{"label", "连接状态"}, {"value", active ? "活跃" : "空闲"}}, Json{{"label", "远端"}, {"value", endpoint}}})}});
}
}
}
@@ -1023,6 +1051,11 @@ private:
return adminive::make_http_success<Json>(data_topology_connection_status(global_, id));
});
}, get);
app.registerHandler(data_topology_path + "/status/runtime_client/{1:feed_id}/{2:client_id}", [this](const drogon::HttpRequestPtr&, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t feed_id, std::uint64_t client_id) {
complete_business_request(callback, [this, feed_id, client_id] {
return adminive::make_http_success<Json>(data_topology_runtime_client_status(global_, feed_id, client_id));
});
}, get);
}
void bind_backend_config(drogon::HttpAppFramework& app) {
const auto get = adminive::make_drogon_constraints(drogon::Get, {});
module/Local_Server/server/Mode_Msg_Buffer.h
+4
View File
@@ -51,6 +51,10 @@ protected:
std::mutex mtx;
std::vector<std::string *> msg_list_cache;
std::vector<std::string *> msg_list;
size_t cache_bytes{};
size_t dispatch_bytes{};
size_t high_water_messages{};
size_t high_water_bytes{};
};
#endif
module/Local_Server/server/With_Loop_Coro.cpp
+64 -18
View File
@@ -14,7 +14,7 @@ void With_Loop_Coro::sync_wait() {
return;
}
while (loop_task->wait_for(std::chrono::milliseconds(0)) != std::future_status::ready) {
std::cout << std::format("{}等待退出 当前状态为{} \n", name, Psc::to_string(this->state));
std::cout << std::format("{}等待退出 当前状态为{} \n", name, Psc::to_string(this->state.load()));
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
std::cout << std::format("{}退出成功！ \n", name);
@@ -22,15 +22,58 @@ void With_Loop_Coro::sync_wait() {
bool With_Loop_Coro::running() const {
return loop_running.load();
}
void With_Loop_Coro::set_state(State state, std::string_view action) {
std::uint64_t With_Loop_Coro::unix_ms() {
return static_cast<std::uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count());
}
void With_Loop_Coro::set_state(State next_state, std::string_view action) {
const auto old_state = state.exchange(next_state);
if (old_state != next_state) {
const auto now = unix_ms();
state_changed_unix_ms_ = now;
if (next_state == State::Loop_Running) {
++start_total_;
last_started_unix_ms_ = now;
active_since_unix_ms_ = now;
}
if (next_state == State::Start && old_state == State::Waiting_Stop_Loop) {
++stop_total_;
last_stopped_unix_ms_ = now;
active_since_unix_ms_ = 0;
}
}
if (!action.empty()) {
std::string name = type + "_" + key;
std::cout << std::format("协程状态机 {} {} {} ==> {} \n", name, action, Psc::to_string(this->state),
Psc::to_string(state));
std::cout << std::format("协程状态机 {} {} {} ==> {} \n", name, action, Psc::to_string(old_state),
Psc::to_string(next_state));
}
this->state = state;
}
With_Loop_Coro::With_Loop_Coro() : stop(0.1) {}
Psc::JSON With_Loop_Coro::lifecycle_state_json() const {
const auto active_since = active_since_unix_ms_.load();
Psc::JSON ret = Psc::JSON::object();
ret.append({"task_state", Psc::to_string(state.load())});
ret.append({"enabled", enabled()});
ret.append({"running", running()});
ret.append({"state_changed_unix_ms", state_changed_unix_ms_.load()});
ret.append({"start_total", start_total_.load()});
ret.append({"stop_total", stop_total_.load()});
ret.append({"exception_total", exception_total_.load()});
ret.append({"last_started_unix_ms", last_started_unix_ms_.load()});
ret.append({"last_stopped_unix_ms", last_stopped_unix_ms_.load()});
ret.append({"active_duration_ms", active_since == 0 ? 0 : unix_ms() - active_since});
{
std::lock_guard lock(exception_mtx_);
ret.append({"last_exception", last_exception_});
}
return ret;
}
void With_Loop_Coro::record_exception(std::string error) {
++exception_total_;
std::lock_guard lock(exception_mtx_);
last_exception_ = std::move(error);
}
With_Loop_Coro::With_Loop_Coro() : stop(0.1) {
state_changed_unix_ms_ = unix_ms();
}
asio::awaitable<void> With_Loop_Coro::run_loop_coro() {
loop_running.store(true, std::memory_order_release);
try {
@@ -39,32 +82,35 @@ asio::awaitable<void> With_Loop_Coro::run_loop_coro() {
co_return;
}
catch (const std::exception& e) {
//loop_running.store(false, std::memory_order_release);
auto str = std::format("{} loop_coro 异常退出: {}\n", key, e.what());
loop_running.store(false, std::memory_order_release);
auto str = std::format("{} loop_coro 异常退出: {}\n", key, Psc::platform_2_utf8(e.what()));
record_exception(str);
std::cerr << str;
set_state(State::Loop_Exception, str);
throw;
}
catch (...) {
//loop_running.store(false, std::memory_order_release);
std::cerr << "loop_coro 未知异常退出\n";
loop_running.store(false, std::memory_order_release);
record_exception("loop_coro unknown exception");
set_state(State::Loop_Exception, "loop_coro unknown exception");
throw;
}
}
asio::awaitable<void> With_Loop_Coro::tick() {
std::string name = type + ":" + key;
if (state == State::Force_Quit) {
const auto current_state = state.load();
if (current_state == State::Force_Quit) {
static Frequency_Limit_Multi flm;
if (flm.test(name)) {
std::cout << std::format("{} 正在强制退出!\n", name);
}
}
if (state == State::Start) {
if (current_state == State::Start) {
if (enabled()) {
set_state(State::Before_Request_Start_Loop);
}
}
else if (state == State::Before_Request_Start_Loop) {
else if (current_state == State::Before_Request_Start_Loop) {
co_await this->_open();
loop_running = enabled();
auto future = asio::co_spawn(Coro::instance()->io, run_loop_coro(), asio::use_future);
@@ -78,17 +124,17 @@ asio::awaitable<void> With_Loop_Coro::tick() {
// });
set_state(State::Waiting_Loop_Start, "开始启动任务");
}
else if (state == State::Waiting_Loop_Start) {
else if (current_state == State::Waiting_Loop_Start) {
if (running()) {
set_state(State::Loop_Running, "启动任务成功！");
}
}
else if (state == State::Loop_Running) {
else if (current_state == State::Loop_Running) {
if (!enabled()) {
set_state(State::Before_Request_Stop_Loop, std::format("{} 检测到 enable变化 异步退出开始！", name));
}
}
else if (state == State::Loop_Exception) {
else if (current_state == State::Loop_Exception) {
bool catch_exception = false;
if (!catch_exception) {
loop_task->get();
@@ -125,11 +171,11 @@ asio::awaitable<void> With_Loop_Coro::tick() {
co_return;
}
}
else if (state == State::Before_Request_Stop_Loop) {
else if (current_state == State::Before_Request_Stop_Loop) {
loop_running = false;
set_state(State::Waiting_Stop_Loop, "退出变量已设置 等待退出");
}
else if (state == State::Waiting_Stop_Loop) {
else if (current_state == State::Waiting_Stop_Loop) {
if (!running()) {
loop_task.reset();
co_await this->_close();
module/Local_Server/server/With_Loop_Coro.h
+16 -1
View File
@@ -7,6 +7,8 @@
#include <future>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>
#include <sstream>
#include <string_view>
#include <thread>
@@ -75,11 +77,24 @@ public:
Loop_Exception,
Before_Request_Stop_Loop,
Waiting_Stop_Loop
} state = State::Start;
};
std::atomic<State> state{State::Start};
void set_state(State state, std::string_view action = "");
[[nodiscard]] Psc::JSON lifecycle_state_json() const;
With_Loop_Coro();
protected:
Psc::Copyable_Atomic<bool> loop_running{};
asio::awaitable<void> run_loop_coro();
Frequency_Limit_Multi stop;
void record_exception(std::string error);
static std::uint64_t unix_ms();
std::atomic<std::uint64_t> state_changed_unix_ms_{};
std::atomic<std::uint64_t> last_started_unix_ms_{};
std::atomic<std::uint64_t> last_stopped_unix_ms_{};
std::atomic<std::uint64_t> active_since_unix_ms_{};
std::atomic<std::uint64_t> start_total_{};
std::atomic<std::uint64_t> stop_total_{};
std::atomic<std::uint64_t> exception_total_{};
mutable std::mutex exception_mtx_;
std::string last_exception_;
};
module/Local_Server/server/io_coro.cpp
+2 -1
View File
@@ -90,7 +90,7 @@ void Coro::stop() {
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));
std::cout << std::format("请求停止 {}:{} 当前状态为{}\n", li->type, li->key, Psc::to_string(li->state.load()));
li->async_stop();
}
}
@@ -182,6 +182,7 @@ asio::awaitable<void> Data_Feed::loop_coro() {
if (!feed->enabled()) {
break;
}
feed->record_dispatched(msg.size());
if (data_feed_debug) {
std::cout << std::format("{} before send {}\n", key, thread_id_str());
}