diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..2da643a --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,96 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +concurrency: + group: ci-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + build-test: + name: build & test (${{ matrix.name }}) + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + include: + - name: linux + os: ubuntu-24.04 + build_dir: _Build_lnx + - name: macos + os: macos-14 + build_dir: _Build_mac + - name: windows + os: windows-2022 + build_dir: _Build_wnd + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.x" + + # Runners ship CMake ~3.31; the kit dependency requires CMake >= 4.0. + - name: Install CMake 4.x + uses: lukka/get-cmake@latest + with: + cmakeVersion: "~4.0" + + # The kit dependency is pulled by CMake FetchContent over SSH + # (git@github.com:RiantZ/kit.git). SSH has no keys in CI, so rewrite that + # URL to plain HTTPS. kit is public, so no token is needed. + - name: Rewrite SSH GitHub URLs to HTTPS + shell: bash + run: | + git config --global \ + url."https://github.com/".insteadOf \ + "git@github.com:" + + - name: Configure & build + shell: bash + run: python scripts/cmake_build.py --build + + - name: Test + shell: bash + run: ctest --test-dir ${{ matrix.build_dir }} -C Release --output-on-failure + + format: + name: clang-format check + runs-on: ubuntu-24.04 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.x" + + # clang-format output drifts between major versions. Pin to 22 to match + # the version developers use locally (Homebrew clang-format); the runner's + # default is v18, which reports spurious violations. + - name: Install clang-format 22 + run: | + wget -q https://apt.llvm.org/llvm.sh + chmod +x llvm.sh + sudo ./llvm.sh 22 + sudo apt-get install -y clang-format-22 + # /usr/local/bin precedes /usr/bin in PATH, so this wins over the + # runner's default clang-format. + sudo ln -sf /usr/bin/clang-format-22 /usr/local/bin/clang-format + clang-format --version + + # code_format.py lives in the kit repo. Clone it outside the p8 checkout + # (into RUNNER_TEMP) so its own sources are not picked up by the + # formatter's recursive scan of the project tree. + - name: Fetch formatter from kit + shell: bash + run: | + git clone --depth 1 \ + "https://github.com/RiantZ/kit.git" \ + "$RUNNER_TEMP/kit" + + - name: Check formatting + shell: bash + run: python "$RUNNER_TEMP/kit/scripts/code_format.py" --check --all diff --git a/CMakeLists.txt b/CMakeLists.txt index 4edbb5c..194498e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,4 +1,4 @@ -cmake_minimum_required(VERSION 3.20...4.0 FATAL_ERROR) +cmake_minimum_required(VERSION 4.0 FATAL_ERROR) project(Protocol8) @@ -54,6 +54,10 @@ set(ROOT_P8_PATH ${PROJECT_SOURCE_DIR} CACHE INTERNAL "") if(MSVC) add_compile_definitions(_UNICODE UNICODE) + # (pulled in via kit) defines min()/max() macros that clobber + # std::min/std::max; NOMINMAX suppresses them. + add_compile_definitions(NOMINMAX) + if(NOT COMMAND set_ide_folder) macro(set_ide_folder _project_name) string(LENGTH ${ROOT_P8_PATH} ROOT_P8_PATH_LEN) @@ -71,6 +75,10 @@ if(MSVC) #enable_cpp_c_compiler_flag_if_supported("/Wall") enable_cpp_c_compiler_flag_if_supported("/W4") enable_cpp_c_compiler_flag_if_supported("/WX") + # C4324: structure padded due to alignas() specifier. The padding is + # intentional (cache-line alignment via alignas(64)), so silence it rather + # than let /WX turn it into an error. + enable_cpp_c_compiler_flag_if_supported("/wd4324") else() enable_c_compiler_flag_if_supported("-std=gnu99") diff --git a/CMakePresets.json b/CMakePresets.json index 03768f3..ab0f2c2 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -1,6 +1,6 @@ { "version": 6, - "cmakeMinimumRequired": { "major": 3, "minor": 21 }, + "cmakeMinimumRequired": { "major": 4, "minor": 0 }, "configurePresets": [ { "name": "windows", diff --git a/README.md b/README.md index 208ae62..395148c 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,12 @@ # Protocol8 + +[![CI](https://github.com/RiantZ/p8/actions/workflows/ci.yml/badge.svg)](https://github.com/RiantZ/p8/actions/workflows/ci.yml) + ## Building ### Prerequisites -- CMake 3.20+ (tested up to 4.0) +- CMake 4.0+ - Python 3 - C++20 compatible compiler (MSVC, GCC, Clang) @@ -78,14 +81,20 @@ Single-thread emit latency (`c_log_perf_test`, 32 batches x 1 000 000 iterations | `send_hello_d_no_attrs` | 26.8 ns | 1 000 000 | | `send_hello_d_3_attrs` (str + i64 + f64) | 39.3 ns | 1 000 000 | -`full_cycle` throughput (init + emit + release) through the in-memory null sink, 1 000 000 iterations per thread: +`full_cycle` throughput (init + emit + release) through the in-memory null sink, +1 000 000 iterations per thread. Numbers are aggregated over 25 runs of the test +(`ns/call` = full-cycle wall time / total calls): + +| Threads | Total calls | Median ns/call | p95 ns/call | Stdev | Throughput (median) | +|---------|-------------|----------------|-------------|-------|---------------------| +| 1 | 1 000 000 | 26.6 ns | 32.8 ns | 3.16 ns | 37.6 M calls/s | +| 2 | 2 000 000 | 13.7 ns | 16.1 ns | 1.85 ns | 72.8 M calls/s | +| 4 | 4 000 000 | 7.9 ns | 11.0 ns | 1.29 ns | 125.9 M calls/s | +| 8 | 8 000 000 | 4.5 ns | 6.5 ns | 0.94 ns | 221.1 M calls/s | -| Threads | Total calls | ns/call | Throughput | -|---------|-------------|---------|------------| -| 1 | 1 000 000 | 22.6 ns | 44.2 M calls/s | -| 2 | 2 000 000 | 12.7 ns | 78.6 M calls/s | -| 4 | 4 000 000 | 9.8 ns | 102.4 M calls/s | -| 8 | 8 000 000 | 6.4 ns | 157.0 M calls/s | +Relative spread (stdev/mean) grows with thread count — from ~12 % at 1 thread to +~19 % at 8 threads — so multi-threaded runs are less stable and p95 sits noticeably +above the median. ### Profiling with Tracy diff --git a/api/p8_protocol.h b/api/p8_protocol.h index 01b4776..a342d33 100644 --- a/api/p8_protocol.h +++ b/api/p8_protocol.h @@ -127,6 +127,18 @@ extern "C" //* padding to align total size on 8 bytes boundary }; + // Serialized representation of s_p8_log_mod + // Fixed-size header; variable-length data follows in the buffer. + // Name (string utf8) include a NUL terminator. + struct s_p8_log_mod_svc + { + s_p8_svc_hdr ms_hdr; // service item header, 4 + uint16_t mu_id; // ID of the module + uint8_t mu_verb; // verbosity == e_p8_level + //* Serialized name + //* padding to align total size on 8 bytes boundary + }; + // Serialized representation of s_p8_log_desc (the immutable log descriptor). // Fixed-size header; variable-length data follows in the buffer. // String lengths are byte counts and do NOT include a NUL terminator. @@ -187,9 +199,12 @@ extern "C" uint16_t mu_args_size; // serialized variable arguments size in bytes // 32 bits - uint16_t mu_size; // total item size in bytes (header + args + attrs + padding) - uint8_t mu_attrs_count; // number of serialized attributes - uint8_t mu_flags; // todo + uint16_t mu_size; // total item size in bytes (header + args + attrs + padding) + uint16_t mu_mod_id; // module ID + + // 16 bits + uint8_t mu_attrs_count; // number of serialized attributes + uint8_t mu_flags; // todo //* Serialized Variable agruments [int32][int64][double][length in bytes (16b) + string data] ... //* Serialized attributes [p8_attr_id + data][...] //* padding to tail of the buffer to 8 bytes boundary, to be sure that following buffer will start from 8b diff --git a/dep/CMakeLists.txt b/dep/CMakeLists.txt index 91bfd12..6832c2a 100644 --- a/dep/CMakeLists.txt +++ b/dep/CMakeLists.txt @@ -1,8 +1,23 @@ include(FetchContent) + +if(CMAKE_VERSION VERSION_GREATER_EQUAL 4.30) + cmake_policy(SET CMP0205 NEW) +endif() + +# SYSTEM: consume kit as a third-party dependency so its headers are included +# with -isystem. That suppresses warnings originating inside kit headers (e.g. +# GCC's -Wclass-memaccess from list.hpp's memset on non-trivial types when a +# kit template is instantiated in a p8 translation unit) without weakening the +# project's -Werror on p8's own code. +# p8 only ever links the static kit archive (kit::kit). Skip building kit's +# shared library so we don't waste build time on an unused DLL/.so (and, on +# Windows, avoid its import lib colliding with the static kit.lib). +set(KIT_BUILD_SHARED OFF CACHE BOOL "" FORCE) FetchContent_Declare( kit GIT_REPOSITORY git@github.com:RiantZ/kit.git GIT_TAG main + SYSTEM ) FetchContent_MakeAvailable(kit) @@ -15,6 +30,7 @@ FetchContent_Declare( nlohmann_json GIT_REPOSITORY https://github.com/nlohmann/json.git GIT_TAG v3.12.0 + SYSTEM ) FetchContent_MakeAvailable(nlohmann_json) @@ -29,7 +45,7 @@ FetchContent_MakeAvailable(xxhash) add_library(xxhash_hdr INTERFACE) add_library(xxhash::xxhash ALIAS xxhash_hdr) -target_include_directories(xxhash_hdr INTERFACE ${xxhash_SOURCE_DIR}) +target_include_directories(xxhash_hdr SYSTEM INTERFACE ${xxhash_SOURCE_DIR}) # Tracy profiler -- optional, controlled by P8_ENABLE_TRACY (OFF by default). # diff --git a/doc/todo.md b/doc/todo.md index c32d8a5..ba48604 100644 --- a/doc/todo.md +++ b/doc/todo.md @@ -1,17 +1,12 @@ # TODO ## Next Sprint -- Добавление для логов модулей и их сериализация -- мониторинг потерь данных - мониторинг в логах, трейсах, метриках и регулярный пулл из core - создание статистики которую можно использовать в тестах -- тест производительности - - логов против принтф в файл из нескольких тредов, замер времени и места на диске - - СДК от Датадога С++ - - СДК от Датадога для профайлинга? - Создание презентации с цифрами (оттолкнуться от ) - Анализ рынка и конкурентов - Потребности клиентов - Архитектура - Что мы предлагаем (p8, microP8, server, datadog integration) +- Добавление режима обратного давления - когда буфера нет - мы ждем - Найдена нестабильность (flaky crash). На втором запросе «100 раз» один из прогонов упал: бинарник P8_RegressionTests был убит (subprocess killed, rc=8) на ~60-й итерации, вывод оборвался на тесте c_writer_registry_test.no_core_writer_is_noop. Воспроизвести не удалось: Изолированный прогон c_writer_registry_test.* — прошёл. @@ -26,6 +21,12 @@ ## Done - add pre-hook for format & auto-install by cmake +- мониторинг потерь данных - мониторинг в логах, трейсах, метриках и регулярный пулл из core - создание статистики которую можно использовать в тестах +- Добавление для логов модулей и их сериализация +- тест производительности + - логов против принтф в файл из нескольких тредов, замер времени и места на диске + - СДК от Датадога С++ + - СДК от Датадога для профайлинга? Создание синков - ответить на вопрос поддержки мульти-синков? **Нет, такую поддержку мы делать не будем** diff --git a/engine/p8_buffer_pool.cpp b/engine/p8_buffer_pool.cpp index 2af40a8..573c1fb 100644 --- a/engine/p8_buffer_pool.cpp +++ b/engine/p8_buffer_pool.cpp @@ -130,6 +130,29 @@ void cp8_buffer_pool::recycle(uint8_t *ip_buf) mo_free.push_last(ip_buf); } +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +void cp8_buffer_pool::recycle(kit::c_lst &io_bufs) +{ + if(0 == io_bufs.size()) + { + return; + } + + std::lock_guard lo_guard(mo_mutex); + + io_bufs.clear( + [this](uint8_t *ip_buf) + { + if(!ip_buf) + { + return; + } + mz_acquired -= mz_buffer_size; + mo_free.push_last(ip_buf); + }, + kit::e_c_lst_pool_policy::e_keep); +} + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// size_t cp8_buffer_pool::get_buffer_size() const { diff --git a/engine/p8_buffer_pool.hpp b/engine/p8_buffer_pool.hpp index fa5d233..862c9e6 100644 --- a/engine/p8_buffer_pool.hpp +++ b/engine/p8_buffer_pool.hpp @@ -56,6 +56,12 @@ class cp8_buffer_pool // Released memory is kept in the pool — it is not given back to the OS. void recycle(uint8_t *ip_buf); + // Batch variant: returns every buffer in io_bufs to the free list under a + // single lock acquisition, then empties io_bufs (keeping its node pool). + // Equivalent to calling recycle() per element but pays the mutex cost once, + // which matters on the worker's flush path where whole batches are recycled. + void recycle(kit::c_lst &io_bufs); + size_t get_buffer_size() const; size_t get_free_count(); diff --git a/engine/p8_config.cpp b/engine/p8_config.cpp index 3c2a2b1..4fd1049 100644 --- a/engine/p8_config.cpp +++ b/engine/p8_config.cpp @@ -31,7 +31,12 @@ extern "C" } #ifdef G_OS_WINDOWS - lp_file = _wfopen(ip_path, L"rb"); + // _wfopen is flagged unsafe (C4996) by MSVC; use the _s variant, which + // sets lp_file to NULL on failure so the shared check below still works. + if(_wfopen_s(&lp_file, ip_path, L"rb") != 0) + { + lp_file = nullptr; + } #else lp_file = std::fopen(ip_path, "rb"); #endif diff --git a/engine/p8_core.cpp b/engine/p8_core.cpp index 894c1f9..c014cdd 100644 --- a/engine/p8_core.cpp +++ b/engine/p8_core.cpp @@ -12,6 +12,7 @@ #include "kit/system.hpp" #include "kit/thread.hpp" #include "kit/time.hpp" +#include "kit/types.h" #include @@ -22,7 +23,6 @@ #include #include #include -#include #include #include #include @@ -59,7 +59,7 @@ static bool parse_size(const char *ip_str, size_t &oz_result) } la_buf[lz_dst] = '\0'; - if(strcasecmp(la_buf, "infinite") == 0) + if(str_casecmp(la_buf, "infinite") == 0) { oz_result = std::numeric_limits::max(); return true; @@ -79,13 +79,13 @@ static bool parse_size(const char *ip_str, size_t &oz_result) return true; } - if(strcasecmp(lp_end, "KB") == 0 || strcasecmp(lp_end, "Ki") == 0) + if(str_casecmp(lp_end, "KB") == 0 || str_casecmp(lp_end, "Ki") == 0) { oz_result = static_cast(lu_val * 1024); return true; } - if(strcasecmp(lp_end, "MB") == 0 || strcasecmp(lp_end, "Mi") == 0) + if(str_casecmp(lp_end, "MB") == 0 || str_casecmp(lp_end, "Mi") == 0) { oz_result = static_cast(lu_val * 1024 * 1024); return true; @@ -248,6 +248,14 @@ cp8_core::~cp8_core() mo_attr_descs.clear(); mo_attr_name_map.clear(); + for(s_p8_log_mod *lp_mod : mo_log_mods) + { + std::free(const_cast(lp_mod->mp_name)); + delete lp_mod; + } + mo_log_mods.clear(); + mo_log_mod_map.clear(); + // drop references to any still-held service buffers; the pool destructor // frees the underlying memory below mo_svc_buffers.clear(); @@ -314,6 +322,12 @@ bool cp8_core::init_buffer_pool(const char *ip_max_memory_size, const char *ip_i return false; } + if(mz_data_buffer_size > (2 << 16)) + { + std::fprintf(stderr, "cp8_core::init_buffer_pool: buffer size > 64KB\n"); + return false; + } + mp_data_pool = new(std::nothrow) cp8_buffer_pool(mz_data_buffer_size, mp_memory_budget); if(!mp_data_pool) { @@ -438,6 +452,18 @@ void cp8_core::worker_main() // best-effort: raising priority typically needs elevated privileges, failure is non-fatal kit::set_thread_priority(kit::e_tp_time_critical); + // Stats-poll cadence: the loop wakes irregularly (event or timeout), so gate + // the drop-counter poll on a monotonic clock rather than an iteration count. + // Precompute the interval in ticks once so the per-iteration test is just a + // subtract-and-compare. + uint64_t lu_stats_numer = 0; + uint64_t lu_stats_denom = 0; + kit::get_hires_ticks_freq(lu_stats_numer, lu_stats_denom); + const uint64_t lu_stats_poll_ticks = lu_stats_numer ? static_cast(P8_CORE_STATS_POLL_INTERVAL_MS) + * 1000000ULL * lu_stats_denom / lu_stats_numer + : 0; + uint64_t lu_stats_last_ticks = kit::get_hires_ticks(); + for(;;) { uint32_t lu_signal = mo_worker_event.wait(P8_CORE_THREAD_TIMEOUT_MS); @@ -454,6 +480,7 @@ void cp8_core::worker_main() if(!lb_skip) { // move buffers from list protected by mutex to local one + mo_svc_mutex.lock(); while(mo_svc_buffers.size()) { @@ -483,6 +510,18 @@ void cp8_core::worker_main() // pull accumulated buffers from every live writer into the same batch drain_writers(lo_ready); + // On the stats cadence, pull-and-reset every writer's drop counters + // into the core accumulators. + if(lu_stats_poll_ticks) + { + uint64_t lu_now_ticks = kit::get_hires_ticks(); + if((lu_now_ticks - lu_stats_last_ticks) >= lu_stats_poll_ticks) + { + poll_dropped_stats(); + lu_stats_last_ticks = lu_now_ticks; + } + } + // Clear the pressure-wake debounce before draining. Every drain (wake, // submit, or timeout) pulls from all writers and relieves pressure, so // the clear is unconditional. It must happen strictly before the flush_ready @@ -620,7 +659,7 @@ p8_attr_id cp8_core::attr_register(const char *ip_name, enum e_p8_attr_type ie_t return P8_ATTR_ERROR_ALLOC_FAILED; } - lp_desc->mp_name = strdup(ip_name); + lp_desc->mp_name = str_dup(ip_name); if(!lp_desc->mp_name) { delete lp_desc; @@ -685,6 +724,112 @@ void cp8_core::sync_attr_cache(std::vector &io_cache) } } +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +bool cp8_core::register_module(const char *ip_name, enum e_p8_level ie_verbosity, p_p8_module *op_module) +{ + if(op_module) + { + *op_module = P8_MODULE_INVALID_ID; + } + + if(!mb_initialized) + { + return false; + } + + if(!ip_name || ip_name[0] == '\0' || !op_module) + { + return false; + } + + std::lock_guard lo_lock(mo_log_mod_mutex); + + auto lo_it = mo_log_mod_map.find(ip_name); + if(lo_it != mo_log_mod_map.end()) + { + // duplicate name: return the existing handle, refresh its verbosity + s_p8_log_mod *lp_existing = lo_it->second; + lp_existing->mb_vervosity.store(ie_verbosity, std::memory_order_relaxed); + *op_module = lp_existing; + return true; + } + + s_p8_log_mod *lp_mod = new(std::nothrow) s_p8_log_mod; + if(!lp_mod) + { + return false; + } + + lp_mod->mp_name = str_dup(ip_name); + if(!lp_mod->mp_name) + { + delete lp_mod; + return false; + } + + lp_mod->mu_id = static_cast(mo_log_mods.size()); + lp_mod->mb_vervosity.store(ie_verbosity, std::memory_order_relaxed); + + mo_log_mods.push_back(lp_mod); + mo_log_mod_map[ip_name] = lp_mod; + + // transmit the module descriptor once so the receiver can resolve mu_mod_id + { + std::lock_guard lo_svc_lock(mo_svc_mutex); + serialize_log_mod(lp_mod); + } + + *op_module = lp_mod; + return true; +} + +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +p_p8_module cp8_core::find_module(const char *ip_name) +{ + if(!mb_initialized) + { + return P8_MODULE_INVALID_ID; + } + + if(!ip_name || ip_name[0] == '\0') + { + return P8_MODULE_INVALID_ID; + } + + std::lock_guard lo_lock(mo_log_mod_mutex); + + auto lo_it = mo_log_mod_map.find(ip_name); + if(lo_it != mo_log_mod_map.end()) + { + return lo_it->second; + } + + return P8_MODULE_INVALID_ID; +} + +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +void cp8_core::set_verbosity(p_p8_module ip_module, enum e_p8_level ie_verbosity) +{ + if(!ip_module) + { + me_default_verbosity.store(ie_verbosity, std::memory_order_relaxed); + return; + } + + reinterpret_cast(ip_module)->mb_vervosity.store(ie_verbosity, std::memory_order_relaxed); +} + +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +enum e_p8_level cp8_core::get_verbosity(p_p8_module ip_module) +{ + if(!ip_module) + { + return me_default_verbosity.load(std::memory_order_relaxed); + } + + return reinterpret_cast(ip_module)->mb_vervosity.load(std::memory_order_relaxed); +} + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// uint8_t *cp8_core::acquire_buffer() { @@ -740,12 +885,16 @@ void cp8_core::release_buffers(kit::c_lst &io_buffers) return; } - if(0 == io_buffers.size()) + const size_t lz_count = io_buffers.size(); + if(0 == lz_count) { return; } - io_buffers.clear([this](uint8_t *ip_buf) { release_buffer(ip_buf); }, kit::e_c_lst_pool_policy::e_keep); + // Single lock acquisition on the pool for the whole batch, then one atomic + // decrement of the outstanding counter — instead of paying both per buffer. + mp_data_pool->recycle(io_buffers); + mu_outstanding_buffers.fetch_sub(lz_count, std::memory_order_relaxed); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -810,6 +959,43 @@ void cp8_core::drain_writers(kit::c_lst &io_data) P8_PROF_PLOT("drain_writers io_data.size", static_cast(io_data.size())); } +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +void cp8_core::poll_dropped_stats() +{ + s_p8_drop_stats lo_sum { 0, 0, 0 }; + + { + std::lock_guard lo_guard(mo_writers_lock); + for(cp8_tls_writer *lp_writer = mp_writers_head; lp_writer; lp_writer = lp_writer->mp_next_writer) + { + s_p8_drop_stats lo_writer = lp_writer->pull_dropped(); + lo_sum.mu_logs += lo_writer.mu_logs; + lo_sum.mu_metrics += lo_writer.mu_metrics; + lo_sum.mu_traces += lo_writer.mu_traces; + } + } + + accumulate_dropped(lo_sum); +} + +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +void cp8_core::accumulate_dropped(const s_p8_drop_stats &ir_stats) +{ + mu_dropped_logs.fetch_add(ir_stats.mu_logs, std::memory_order_relaxed); + mu_dropped_metrics.fetch_add(ir_stats.mu_metrics, std::memory_order_relaxed); + mu_dropped_traces.fetch_add(ir_stats.mu_traces, std::memory_order_relaxed); +} + +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +s_p8_drop_stats cp8_core::get_dropped_stats() const +{ + s_p8_drop_stats lo_stats; + lo_stats.mu_logs = mu_dropped_logs.load(std::memory_order_relaxed); + lo_stats.mu_metrics = mu_dropped_metrics.load(std::memory_order_relaxed); + lo_stats.mu_traces = mu_dropped_traces.load(std::memory_order_relaxed); + return lo_stats; +} + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// void cp8_core::flush_ready(kit::c_lst &io_ready) { @@ -839,8 +1025,9 @@ void cp8_core::flush_ready(kit::c_lst &io_ready) mp_sink->write_data(io_ready); - // recycle the consumed data buffers back to the pool - io_ready.clear([this](uint8_t *ip_buf) { release_buffer(ip_buf); }, kit::e_c_lst_pool_policy::e_keep); + // recycle the consumed data buffers back to the pool in one batch (single + // pool-mutex acquisition for the whole flush) + release_buffers(io_ready); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -1003,6 +1190,42 @@ void cp8_core::serialize_log_desc(const struct s_p8_log_desc *ip_desc) // trailing padding is already zeroed by svc_reserve } +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +void cp8_core::serialize_log_mod(const s_p8_log_mod *ip_mod) +{ + if(!ip_mod) + { + return; + } + + // truncate over-long names to the service-string boundary (length modulo) + size_t lz_name = ip_mod->mp_name ? strlen(ip_mod->mp_name) % P8_SVC_STR_MAX_LEN : 0; + + // fixed header + NUL-terminated name, padded so the whole entry is 8-aligned + size_t lz_padded = P8_ALIGN_UP_8(sizeof(s_p8_log_mod_svc) + lz_name + 1); + + uint8_t *lp_dst = svc_reserve(lz_padded); + if(!lp_dst) + { + return; + } + + s_p8_log_mod_svc *lp_entry = reinterpret_cast(lp_dst); + lp_entry->ms_hdr.mu_packet_type = P8_PACKET_SERVICE; + lp_entry->ms_hdr.mu_svc_type = P8_SVC_TYPE_MODULE; + lp_entry->ms_hdr.mu_size = static_cast(lz_padded); + lp_entry->mu_id = ip_mod->mu_id; + lp_entry->mu_verb = static_cast(ip_mod->mb_vervosity.load(std::memory_order_relaxed)); + + uint8_t *lp_name = lp_dst + sizeof(s_p8_log_mod_svc); + if(lz_name) + { + memcpy(lp_name, ip_mod->mp_name, lz_name); + } + lp_name[lz_name] = '\0'; + // trailing padding is already zeroed by svc_reserve +} + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// struct s_p8_log_desc *cp8_core::resolve_log_desc(uint64_t iu_hash, const char *ip_file, @@ -1264,7 +1487,9 @@ uint32_t p8_test_get_instance_count() //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// size_t p8_test_get_buffer_size() { - return gp_instance ? gp_instance->mz_data_buffer_size : 0; + // Compile-time pool geometry: return it directly so tests can size their + // configs as multiples of the buffer size before any instance exists. + return cp8_core::mz_data_buffer_size; } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -1366,6 +1591,12 @@ cp8_tls_writer *p8_test_get_writers_head() return gp_instance ? gp_instance->get_writers_head() : nullptr; } +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +s_p8_drop_stats p8_test_get_dropped_stats() +{ + return gp_instance ? gp_instance->get_dropped_stats() : s_p8_drop_stats { 0, 0, 0 }; +} + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// std::vector> cp8_core::get_service_buffers() { diff --git a/engine/p8_core.hpp b/engine/p8_core.hpp index a217497..1d07485 100644 --- a/engine/p8_core.hpp +++ b/engine/p8_core.hpp @@ -19,18 +19,34 @@ #include #include -#define P8_CORE_ACQUIRE_TIMEOUT_MS 100 -#define P8_CORE_THREAD_TIMEOUT_MS 50 +#define P8_CORE_ACQUIRE_TIMEOUT_MS 100 +#define P8_CORE_THREAD_TIMEOUT_MS 50 + +// Cadence at which the worker pulls-and-resets the per-writer drop counters and +// folds them into the core loss accumulators. The loop wakes irregularly, so +// this is enforced against a monotonic clock, not an iteration count. +#define P8_CORE_STATS_POLL_INTERVAL_MS 250 // When free data-buffer memory drops below this percentage of the total // *allocated* pool (buffers that actually exist, i.e. free + outstanding — not // the budget cap), acquire_buffer wakes the worker so it can pull accumulated // buffers from all writers. Measuring against allocated memory keeps an // infinite/default budget from reading as permanent pressure. -#define P8_CORE_FREE_MEM_PERCENT 25 +#define P8_CORE_FREE_MEM_PERCENT 75 class cp8_tls_writer; struct s_p8_log_desc; +struct s_p8_log_mod; + +// Aggregated count of telemetry elements dropped before reaching the sink, +// split by kind. Returned by cp8_core::get_dropped_stats() and by each writer's +// pull_dropped(). +struct s_p8_drop_stats +{ + uint64_t mu_logs; + uint64_t mu_metrics; + uint64_t mu_traces; +}; struct s_p8_attr_desc { @@ -61,6 +77,12 @@ class cp8_core void register_writer(cp8_tls_writer *ip_writer); void unregister_writer(cp8_tls_writer *ip_writer); + // loss statistics: running totals of dropped logs/metrics/traces. Both the + // worker's periodic poll and each writer's destructor flush feed + // accumulate_dropped; get_dropped_stats returns a lock-free snapshot. + void accumulate_dropped(const s_p8_drop_stats &ir_stats); + s_p8_drop_stats get_dropped_stats() const; + // thread bool register_current_thread(const char *ip_name); void unregister_current_thread(); @@ -70,6 +92,12 @@ class cp8_core p8_attr_id attr_get(const char *ip_name) const; void sync_attr_cache(std::vector &io_cache); + // log modules + bool register_module(const char *ip_name, enum e_p8_level ie_verbosity, p_p8_module *op_module); + p_p8_module find_module(const char *ip_name); + void set_verbosity(p_p8_module ip_module, enum e_p8_level ie_verbosity); + enum e_p8_level get_verbosity(p_p8_module ip_module); + // buffer pool static size_t get_buffer_size(); uint8_t *acquire_buffer(); @@ -128,17 +156,23 @@ class cp8_core // registry order. Takes mo_writers_lock and each writer's lock internally. void drain_writers(kit::c_lst &io_data); + // Pull-and-reset every registered writer's drop counters and fold them into + // the core accumulators. Runs on the worker's P8_CORE_STATS_POLL_INTERVAL_MS + // cadence. Takes mo_writers_lock; the per-writer pull is lock-free. + void poll_dropped_stats(); + // Write a batch of ready data buffers to the sink and recycle them. Single // choke point for both the worker-pull path and the writer-shutdown ready // queue, and the only place data buffers are captured under P8_TESTING. void flush_ready(kit::c_lst &io_ready); - // service-data serialization (log + attr descriptors). All helpers below - // assume mo_svc_mutex is held by the caller. + // service-data serialization (log + attr + module descriptors). All helpers + // below assume mo_svc_mutex is held by the caller. s_p8_svc_buf *svc_acquire_new(); uint8_t *svc_reserve(size_t iz_padded); void serialize_attr_desc(const s_p8_attr_desc *ip_desc); void serialize_log_desc(const struct s_p8_log_desc *ip_desc); + void serialize_log_mod(const s_p8_log_mod *ip_mod); bool mb_initialized = false; std::atomic mu_ref_count { 1 }; @@ -162,6 +196,13 @@ class cp8_core // acquire_buffer to detect pool pressure and wake the worker. std::atomic mu_outstanding_buffers { 0 }; + // Running totals of dropped telemetry elements, fed by poll_dropped_stats() + // on the worker cadence and by each writer's destructor flush. Monotonic; + // read lock-free via get_dropped_stats(). Only logs are wired today. + std::atomic mu_dropped_logs { 0 }; + std::atomic mu_dropped_metrics { 0 }; + std::atomic mu_dropped_traces { 0 }; + // log descriptor registry (global, shared across all TLS cp8_log instances) std::map mo_log_descs; std::mutex mo_log_desc_mutex; @@ -171,6 +212,14 @@ class cp8_core std::unordered_map mo_attr_name_map; mutable std::mutex mo_attr_mutex; + // module registry (global, mutex-protected). The handle handed to callers is + // a pointer to the owned s_p8_log_mod; me_default_verbosity backs the + // null-module (whole-p8) case of set/get_verbosity. + std::vector mo_log_mods; + std::unordered_map mo_log_mod_map; + std::mutex mo_log_mod_mutex; + std::atomic me_default_verbosity { e_p8_trace0 }; + // serialized service data (log + attr descriptors), drained by the worker // thread. The last element is the current in-progress buffer; the earlier // elements are full. Each entry inside a buffer is 8-byte aligned. @@ -242,4 +291,5 @@ size_t p8_test_get_writer_count(); cp8_tls_writer *p8_test_get_writers_head(); std::vector> p8_test_get_service_buffers(); void p8_test_drain_writers(); +s_p8_drop_stats p8_test_get_dropped_stats(); #endif diff --git a/engine/p8_log.cpp b/engine/p8_log.cpp index 3f6247c..75a0e10 100644 --- a/engine/p8_log.cpp +++ b/engine/p8_log.cpp @@ -345,47 +345,12 @@ size_t cp8_log::parse_format_string(struct s_p8_log_varg *op_args, size_t iz_arg return lz_count; } -static thread_local cp8_log go_tls_log; - //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// cp8_log::cp8_log() : cp8_tls_writer(&mo_lock) { } -//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// -void cp8_log::set_verbosity(p_p8_module ip_module, enum e_p8_level ie_verbosity) -{ - (void)ip_module; - (void)ie_verbosity; -} - -//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// -enum e_p8_level cp8_log::get_verbosity(p_p8_module ip_module) -{ - (void)ip_module; - return e_p8_trace0; -} - -//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// -bool cp8_log::register_module(const char *ip_name, enum e_p8_level ie_verbosity, p_p8_module *op_module) -{ - (void)ip_name; - (void)ie_verbosity; - if(op_module) - { - *op_module = P8_MODULE_INVALID_ID; - } - return false; -} - -//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// -p_p8_module cp8_log::find_module(const char *ip_name) -{ - (void)ip_name; - return P8_MODULE_INVALID_ID; -} - //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// bool cp8_log::send(enum e_p8_level ie_level, p_p8_module ip_module, @@ -407,10 +372,24 @@ bool cp8_log::send(enum e_p8_level ie_level, size_t lz_attrs_written = 0; uint8_t lu_attrs_count = 0; size_t lz_frag_commit = 0; + uint16_t lu_mod_id = 0xFFFF; + enum e_p8_level le_min = e_p8_trace0; + + // verbosity gate: reject records below the module's threshold (or the whole-p8 default for a null module) + if(ip_module) + { + le_min = reinterpret_cast(ip_module)->mb_vervosity.load(std::memory_order_relaxed); + lu_mod_id = reinterpret_cast(ip_module)->mu_id; + } + else if(mp_core) + { + le_min = mp_core->get_verbosity(nullptr); + } - // TODO: need to store in s_p8_log_item_hdr - // TODO: need to check verbosity and drop if below min verbosity - (void)ip_module; + if(ie_level < le_min) + { + return false; + } std::lock_guard lo_guard(mo_lock); @@ -436,6 +415,7 @@ bool cp8_log::send(enum e_p8_level ie_level, mp_buffer = mp_core->acquire_buffer(); if(!mp_buffer) [[unlikely]] { + mu_dropped_logs.fetch_add(1, std::memory_order_relaxed); return false; } s_p8_data_buf_hdr *lp_buf_hdr = reinterpret_cast(mp_buffer); @@ -483,8 +463,9 @@ bool cp8_log::send(enum e_p8_level ie_level, lp_hdr->mu_trace_id = iu_trace_id; lp_hdr->mu_thread_id = mu_thread_id; lp_hdr->mu_level = static_cast(ie_level); - lp_hdr->mu_processor = 0; + lp_hdr->mu_processor = 0; // TODO: get current processor CORE lp_hdr->mu_attrs_count = 0; + lp_hdr->mu_mod_id = lu_mod_id; lp_hdr->mu_flags = 0; lp_dst = lp_base + sizeof(struct s_p8_log_item_dat); @@ -696,7 +677,7 @@ bool cp8_log::send(enum e_p8_level ie_level, // the buffers this record added, leaving earlier complete records intact. lp_hdr->mu_args_size = static_cast(lz_args_written); lp_hdr->mu_attrs_count = lu_attrs_count; - lp_hdr->mu_size = static_cast(sizeof(s_p8_log_item_dat) + lz_args_written + lz_attrs_written); + lp_hdr->mu_size = static_cast(sizeof(s_p8_log_item_dat) + lz_args_written + lz_attrs_written); while(mo_fragments.size() > lz_frag_commit) { @@ -704,7 +685,7 @@ bool cp8_log::send(enum e_p8_level ie_level, } mp_buffer = nullptr; mz_buf_used = 0; - ++mu_dropped_records; + mu_dropped_logs.fetch_add(1, std::memory_order_relaxed); return false; } @@ -714,25 +695,48 @@ extern "C" //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// void p8_log_set_verbosity(p_p8_module ip_module, enum e_p8_level ie_verbosity) { - go_tls_log.set_verbosity(ip_module, ie_verbosity); + cp8_core *lp_core = cp8_core::get_global_core(P8_CORE_ACQUIRE_TIMEOUT_MS); + if(lp_core) + { + lp_core->set_verbosity(ip_module, ie_verbosity); + } } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// enum e_p8_level p8_log_get_verbosity(p_p8_module ip_module) { - return go_tls_log.get_verbosity(ip_module); + cp8_core *lp_core = cp8_core::get_global_core(P8_CORE_ACQUIRE_TIMEOUT_MS); + if(lp_core) + { + return lp_core->get_verbosity(ip_module); + } + return e_p8_trace0; } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// bool p8_log_register_module(const char *ip_name, enum e_p8_level ie_verbosity, p_p8_module *op_module) { - return go_tls_log.register_module(ip_name, ie_verbosity, op_module); + cp8_core *lp_core = cp8_core::get_global_core(P8_CORE_ACQUIRE_TIMEOUT_MS); + if(!lp_core) + { + if(op_module) + { + *op_module = P8_MODULE_INVALID_ID; + } + return false; + } + return lp_core->register_module(ip_name, ie_verbosity, op_module); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// p_p8_module p8_log_find_module(const char *ip_name) { - return go_tls_log.find_module(ip_name); + cp8_core *lp_core = cp8_core::get_global_core(P8_CORE_ACQUIRE_TIMEOUT_MS); + if(!lp_core) + { + return P8_MODULE_INVALID_ID; + } + return lp_core->find_module(ip_name); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -747,9 +751,10 @@ extern "C" const char *ip_format, ...) { - va_list lo_args; + thread_local cp8_log lo_tls_log; + va_list lo_args; va_start(lo_args, ip_format); - bool lb_result = go_tls_log.send(ie_level, + bool lb_result = lo_tls_log.send(ie_level, ip_module, iu_trace_id, iu_line, @@ -775,7 +780,8 @@ extern "C" const char *ip_format, va_list *ip_va_list) { - return go_tls_log.send(ie_level, + thread_local cp8_log lo_tls_log; + return lo_tls_log.send(ie_level, ip_module, iu_trace_id, iu_line, @@ -788,11 +794,3 @@ extern "C" } } // extern "C" - -#ifdef P8_TESTING -//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// -uint64_t p8_test_get_tls_dropped_records() -{ - return go_tls_log.get_dropped_records(); -} -#endif diff --git a/engine/p8_log.hpp b/engine/p8_log.hpp index 531ed1e..c9da7e0 100644 --- a/engine/p8_log.hpp +++ b/engine/p8_log.hpp @@ -21,16 +21,18 @@ struct s_p8_log_desc s_p8_log_varg ma_args[P8_LOG_MAX_ARGS]; }; +struct s_p8_log_mod +{ + const char *mp_name; + uint16_t mu_id; + alignas(64) std::atomic mb_vervosity { e_p8_trace0 }; +}; + class cp8_log : public cp8_tls_writer { public: cp8_log(); - void set_verbosity(p_p8_module ip_module, enum e_p8_level ie_verbosity); - enum e_p8_level get_verbosity(p_p8_module ip_module); - - bool register_module(const char *ip_name, enum e_p8_level ie_verbosity, p_p8_module *op_module); - p_p8_module find_module(const char *ip_name); bool send(enum e_p8_level ie_level, p_p8_module ip_module, @@ -45,21 +47,7 @@ class cp8_log : public cp8_tls_writer static size_t parse_format_string(struct s_p8_log_varg *op_args, size_t iz_args_max, const char *ip_format); - // Number of logical records dropped because the buffer pool was exhausted - // mid-serialization. Accessed only under mo_lock. - uint64_t get_dropped_records() const - { - return mu_dropped_records; - } - private: kit::c_spin_lock mo_lock; std::map mo_desc_map; - uint64_t mu_dropped_records = 0; }; - -#ifdef P8_TESTING -// Dropped-record count for the calling thread's TLS log writer. Must be called -// on the same thread that performed the sends. -uint64_t p8_test_get_tls_dropped_records(); -#endif diff --git a/engine/p8_sink_null.cpp b/engine/p8_sink_null.cpp index 5edd8e2..45ea553 100644 --- a/engine/p8_sink_null.cpp +++ b/engine/p8_sink_null.cpp @@ -4,7 +4,6 @@ #include #include #include -#include //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// bool cp8_sink_null::open() diff --git a/engine/p8_static_checks.cpp b/engine/p8_static_checks.cpp index e53549a..05ac584 100644 --- a/engine/p8_static_checks.cpp +++ b/engine/p8_static_checks.cpp @@ -105,9 +105,9 @@ static_assert(offsetof(struct s_p8_log_item_dat, mu_processor) == 29, static_assert(offsetof(struct s_p8_log_item_dat, mu_args_size) == 30, "s_p8_log_item_dat::mu_args_size offset changed"); static_assert(offsetof(struct s_p8_log_item_dat, mu_size) == 32, "s_p8_log_item_dat::mu_size offset changed"); -static_assert(offsetof(struct s_p8_log_item_dat, mu_attrs_count) == 34, +static_assert(offsetof(struct s_p8_log_item_dat, mu_attrs_count) == 36, "s_p8_log_item_dat::mu_attrs_count offset changed"); -static_assert(offsetof(struct s_p8_log_item_dat, mu_flags) == 35, "s_p8_log_item_dat::mu_flags offset changed"); +static_assert(offsetof(struct s_p8_log_item_dat, mu_flags) == 37, "s_p8_log_item_dat::mu_flags offset changed"); //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Attribute value type contract. The public enum e_p8_attr_type (p8_client_api.h) is the source of truth the caller diff --git a/engine/p8_tls_writer.cpp b/engine/p8_tls_writer.cpp index 49b33c8..41c2b41 100644 --- a/engine/p8_tls_writer.cpp +++ b/engine/p8_tls_writer.cpp @@ -27,6 +27,9 @@ cp8_tls_writer::~cp8_tls_writer() if(mp_core) { core_push(); + // Flush any residual drop counts before leaving the registry so a + // short-lived thread's losses are never lost. + mp_core->accumulate_dropped(pull_dropped()); mp_core->unregister_writer(this); mp_core->release(); mp_core = nullptr; @@ -147,6 +150,16 @@ void cp8_tls_writer::pull(kit::c_lst &io_data) } } +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +s_p8_drop_stats cp8_tls_writer::pull_dropped() +{ + s_p8_drop_stats lo_stats; + lo_stats.mu_logs = mu_dropped_logs.exchange(0, std::memory_order_relaxed); + lo_stats.mu_metrics = mu_dropped_metrics.exchange(0, std::memory_order_relaxed); + lo_stats.mu_traces = mu_dropped_traces.exchange(0, std::memory_order_relaxed); + return lo_stats; +} + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// s_p8_attrs_result cp8_tls_writer::serialize_attrs(uint8_t *op_dst, const uint8_t *ip_buf_end, diff --git a/engine/p8_tls_writer.hpp b/engine/p8_tls_writer.hpp index d3f4e26..ab2d1c8 100644 --- a/engine/p8_tls_writer.hpp +++ b/engine/p8_tls_writer.hpp @@ -2,6 +2,7 @@ #include "p8_core.hpp" +#include #include #include "kit/spin_lock.hpp" @@ -38,6 +39,11 @@ class cp8_tls_writer // owning lock is taken internally, so this never races with send(). void pull(kit::c_lst &io_data); + // Atomically read-and-reset the three drop counters and return them. Uses + // atomic exchange (no spin lock), so it never contends with send() on the + // hot path and can run while the writer is live. + s_p8_drop_stats pull_dropped(); + static size_t serialize_utf8_string(uint8_t *op_dst, size_t iz_avail, const char *ip_str); bool copy_fragmented(uint8_t *&io_dst, @@ -70,6 +76,15 @@ class cp8_tls_writer uint32_t mu_thread_id = 0; kit::c_spin_lock *mp_lock = nullptr; + // Per-thread dropped-element counters, incremented on the hot path at each + // discard site (single relaxed atomic add). The core pulls-and-resets them + // via pull_dropped() on its stats cadence and on destruction. Each subclass + // bumps the counter for its own kind; only cp8_log does so today, so metrics + // and traces stay 0 until those become writers. + std::atomic mu_dropped_logs { 0 }; + std::atomic mu_dropped_metrics { 0 }; + std::atomic mu_dropped_traces { 0 }; + cp8_tls_writer *mp_next_writer = nullptr; cp8_tls_writer *mp_prev_writer = nullptr; friend class cp8_core; diff --git a/examples/logs_file_sink/CMakeLists.txt b/examples/logs_file_sink/CMakeLists.txt index a40da55..663474b 100644 --- a/examples/logs_file_sink/CMakeLists.txt +++ b/examples/logs_file_sink/CMakeLists.txt @@ -1,4 +1,4 @@ -cmake_minimum_required(VERSION 3.20...4.0) +cmake_minimum_required(VERSION 4.0) project(P8_Example_LogsFileSink) diff --git a/tests/regression/CMakeLists.txt b/tests/regression/CMakeLists.txt index f5c0832..9db90de 100644 --- a/tests/regression/CMakeLists.txt +++ b/tests/regression/CMakeLists.txt @@ -1,4 +1,4 @@ -cmake_minimum_required(VERSION 3.20...4.0) +cmake_minimum_required(VERSION 4.0) project(P8_RegressionTests) @@ -15,6 +15,20 @@ FetchContent_Declare( set(gtest_force_shared_crt ON CACHE BOOL "" FORCE) FetchContent_MakeAvailable(googletest) +# googletest is third-party; do not subject it to the project's strict +# -Wall -Wextra -pedantic -Werror (or /W4 /WX) flags inherited via the global +# CMAKE_CXX_FLAGS, or its own sources fail to compile (e.g. -Wcharacter-conversion +# on newer Clang). Mirrors the Tracy handling in dep/CMakeLists.txt. +foreach(gtest_target gtest gtest_main gmock gmock_main) + if(TARGET ${gtest_target}) + if(MSVC) + target_compile_options(${gtest_target} PRIVATE /w) + else() + target_compile_options(${gtest_target} PRIVATE -w) + endif() + endif() +endforeach() + add_executable(${PROJECT_NAME} main.cpp p8_attr_test.cpp @@ -26,6 +40,7 @@ add_executable(${PROJECT_NAME} p8_log_format_test.cpp p8_log_fragment_test.cpp p8_log_content_test.cpp + p8_log_module_test.cpp p8_log_perf_test.cpp p8_writer_registry_test.cpp p8_svc_serialize_test.cpp diff --git a/tests/regression/p8_attr_test.cpp b/tests/regression/p8_attr_test.cpp index 2df3716..809231e 100644 --- a/tests/regression/p8_attr_test.cpp +++ b/tests/regression/p8_attr_test.cpp @@ -3,6 +3,7 @@ #include +#include #include #include #include diff --git a/tests/regression/p8_core_test.cpp b/tests/regression/p8_core_test.cpp index 8c77cdf..73625ec 100644 --- a/tests/regression/p8_core_test.cpp +++ b/tests/regression/p8_core_test.cpp @@ -15,6 +15,19 @@ #include #include +namespace +{ +// Build a memory config whose sizes are expressed in whole buffers, so the +// tests stay correct regardless of the compile-time buffer size. +std::string make_mem_config(size_t iz_max_buffers, size_t iz_initial_buffers) +{ + const size_t lz_bufsz = p8_test_get_buffer_size(); + return std::string("{\"") + P8_CFG_KEY_MAX_MEMORY_SIZE + "\": \"" + std::to_string(iz_max_buffers * lz_bufsz) + + "\",\"" + P8_CFG_KEY_INITIAL_MEMORY_SIZE + "\": \"" + std::to_string(iz_initial_buffers * lz_bufsz) + + "\"}"; +} +} // namespace + class c_p8_core_test : public ::testing::Test { protected: @@ -102,24 +115,21 @@ TEST_F(c_p8_core_test, invalid_json_fails) TEST_F(c_p8_core_test, buffer_pool_preallocated) { + const std::string ls_config = make_mem_config(/*max*/ 16, /*initial*/ 8); struct s_p8_config lo_config = {}; - lo_config.mp_json_config = "{" - "\"" P8_CFG_KEY_MAX_MEMORY_SIZE "\": \"128KB\"," - "\"" P8_CFG_KEY_INITIAL_MEMORY_SIZE "\": \"64KB\"" - "}"; + lo_config.mp_json_config = ls_config.c_str(); EXPECT_TRUE(p8_initialize(&lo_config)); - EXPECT_EQ(p8_test_get_buffer_size(), 8192u); + EXPECT_GT(p8_test_get_buffer_size(), 0u); EXPECT_EQ(p8_test_get_free_buffers_count(), 8u); } TEST_F(c_p8_core_test, buffer_pool_initial_clamped_to_max) { + // initial (256 buffers) far exceeds max (2 buffers) → clamped to max + const std::string ls_config = make_mem_config(/*max*/ 2, /*initial*/ 256); struct s_p8_config lo_config = {}; - lo_config.mp_json_config = "{" - "\"" P8_CFG_KEY_MAX_MEMORY_SIZE "\": \"16KB\"," - "\"" P8_CFG_KEY_INITIAL_MEMORY_SIZE "\": \"2MB\"" - "}"; + lo_config.mp_json_config = ls_config.c_str(); EXPECT_TRUE(p8_initialize(&lo_config)); EXPECT_EQ(p8_test_get_free_buffers_count(), 2u); @@ -146,11 +156,10 @@ TEST_F(c_p8_core_test, buffer_acquire_release) TEST_F(c_p8_core_test, buffer_acquire_on_demand) { + // 2 pre-allocated, may grow on demand up to 3 (max) + const std::string ls_config = make_mem_config(/*max*/ 3, /*initial*/ 2); struct s_p8_config lo_config = {}; - lo_config.mp_json_config = "{" - "\"" P8_CFG_KEY_MAX_MEMORY_SIZE "\": \"24KB\"," - "\"" P8_CFG_KEY_INITIAL_MEMORY_SIZE "\": \"16KB\"" - "}"; + lo_config.mp_json_config = ls_config.c_str(); EXPECT_TRUE(p8_initialize(&lo_config)); EXPECT_EQ(p8_test_get_free_buffers_count(), 2u); @@ -174,11 +183,10 @@ TEST_F(c_p8_core_test, buffer_acquire_on_demand) TEST_F(c_p8_core_test, buffer_acquire_on_demand_within_limit) { + // 8 pre-allocated, generous max so on-demand growth stays within the limit + const std::string ls_config = make_mem_config(/*max*/ 128, /*initial*/ 8); struct s_p8_config lo_config = {}; - lo_config.mp_json_config = "{" - "\"" P8_CFG_KEY_MAX_MEMORY_SIZE "\": \"1MB\"," - "\"" P8_CFG_KEY_INITIAL_MEMORY_SIZE "\": \"64KB\"" - "}"; + lo_config.mp_json_config = ls_config.c_str(); EXPECT_TRUE(p8_initialize(&lo_config)); @@ -278,7 +286,10 @@ TEST_F(c_p8_core_test, get_global_core_timeout_no_init) EXPECT_EQ(lp_core, nullptr); EXPECT_GE(lo_elapsed.count(), 50); - EXPECT_LE(lo_elapsed.count(), 200); + // Generous upper bound: the poll loop sleeps in 10 ms steps and sleep_for can + // overshoot heavily on loaded CI runners. This bound only guards against a hang, + // not precise timing (the >= 50 check verifies the timeout is honored). + EXPECT_LE(lo_elapsed.count(), 5000); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// diff --git a/tests/regression/p8_log_content_test.cpp b/tests/regression/p8_log_content_test.cpp index 668e3da..950154e 100644 --- a/tests/regression/p8_log_content_test.cpp +++ b/tests/regression/p8_log_content_test.cpp @@ -6,12 +6,38 @@ #include +#include #include #include #include #include #include +namespace +{ +// Build a memory config sized in whole buffers so the fixture stays correct +// regardless of the compile-time buffer size. +std::string make_mem_config(size_t iz_buffers) +{ + const size_t lz_bytes = iz_buffers * p8_test_get_buffer_size(); + const std::string ls = std::to_string(lz_bytes); + return std::string("{\"") + P8_CFG_KEY_MAX_MEMORY_SIZE + "\": \"" + ls + "\",\"" + P8_CFG_KEY_INITIAL_MEMORY_SIZE + + "\": \"" + ls + "\"}"; +} + +// Largest single-string payload that still spans more than one buffer (when the +// buffer is smaller than the wire cap) while keeping the whole log item within +// the uint16 mu_size field (item header + length prefix + body + padding must +// stay <= UINT16_MAX). The 1 KB headroom covers the item header, extra fixed +// args, and any attribute prefix across the fragmentation tests. Keeps them +// exercising real multi-buffer splits regardless of the compile-time buffer +// size (a 32 KB buffer still fragments a ~64 KB body across buffers). +size_t frag_string_len(size_t iz_buf_sz) +{ + return std::min(iz_buf_sz * 2, static_cast(UINT16_MAX) - 1024); +} +} // namespace + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // helpers: reassemble captured buffers into a linear payload //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -100,20 +126,21 @@ static s_log_content_parsed parse_captured_buffers() return lo_result; } -template static t_val read_val(const std::vector &io_payload, size_t &ioz_offset) +template +static t_val read_val(const std::vector &io_payload, size_t &ioz_offset, size_t iz_increment_bytes) { t_val lo_val = {}; if(ioz_offset + sizeof(t_val) <= io_payload.size()) { memcpy(&lo_val, io_payload.data() + ioz_offset, sizeof(t_val)); } - ioz_offset += sizeof(t_val); + ioz_offset += iz_increment_bytes; return lo_val; } static std::string read_string(const std::vector &io_payload, size_t &ioz_offset) { - uint16_t lu_len = read_val(io_payload, ioz_offset); + uint16_t lu_len = read_val(io_payload, ioz_offset, sizeof(uint16_t)); std::string lo_str; if(lu_len > 0 && ioz_offset + lu_len <= io_payload.size()) { @@ -125,7 +152,7 @@ static std::string read_string(const std::vector &io_payload, size_t &i static uint16_t read_string_len(const std::vector &io_payload, size_t &ioz_offset) { - return read_val(io_payload, ioz_offset); + return read_val(io_payload, ioz_offset, sizeof(uint16_t)); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -242,11 +269,9 @@ class c_log_content_test : public ::testing::Test protected: void SetUp() override { + const std::string ls_config = make_mem_config(/*buffers*/ 16); struct s_p8_config lo_config = {}; - lo_config.mp_json_config = "{" - "\"" P8_CFG_KEY_MAX_MEMORY_SIZE "\": \"128KB\"," - "\"" P8_CFG_KEY_INITIAL_MEMORY_SIZE "\": \"128KB\"" - "}"; + lo_config.mp_json_config = ls_config.c_str(); ASSERT_TRUE(p8_initialize(&lo_config)); p8_test_enable_buffer_capture(); } @@ -631,7 +656,7 @@ TEST_F(c_log_content_test, payload_int32) auto lo_parsed = parse_captured_buffers(); size_t lz_off = 0; - uint64_t lu_val = read_val(lo_parsed.mo_args_payload, lz_off); + uint32_t lu_val = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint32_t)); EXPECT_EQ(lu_val, 42u); } @@ -662,7 +687,7 @@ TEST_F(c_log_content_test, payload_int64) auto lo_parsed = parse_captured_buffers(); size_t lz_off = 0; - uint64_t lu_val = read_val(lo_parsed.mo_args_payload, lz_off); + uint64_t lu_val = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint64_t)); EXPECT_EQ(lu_val, static_cast(li_expected)); } @@ -691,7 +716,7 @@ TEST_F(c_log_content_test, payload_double) auto lo_parsed = parse_captured_buffers(); size_t lz_off = 0; - double ld_val = read_val(lo_parsed.mo_args_payload, lz_off); + double ld_val = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(double)); EXPECT_EQ(memcmp(&ld_val, &ld_expected, sizeof(double)), 0); } @@ -800,13 +825,13 @@ TEST_F(c_log_content_test, payload_multi_args) auto lo_parsed = parse_captured_buffers(); size_t lz_off = 0; - uint64_t lu_int1 = read_val(lo_parsed.mo_args_payload, lz_off); + uint32_t lu_int1 = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint32_t)); EXPECT_EQ(lu_int1, 42u); std::string lo_str = read_string(lo_parsed.mo_args_payload, lz_off); EXPECT_EQ(lo_str, "test"); - uint64_t lu_int2 = read_val(lo_parsed.mo_args_payload, lz_off); + uint64_t lu_int2 = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint64_t)); EXPECT_EQ(lu_int2, 99u); } @@ -835,7 +860,7 @@ TEST_F(c_log_content_test, payload_pointer) auto lo_parsed = parse_captured_buffers(); size_t lz_off = 0; - uint64_t lu_val = read_val(lo_parsed.mo_args_payload, lz_off); + uint64_t lu_val = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(void *)); EXPECT_EQ(lu_val, static_cast(reinterpret_cast(lp_addr))); } @@ -886,11 +911,12 @@ TEST_F(c_log_content_test, fragment_flag_on_continuation) TEST_F(c_log_content_test, fragment_string_content) { - size_t lz_buf_sz = p8_test_get_buffer_size(); + size_t lz_buf_sz = p8_test_get_buffer_size(); - std::string lo_pattern; - lo_pattern.reserve(lz_buf_sz * 2); - for(size_t lz_i = 0; lz_i < lz_buf_sz * 2; ++lz_i) + const size_t lz_len = frag_string_len(lz_buf_sz); + std::string lo_pattern; + lo_pattern.reserve(lz_len); + for(size_t lz_i = 0; lz_i < lz_len; ++lz_i) { lo_pattern.push_back(static_cast('A' + (lz_i % 26))); } @@ -923,9 +949,14 @@ TEST_F(c_log_content_test, fragment_string_content) TEST_F(c_log_content_test, fragment_args_size_spans_buffers) { - size_t lz_buf_sz = p8_test_get_buffer_size(); + size_t lz_buf_sz = p8_test_get_buffer_size(); - std::string lo_large(lz_buf_sz * 2, 'X'); + // A single wire string is length-prefixed with a uint16 and mu_args_size is + // itself a uint16, so cap the payload so (prefix + body) stays representable. + // Whenever the buffer is smaller than this cap the string still spans >1 + // buffer; when it is larger the record simply stays within one buffer. + const size_t lz_len = std::min(lz_buf_sz * 2, static_cast(UINT16_MAX) - sizeof(uint16_t)); + std::string lo_large(lz_len, 'X'); auto lo_ctx = run_send_in_thread( [&lo_large]() @@ -991,7 +1022,7 @@ TEST_F(c_log_content_test, fragment_fixed_then_string) { size_t lz_buf_sz = p8_test_get_buffer_size(); - std::string lo_large(lz_buf_sz * 2, 'C'); + std::string lo_large(frag_string_len(lz_buf_sz), 'C'); auto lo_ctx = run_send_in_thread( [&lo_large]() @@ -1015,7 +1046,7 @@ TEST_F(c_log_content_test, fragment_fixed_then_string) auto lo_parsed = parse_captured_buffers(); size_t lz_off = 0; - uint64_t lu_int = read_val(lo_parsed.mo_args_payload, lz_off); + uint32_t lu_int = (uint32_t)read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint32_t)); EXPECT_EQ(lu_int, 777u); std::string lo_str = read_string(lo_parsed.mo_args_payload, lz_off); @@ -1025,8 +1056,9 @@ TEST_F(c_log_content_test, fragment_fixed_then_string) TEST_F(c_log_content_test, fragment_many_buffers) { size_t lz_buf_sz = p8_test_get_buffer_size(); - size_t lz_str_sz = lz_buf_sz * 7; - ASSERT_LE(lz_str_sz, static_cast(UINT16_MAX)); + // A single wire string is capped at UINT16_MAX, so it can only span as many + // buffers as that cap allows for the current buffer size. + size_t lz_str_sz = std::min(lz_buf_sz * 7, static_cast(UINT16_MAX)); std::string lo_huge; lo_huge.reserve(lz_str_sz); @@ -1060,7 +1092,11 @@ TEST_F(c_log_content_test, fragment_many_buffers) EXPECT_EQ(lo_result.size(), lo_huge.size()); EXPECT_EQ(lo_result, lo_huge); - EXPECT_GE(lo_parsed.mo_all_captured.size(), 7u); + // Each buffer holds less than a full buffer of payload (headers take space), + // so the body occupies at least this many buffers. Derived from the buffer + // size so the expectation tracks the compile-time geometry (7 at 8 KB). + const size_t lz_expected_bufs = std::max(1, lz_str_sz / lz_buf_sz); + EXPECT_GE(lo_parsed.mo_all_captured.size(), lz_expected_bufs); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// @@ -1102,10 +1138,10 @@ TEST_F(c_log_content_test, attr_numeric_i64) size_t lz_off = lz_args_end; - p8_attr_id li_id = read_val(lo_parsed.mo_args_payload, lz_off); + p8_attr_id li_id = read_val(lo_parsed.mo_args_payload, lz_off, sizeof(p8_attr_id)); EXPECT_EQ(li_id, li_attr); - uint64_t lu_val = read_val(lo_parsed.mo_args_payload, lz_off); + uint64_t lu_val = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint64_t)); int64_t li_val = 0; memcpy(&li_val, &lu_val, sizeof(int64_t)); EXPECT_EQ(li_val, 12345LL); @@ -1144,7 +1180,7 @@ TEST_F(c_log_content_test, attr_numeric_f64) lz_off += sizeof(p8_attr_id); double ld_expected = 2.718; - uint64_t lu_raw = read_val(lo_parsed.mo_args_payload, lz_off); + uint64_t lu_raw = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint64_t)); double ld_actual = 0; memcpy(&ld_actual, &lu_raw, sizeof(double)); EXPECT_EQ(memcmp(&ld_actual, &ld_expected, sizeof(double)), 0); @@ -1181,7 +1217,7 @@ TEST_F(c_log_content_test, attr_string) size_t lz_off = lo_parsed.mo_item_hdr.mu_args_size; EXPECT_EQ(lo_parsed.mo_item_hdr.mu_attrs_count, 1u); - p8_attr_id li_id = read_val(lo_parsed.mo_args_payload, lz_off); + p8_attr_id li_id = read_val(lo_parsed.mo_args_payload, lz_off, sizeof(p8_attr_id)); EXPECT_EQ(li_id, li_attr); std::string lo_str = read_string(lo_parsed.mo_args_payload, lz_off); @@ -1228,17 +1264,17 @@ TEST_F(c_log_content_test, attr_count_multiple) size_t lz_off = lo_parsed.mo_item_hdr.mu_args_size; - p8_attr_id li_id1 = read_val(lo_parsed.mo_args_payload, lz_off); + p8_attr_id li_id1 = read_val(lo_parsed.mo_args_payload, lz_off, sizeof(p8_attr_id)); EXPECT_EQ(li_id1, li_a1); - uint64_t lu_v1 = read_val(lo_parsed.mo_args_payload, lz_off); + uint64_t lu_v1 = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint64_t)); EXPECT_EQ(static_cast(lu_v1), 100); - p8_attr_id li_id2 = read_val(lo_parsed.mo_args_payload, lz_off); + p8_attr_id li_id2 = read_val(lo_parsed.mo_args_payload, lz_off, sizeof(p8_attr_id)); EXPECT_EQ(li_id2, li_a2); - uint64_t lu_v2 = read_val(lo_parsed.mo_args_payload, lz_off); + uint64_t lu_v2 = read_val(lo_parsed.mo_args_payload, lz_off, P8_SIZE_OF_ARG(uint64_t)); EXPECT_EQ(lu_v2, 200u); - p8_attr_id li_id3 = read_val(lo_parsed.mo_args_payload, lz_off); + p8_attr_id li_id3 = read_val(lo_parsed.mo_args_payload, lz_off, sizeof(p8_attr_id)); EXPECT_EQ(li_id3, li_a3); std::string lo_str = read_string(lo_parsed.mo_args_payload, lz_off); EXPECT_EQ(lo_str, "attr3"); @@ -1289,7 +1325,7 @@ TEST_F(c_log_content_test, attr_string_fragments) p8_attr_id li_attr = p8_attr_register("big_attr_str", e_p8_attr_str); ASSERT_TRUE(P8_IS_ATTR_VALID(li_attr)); - std::string lo_big_val(lz_buf_sz * 2, 'V'); + std::string lo_big_val(frag_string_len(lz_buf_sz), 'V'); struct s_p8_attr_val lo_av = {}; lo_av.m_id = li_attr; @@ -1317,7 +1353,7 @@ TEST_F(c_log_content_test, attr_string_fragments) size_t lz_off = lo_parsed.mo_item_hdr.mu_args_size; EXPECT_EQ(lo_parsed.mo_item_hdr.mu_attrs_count, 1u); - p8_attr_id li_id = read_val(lo_parsed.mo_args_payload, lz_off); + p8_attr_id li_id = read_val(lo_parsed.mo_args_payload, lz_off, sizeof(p8_attr_id)); EXPECT_EQ(li_id, li_attr); std::string lo_result = read_string(lo_parsed.mo_args_payload, lz_off); @@ -1453,7 +1489,7 @@ TEST_F(c_log_content_test, multi_send_three_items) expected_item_size(sizeof(s_p8_log_item_dat) + lo_items[0].mo_hdr.mu_args_size)); { size_t lz_off = 0; - uint64_t lu_val = read_val(lo_items[0].mo_payload, lz_off); + uint32_t lu_val = read_val(lo_items[0].mo_payload, lz_off, P8_SIZE_OF_ARG(uint32_t)); EXPECT_EQ(lu_val, 100u); } @@ -1475,10 +1511,10 @@ TEST_F(c_log_content_test, multi_send_three_items) EXPECT_EQ(lo_items[2].mo_hdr.mu_attrs_count, 0u); { size_t lz_off = 0; - uint64_t lu_val = read_val(lo_items[2].mo_payload, lz_off); + uint32_t lu_val = read_val(lo_items[2].mo_payload, lz_off, P8_SIZE_OF_ARG(uint32_t)); EXPECT_EQ(lu_val, 42u); - double ld_val = read_val(lo_items[2].mo_payload, lz_off); + double ld_val = read_val(lo_items[2].mo_payload, lz_off, P8_SIZE_OF_ARG(double)); double ld_expected = 3.14; EXPECT_EQ(memcmp(&ld_val, &ld_expected, sizeof(double)), 0); } @@ -1538,7 +1574,7 @@ TEST_F(c_log_content_test, item_size_8_byte_aligned) TEST_F(c_log_content_test, multi_send_with_fragmentation) { size_t lz_buf_sz = p8_test_get_buffer_size(); - std::string lo_large(lz_buf_sz * 2, 'X'); + std::string lo_large(frag_string_len(lz_buf_sz), 'X'); bool lb_r1 = false, lb_r2 = false, lb_r3 = false; @@ -1589,7 +1625,7 @@ TEST_F(c_log_content_test, multi_send_with_fragmentation) EXPECT_EQ(lo_items[0].mo_hdr.mu_trace_id, 10u); { size_t lz_off = 0; - uint64_t lu_val = read_val(lo_items[0].mo_payload, lz_off); + uint32_t lu_val = read_val(lo_items[0].mo_payload, lz_off, P8_SIZE_OF_ARG(uint32_t)); EXPECT_EQ(lu_val, 1u); } @@ -1606,7 +1642,7 @@ TEST_F(c_log_content_test, multi_send_with_fragmentation) EXPECT_EQ(lo_items[2].mo_hdr.mu_trace_id, 30u); { size_t lz_off = 0; - uint64_t lu_val = read_val(lo_items[2].mo_payload, lz_off); + uint32_t lu_val = read_val(lo_items[2].mo_payload, lz_off, P8_SIZE_OF_ARG(uint32_t)); EXPECT_EQ(lu_val, 3u); } @@ -1668,13 +1704,13 @@ TEST_F(c_log_content_test, multi_send_with_attrs) EXPECT_EQ(lo_items[0].mo_hdr.mu_attrs_count, 1u); { size_t lz_off = 0; - uint64_t lu_val = read_val(lo_items[0].mo_payload, lz_off); + uint32_t lu_val = read_val(lo_items[0].mo_payload, lz_off, P8_SIZE_OF_ARG(uint32_t)); EXPECT_EQ(lu_val, 42u); lz_off = lo_items[0].mo_hdr.mu_args_size; - p8_attr_id li_id = read_val(lo_items[0].mo_payload, lz_off); + p8_attr_id li_id = read_val(lo_items[0].mo_payload, lz_off, sizeof(p8_attr_id)); EXPECT_EQ(li_id, li_attr_int); - uint64_t lu_attr = read_val(lo_items[0].mo_payload, lz_off); + uint64_t lu_attr = read_val(lo_items[0].mo_payload, lz_off, P8_SIZE_OF_ARG(uint64_t)); EXPECT_EQ(static_cast(lu_attr), 777); } @@ -1687,9 +1723,108 @@ TEST_F(c_log_content_test, multi_send_with_attrs) EXPECT_EQ(lo_str, "test"); lz_off = lo_items[1].mo_hdr.mu_args_size; - p8_attr_id li_id = read_val(lo_items[1].mo_payload, lz_off); + p8_attr_id li_id = read_val(lo_items[1].mo_payload, lz_off, sizeof(p8_attr_id)); EXPECT_EQ(li_id, li_attr_str); std::string lo_attr = read_string(lo_items[1].mo_payload, lz_off); EXPECT_EQ(lo_attr, "label_value"); } } + +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +// Group: verbosity gate in cp8_log::send() +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + +TEST_F(c_log_content_test, verbosity_drops_below_module_threshold) +{ + p_p8_module lp_mod = P8_MODULE_INVALID_ID; + ASSERT_TRUE(p8_log_register_module("gate", e_p8_error0, &lp_mod)); + ASSERT_NE(lp_mod, P8_MODULE_INVALID_ID); + + // below the module threshold: rejected before any buffer is touched + auto lo_below = run_send_in_thread( + [lp_mod]() + { + return p8_log_sent(e_p8_trace0, + lp_mod, + 0, + static_cast(__LINE__), + __FILE__, + __FUNCTION__, + 0, + nullptr, + "%d", + 1); + }, + __LINE__, + __FILE__); + EXPECT_FALSE(lo_below.mb_result); + EXPECT_TRUE(parse_captured_buffers().mo_all_captured.empty()); + + p8_test_clear_captured_buffers(); + + // at the module threshold: accepted and emitted + auto lo_at = run_send_in_thread( + [lp_mod]() + { + return p8_log_sent(e_p8_error0, + lp_mod, + 0, + static_cast(__LINE__), + __FILE__, + __FUNCTION__, + 0, + nullptr, + "%d", + 2); + }, + __LINE__, + __FILE__); + EXPECT_TRUE(lo_at.mb_result); + EXPECT_FALSE(parse_captured_buffers().mo_all_captured.empty()); +} + +TEST_F(c_log_content_test, verbosity_default_gates_null_module) +{ + // a null module is gated by the whole-p8 default threshold + p8_log_set_verbosity(nullptr, e_p8_error0); + + auto lo_below = run_send_in_thread( + []() + { + return p8_log_sent(e_p8_trace0, + nullptr, + 0, + static_cast(__LINE__), + __FILE__, + __FUNCTION__, + 0, + nullptr, + "%d", + 1); + }, + __LINE__, + __FILE__); + EXPECT_FALSE(lo_below.mb_result); + EXPECT_TRUE(parse_captured_buffers().mo_all_captured.empty()); + + p8_test_clear_captured_buffers(); + + auto lo_at = run_send_in_thread( + []() + { + return p8_log_sent(e_p8_critical0, + nullptr, + 0, + static_cast(__LINE__), + __FILE__, + __FUNCTION__, + 0, + nullptr, + "%d", + 2); + }, + __LINE__, + __FILE__); + EXPECT_TRUE(lo_at.mb_result); + EXPECT_FALSE(parse_captured_buffers().mo_all_captured.empty()); +} diff --git a/tests/regression/p8_log_fragment_test.cpp b/tests/regression/p8_log_fragment_test.cpp index 9af9f0e..28f17ed 100644 --- a/tests/regression/p8_log_fragment_test.cpp +++ b/tests/regression/p8_log_fragment_test.cpp @@ -11,16 +11,27 @@ #include #include +namespace +{ +// Build a memory config sized in whole buffers so the fixture stays correct +// regardless of the compile-time buffer size. +std::string make_mem_config(size_t iz_buffers) +{ + const size_t lz_bytes = iz_buffers * p8_test_get_buffer_size(); + const std::string ls = std::to_string(lz_bytes); + return std::string("{\"") + P8_CFG_KEY_MAX_MEMORY_SIZE + "\": \"" + ls + "\",\"" + P8_CFG_KEY_INITIAL_MEMORY_SIZE + + "\": \"" + ls + "\"}"; +} +} // namespace + class c_log_fragment_test : public ::testing::Test { protected: void SetUp() override { + const std::string ls_config = make_mem_config(/*buffers*/ 16); struct s_p8_config lo_config = {}; - lo_config.mp_json_config = "{" - "\"" P8_CFG_KEY_MAX_MEMORY_SIZE "\": \"128KB\"," - "\"" P8_CFG_KEY_INITIAL_MEMORY_SIZE "\": \"128KB\"" - "}"; + lo_config.mp_json_config = ls_config.c_str(); ASSERT_TRUE(p8_initialize(&lo_config)); } @@ -169,39 +180,37 @@ TEST_F(c_log_fragment_test, discard_when_pool_exhausted) { p8_release(); + // single-buffer pool: a record that needs a second buffer must be discarded + const std::string ls_config = make_mem_config(/*buffers*/ 1); struct s_p8_config lo_config = {}; - lo_config.mp_json_config = "{" - "\"" P8_CFG_KEY_MAX_MEMORY_SIZE "\": \"8KB\"," - "\"" P8_CFG_KEY_INITIAL_MEMORY_SIZE "\": \"8KB\"" - "}"; + lo_config.mp_json_config = ls_config.c_str(); ASSERT_TRUE(p8_initialize(&lo_config)); ASSERT_EQ(p8_test_get_free_buffers_count(), 1u); - bool lb_result = true; - uint64_t lu_dropped = 0; + bool lb_result = true; std::thread lo_thread( - [&lb_result, &lu_dropped]() + [&lb_result]() { size_t lz_buf_sz = p8_test_get_buffer_size(); std::string lo_huge(lz_buf_sz * 2, 'D'); - lb_result = p8_log_sent(e_p8_trace0, - nullptr, - 0, - static_cast(__LINE__), - __FILE__, - __FUNCTION__, - 0, - nullptr, - "%s", - lo_huge.c_str()); - lu_dropped = p8_test_get_tls_dropped_records(); + lb_result = p8_log_sent(e_p8_trace0, + nullptr, + 0, + static_cast(__LINE__), + __FILE__, + __FUNCTION__, + 0, + nullptr, + "%s", + lo_huge.c_str()); }); lo_thread.join(); EXPECT_FALSE(lb_result); - // the discarded record must be accounted for in the per-writer counter - EXPECT_EQ(lu_dropped, 1u); + // the discarded record is now visible through the core loss statistics, + // flushed there when the producer thread's writer was destroyed on join + EXPECT_EQ(p8_test_get_dropped_stats().mu_logs, 1u); } //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// diff --git a/tests/regression/p8_log_module_test.cpp b/tests/regression/p8_log_module_test.cpp new file mode 100644 index 0000000..118bbc1 --- /dev/null +++ b/tests/regression/p8_log_module_test.cpp @@ -0,0 +1,102 @@ +#include "p8_client_api.h" + +#include + +// Registry-level tests for the log-module API (register/find + verbosity +// get/set). These exercise only the global module registry in cp8_core and do +// not emit any records, so a bare "{}" config is enough to bring the core up. +class c_log_module_test : public ::testing::Test +{ +protected: + void SetUp() override + { + struct s_p8_config lo_config = {}; + lo_config.mp_json_config = "{}"; + ASSERT_TRUE(p8_initialize(&lo_config)); + } + + void TearDown() override + { + p8_release(); + } +}; + +TEST_F(c_log_module_test, register_new_returns_handle) +{ + p_p8_module lp_mod = P8_MODULE_INVALID_ID; + EXPECT_TRUE(p8_log_register_module("mod_a", e_p8_info0, &lp_mod)); + EXPECT_NE(lp_mod, P8_MODULE_INVALID_ID); +} + +TEST_F(c_log_module_test, register_stores_requested_verbosity) +{ + p_p8_module lp_mod = P8_MODULE_INVALID_ID; + ASSERT_TRUE(p8_log_register_module("mod_v", e_p8_warning0, &lp_mod)); + EXPECT_EQ(p8_log_get_verbosity(lp_mod), e_p8_warning0); +} + +TEST_F(c_log_module_test, register_duplicate_returns_same_handle) +{ + p_p8_module lp_first = P8_MODULE_INVALID_ID; + p_p8_module lp_second = P8_MODULE_INVALID_ID; + ASSERT_TRUE(p8_log_register_module("dup", e_p8_info0, &lp_first)); + ASSERT_TRUE(p8_log_register_module("dup", e_p8_info0, &lp_second)); + EXPECT_EQ(lp_first, lp_second); +} + +TEST_F(c_log_module_test, register_duplicate_updates_verbosity) +{ + p_p8_module lp_first = P8_MODULE_INVALID_ID; + p_p8_module lp_second = P8_MODULE_INVALID_ID; + ASSERT_TRUE(p8_log_register_module("dup_v", e_p8_info0, &lp_first)); + ASSERT_TRUE(p8_log_register_module("dup_v", e_p8_error0, &lp_second)); + EXPECT_EQ(lp_first, lp_second); + EXPECT_EQ(p8_log_get_verbosity(lp_first), e_p8_error0); +} + +TEST_F(c_log_module_test, find_existing_returns_registered_handle) +{ + p_p8_module lp_reg = P8_MODULE_INVALID_ID; + ASSERT_TRUE(p8_log_register_module("findme", e_p8_debug0, &lp_reg)); + EXPECT_EQ(p8_log_find_module("findme"), lp_reg); +} + +TEST_F(c_log_module_test, find_missing_returns_invalid) +{ + EXPECT_EQ(p8_log_find_module("does_not_exist"), P8_MODULE_INVALID_ID); +} + +TEST_F(c_log_module_test, set_get_verbosity_on_module) +{ + p_p8_module lp_mod = P8_MODULE_INVALID_ID; + ASSERT_TRUE(p8_log_register_module("setget", e_p8_trace0, &lp_mod)); + p8_log_set_verbosity(lp_mod, e_p8_warning0); + EXPECT_EQ(p8_log_get_verbosity(lp_mod), e_p8_warning0); +} + +TEST_F(c_log_module_test, set_get_default_verbosity_null_module) +{ + p8_log_set_verbosity(nullptr, e_p8_error0); + EXPECT_EQ(p8_log_get_verbosity(nullptr), e_p8_error0); +} + +TEST_F(c_log_module_test, register_null_name_fails_and_clears_out) +{ + int li_dummy = 0; + p_p8_module lp_mod = &li_dummy; + EXPECT_FALSE(p8_log_register_module(nullptr, e_p8_info0, &lp_mod)); + EXPECT_EQ(lp_mod, P8_MODULE_INVALID_ID); +} + +TEST_F(c_log_module_test, register_empty_name_fails_and_clears_out) +{ + int li_dummy = 0; + p_p8_module lp_mod = &li_dummy; + EXPECT_FALSE(p8_log_register_module("", e_p8_info0, &lp_mod)); + EXPECT_EQ(lp_mod, P8_MODULE_INVALID_ID); +} + +TEST_F(c_log_module_test, register_null_out_fails) +{ + EXPECT_FALSE(p8_log_register_module("noout", e_p8_info0, nullptr)); +} diff --git a/tests/regression/p8_memory_budget_test.cpp b/tests/regression/p8_memory_budget_test.cpp index 28a2508..2d442a3 100644 --- a/tests/regression/p8_memory_budget_test.cpp +++ b/tests/regression/p8_memory_budget_test.cpp @@ -163,12 +163,60 @@ TEST_F(c_p8_memory_budget_test, concurrent_reserve_release_stays_within_limit) EXPECT_EQ(lu_concurrent_holders.load(), 0u); EXPECT_LE(lu_max_holders.load(), lz_max_slots); - // sanity: under heavy contention with capped slots, both paths must be exercised + // sanity: the stress loop must have exercised the success path size_t lz_success = lu_total_success.load(); size_t lz_failure = lu_total_failure.load(); EXPECT_EQ(lz_success + lz_failure, lz_thread_count * lz_iterations); EXPECT_GT(lz_success, 0u) << "no successful reservations — test degenerate"; - EXPECT_GT(lz_failure, 0u) << "no failed reservations — contention too low to validate try_reserve"; + + // Deterministic rejection burst: every thread attempts a single reserve at + // once and holds it until all threads have attempted, so the budget stays + // saturated for the whole burst. With more threads than slots, exactly + // lz_max_slots reservations succeed and the rest MUST be rejected — this + // validates the try_reserve failure path without relying on scheduling + // timing (the loop above cannot guarantee overlap on a fast machine). + std::atomic lu_burst_success { 0 }; + std::atomic lu_burst_failure { 0 }; + std::latch lo_burst_latch(lz_thread_count); + std::latch lo_attempt_done(lz_thread_count); + std::vector lo_burst_threads; + lo_burst_threads.reserve(lz_thread_count); + + for(size_t lz_i = 0; lz_i < lz_thread_count; ++lz_i) + { + lo_burst_threads.emplace_back( + [&]() + { + lo_burst_latch.arrive_and_wait(); + bool lb_reserved = lo_budget.try_reserve(lz_chunk_size); + if(lb_reserved) + { + lu_burst_success.fetch_add(1, std::memory_order_relaxed); + } + else + { + lu_burst_failure.fetch_add(1, std::memory_order_relaxed); + } + + // hold the slot until every thread has attempted, keeping the + // budget saturated so rejections are guaranteed + lo_attempt_done.arrive_and_wait(); + if(lb_reserved) + { + lo_budget.release(lz_chunk_size); + } + }); + } + + for(auto &lo_t : lo_burst_threads) + { + lo_t.join(); + } + + EXPECT_EQ(lu_burst_success.load(), lz_max_slots) << "budget admitted the wrong number of concurrent reservations"; + EXPECT_EQ(lu_burst_failure.load(), lz_thread_count - lz_max_slots) + << "try_reserve failed to reject over-capacity requests"; + EXPECT_EQ(lo_budget.get_used(), 0u) << "leak after burst reserve/release"; } TEST_F(c_p8_memory_budget_test, concurrent_reserve_release_exact_accounting) diff --git a/tests/regression/p8_pull_test.cpp b/tests/regression/p8_pull_test.cpp index 3c30c78..5d1fa8e 100644 --- a/tests/regression/p8_pull_test.cpp +++ b/tests/regression/p8_pull_test.cpp @@ -12,6 +12,19 @@ #include #include +namespace +{ +// Build a memory config sized in whole buffers so the tests stay correct +// regardless of the compile-time buffer size. +std::string make_mem_config(size_t iz_buffers) +{ + const size_t lz_bytes = iz_buffers * p8_test_get_buffer_size(); + const std::string ls = std::to_string(lz_bytes); + return std::string("{\"") + P8_CFG_KEY_MAX_MEMORY_SIZE + "\": \"" + ls + "\",\"" + P8_CFG_KEY_INITIAL_MEMORY_SIZE + + "\": \"" + ls + "\"}"; +} +} // namespace + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Tests for the pull-based drain architecture: writers retain filled buffers // locally and the core pulls them via cp8_core::drain_writers (exercised here @@ -140,21 +153,21 @@ TEST_F(c_pull_test, drain_pulls_from_multiple_live_writers) TEST_F(c_pull_test, discard_preserves_earlier_records) { - // 4-buffer pool: two records fill+park two buffers, the third record - // exhausts the remaining pool mid-serialization and must be discarded. - init_core("{" - "\"" P8_CFG_KEY_MAX_MEMORY_SIZE "\": \"32KB\"," - "\"" P8_CFG_KEY_INITIAL_MEMORY_SIZE "\": \"32KB\"" - "}"); + // 3-buffer pool: two records fill+park two buffers, the third record + // exhausts the single remaining buffer mid-serialization and must be + // discarded. A single wire string is capped at UINT16_MAX, so a record can + // span at most ~2 buffers when the buffer is large; a 3-buffer pool keeps + // the exhaustion reachable for any compile-time buffer size. + const std::string ls_config = make_mem_config(/*buffers*/ 3); + init_core(ls_config.c_str()); p8_test_enable_buffer_capture(); - ASSERT_EQ(p8_test_get_free_buffers_count(), 4u); + ASSERT_EQ(p8_test_get_free_buffers_count(), 3u); - bool lb_r3 = true; - uint64_t lu_dropped = 0; - size_t lz_capture = 0; + bool lb_r3 = true; + size_t lz_capture = 0; std::thread lo_thread( - [&lb_r3, &lu_dropped, &lz_capture]() + [&lb_r3, &lz_capture]() { const size_t lz_buf_sz = p8_test_get_buffer_size(); // size the string so the record all but fills its buffer, leaving @@ -189,18 +202,16 @@ TEST_F(c_pull_test, discard_preserves_earlier_records) // R3 -> huge, exhausts the remaining pool and is discarded std::string lo_huge(lz_buf_sz * 3, 'Z'); - lb_r3 = p8_log_sent(e_p8_trace0, - nullptr, - 0, - static_cast(__LINE__), - __FILE__, - __FUNCTION__, - 0, - nullptr, - "%s", - lo_huge.c_str()); - - lu_dropped = p8_test_get_tls_dropped_records(); + lb_r3 = p8_log_sent(e_p8_trace0, + nullptr, + 0, + static_cast(__LINE__), + __FILE__, + __FUNCTION__, + 0, + nullptr, + "%s", + lo_huge.c_str()); // pull the survivors from this still-live writer p8_test_drain_writers(); @@ -209,7 +220,9 @@ TEST_F(c_pull_test, discard_preserves_earlier_records) lo_thread.join(); EXPECT_FALSE(lb_r3); - EXPECT_EQ(lu_dropped, 1u); + // the discarded record is now visible through the core loss statistics, + // flushed there when the producer thread's writer was destroyed on join + EXPECT_EQ(p8_test_get_dropped_stats().mu_logs, 1u); // the two earlier complete records (buffers A and B) were not rolled back EXPECT_GE(lz_capture, 2u); } diff --git a/tests/regression/p8_svc_serialize_test.cpp b/tests/regression/p8_svc_serialize_test.cpp index 718ff46..5adb89a 100644 --- a/tests/regression/p8_svc_serialize_test.cpp +++ b/tests/regression/p8_svc_serialize_test.cpp @@ -297,3 +297,65 @@ TEST_F(c_p8_svc_serialize_test, multiple_entries_are_each_8_byte_aligned) EXPECT_EQ(lr_entry.mo_bytes.size() % 8u, 0u); } } + +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// +// module descriptor serialization +//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + +TEST_F(c_p8_svc_serialize_test, register_module_serializes_entry) +{ + p_p8_module lp_mod = P8_MODULE_INVALID_ID; + ASSERT_TRUE(p8_log_register_module("net", e_p8_warning0, &lp_mod)); + ASSERT_NE(lp_mod, P8_MODULE_INVALID_ID); + + auto lo_entries = collect_service_entries(); + + int li_found = 0; + for(const auto &lr_entry : lo_entries) + { + if(lr_entry.mu_type != P8_SVC_TYPE_MODULE) + { + continue; + } + + s_p8_log_mod_svc lo_mod = {}; + ASSERT_GE(lr_entry.mo_bytes.size(), sizeof(s_p8_log_mod_svc)); + memcpy(&lo_mod, lr_entry.mo_bytes.data(), sizeof(lo_mod)); + + if(read_cstr(lr_entry.mo_bytes, sizeof(s_p8_log_mod_svc)) != "net") + { + continue; + } + + ++li_found; + + EXPECT_EQ(lo_mod.ms_hdr.mu_size % 8u, 0u); + EXPECT_EQ(lo_mod.ms_hdr.mu_size, lr_entry.mo_bytes.size()); + EXPECT_EQ(lo_mod.mu_verb, static_cast(e_p8_warning0)); + } + + EXPECT_EQ(li_found, 1); +} + +TEST_F(c_p8_svc_serialize_test, register_duplicate_module_serializes_once) +{ + p_p8_module lp_a = P8_MODULE_INVALID_ID; + p_p8_module lp_b = P8_MODULE_INVALID_ID; + ASSERT_TRUE(p8_log_register_module("dup_mod", e_p8_info0, &lp_a)); + ASSERT_TRUE(p8_log_register_module("dup_mod", e_p8_error0, &lp_b)); + EXPECT_EQ(lp_a, lp_b); + + auto lo_entries = collect_service_entries(); + + int li_count = 0; + for(const auto &lr_entry : lo_entries) + { + if(lr_entry.mu_type == P8_SVC_TYPE_MODULE + && read_cstr(lr_entry.mo_bytes, sizeof(s_p8_log_mod_svc)) == "dup_mod") + { + ++li_count; + } + } + + EXPECT_EQ(li_count, 1); +}