From fa1e7e24fcaa49888e1405b5f19683899577efe4 Mon Sep 17 00:00:00 2001 From: wyc <1104749580@qq.com> Date: Fri, 28 Aug 2026 11:49:06 +0800 Subject: [PATCH] =?UTF-8?q?=E5=86=85=E5=AD=98=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + main.cmake | 17 +- module/Local_Server/DSP/DSP_Config.cpp | 7 +- module/Local_Server/DSP/DSP_Config.h | 39 ++- module/Local_Server/Data_Feed/Data_Feed.cpp | 33 +-- module/Local_Server/Data_Feed/Data_Feed.h | 3 +- .../Local_Server/Data_Source/Data_Source.cpp | 144 ++++++++--- module/Local_Server/Data_Source/Data_Source.h | 28 ++- .../Data_Source/Data_Source_Handler.cpp | 73 ++---- .../Data_Source/Data_Source_Handler.h | 2 + module/Local_Server/Data_Source/Database.cpp | 16 +- .../External_Database/External_Database.cpp | 24 +- module/Local_Server/server/Global.cpp | 11 +- module/Local_Server/server/Memory_Monitor.cpp | 227 ++++++++++++++++++ module/Local_Server/server/Memory_Monitor.h | 35 +++ .../Local_Server/server/Memory_Resource.cpp | 171 +++++++++++++ module/Local_Server/server/Memory_Resource.h | 21 ++ module/Local_Server/server/Mode_Msg_Buffer.h | 11 +- module/Local_Server/server/io_coro.cpp | 20 +- module/Local_Server/server/server.cpp | 21 +- module/Local_Server_main.cpp | 55 +---- 21 files changed, 722 insertions(+), 237 deletions(-) create mode 100644 module/Local_Server/server/Memory_Monitor.cpp create mode 100644 module/Local_Server/server/Memory_Monitor.h create mode 100644 module/Local_Server/server/Memory_Resource.cpp create mode 100644 module/Local_Server/server/Memory_Resource.h diff --git a/.gitignore b/.gitignore index 9acb5b3..1573cd8 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ /third_party/eacp_webapp/.playwright-cli/ /.playwright-cli/ /output/ +/third_party/mimalloc/ diff --git a/main.cmake b/main.cmake index 57d7000..e43019b 100644 --- a/main.cmake +++ b/main.cmake @@ -1,6 +1,6 @@ get_filename_component(name ${CMAKE_CURRENT_LIST_DIR} NAME) -set(ecap_rely global::drogon global::libarchive global::SQLiteCpp) +set(ecap_rely global::drogon global::libarchive global::SQLiteCpp global::mimalloc) library_get_missing_with_rely(ecap_dependencies_miss ${ecap_rely}) if (ecap_dependencies_miss) @@ -10,6 +10,11 @@ endif () rcl_load_dependency_environment(${ecap_rely}) find_package(Drogon) find_package(SQLiteCpp CONFIG REQUIRED) +set(ecap_runtime_optimization_default OFF) +if (CMAKE_SYSTEM_NAME STREQUAL "Linux") + set(ecap_runtime_optimization_default ON) +endif () +option(ECAP_SERVER_RUNTIME_OPTIMIZATION "Optimize the deployed server while retaining debug symbols" ${ecap_runtime_optimization_default}) set(Core Psc_Core_Static) set(md "${CMAKE_CURRENT_LIST_DIR}/module") set(data_dir ${CMAKE_CURRENT_LIST_DIR}/data) @@ -30,6 +35,15 @@ target_include_directories(mlat_source PUBLIC "${md}") target_link_libraries(mlat_source PUBLIC ${Core}) append_glob_source(ecap_server_srcs "${md}/Local_Server") add_executable(ecap_server "${md}/Local_Server_main.cpp" ${ecap_server_srcs}) +if (ECAP_SERVER_RUNTIME_OPTIMIZATION) + foreach (ecap_optimized_target ecap_server SSR Psc_Core_Static) + if (MSVC) + target_compile_options(${ecap_optimized_target} PRIVATE /O2) + else () + target_compile_options(${ecap_optimized_target} PRIVATE -O2 -fno-omit-frame-pointer) + endif () + endforeach () +endif () set(ecap_version "v1.0.5") set(ecap_build_version_header "${CMAKE_CURRENT_BINARY_DIR}/generated/ecap_server/Build_Version.h") set(ecap_build_version_source "${CMAKE_CURRENT_BINARY_DIR}/generated/ecap_server/Build_Version.cpp") @@ -65,6 +79,7 @@ target_link_libraries(ecap_server PUBLIC ${Core}) target_link_libraries(ecap_server PRIVATE Drogon::Drogon) target_link_libraries(ecap_server PRIVATE SQLiteCpp) target_link_libraries(ecap_server PRIVATE ${ZLIB_LIBRARY}) +target_link_libraries(ecap_server PRIVATE mimalloc-static) if (MSVC) target_compile_options(ecap_server PRIVATE /bigobj /utf-8) endif () diff --git a/module/Local_Server/DSP/DSP_Config.cpp b/module/Local_Server/DSP/DSP_Config.cpp index 4204bec..15b15bd 100644 --- a/module/Local_Server/DSP/DSP_Config.cpp +++ b/module/Local_Server/DSP/DSP_Config.cpp @@ -205,18 +205,15 @@ void DSP_Config::init_env() { #else manager.test_and_start_thread("iq回调处理", [this, handle_iq]( std::atomic& running) { - Frequency_Limit_ST limit(frame_rate); while (running) { - auto list = iq_memory_buffer.get_all(); + auto list = iq_memory_buffer.wait_all(running); if (debug && list.size() > 1000) { std::ostringstream oss; oss << "警告： 数据过多 当前数据队列长度:" << list.size() << std::endl; server_logger->c_debug({}, {}, oss.str()); } - if (list.empty()) { - std::this_thread::sleep_for(std::chrono::microseconds(1)); + if (list.empty()) continue; - } auto wi = Ws_Mgr::instance(); if (!wi) continue; diff --git a/module/Local_Server/DSP/DSP_Config.h b/module/Local_Server/DSP/DSP_Config.h index 10bde1f..033b728 100644 --- a/module/Local_Server/DSP/DSP_Config.h +++ b/module/Local_Server/DSP/DSP_Config.h @@ -1,5 +1,9 @@ #ifndef PLANTSUNCAT_DSP2_H #define PLANTSUNCAT_DSP2_H +#include +#include +#include +#include #include "../server/WebSocket_Manager.h" #include "global.h" class Global; @@ -39,19 +43,36 @@ public: void set_fft_point_number(uint64_t fft_point_number, bool force = false); std::atomic record_iq_stream_ing = false; struct Memory_Buffer { - std::vector get_all() { - std::vector empty; - std::lock_guard lock(m); - std::swap(empty, data); - return empty; + using Batch = std::pmr::deque; + Batch wait_all(const std::atomic& running) { + std::unique_lock lock(m); + ready.wait_for(lock, std::chrono::milliseconds(100), [&] { + return !data.empty() || !running.load(std::memory_order_acquire); + }); + Batch result(data.get_allocator().resource()); + result.swap(data); + return result; } void push(void* pdata, uint64_t length) { - std::lock_guard lock(m); - data.emplace_back((char*)pdata, length); + { + std::lock_guard lock(m); + if (data.size() == max_frames) { + data.pop_front(); + ++dropped_frames; + } + data.emplace_back(static_cast(pdata), length); + } + ready.notify_one(); + } + std::uint64_t dropped() const { + return dropped_frames.load(std::memory_order_relaxed); } protected: - std::vector data; - Psc::Spin_Lock m; + static constexpr std::size_t max_frames = 4096; + Batch data; + std::atomic dropped_frames{}; + std::mutex m; + std::condition_variable ready; }; #if USE_PROCESS void adsb_set_less_30Mhz_ddc_param(double center_freq, double sample_rate, double band_width, int sw); diff --git a/module/Local_Server/Data_Feed/Data_Feed.cpp b/module/Local_Server/Data_Feed/Data_Feed.cpp index 18218f5..bb8418c 100644 --- a/module/Local_Server/Data_Feed/Data_Feed.cpp +++ b/module/Local_Server/Data_Feed/Data_Feed.cpp @@ -66,47 +66,42 @@ void Data_feed_Config::server(Global* g) { void BIN_Msg_Buffer::push(std::string_view msg) { 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; }); std::lock_guard g(mtx); - auto& pool = Global::instance()->mode_acs.data_feed_config.pool_; - if (msg_list_cache.empty() || msg_list_cache.back()->size() > output_packet_size) { - auto t = pool.get(); - t->append(msg); - msg_list_cache.push_back(t); + if (msg_list_cache.empty() || msg_list_cache.back().size() > output_packet_size) { + msg_list_cache.emplace_back(); + msg_list_cache.back().append(msg); } else { - msg_list_cache.back()->append(msg); + 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() { +BIN_Msg_Buffer::Batch BIN_Msg_Buffer::take_all() { + Batch result(msg_list_cache.get_allocator().resource()); std::lock_guard g(mtx); - std::swap(msg_list_cache, msg_list); - msg_list_cache.clear(); + result.swap(msg_list_cache); + dispatch_batches = result.size(); dispatch_bytes = cache_bytes; cache_bytes = 0; mode_s_msg_num = 0; mode_other_msg_num = 0; - return msg_list; + return result; } BIN_Msg_Buffer::BIN_Msg_Buffer() { msg_list_cache.reserve(8 * 1024); - msg_list.reserve(8 * 1024); } 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, current_cache_bytes, current_dispatch_bytes, current_high_water_messages, current_high_water_bytes, pool_capacity, pool_free_count; + size_t cache_size, current_cache_bytes, current_dispatch_batches, current_dispatch_bytes, current_high_water_messages, current_high_water_bytes; { 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_batches = dispatch_batches; 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(); } Psc::JSON ret = Psc::JSON::object(); ret.append({"mode_s_pending", mode_s_msg_num.load()}); @@ -114,12 +109,10 @@ Psc::JSON BIN_Msg_Buffer::state_json() { 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_batches", current_dispatch_batches}); 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)}); + ret.append({"allocator", "std::pmr/mimalloc"}); return ret; } diff --git a/module/Local_Server/Data_Feed/Data_Feed.h b/module/Local_Server/Data_Feed/Data_Feed.h index f847621..7caf2e1 100644 --- a/module/Local_Server/Data_Feed/Data_Feed.h +++ b/module/Local_Server/Data_Feed/Data_Feed.h @@ -165,6 +165,8 @@ public: const auto value = specific.read([](const auto& value) { return value; }); svr.set_connect_user_buffer_size(value.connect_user_buffer_size); svr.set_connect_system_buffer_size(value.connect_system_buffer_size); + svr.set_recv_system_buffer_size(value.connect_system_buffer_size); + svr.set_receive_buffering(false); svr.set_tcp_no_delay(false); svr.create(); co_await svr.listen_coro("0.0.0.0", value.port); @@ -405,7 +407,6 @@ struct Type_Descriptor { class Data_feed_Config { public: adminive::Managed_Value settings; - Psc::String_Pool pool_ = Psc::String_Pool(1600, 8 * 1024); // 8192 * 1600 12.5 MB Ordered_Map> map; Data_feed_Config() = default; void init(const Psc::JSON* that_json) { diff --git a/module/Local_Server/Data_Source/Data_Source.cpp b/module/Local_Server/Data_Source/Data_Source.cpp index b1ccebe..8b9b7e5 100644 --- a/module/Local_Server/Data_Source/Data_Source.cpp +++ b/module/Local_Server/Data_Source/Data_Source.cpp @@ -1,4 +1,5 @@ #include "Data_Source.h" +#include #include #include #include "../server/Global.h" @@ -155,6 +156,18 @@ std::string bin_format(std::string_view hex, time_t t) { SSR::MLAT_timestamp a(sec, 0); return create_Binary_Format_memory(hex, 0, &a); } +static int hex_digit_value(unsigned char c) { + if (c >= '0' && c <= '9') + return c - '0'; + if (c >= 'a' && c <= 'f') + return c - 'a' + 10; + if (c >= 'A' && c <= 'F') + return c - 'A' + 10; + return -1; +} +static bool is_hex_separator(unsigned char c) { + return c == ' ' || c == '\t' || c == '\n' || c == '\r' || c == '\f' || c == '\v'; +} std::optional File_Data_Source::get_raw_line(int& ret_index) { if (index == 0 && part_infos.empty()) { ret_index = -1; @@ -204,31 +217,35 @@ std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, in // 10 01 AA 14 61 7E 5E 02 E6 0F 38 7D 87 08 1A 32 10 01 AD B5 64 F7 63 5D 78 // 0F 9D BA 93 BA else if (data_type == File_Data_Type::BIN_Blank_Text) { - std::string info(log_info); - std::string result = ds->last_char; - ds->last_char = ""; - for (char c : info) { - if (std::isxdigit(c)) { - // 会判断字符 c 是否是十六进制数字字符 - result += c; + int high = -1; + char high_char{}; + ret.reserve((log_info.size() + ds->last_char.size()) / 2); + if (!ds->last_char.empty()) { + high_char = ds->last_char[0]; + high = hex_digit_value(static_cast(high_char)); + } + ds->last_char.clear(); + for (char c : log_info) { + const auto value = hex_digit_value(static_cast(c)); + if (value >= 0) { + if (high < 0) { + high = value; + high_char = c; + } + else { + ret.push_back(static_cast((high << 4) | value)); + high = -1; + } } - else if (std::isspace(c)) { - // - } - else { + else if (!is_hex_separator(static_cast(c))) { std::ostringstream oss; oss << "存在非法字符[" << c << "]: value:" << (int)c << " in：" << log_info << LOG_POS; server_logger->c_debug("非法字符：", {}, oss.str()); return ""; } } - auto size = result.size(); - if (size % 2 != 0) { - // 保存最后一个字符 - ds->last_char = result.empty() ? "" : std::string(1, result.back()); - result.pop_back(); - } - ret = hex2mem(result); + if (high >= 0) + ds->last_char.assign(1, high_char); } else if (data_type == File_Data_Type::BIN_Blank_One_Line_With_Escape) { auto size = log_info.size(); @@ -406,25 +423,86 @@ void File_Data_Source::handle_mode_s(std::shared_ptr& msg) { } } void Dll_Data_Source::origin_data_transform_mode_data(std::string& mode_data) { - auto start = std::chrono::steady_clock::now(); if (read_func_ptr == nullptr) return; 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); - read_statistics.update(len); - std::string data(buf, len); - if (data.empty()) { - auto end = std::chrono::steady_clock::now(); - auto cost = std::chrono::duration_cast(end - start).count(); - // std::cout << "Dll_Data_Source::read cost: " << cost << " us\n"; - return; + const auto normal_read_size = config.buffer_size; + const auto drain_read_size = std::max(normal_read_size, backlog_read_size); + std::size_t raw_size = 0; + drop_mode_ac_for_batch_ = false; + auto read = [&](std::size_t request_size) { + const auto required_size = raw_size + request_size; + if (buffer.size() < required_size) + buffer.resize(required_size); + const auto length = read_func_ptr(reinterpret_cast(buffer.data() + raw_size), request_size); + read_statistics.update(length); + raw_size += length; + last_read_request_bytes_ = request_size; + return length; + }; + if (backlog_active_.load(std::memory_order_relaxed)) { + drop_mode_ac_for_batch_ = true; + const auto length = read(drain_read_size); + backlog_active_.store(length == drain_read_size, std::memory_order_relaxed); } - auto ret = get_true_from_raw_line(this, config.data_type, -1, data); - auto end = std::chrono::steady_clock::now(); - auto cost = std::chrono::duration_cast(end - start).count(); - // std::cout << "Dll_Data_Source::read cost: " << cost << " us\n"; - mode_data = ret; + else { + const auto first_length = read(normal_read_size); + if (first_length == normal_read_size) { + const auto probe_length = read(drain_read_size); + if (probe_length >= normal_read_size) { + drop_mode_ac_for_batch_ = true; + backlog_active_.store(probe_length == drain_read_size, std::memory_order_relaxed); + backlog_events_total_.fetch_add(1, std::memory_order_relaxed); + } + } + } + last_batch_raw_bytes_.store(raw_size, std::memory_order_relaxed); + auto largest = largest_batch_raw_bytes_.load(std::memory_order_relaxed); + while (largest < raw_size && !largest_batch_raw_bytes_.compare_exchange_weak(largest, raw_size, std::memory_order_relaxed)) { + } + if (drop_mode_ac_for_batch_) { + backlog_batches_total_.fetch_add(1, std::memory_order_relaxed); + backlog_raw_bytes_total_.fetch_add(raw_size, std::memory_order_relaxed); + } + if (raw_size == 0) + return; + mode_data = get_true_from_raw_line(this, config.data_type, -1, + std::string_view(reinterpret_cast(buffer.data()), raw_size)); +} +bool Dll_Data_Source::discard_input_packet(std::string_view packet) { + if (!drop_mode_ac_for_batch_ || packet.size() < 2 || packet[1] != SSR::Msg::AC) + return false; + dropped_mode_ac_frames_total_.fetch_add(1, std::memory_order_relaxed); + dropped_mode_ac_bytes_total_.fetch_add(packet.size(), std::memory_order_relaxed); + return true; +} +Psc::JSON Dll_Data_Source::backlog_state_json() const { + const auto normal_read_size = specific.member<&Dll_Data_Source_Data::buffer_size>().read([](const auto& value) { + return value; + }); + Psc::JSON result = Psc::JSON::object(); + result.append({"active", backlog_active_.load(std::memory_order_relaxed)}); + result.append({"normal_read_bytes", normal_read_size}); + result.append({"drain_read_bytes", std::max(normal_read_size, backlog_read_size)}); + result.append({"last_read_request_bytes", last_read_request_bytes_.load(std::memory_order_relaxed)}); + result.append({"last_batch_raw_bytes", last_batch_raw_bytes_.load(std::memory_order_relaxed)}); + result.append({"largest_batch_raw_bytes", largest_batch_raw_bytes_.load(std::memory_order_relaxed)}); + result.append({"events_total", backlog_events_total_.load(std::memory_order_relaxed)}); + result.append({"batches_total", backlog_batches_total_.load(std::memory_order_relaxed)}); + result.append({"raw_bytes_total", backlog_raw_bytes_total_.load(std::memory_order_relaxed)}); + result.append({"dropped_mode_ac_frames_total", dropped_mode_ac_frames_total_.load(std::memory_order_relaxed)}); + result.append({"dropped_mode_ac_bytes_total", dropped_mode_ac_bytes_total_.load(std::memory_order_relaxed)}); + return result; +} +Psc::JSON Dll_Data_Source::get_custom_state_json() { + Psc::JSON result = Psc::JSON::object(); + { + std::lock_guard lock(state_mtx); + result.append({"state", state}); + } + result.append({"library_read", read_statistics.to_json()}); + result.append({"backlog", backlog_state_json()}); + return result; } asio::awaitable Shared_Memory_Data_Source::_open() { const auto config = specific.read([](const auto& value) { return value; }); diff --git a/module/Local_Server/Data_Source/Data_Source.h b/module/Local_Server/Data_Source/Data_Source.h index 7e4f813..44e48d8 100644 --- a/module/Local_Server/Data_Source/Data_Source.h +++ b/module/Local_Server/Data_Source/Data_Source.h @@ -1,4 +1,5 @@ #pragma once +#include #include #include "Data_Source_Handler.h" namespace Psc { @@ -176,6 +177,9 @@ public: virtual asio::awaitable handle_in_loop_coro() { co_return; } + virtual bool discard_input_packet(std::string_view) { + return false; + } Frequency_Limit statistic_fl; Psc::JSON statistic_json() { Psc::JSON ret = With_Loop_Coro::to_base_json(); @@ -459,19 +463,14 @@ struct Type_Descriptor { class Dll_Data_Source : public Data_Source { public: adminive::Managed_Value specific; - Psc::JSON get_custom_state_json() override { - 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::JSON get_custom_state_json() override; + Psc::JSON backlog_state_json() const; Psc::Throughput_Statistics read_statistics; Dll_Data_Source() { type = "Dll_Data_Source"; } ~Dll_Data_Source() override = default; - std::vector buffer; + std::pmr::vector buffer; void from_json(const Psc::JSON* that_json) override { Data_Source::from_json(that_json); const auto buffer_size = specific.write([&](auto& value) { @@ -521,7 +520,20 @@ public: } co_return; } + bool discard_input_packet(std::string_view packet) override; void origin_data_transform_mode_data(std::string& data) override; +private: + static constexpr std::size_t backlog_read_size = 8 * 1024 * 1024; + bool drop_mode_ac_for_batch_{}; + std::atomic_bool backlog_active_{}; + std::atomic backlog_events_total_{}; + std::atomic backlog_batches_total_{}; + std::atomic backlog_raw_bytes_total_{}; + std::atomic dropped_mode_ac_frames_total_{}; + std::atomic dropped_mode_ac_bytes_total_{}; + std::atomic last_batch_raw_bytes_{}; + std::atomic largest_batch_raw_bytes_{}; + std::atomic last_read_request_bytes_{}; }; class Shared_Memory_Data_Source_Data { public: diff --git a/module/Local_Server/Data_Source/Data_Source_Handler.cpp b/module/Local_Server/Data_Source/Data_Source_Handler.cpp index c71e99e..f6c51c6 100644 --- a/module/Local_Server/Data_Source/Data_Source_Handler.cpp +++ b/module/Local_Server/Data_Source/Data_Source_Handler.cpp @@ -15,14 +15,16 @@ std::shared_ptr Data_Source_Handler::create_msg(std::string_view packe std::cout << "first:" << mem2hex(std::string(packet)) << std::endl; return nullptr; } - std::string error_len = source->key + " " + LOG_POS_SIMPLE + std::string(" ") + "mode_s_error_length"; + auto error_len = [&] { + return source->key + " " + LOG_POS_SIMPLE + std::string(" ") + "mode_s_error_length"; + }; if (packet.size() < 2) return nullptr; auto mt = packet[1]; if (mt == SSR::Msg::AC) { if (packet.size() != SSR::Msg::AC_len) { auto packet_hex = mem2hex(std::string(packet)); - SSR::mode_s_logger->c_debug(error_len, {}, + SSR::mode_s_logger->c_debug(error_len(), {}, std::to_string(mt) + " size:" + std::to_string(packet.size()) + " hex:" + packet_hex + " should:" + std::to_string(SSR::Msg::AC_len)); mode_ac_statistic.add_length_error(); @@ -34,7 +36,7 @@ std::shared_ptr Data_Source_Handler::create_msg(std::string_view packe if (mt == SSR::Msg::S7) { if (packet.size() != SSR::Mode_S_Msg::S7_len) { auto packet_hex = mem2hex(std::string(packet)); - SSR::mode_s_logger->c_debug(error_len, {}, + SSR::mode_s_logger->c_debug(error_len(), {}, std::to_string(mt) + " size:" + std::to_string(packet.size()) + " hex:" + packet_hex + " should:" + std::to_string(SSR::Msg::S7_len)); mode_s_statistic.add_length_error("[length error] Msg::S7 handle_mode_s_source"); @@ -45,7 +47,7 @@ std::shared_ptr Data_Source_Handler::create_msg(std::string_view packe if (mt == SSR::Msg::S14) { if (packet.size() != SSR::Msg::S14_len) { auto packet_hex = mem2hex(std::string(packet)); - SSR::mode_s_logger->c_debug(error_len, {}, + SSR::mode_s_logger->c_debug(error_len(), {}, std::to_string(mt) + " size:" + std::to_string(packet.size()) + " hex:" + packet_hex + " should:" + std::to_string(SSR::Msg::S14_len)); mode_s_statistic.add_length_error("[length error] Msg::S14 handle_mode_s_source"); @@ -57,7 +59,7 @@ std::shared_ptr Data_Source_Handler::create_msg(std::string_view packe return nullptr; // 不知道如何解析跳过 if (packet.size() != SSR::Msg::Radarcape_status_len) { auto packet_hex = mem2hex(std::string(packet)); - SSR::mode_s_logger->c_debug(error_len, {}, + SSR::mode_s_logger->c_debug(error_len(), {}, std::to_string(mt) + " size:" + std::to_string(packet.size()) + " hex:" + packet_hex + " should:" + std::to_string(SSR::Msg::Radarcape_status_len)); mode_s_statistic.add_length_error("[length error] Msg::Radarcape_status_len handle_mode_s_source"); @@ -70,7 +72,7 @@ std::shared_ptr Data_Source_Handler::create_msg(std::string_view packe auto packet_hex = mem2hex(std::string(packet)); // 找不到协议 size:23 hex:1A34195F0000001500B502FF06F423CE50000090000000 // should:5 - SSR::mode_s_logger->c_debug(error_len, {}, + SSR::mode_s_logger->c_debug(error_len(), {}, Psc::to_string(source->type) + " HULC size:" + std::to_string(packet.size()) + " hex:" + packet_hex); return nullptr; @@ -94,14 +96,9 @@ size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data) std::cout << " after origin_data_transform_mode_data " << VAR_STR_2(name, mode_data.size()) << key << std::endl; } - static Value_Growth_Multi_T mt; - auto t = key; if (data_source_debug) { std::cout << name << " 读取到:" << VAR_STR_2(key, mode_data.size()) << " " << std::endl; } - if (mt.test(key, mode_data.size())) { - std::cout << VAR_STR_2(key, mode_data.size()) << " 数据增长过快，可能内存积压" << std::endl; - } size_t ret = 0; if (mode_data.empty()) { return ret; @@ -120,6 +117,8 @@ size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data) } auto handle_packet = [this, source, &ret](std::string& packet) { ret++; + if (source->discard_input_packet(packet)) + return; auto msg = create_msg(packet); if (!msg) return; @@ -145,30 +144,14 @@ size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data) source->push_to_feed(msg); } }; - auto f = [this, source, handle_packet](std::string& packet) { - const auto wait = Global::instance()->mode_acs.settings.member<&Mode_ACS_Config_Data::wait_process_msg>().read([](const auto& value) { return value; }); - if (wait) { - handle_packet(packet); - } - else { - auto co = Coro::instance(); - auto executor = co->process_data->get_executor(); - asio::post(executor, [handle_packet, packet = std::move(packet)]() mutable { handle_packet(packet); }); - } + auto f = [handle_packet](std::string& packet) { + handle_packet(packet); }; SSR::Binary_Format_handle_buffer(source->buffer, mode_data, f); return ret; } void Data_Source_Handler::handle_mode_s(std::shared_ptr& mode_s_msg) { auto source = ds(this); - std::string t = - mode_s_msg->mlat_timestamp.to_memory() + mode_s_msg->signal_level + Psc::hex2mem(mode_s_msg->msg_hex); - auto p = mode_s_msg->packet.substr(2); - if (t != p) { - auto data = std::format("编解码出问题了(组合的数据，拼接的数据) hex:[{}] [{}] bin:[{}] [{}]", - mem2hex(t, true, " "), mem2hex(mode_s_msg->packet, true, " "), t, mode_s_msg->packet); - Psc::fail_fast(data); - } const auto mode_config = Global::instance()->mode_acs.settings.read([](const auto& value) { return std::tuple{value.time_space_filter, value.speed_filter, value.aircraft_change_list_adsb_range_filter, value.aircraft_change_list_adsb_range_factor, value.max_speed_m_s, value.air_pos_timeout, value.surface_pos_timeout}; }); @@ -181,9 +164,11 @@ void Data_Source_Handler::handle_mode_s(std::shared_ptr& mode_s if (!base_station_pos && base_station_has_valid_position) base_station_pos = SSR::Position_3D{lat, lon, alt}; SSR::ADS_B_T::Constraint air_constraint{max_speed_m_s, air_pos_timeout, SSR::cpr_cb, time_space_filter, - speed_filter, use_system_time, base_station_pos, alt, range_filter, range_factor}; + speed_filter, use_system_time, base_station_pos, alt, range_filter, range_factor, + false}; SSR::ADS_B_T::Constraint surface_constraint{max_speed_m_s, surface_pos_timeout, SSR::cpr_cb, time_space_filter, - speed_filter, use_system_time, base_station_pos, alt, range_filter, range_factor}; + speed_filter, use_system_time, base_station_pos, alt, range_filter, range_factor, + false}; SSR::parse_mode_s_bin(source.get(), mode_s_msg, base_station_pos, air_constraint, surface_constraint); auto base = source->get_aircraft(mode_s_msg->icao); if (base) { @@ -271,7 +256,7 @@ void Data_Source_Handler::refresh_data_feed_key_list() { std::vector next; for (const auto& feed : Global::instance()->mode_acs.data_feed_config.map.list()) { if (feed->enabled() && feed->source_key == key) - next.push_back({feed->key}); + next.push_back({feed->key, feed}); } std::lock_guard lock(cdf_mtx); for (auto& item : next) { @@ -287,14 +272,8 @@ void Data_Source_Handler::refresh_data_feed_key_list() { } std::optional convert_to_send_format(Data_Source_Handler* ds, const std::shared_ptr& feed, - const std::shared_ptr& msg) { + const std::shared_ptr& msg) { auto& type = msg->type; - auto fs = dynamic_cast(msg->source.get()); - if (fs) { - if (fs->specific.member<&File_Data_Source_Data::play_mode>().read([](const auto& value) { return value; }) == Play_Mode::analysis) { - // std::cout << fs->key << " analysis i ==" << i << std::endl; - } - } // 出口输出的条件 const auto feed_format = feed->output_format.read([](const auto& value) { return value; }); const auto output_format = feed_format.type; @@ -409,20 +388,8 @@ 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 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; - break; - } - std::shared_ptr& feed = opt_feed.value(); + for (const auto& item : cached_data_feed_key_list) { + const auto& feed = item.feed; if (!feed->enabled()) continue; const bool mode_s = msg->type == SSR::Mode_Msg::T::S7 || msg->type == SSR::Mode_Msg::T::S14; diff --git a/module/Local_Server/Data_Source/Data_Source_Handler.h b/module/Local_Server/Data_Source/Data_Source_Handler.h index 09a98e6..8863df9 100644 --- a/module/Local_Server/Data_Source/Data_Source_Handler.h +++ b/module/Local_Server/Data_Source/Data_Source_Handler.h @@ -10,6 +10,7 @@ #include "Psc_Cpp_Core/Statistics/Statistics.h" #include "Psc_Cpp_Core/transmit_protocol/core/statistics.h" #include "global.h" +class Data_Feed; // 这个函数专门装cpu密集函数 放到线程池去处理 class Data_Source_Handler : public DataBase { public: @@ -28,6 +29,7 @@ public: void refresh_data_feed_key_list(); struct Cached_Source_Info { std::string key; + std::shared_ptr feed; Psc::Value_Statistics mode_s_cache_num; Psc::Value_Statistics mode_other_cache_num; }; diff --git a/module/Local_Server/Data_Source/Database.cpp b/module/Local_Server/Data_Source/Database.cpp index b55fb3b..66b78d5 100644 --- a/module/Local_Server/Data_Source/Database.cpp +++ b/module/Local_Server/Data_Source/Database.cpp @@ -417,17 +417,15 @@ DataBase::create_aircraft(std::string_view icao) return ret; } -void DataBase::delete_timeout_aircraft() -{ - aircraft_map.remove_cond([](const std::shared_ptr& item) - { - if (item->timestamp == 0) - { +void DataBase::delete_timeout_aircraft() { + const auto now = std::time(nullptr); + const auto timeout_seconds = Global::instance()->mode_acs.settings.member<&Mode_ACS_Config_Data::timeout_seconds>().read( + [](const auto& value) { return value; }); + aircraft_map.remove_cond([now, timeout_seconds](const std::shared_ptr& item) { + if (item->timestamp == 0) { std::cout << "未初始化的 timestamp " << LOG_POS << std::endl; } - long long time = std::time(nullptr) - item->timestamp; - return time > Global::instance()->mode_acs.settings.member<&Mode_ACS_Config_Data::timeout_seconds>().read( - [](const auto& value) { return value; }); + return now - item->timestamp > timeout_seconds; }); } diff --git a/module/Local_Server/External_Database/External_Database.cpp b/module/Local_Server/External_Database/External_Database.cpp index 70c2f99..36105d9 100644 --- a/module/Local_Server/External_Database/External_Database.cpp +++ b/module/Local_Server/External_Database/External_Database.cpp @@ -205,38 +205,30 @@ External_Resource_Status External_Resources::clear_table(SQLite::Database& db, c } std::optional -External_Resources::query_one(SQLite::Database& db, const std::vector& primary_key_values) const -{ +External_Resources::query_one(SQLite::Database& db, const std::vector& primary_key_values) const { const auto& columns = primary_key_columns(); - if (columns.size() != primary_key_values.size()) - { + if (columns.size() != primary_key_values.size()) { throw std::invalid_argument("primary key value count mismatch for resource: " + name); } std::string sql = "SELECT * FROM " + External_Database_Utils::quote_identifier(name) + " WHERE "; - for (std::size_t index = 0; index < columns.size(); ++index) - { + for (std::size_t index = 0; index < columns.size(); ++index) { if (index != 0) sql += " AND "; - sql += External_Database_Utils::quote_identifier(columns[index]) + " = ? COLLATE NOCASE"; + sql += External_Database_Utils::quote_identifier(columns[index]) + " = ?"; } sql += " LIMIT 1"; SQLite::Statement statement(db, sql); - for (std::size_t index = 0; index < primary_key_values.size(); ++index) - { + for (std::size_t index = 0; index < primary_key_values.size(); ++index) { statement.bind(static_cast(index + 1), External_Database_Utils::normalize_key(primary_key_values[index])); } if (!statement.executeStep()) return std::nullopt; External_Database_Row row; - for (int index = 0; index < statement.getColumnCount(); ++index) - { + for (int index = 0; index < statement.getColumnCount(); ++index) { const auto name = statement.getColumnName(index); - if (statement.isColumnNull(index)) - { + if (statement.isColumnNull(index)) { row.columns.emplace(name, std::nullopt); - } - else - { + } else { row.columns.emplace(name, statement.getColumn(index).getString()); } } diff --git a/module/Local_Server/server/Global.cpp b/module/Local_Server/server/Global.cpp index 4066e90..fd834a8 100644 --- a/module/Local_Server/server/Global.cpp +++ b/module/Local_Server/server/Global.cpp @@ -65,6 +65,8 @@ Global::Global() : Config() { SSR::mode_s_logger = new BaseLogger("@/logs/mode_s", MB * size_MB * 10, piece_num); dsp_logger = new BaseLogger("@/logs/dsp", MB * size_MB, piece_num); server_logger = new BaseLogger("@/logs/server", MB * size_MB, piece_num); + SSR::parse_pos_call_back = nullptr; + SSR::parse_ok_json = nullptr; asio_socket::socket_logger->set_redirect([](const BaseLogger::Log_Info& log_info) { auto msg = "[" + std::string("func") + "~" + log_info.func_pattern + "] " + log_info.log_type.to_string() + log_info.content; @@ -77,9 +79,7 @@ Global::Global() : Config() { }); from_base_json(&j); #ifdef Cache_Some_Mode_S - auto old_parse_ok_json = SSR::parse_ok_json; - SSR::parse_ok_json = [old_parse_ok_json](Aircraft_Info* aircraft_info, SSR::P_S mode_s_msg, Psc::JSON json) { - old_parse_ok_json(aircraft_info, mode_s_msg, json); + SSR::parse_ok_json = [](Aircraft_Info* aircraft_info, SSR::P_S mode_s_msg, Psc::JSON) { auto f = static_cast(aircraft_info); f->add_msg_limit(mode_s_msg); }; @@ -88,7 +88,10 @@ Global::Global() : Config() { auto src = std::dynamic_pointer_cast(msg->source); Mode_S_Statistic_Data* ss = &src->mode_s_statistic; ss->add_cpr_error(t, err_type); - SSR::mode_s_logger->debug("CPR/" + to_string(t) + "/" + to_string(err_type), {}, log); + if (err_type == SSR::CPR_Ret_Type::Speed_Error || err_type == SSR::CPR_Ret_Type::Inter_Error || + err_type == SSR::CPR_Ret_Type::Time_Space_Too_Long || + err_type == SSR::CPR_Ret_Type::Out_of_Maximum_Detection_Range) + SSR::mode_s_logger->debug("CPR/" + to_string(t) + "/" + to_string(err_type), {}, log); }; // 处理错误 SSR::parse_call_back = [this](SSR::Parse_Call_Back_Type type, SSR::Aircraft_Info* info, std::string_view field_name, diff --git a/module/Local_Server/server/Memory_Monitor.cpp b/module/Local_Server/server/Memory_Monitor.cpp new file mode 100644 index 0000000..8f5ab56 --- /dev/null +++ b/module/Local_Server/server/Memory_Monitor.cpp @@ -0,0 +1,227 @@ +#include "Memory_Monitor.h" +#include +#include +#include +#include +#include +#include +#include +#include "Memory_Resource.h" +namespace { +constexpr std::uint64_t kibibyte = 1024; +constexpr std::uint64_t mebibyte = 1024 * kibibyte; +constexpr std::size_t history_capacity = 600; +constexpr std::size_t response_history_capacity = 120; +constexpr std::uint64_t growth_warmup_ms = 10 * 60 * 1000; +constexpr std::uint64_t growth_window_ms = 15 * 60 * 1000; +constexpr std::uint64_t recent_growth_window_ms = 5 * 60 * 1000; +constexpr std::uint64_t minimum_growth_window_ms = 10 * 60 * 1000; +constexpr std::uint64_t minimum_recent_growth_window_ms = 4 * 60 * 1000; +std::uint64_t now_milliseconds() { + return static_cast(std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count()); +} +std::uint64_t parse_kibibytes(std::string_view line) { + const auto separator = line.find(':'); + if (separator == std::string_view::npos) + return 0; + std::uint64_t value{}; + std::istringstream stream(std::string(line.substr(separator + 1))); + stream >> value; + return value * kibibyte; +} +#if defined(ECAP_TARGET_LINUX) +void read_linux_process_memory(Process_Memory_Snapshot& result) { + std::ifstream stream("/proc/self/status"); + std::string line; + while (std::getline(stream, line)) { + if (line.starts_with("VmRSS:")) + result.rss_bytes = parse_kibibytes(line); + else if (line.starts_with("VmHWM:")) + result.peak_rss_bytes = parse_kibibytes(line); + else if (line.starts_with("RssAnon:")) + result.anonymous_rss_bytes = parse_kibibytes(line); + else if (line.starts_with("RssFile:")) + result.file_rss_bytes = parse_kibibytes(line); + else if (line.starts_with("RssShmem:")) + result.shared_rss_bytes = parse_kibibytes(line); + else if (line.starts_with("VmSize:")) + result.virtual_bytes = parse_kibibytes(line); + else if (line.starts_with("VmSwap:")) + result.swap_bytes = parse_kibibytes(line); + } +} +void read_linux_system_memory(Process_Memory_Snapshot& result) { + std::ifstream stream("/proc/meminfo"); + std::string line; + while (std::getline(stream, line)) { + if (line.starts_with("MemTotal:")) + result.system_total_bytes = parse_kibibytes(line); + else if (line.starts_with("MemAvailable:")) + result.system_available_bytes = parse_kibibytes(line); + } +} +#endif +struct Growth_Trend { + std::int64_t bytes_per_hour{}; + std::uint64_t window_ms{}; +}; +template +Growth_Trend growth_trend(const std::pmr::deque& history, Getter getter, + std::uint64_t maximum_window_ms) { + if (history.size() < 2) + return {}; + const auto latest_timestamp = history.back().timestamp_ms; + const auto warmup_end = history.front().timestamp_ms + growth_warmup_ms; + if (latest_timestamp <= warmup_end) + return {}; + const auto tail_start = latest_timestamp > maximum_window_ms ? latest_timestamp - maximum_window_ms : 0; + const auto analysis_start = std::max(warmup_end, tail_start); + const auto first = std::find_if(history.begin(), history.end(), [analysis_start](const auto& sample) { + return sample.timestamp_ms >= analysis_start; + }); + if (first == history.end() || std::next(first) == history.end()) + return {}; + const auto origin = first->timestamp_ms; + long double sum_x = 0; + long double sum_y = 0; + for (auto iter = first; iter != history.end(); ++iter) { + const auto& sample = *iter; + sum_x += static_cast(sample.timestamp_ms - origin) / 1000.0L; + sum_y += static_cast(getter(sample)); + } + const auto count = static_cast(std::distance(first, history.end())); + const auto mean_x = sum_x / count; + const auto mean_y = sum_y / count; + long double numerator = 0; + long double denominator = 0; + for (auto iter = first; iter != history.end(); ++iter) { + const auto& sample = *iter; + const auto x = static_cast(sample.timestamp_ms - origin) / 1000.0L - mean_x; + const auto y = static_cast(getter(sample)) - mean_y; + numerator += x * y; + denominator += x * x; + } + if (denominator == 0) + return {}; + return {static_cast(numerator / denominator * 3600.0L), latest_timestamp - origin}; +} +Psc::JSON snapshot_json(const Process_Memory_Snapshot& value) { + Psc::JSON result = Psc::JSON::object(); + result.append({"timestamp_ms", value.timestamp_ms}); + result.append({"rss_bytes", value.rss_bytes}); + result.append({"peak_rss_bytes", value.peak_rss_bytes}); + result.append({"anonymous_rss_bytes", value.anonymous_rss_bytes}); + result.append({"file_rss_bytes", value.file_rss_bytes}); + result.append({"shared_rss_bytes", value.shared_rss_bytes}); + result.append({"virtual_bytes", value.virtual_bytes}); + result.append({"swap_bytes", value.swap_bytes}); + result.append({"commit_bytes", value.commit_bytes}); + result.append({"peak_commit_bytes", value.peak_commit_bytes}); + result.append({"system_total_bytes", value.system_total_bytes}); + result.append({"system_available_bytes", value.system_available_bytes}); + result.append({"page_faults", value.page_faults}); + result.append({"allocator_live_bytes", value.allocator_live_bytes}); + result.append({"allocator_peak_live_bytes", value.allocator_peak_live_bytes}); + result.append({"pmr_live_bytes", value.pmr_live_bytes}); + result.append({"pmr_peak_live_bytes", value.pmr_peak_live_bytes}); + return result; +} +} +Memory_Monitor::Memory_Monitor() : history_(server_memory_resource()) { +} +Memory_Monitor& Memory_Monitor::instance() { + static Memory_Monitor monitor; + return monitor; +} +void Memory_Monitor::sample() { + Process_Memory_Snapshot result; + result.timestamp_ms = now_milliseconds(); + std::size_t elapsed{}; + std::size_t user{}; + std::size_t system{}; + std::size_t rss{}; + std::size_t peak_rss{}; + std::size_t commit{}; + std::size_t peak_commit{}; + std::size_t page_faults{}; + mi_process_info(&elapsed, &user, &system, &rss, &peak_rss, &commit, &peak_commit, &page_faults); + result.rss_bytes = rss; + result.peak_rss_bytes = peak_rss; + result.commit_bytes = commit; + result.peak_commit_bytes = peak_commit; + result.page_faults = page_faults; + result.system_total_bytes = Psc::get_system_memory(); +#if defined(ECAP_TARGET_LINUX) + read_linux_process_memory(result); + read_linux_system_memory(result); +#endif + const auto allocator = server_allocator_snapshot(); + result.allocator_live_bytes = allocator.live_bytes; + result.allocator_peak_live_bytes = allocator.peak_live_bytes; + result.pmr_live_bytes = allocator.pmr_live_bytes; + result.pmr_peak_live_bytes = allocator.pmr_peak_live_bytes; + std::lock_guard lock(mutex_); + history_.push_back(result); + if (history_.size() > history_capacity) + history_.pop_front(); +} +Psc::JSON Memory_Monitor::state_json() const { + std::lock_guard lock(mutex_); + Psc::JSON result = Psc::JSON::object(); + if (history_.empty()) + return result; + const auto& latest = history_.back(); + const auto window_ms = latest.timestamp_ms - history_.front().timestamp_ms; + const auto rss_growth = growth_trend(history_, [](const auto& sample) { return sample.rss_bytes; }, growth_window_ms); + const auto anonymous_growth = growth_trend(history_, [](const auto& sample) { return sample.anonymous_rss_bytes; }, + growth_window_ms); + const auto recent_anonymous_growth = growth_trend(history_, [](const auto& sample) { + return sample.anonymous_rss_bytes; + }, recent_growth_window_ms); + const auto allocator_growth = growth_trend(history_, [](const auto& sample) { + return sample.allocator_live_bytes; + }, growth_window_ms); + const auto process_percent = latest.system_total_bytes == 0 ? 0.0 : + static_cast(latest.rss_bytes) * 100.0 / static_cast(latest.system_total_bytes); + const auto available_percent = latest.system_total_bytes == 0 ? 0.0 : + static_cast(latest.system_available_bytes) * 100.0 / static_cast(latest.system_total_bytes); + std::string pressure = "normal"; + if (process_percent >= 75.0 || (available_percent > 0.0 && available_percent <= 5.0)) + pressure = "critical"; + else if (process_percent >= 50.0 || (available_percent > 0.0 && available_percent <= 15.0)) + pressure = "warning"; + const bool sustained_growth = anonymous_growth.window_ms >= minimum_growth_window_ms && + recent_anonymous_growth.window_ms >= minimum_recent_growth_window_ms && + anonymous_growth.bytes_per_hour >= static_cast(16 * mebibyte) && + recent_anonymous_growth.bytes_per_hour >= static_cast(16 * mebibyte); + const auto allocator = server_allocator_snapshot(); + result.append({"allocator", "mimalloc"}); + result.append({"allocator_version", allocator.allocator_version}); + result.append({"current", snapshot_json(latest)}); + result.append({"process_memory_percent", process_percent}); + result.append({"system_available_percent", available_percent}); + result.append({"pressure", pressure}); + result.append({"sample_count", history_.size()}); + result.append({"window_seconds", window_ms / 1000}); + result.append({"growth_window_seconds", anonymous_growth.window_ms / 1000}); + result.append({"rss_growth_bytes_per_hour", rss_growth.bytes_per_hour}); + result.append({"anonymous_growth_bytes_per_hour", anonymous_growth.bytes_per_hour}); + result.append({"recent_anonymous_growth_bytes_per_hour", recent_anonymous_growth.bytes_per_hour}); + result.append({"allocator_growth_bytes_per_hour", allocator_growth.bytes_per_hour}); + result.append({"sustained_growth", sustained_growth}); + result.append({"allocated_bytes_total", allocator.allocated_bytes_total}); + result.append({"allocation_count", allocator.allocation_count}); + result.append({"deallocation_count", allocator.deallocation_count}); + result.append({"pmr_allocated_bytes_total", allocator.pmr_allocated_bytes_total}); + result.append({"pmr_allocation_count", allocator.pmr_allocation_count}); + result.append({"pmr_deallocation_count", allocator.pmr_deallocation_count}); + Psc::JSON history = Psc::JSON::array(); + const auto stride = std::max(1, (history_.size() + response_history_capacity - 1) / response_history_capacity); + for (std::size_t index = 0; index < history_.size(); index += stride) + history.append(snapshot_json(history_[index])); + if ((history_.size() - 1) % stride != 0) + history.append(snapshot_json(history_.back())); + result.append({"history", history}); + return result; +} diff --git a/module/Local_Server/server/Memory_Monitor.h b/module/Local_Server/server/Memory_Monitor.h new file mode 100644 index 0000000..5e1dcd1 --- /dev/null +++ b/module/Local_Server/server/Memory_Monitor.h @@ -0,0 +1,35 @@ +#pragma once +#include +#include +#include +#include +#include "../global_include.h" +struct Process_Memory_Snapshot { + std::uint64_t timestamp_ms{}; + std::uint64_t rss_bytes{}; + std::uint64_t peak_rss_bytes{}; + std::uint64_t anonymous_rss_bytes{}; + std::uint64_t file_rss_bytes{}; + std::uint64_t shared_rss_bytes{}; + std::uint64_t virtual_bytes{}; + std::uint64_t swap_bytes{}; + std::uint64_t commit_bytes{}; + std::uint64_t peak_commit_bytes{}; + std::uint64_t system_total_bytes{}; + std::uint64_t system_available_bytes{}; + std::uint64_t page_faults{}; + std::uint64_t allocator_live_bytes{}; + std::uint64_t allocator_peak_live_bytes{}; + std::uint64_t pmr_live_bytes{}; + std::uint64_t pmr_peak_live_bytes{}; +}; +class Memory_Monitor { +public: + static Memory_Monitor& instance(); + void sample(); + Psc::JSON state_json() const; +private: + Memory_Monitor(); + mutable std::mutex mutex_; + std::pmr::deque history_; +}; diff --git a/module/Local_Server/server/Memory_Resource.cpp b/module/Local_Server/server/Memory_Resource.cpp new file mode 100644 index 0000000..9eb4253 --- /dev/null +++ b/module/Local_Server/server/Memory_Resource.cpp @@ -0,0 +1,171 @@ +#include "Memory_Resource.h" +#include +#include +#include +namespace { +struct Allocation_Counters { + std::atomic live_bytes{}; + std::atomic peak_live_bytes{}; + std::atomic allocated_bytes_total{}; + std::atomic allocation_count{}; + std::atomic deallocation_count{}; + std::atomic pmr_live_bytes{}; + std::atomic pmr_peak_live_bytes{}; + std::atomic pmr_allocated_bytes_total{}; + std::atomic pmr_allocation_count{}; + std::atomic pmr_deallocation_count{}; +}; +constinit Allocation_Counters allocation_counters; +void update_peak(std::atomic& peak, std::uint64_t value) { + auto current = peak.load(std::memory_order_relaxed); + while (current < value && !peak.compare_exchange_weak(current, value, std::memory_order_relaxed)) { + } +} +void record_allocation(void* pointer, bool pmr) { + if (pointer == nullptr) + return; + const auto bytes = static_cast(mi_usable_size(pointer)); + const auto live = allocation_counters.live_bytes.fetch_add(bytes, std::memory_order_relaxed) + bytes; + allocation_counters.allocated_bytes_total.fetch_add(bytes, std::memory_order_relaxed); + allocation_counters.allocation_count.fetch_add(1, std::memory_order_relaxed); + update_peak(allocation_counters.peak_live_bytes, live); + if (!pmr) + return; + const auto pmr_live = allocation_counters.pmr_live_bytes.fetch_add(bytes, std::memory_order_relaxed) + bytes; + allocation_counters.pmr_allocated_bytes_total.fetch_add(bytes, std::memory_order_relaxed); + allocation_counters.pmr_allocation_count.fetch_add(1, std::memory_order_relaxed); + update_peak(allocation_counters.pmr_peak_live_bytes, pmr_live); +} +void record_deallocation(void* pointer, bool pmr) { + if (pointer == nullptr) + return; + const auto bytes = static_cast(mi_usable_size(pointer)); + allocation_counters.live_bytes.fetch_sub(bytes, std::memory_order_relaxed); + allocation_counters.deallocation_count.fetch_add(1, std::memory_order_relaxed); + if (!pmr) + return; + allocation_counters.pmr_live_bytes.fetch_sub(bytes, std::memory_order_relaxed); + allocation_counters.pmr_deallocation_count.fetch_add(1, std::memory_order_relaxed); +} +void free_tracked(void* pointer) noexcept { + record_deallocation(pointer, false); + mi_free(pointer); +} +class Mimalloc_Memory_Resource final : public std::pmr::memory_resource { +protected: + void* do_allocate(std::size_t bytes, std::size_t alignment) override { + auto pointer = mi_new_aligned(bytes, alignment); + record_allocation(pointer, true); + return pointer; + } + void do_deallocate(void* pointer, std::size_t, std::size_t) override { + record_deallocation(pointer, true); + mi_free(pointer); + } + bool do_is_equal(const std::pmr::memory_resource& other) const noexcept override { + return this == &other; + } +}; +} +void* operator new(std::size_t bytes) { + auto pointer = mi_new(bytes); + record_allocation(pointer, false); + return pointer; +} +void* operator new[](std::size_t bytes) { + auto pointer = mi_new(bytes); + record_allocation(pointer, false); + return pointer; +} +void* operator new(std::size_t bytes, const std::nothrow_t&) noexcept { + auto pointer = mi_new_nothrow(bytes); + record_allocation(pointer, false); + return pointer; +} +void* operator new[](std::size_t bytes, const std::nothrow_t&) noexcept { + auto pointer = mi_new_nothrow(bytes); + record_allocation(pointer, false); + return pointer; +} +void operator delete(void* pointer) noexcept { + free_tracked(pointer); +} +void operator delete[](void* pointer) noexcept { + free_tracked(pointer); +} +void operator delete(void* pointer, const std::nothrow_t&) noexcept { + free_tracked(pointer); +} +void operator delete[](void* pointer, const std::nothrow_t&) noexcept { + free_tracked(pointer); +} +void operator delete(void* pointer, std::size_t) noexcept { + free_tracked(pointer); +} +void operator delete[](void* pointer, std::size_t) noexcept { + free_tracked(pointer); +} +void* operator new(std::size_t bytes, std::align_val_t alignment) { + auto pointer = mi_new_aligned(bytes, static_cast(alignment)); + record_allocation(pointer, false); + return pointer; +} +void* operator new[](std::size_t bytes, std::align_val_t alignment) { + auto pointer = mi_new_aligned(bytes, static_cast(alignment)); + record_allocation(pointer, false); + return pointer; +} +void* operator new(std::size_t bytes, std::align_val_t alignment, const std::nothrow_t&) noexcept { + auto pointer = mi_new_aligned_nothrow(bytes, static_cast(alignment)); + record_allocation(pointer, false); + return pointer; +} +void* operator new[](std::size_t bytes, std::align_val_t alignment, const std::nothrow_t&) noexcept { + auto pointer = mi_new_aligned_nothrow(bytes, static_cast(alignment)); + record_allocation(pointer, false); + return pointer; +} +void operator delete(void* pointer, std::align_val_t) noexcept { + free_tracked(pointer); +} +void operator delete[](void* pointer, std::align_val_t) noexcept { + free_tracked(pointer); +} +void operator delete(void* pointer, std::size_t, std::align_val_t) noexcept { + free_tracked(pointer); +} +void operator delete[](void* pointer, std::size_t, std::align_val_t) noexcept { + free_tracked(pointer); +} +void operator delete(void* pointer, std::align_val_t, const std::nothrow_t&) noexcept { + free_tracked(pointer); +} +void operator delete[](void* pointer, std::align_val_t, const std::nothrow_t&) noexcept { + free_tracked(pointer); +} +std::pmr::memory_resource* server_memory_resource() { + static Mimalloc_Memory_Resource resource; + return &resource; +} +void initialize_server_memory_resource() { + mi_process_init(); + std::pmr::set_default_resource(server_memory_resource()); +} +Server_Allocator_Snapshot server_allocator_snapshot() { + Server_Allocator_Snapshot result; + result.live_bytes = allocation_counters.live_bytes.load(std::memory_order_relaxed); + result.peak_live_bytes = allocation_counters.peak_live_bytes.load(std::memory_order_relaxed); + result.allocated_bytes_total = allocation_counters.allocated_bytes_total.load(std::memory_order_relaxed); + result.allocation_count = allocation_counters.allocation_count.load(std::memory_order_relaxed); + result.deallocation_count = allocation_counters.deallocation_count.load(std::memory_order_relaxed); + result.pmr_live_bytes = allocation_counters.pmr_live_bytes.load(std::memory_order_relaxed); + result.pmr_peak_live_bytes = allocation_counters.pmr_peak_live_bytes.load(std::memory_order_relaxed); + result.pmr_allocated_bytes_total = allocation_counters.pmr_allocated_bytes_total.load(std::memory_order_relaxed); + result.pmr_allocation_count = allocation_counters.pmr_allocation_count.load(std::memory_order_relaxed); + result.pmr_deallocation_count = allocation_counters.pmr_deallocation_count.load(std::memory_order_relaxed); + result.allocator_version = mi_version(); + return result; +} +void collect_server_memory(bool force) { + mi_collect(force); +} diff --git a/module/Local_Server/server/Memory_Resource.h b/module/Local_Server/server/Memory_Resource.h new file mode 100644 index 0000000..26b3cd3 --- /dev/null +++ b/module/Local_Server/server/Memory_Resource.h @@ -0,0 +1,21 @@ +#pragma once +#include +#include +#include +struct Server_Allocator_Snapshot { + std::uint64_t live_bytes{}; + std::uint64_t peak_live_bytes{}; + std::uint64_t allocated_bytes_total{}; + std::uint64_t allocation_count{}; + std::uint64_t deallocation_count{}; + std::uint64_t pmr_live_bytes{}; + std::uint64_t pmr_peak_live_bytes{}; + std::uint64_t pmr_allocated_bytes_total{}; + std::uint64_t pmr_allocation_count{}; + std::uint64_t pmr_deallocation_count{}; + int allocator_version{}; +}; +void initialize_server_memory_resource(); +std::pmr::memory_resource* server_memory_resource(); +Server_Allocator_Snapshot server_allocator_snapshot(); +void collect_server_memory(bool force); diff --git a/module/Local_Server/server/Mode_Msg_Buffer.h b/module/Local_Server/server/Mode_Msg_Buffer.h index 6cf9a35..3fea7e5 100644 --- a/module/Local_Server/server/Mode_Msg_Buffer.h +++ b/module/Local_Server/server/Mode_Msg_Buffer.h @@ -2,6 +2,7 @@ #define BIN_Msg_Buffer_H #include "../global_include.h" +#include #include #include @@ -38,20 +39,18 @@ protected: }; struct BIN_Msg_Buffer { + using Batch = std::pmr::vector; void push(std::string_view msg); - const std::vector &get_all(); - + Batch take_all(); std::atomic mode_s_msg_num{}; std::atomic mode_other_msg_num{}; BIN_Msg_Buffer(); - Psc::JSON state_json(); - protected: std::mutex mtx; - std::vector msg_list_cache; - std::vector msg_list; + Batch msg_list_cache; size_t cache_bytes{}; + size_t dispatch_batches{}; size_t dispatch_bytes{}; size_t high_water_messages{}; size_t high_water_bytes{}; diff --git a/module/Local_Server/server/io_coro.cpp b/module/Local_Server/server/io_coro.cpp index 852094b..583085c 100644 --- a/module/Local_Server/server/io_coro.cpp +++ b/module/Local_Server/server/io_coro.cpp @@ -136,13 +136,11 @@ asio::awaitable Data_Feed::loop_coro() { auto feed = this; auto co = Coro::instance(); auto& mode_acs = Global::instance()->mode_acs; - auto& cfg = mode_acs.data_feed_config; - auto& pool = cfg.pool_; while (feed->running()) { if (data_feed_debug) { std::cout << std::format("{} feed loop begin {}\n", key, thread_id_str()); } - std::vector messages; + BIN_Msg_Buffer::Batch messages; { if (!feed->running() || !feed->registered()) { break; @@ -150,26 +148,17 @@ asio::awaitable Data_Feed::loop_coro() { co_await feed->handle_in_loop_coro(); auto s_num = feed->msg_buffer.mode_s_msg_num.load(); auto other_num = feed->msg_buffer.mode_other_msg_num.load(); - const std::vector& all = feed->msg_buffer.get_all(); - Pool_Guard pg(&pool, all); + messages = feed->msg_buffer.take_all(); const auto [num, monitor_msg_live] = mode_acs.settings.read([](const auto& value) { return std::pair{value.report_data_feed_msg_mum, value.monitor_msg_live}; }); - if (monitor_msg_live && num != 0 && all.size() > num && too_many_msg_limit.test()) { + if (monitor_msg_live && num != 0 && messages.size() > num && too_many_msg_limit.test()) { std::ostringstream oss; oss << "feed_key:" << feed->key << " "; oss << "recv_s:" << s_num << " "; oss << "recv_other:" << other_num << " "; std::cout << oss.str() << std::endl; } - messages.reserve(all.size()); - for (const auto* str : all) { - if (!str || str->empty()) { - std::cout << "empty feed message:" << feed->key << std::endl; - continue; - } - messages.emplace_back(*str); - } } if (messages.empty()) { co_await co->sleep_for(std::chrono::milliseconds(10)); @@ -186,7 +175,7 @@ asio::awaitable Data_Feed::loop_coro() { if (data_feed_debug) { std::cout << std::format("{} before send {}\n", key, thread_id_str()); } - co_await feed->send_coro(msg); + co_await feed->send_coro(std::string_view(msg.data(), msg.size())); if (data_feed_debug) { std::cout << std::format("{} after send {}\n", key, thread_id_str()); } @@ -292,7 +281,6 @@ asio::awaitable Data_Source::loop_coro() { ask_sleep = false; } } - std::cout << std::flush; } std::cout << std::format("{} source loop exit {}\n", key, thread_id_str()); co_return; diff --git a/module/Local_Server/server/server.cpp b/module/Local_Server/server/server.cpp index c09b04a..22511e1 100644 --- a/module/Local_Server/server/server.cpp +++ b/module/Local_Server/server/server.cpp @@ -5,6 +5,7 @@ #include #include #include "Global.h" +#include "Memory_Monitor.h" #include "Performance_Monitor.h" namespace { constexpr std::size_t kMaxLogRequestBodySize = 8 * 1024; @@ -174,6 +175,7 @@ void Global::init_web_server() { add_path(R"(^/map3d)"); add_path(R"(^/aircraftlist)"); add_path(R"(^/topology)"); + add_path(R"(^/memory)"); drogon::app().registerHandlerViaRegex( R"(^/ui(/.*)?$)", [return_file](const drogon::HttpRequestPtr& req, std::function&& callback) { @@ -185,7 +187,7 @@ void Global::init_web_server() { } auto is_spa_path = [](std::string_view value) { return value == "/" || value == "/map" || value == "/map3d" || value == "/settings" || - value == "/aircraftlist" || value == "/topology"; + value == "/aircraftlist" || value == "/topology" || value == "/memory"; }; if (is_spa_path(rp)) { path += "/index.html"; @@ -196,12 +198,6 @@ void Global::init_web_server() { callback(return_file(path)); }); State_Report::instance(); - svr.Get(R"(/get_debug_info)", [](HTTP_Param) { - JSON arr = JSON::array(); - auto& sp = Global::instance()->mode_acs.data_feed_config.pool_; - arr.append(JSON::object({{"remain_size", sp.remain_size()}, {"free_size", sp.free_size()}})); - res->setBody(arr.to_json_string("", "\n", "")); - }); database_server(this); external_resources_manager.server(this); mode_acs.server(this); @@ -389,6 +385,17 @@ void Global::init_web_server() { // ret.append({"基站信息", base_station.to_Json()}); res->setBody(warp(ret).to_json_string()); }); + svr.Post(api + "memory_status", [this](HTTP_Param) { + auto result = Memory_Monitor::instance().state_json(); + JSON sources = JSON::object(); + for (const auto& source : mode_acs.data_source_config.map.list()) { + if (const auto dll_source = dynamic_cast(source.get())) + sources.append({source->key, dll_source->backlog_state_json()}); + } + result.append({"dll_sources", sources}); + result.append({"iq_dropped_frames_total", dsp_config.iq_memory_buffer.dropped()}); + res->setBody(warp(result).to_json_string()); + }); svr.Post(api + "version", [this](HTTP_Param) { auto ret = JSON::object(); ret.append({"version", version}); diff --git a/module/Local_Server_main.cpp b/module/Local_Server_main.cpp index d501a7f..5a64c0e 100644 --- a/module/Local_Server_main.cpp +++ b/module/Local_Server_main.cpp @@ -10,46 +10,9 @@ #include "Local_Server/Data_Source/Data_Source.h" #include "Local_Server/State_Report/State_Report.h" #include "Local_Server/server/Global.h" +#include "Local_Server/server/Memory_Monitor.h" +#include "Local_Server/server/Memory_Resource.h" #include "Local_Server/server/io_coro.h" -// ./server ./config -#ifdef __linux__ -#include -#endif -struct Catch_Memory { - Catch_Memory() { - sm = (std::int64_t)get_system_memory(); - cur = (std::int64_t)get_process_memory(); - } - std::int64_t MB = 1024 * 1024; - bool catch_ok() { - auto old_memory = cur; - cur = (std::int64_t)get_process_memory(); - auto change_mb = (cur - old_memory) / MB; - auto cur_mb = cur / MB; - // std::cout << get_current_date_string() << " 时内存" << cur_mb << "MB " - // << " 变更" << change_mb << "MB" << std::endl; - if (cur > memory_max) { - auto rate = (double)cur / (double)sm; - std::cout << get_current_date_string() + " 抓住内存暴涨:" << cur_mb << "MB" << " 变更" << change_mb - << "MB" << std::endl; - std::cout << "\t内存占用" << rate << "%" << std::endl; - stop_program = SIGINT; -#if WIN32 -#else - system("top"); -#endif - return true; - } - return false; - } -#if WIN32 - std::int64_t memory_max = 1 * 1024 * 1024 * 1024; -#else - std::int64_t memory_max = 1 * 1024 * 1024 * 1024; -#endif - std::int64_t cur; - std::int64_t sm; -}; int psc_main(int argc, char* argv[]) { std::cout << "wyc_main" << std::endl; if (argc >= 2) { @@ -133,22 +96,14 @@ int psc_main(int argc, char* argv[]) { g->dsp_config.init_env(); auto coro = Coro::instance(); coro->start(); - // Catch_Memory cm; while (stop_program == 0) { for (auto& ds : g->mode_acs.data_source_config.map.list()) { ds->delete_timeout_aircraft(); } -#ifdef __linux__ if (g->mode_acs.settings.member<&Mode_ACS_Config_Data::monitor_msg_live>().snapshot()) - malloc_trim(0); -#endif + collect_server_memory(false); + Memory_Monitor::instance().sample(); std::this_thread::sleep_for(std::chrono::seconds(3)); - if (Global::instance()->init_ok) { - // if (cm.catch_ok()) { - // stop_program = SIGINT; - // break; - // } - } } coro->stop(); if (stop_program == SIGINT || stop_program == SIGTERM) { @@ -157,5 +112,7 @@ int psc_main(int argc, char* argv[]) { return 0; } int main(int argc, char* argv[]) { + initialize_server_memory_resource(); + Memory_Monitor::instance().sample(); return redict_main_with_gtest(argc, argv, psc_main); }