From d28edab2d70e3a04e920ef9a21a20f3a733d9c88 Mon Sep 17 00:00:00 2001 From: wyc <1104749580@qq.com> Date: Mon, 10 Aug 2026 16:36:12 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=AE=8Cbug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- module/Local_Server/Data_Feed/Data_Feed.cpp | 28 +++- module/Local_Server/Data_Feed/Data_Feed.h | 91 +++++++++---- .../Local_Server/Data_Source/Data_Source.cpp | 27 +++- module/Local_Server/Data_Source/Data_Source.h | 65 +++++++-- .../Data_Source/Data_Source_Handler.cpp | 42 +++--- .../Data_Source/Data_Source_Handler.h | 3 +- module/Local_Server/global_include.h | 128 ++++++++++-------- .../Local_Server/server/Adminive_Config.cpp | 57 ++++++-- module/Local_Server/server/Mode_Msg_Buffer.h | 4 + module/Local_Server/server/With_Loop_Coro.cpp | 82 ++++++++--- module/Local_Server/server/With_Loop_Coro.h | 17 ++- module/Local_Server/server/io_coro.cpp | 3 +- 12 files changed, 389 insertions(+), 158 deletions(-) diff --git a/module/Local_Server/Data_Feed/Data_Feed.cpp b/module/Local_Server/Data_Feed/Data_Feed.cpp index d526a7c..18218f5 100644 --- a/module/Local_Server/Data_Feed/Data_Feed.cpp +++ b/module/Local_Server/Data_Feed/Data_Feed.cpp @@ -33,6 +33,7 @@ asio::awaitable 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& BIN_Msg_Buffer::get_all() { std::lock_guard 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 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(pool_capacity - pool_free_count) * 100.0 / static_cast(pool_capacity)}); + return ret; } diff --git a/module/Local_Server/Data_Feed/Data_Feed.h b/module/Local_Server/Data_Feed/Data_Feed.h index 8ad0e43..c2a5e99 100644 --- a/module/Local_Server/Data_Feed/Data_Feed.h +++ b/module/Local_Server/Data_Feed/Data_Feed.h @@ -33,6 +33,12 @@ public: BIN_Msg_Buffer msg_buffer{}; adminive::Managed_Value 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 mode_s_limit_drop_total{}; + std::atomic mode_other_limit_drop_total{}; asio::awaitable 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 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 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 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 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 handle_in_loop_coro() override { co_await cli.tick_coro(); diff --git a/module/Local_Server/Data_Source/Data_Source.cpp b/module/Local_Server/Data_Source/Data_Source.cpp index 0c4bbfe..8132c9e 100644 --- a/module/Local_Server/Data_Source/Data_Source.cpp +++ b/module/Local_Server/Data_Source/Data_Source.cpp @@ -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 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 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 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(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 Shared_Memory_Data_Source::_open() { const auto config = specific.read([](const auto& value) { return value; }); sm = std::make_unique(); sm->init(config.shared_memory_name, config.shared_memory_size); + shared_memory_open = true; co_return; } asio::awaitable 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); diff --git a/module/Local_Server/Data_Source/Data_Source.h b/module/Local_Server/Data_Source/Data_Source.h index 6bccef1..94d5cbc 100644 --- a/module/Local_Server/Data_Source/Data_Source.h +++ b/module/Local_Server/Data_Source/Data_Source.h @@ -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 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 _close() override { if (serial) serial->close(); + serial_open = false; co_return; } asio::awaitable handle_in_loop_coro() override { @@ -321,6 +329,7 @@ public: }); } std::unique_ptr 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::max(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(current_index) * 100.0 / static_cast(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 msg; }; Player_Clock player_clock; + Psc::Throughput_Statistics raw_read_statistics; + Psc::Throughput_Statistics conversion_dropped_statistics; + std::atomic loop_total{}; std::shared_ptr pre_cache = nullptr; std::deque> cache_list; std::string get_true_file_path() const { @@ -432,11 +458,13 @@ class Dll_Data_Source : public Data_Source { public: adminive::Managed_Value 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 _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 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 sm; + std::atomic_bool shared_memory_open{}; }; inline std::shared_ptr create_data_source_from_type(std::string_view t) { std::shared_ptr ret{}; diff --git a/module/Local_Server/Data_Source/Data_Source_Handler.cpp b/module/Local_Server/Data_Source/Data_Source_Handler.cpp index 1b89563..c71e99e 100644 --- a/module/Local_Server/Data_Source/Data_Source_Handler.cpp +++ b/module/Local_Server/Data_Source/Data_Source_Handler.cpp @@ -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& 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 list; - for (const auto& item : cached_data_feed_key_list) { - auto& key = item.key; + std::vector 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& msg) { std::shared_ptr& 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 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& 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& 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 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 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()) { diff --git a/module/Local_Server/Data_Source/Data_Source_Handler.h b/module/Local_Server/Data_Source/Data_Source_Handler.h index 909c05a..09a98e6 100644 --- a/module/Local_Server/Data_Source/Data_Source_Handler.h +++ b/module/Local_Server/Data_Source/Data_Source_Handler.h @@ -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& msg); diff --git a/module/Local_Server/global_include.h b/module/Local_Server/global_include.h index 4f8c8ec..05441a5 100644 --- a/module/Local_Server/global_include.h +++ b/module/Local_Server/global_include.h @@ -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(crc_error_num) / static_cast(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 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(crc_error_num) * 100.0 / static_cast(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 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(num) / static_cast(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(num) / static_cast(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 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(crc_error_num) / static_cast(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(crc_error_num) * 100.0 / static_cast(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(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 length_error_num{}; std::map DF_Statistic_Data_map; CPR_Percentage_Statistic_Data cpr_statistic_data; + std::mutex mtx; }; template class Immutable : public T {}; diff --git a/module/Local_Server/server/Adminive_Config.cpp b/module/Local_Server/server/Adminive_Config.cpp index fb01cc4..4784d08 100644 --- a/module/Local_Server/server/Adminive_Config.cpp +++ b/module/Local_Server/server/Adminive_Config.cpp @@ -453,7 +453,7 @@ Data_Source_Row source_row(const std::shared_ptr& 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(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& 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(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(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(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(feed.get())) { + const auto client = tcp->svr.get_client(static_cast(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(client->statistics.to_json().to_json_string())}}; + } + if(const auto udp = dynamic_cast(feed.get())) { + const auto client = udp->svr.get_client(static_cast(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(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(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(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(source.get())) { @@ -691,7 +713,7 @@ Json data_topology(Global* global) { const auto id = "data_feed:" + feed->key; auto config = adminive::to_frontend_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(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(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(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&& 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(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, {}); diff --git a/module/Local_Server/server/Mode_Msg_Buffer.h b/module/Local_Server/server/Mode_Msg_Buffer.h index 709bae4..6cf9a35 100644 --- a/module/Local_Server/server/Mode_Msg_Buffer.h +++ b/module/Local_Server/server/Mode_Msg_Buffer.h @@ -51,6 +51,10 @@ protected: std::mutex mtx; std::vector msg_list_cache; std::vector msg_list; + size_t cache_bytes{}; + size_t dispatch_bytes{}; + size_t high_water_messages{}; + size_t high_water_bytes{}; }; #endif diff --git a/module/Local_Server/server/With_Loop_Coro.cpp b/module/Local_Server/server/With_Loop_Coro.cpp index d5e7b4a..78d5e74 100644 --- a/module/Local_Server/server/With_Loop_Coro.cpp +++ b/module/Local_Server/server/With_Loop_Coro.cpp @@ -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::chrono::duration_cast(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 With_Loop_Coro::run_loop_coro() { loop_running.store(true, std::memory_order_release); try { @@ -39,32 +82,35 @@ asio::awaitable 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 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 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 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(); diff --git a/module/Local_Server/server/With_Loop_Coro.h b/module/Local_Server/server/With_Loop_Coro.h index 3948922..86ae57b 100644 --- a/module/Local_Server/server/With_Loop_Coro.h +++ b/module/Local_Server/server/With_Loop_Coro.h @@ -7,6 +7,8 @@ #include #include #include +#include +#include #include #include #include @@ -75,11 +77,24 @@ public: Loop_Exception, Before_Request_Stop_Loop, Waiting_Stop_Loop - } state = State::Start; + }; + std::atomic 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 loop_running{}; asio::awaitable run_loop_coro(); Frequency_Limit_Multi stop; + void record_exception(std::string error); + static std::uint64_t unix_ms(); + std::atomic state_changed_unix_ms_{}; + std::atomic last_started_unix_ms_{}; + std::atomic last_stopped_unix_ms_{}; + std::atomic active_since_unix_ms_{}; + std::atomic start_total_{}; + std::atomic stop_total_{}; + std::atomic exception_total_{}; + mutable std::mutex exception_mtx_; + std::string last_exception_; }; diff --git a/module/Local_Server/server/io_coro.cpp b/module/Local_Server/server/io_coro.cpp index b75d2a2..852094b 100644 --- a/module/Local_Server/server/io_coro.cpp +++ b/module/Local_Server/server/io_coro.cpp @@ -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 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()); }