diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0c20ff88..0b5dd776 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,6 +55,7 @@ jobs: - client-sdk-rust - client-sdk-rust/** - cmake/** + - scripts/memory-tester.sh - scripts/set-test-tokens.sh - scripts/run-with-backtrace.sh - CMakeLists.txt diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index efeaf448..1060eeb2 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -357,7 +357,11 @@ jobs: --gtest_output="xml:${{ env.BUILD_DIR }}\unit-test-results.xml" - name: Start livekit-server - if: matrix.e2e-testing && (inputs.integration_repeat > 0 || inputs.run_stress_tests) + if: >- + matrix.e2e-testing && + (inputs.integration_repeat > 0 || + inputs.run_stress_tests || + (matrix.name == 'ubuntu-24.04-x64' && inputs.build_type == 'release')) id: livekit_server uses: livekit/dev-server-action@5d4d5337a875e2d1afd37bed03c601d159dab002 # v1.1.1 with: @@ -368,7 +372,11 @@ jobs: # Needed by token helper script - name: Install livekit-cli - if: matrix.e2e-testing && (inputs.integration_repeat > 0 || inputs.run_stress_tests) + if: >- + matrix.e2e-testing && + (inputs.integration_repeat > 0 || + inputs.run_stress_tests || + (matrix.name == 'ubuntu-24.04-x64' && inputs.build_type == 'release')) shell: bash env: # Windows installs lk via `gh api` / `gh release download`, which need this env var @@ -425,6 +433,21 @@ jobs: --gtest_recreate_environments_when_repeating=1 \ --gtest_output=xml:${{ env.BUILD_DIR }}/integration-test-results.xml + - name: Run memory tester + if: matrix.name == 'ubuntu-24.04-x64' && inputs.build_type == 'release' + timeout-minutes: 10 + shell: bash + run: | + set -euo pipefail + source scripts/set-test-tokens.sh + scripts/memory-tester.sh \ + --build-dir "${{ env.BUILD_DIR }}" \ + --scenario all \ + --iterations 100 \ + --warmup 20 \ + --max-rss-growth-mib 8 \ + --max-thread-growth 0 + - name: Run stress tests if: matrix.e2e-testing && inputs.run_stress_tests timeout-minutes: ${{ inputs.stress_timeout_minutes }} diff --git a/docs/testing.md b/docs/testing.md index 33919c0e..eb66f5da 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -1,7 +1,21 @@ # Testing -The SDK includes integration and stress tests using -[Google Test](https://github.com/google/googletest). +The SDK includes unit, integration and stress tests using +[Google Test](https://github.com/google/googletest). It also +includes a memory-leak/resource checking tester. + +Unit, integration, and memory tests are run on every PR in CI. +Stress tests are reserved for nightly tests in CI due to their +length. + +## Test binaries + +| Executable | Description | +|------------|-------------| +| `livekit_unit_tests` | Unit tests (no server required) | +| `livekit_integration_tests` | Integration tests (server required) | +| `livekit_stress_tests` | Long-running stress tests (server required) | +| `livekit_memory_tester` | Resource leak tester | ## Building the test binaries @@ -17,33 +31,35 @@ The SDK includes integration and stress tests using .\build.cmd release-tests ``` -## Running tests +After building, test binaries are available at: -After building, run tests using `ctest` or invoke the binaries directly: +```bash +# Linux/macOS +/bin/ + +# Windows +\bin\ +``` + +## Running tests ```bash -# Run all tests via ctest +# Run all tests via ctest (warning, runs all GTest-based tests including stress) cd build-debug ctest --output-on-failure # Or run test executables directly +./build-debug/bin/livekit_unit_tests ./build-debug/bin/livekit_integration_tests ./build-debug/bin/livekit_stress_tests # Run specific test suites +./build-debug/bin/livekit_unit_tests --gtest_filter="*Room*" ./build-debug/bin/livekit_integration_tests --gtest_filter="*Rpc*" ./build-debug/bin/livekit_stress_tests --gtest_filter="*MaxPayloadStress*" ``` -__Note:__ The tests require tokens and a running LiveKit server. See the section below for details. - -## Test binaries - -| Executable | Description | -|------------|-------------| -| `livekit_unit_tests` | Pure unit tests (no server required) | -| `livekit_integration_tests` | Quick tests (~1-2 minutes) for SDK functionality | -| `livekit_stress_tests` | Long-running tests (configurable, default 1 hour) | +__Note:__ The integration and stress tests require tokens and a running LiveKit server. See the section below for details. ## Running a local LiveKit server for tests @@ -52,9 +68,10 @@ path is `livekit-server --dev`, which uses the well-known dev API key/secret (`devkey` / `secret`). Install [`livekit-server`](https://docs.livekit.io/home/self-hosting/local/) -and start it with data tracks enabled: +and start it with data blobs enabled: ```bash +LIVEKIT_CONFIG="enable_participant_data_blob: true" livekit-server --dev ``` @@ -106,7 +123,69 @@ export LIVEKIT_TOKEN_B="$(lk token create --api-key devkey --api-secret secret - - **RPC**: round-trip calls, max payload (15 KB), timeouts, errors, concurrent calls. - **Stress**: high throughput, bidirectional RPC, memory pressure. -## Memory checks (valgrind) +## Memory tester + +> Note: Linux only + +The memory tester binary catches resource leak regressions in an +automated fashion, and is invoked in CI. This single binary runs individual +or groups of scenarios: + +- `audio-track` repeatedly initializes the SDK, connects, publishes an audio + track, captures synthetic PCM frames, unpublishes, disconnects, and shuts down. +- `video-track` repeatedly initializes the SDK, connects, publishes a video + track, captures synthetic I420 frames, unpublishes, disconnects, and shuts down. +- `data-track` repeatedly connects, publishes a data track, pushes synthetic + payloads, unpublishes it, and disconnects. +- `room-reuse` repeatedly connects and disconnects the same room object. +- `room-client-leave` repeatedly connects, disconnects, and destroys a room. +- `room-server-delete` repeatedly connects, publishes one synthetic video + track, captures I420 frames without unpublishing, and uses `lk` to delete + the local test room. It then verifies the disconnect and end-of-stream + callbacks. + +Each scenario reports process RSS and thread count at cycle 0, after warmup +cycle 20, and after final cycle 100. Checkpoint sampling waits for the thread +count to remain stable for 500 ms so short-lived teardown work does not cause a +false failure. The gate applies to warmup-to-final growth and fails on thread +growth or more than the configured amount of RSS growth. If threads grow, the +tester also reports when the final sustained growth began and groups the warmup +and final threads by their Linux thread names. + +The binary can be run directly, or via a helper script at `scripts/memory-tester.sh`, +which invokes the binary but also has GitHub actions support for a results summary table. +The helper defaults `MALLOC_ARENA_MAX` to `1` when launching each tester process. +This limits `glibc`'s allocator arenas, reducing Linux thread-churn RSS noise, but it is +not evidence that the default allocator returns all freed pages to the OS. + +Start `livekit-server --dev`, install the `lk` CLI, and source the local test +tokens before running any scenario: + +```bash +./build.sh release-tests +source scripts/set-test-tokens.sh +scripts/memory-tester.sh --scenario audio-track \ + --iterations 100 --warmup 20 \ + --max-rss-growth-mib 8 --max-thread-growth 0 +scripts/memory-tester.sh --scenario video-track \ + --iterations 100 --warmup 20 \ + --max-rss-growth-mib 8 --max-thread-growth 0 +``` + +Use `--scenario all` to run all six scenarios, each in a fresh tester process: + +```bash +scripts/memory-tester.sh --scenario data-track +scripts/memory-tester.sh --scenario room-reuse +scripts/memory-tester.sh --scenario room-client-leave +scripts/memory-tester.sh --scenario room-server-delete +scripts/memory-tester.sh --scenario all +``` + +The limits are command-line options so a longer local run can use different +budgets. + +## Additional memory checks (valgrind) Run `valgrind` against the test binaries to check for memory leaks and other issues. See [tools.md](tools.md) for the recipe. diff --git a/scripts/memory-tester.sh b/scripts/memory-tester.sh new file mode 100755 index 00000000..936f48e6 --- /dev/null +++ b/scripts/memory-tester.sh @@ -0,0 +1,163 @@ +#!/usr/bin/env bash +# +# Copyright 2026 LiveKit +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +set -euo pipefail + +usage() { + cat <<'EOF' +Usage: memory-tester.sh [OPTIONS] + +Run the Linux memory tester and append a table to the +GitHub step summary when GITHUB_STEP_SUMMARY is set. + +Options: + --build-dir PATH Build directory (default: build-release) + --scenario NAME audio-track, video-track, data-track, + room-reuse, room-client-leave, + room-server-delete, or all (default: audio-track) + --iterations N Lifecycle cycles (default: 100) + --warmup N Baseline cycle (default: 20) + --max-rss-growth-mib N Allowed RSS growth (default: 8) + --max-thread-growth N Allowed thread growth (default: 0) + -h, --help Show this help + +Environment: + MEMORY_TESTER_BUILD_DIR, MEMORY_TESTER_SCENARIO, + MEMORY_TESTER_ITERATIONS, + MEMORY_TESTER_WARMUP, MEMORY_TESTER_MAX_RSS_GROWTH_MIB, + MEMORY_TESTER_MAX_THREAD_GROWTH, MALLOC_ARENA_MAX. + All scenarios require LIVEKIT_URL and LIVEKIT_TOKEN_A. +EOF +} + +repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd -P)" +build_dir="${MEMORY_TESTER_BUILD_DIR:-build-release}" +scenario="${MEMORY_TESTER_SCENARIO:-audio-track}" +iterations="${MEMORY_TESTER_ITERATIONS:-100}" +warmup="${MEMORY_TESTER_WARMUP:-20}" +max_rss_growth_mib="${MEMORY_TESTER_MAX_RSS_GROWTH_MIB:-8}" +max_thread_growth="${MEMORY_TESTER_MAX_THREAD_GROWTH:-0}" + +while (($#)); do + case "$1" in + --build-dir) build_dir="$2"; shift 2 ;; + --scenario) scenario="$2"; shift 2 ;; + --iterations) iterations="$2"; shift 2 ;; + --warmup) warmup="$2"; shift 2 ;; + --max-rss-growth-mib) max_rss_growth_mib="$2"; shift 2 ;; + --max-thread-growth) max_thread_growth="$2"; shift 2 ;; + -h|--help) usage; exit 0 ;; + *) echo "ERROR: unknown option: $1" >&2; usage >&2; exit 2 ;; + esac +done + +if [[ "${build_dir}" != /* ]]; then + build_dir="${repo_root}/${build_dir}" +fi +tester="${build_dir}/bin/livekit_memory_tester" +if [[ ! -x "${tester}" ]]; then + echo "ERROR: ${tester} not found. Run ./build.sh release-tests first." >&2 + exit 1 +fi + +case "${scenario}" in + audio-track|video-track|data-track|room-reuse|room-client-leave|room-server-delete) + scenarios=("${scenario}") + ;; + all) + scenarios=(audio-track video-track data-track room-reuse room-client-leave room-server-delete) + ;; + *) + echo "ERROR: invalid scenario: ${scenario}" >&2 + usage >&2 + exit 2 + ;; +esac + +for current_scenario in "${scenarios[@]}"; do + if [[ -z "${LIVEKIT_URL:-}" || -z "${LIVEKIT_TOKEN_A:-}" ]]; then + echo "ERROR: ${current_scenario} requires LIVEKIT_URL and LIVEKIT_TOKEN_A." >&2 + echo "Run: source scripts/set-test-tokens.sh" >&2 + exit 1 + fi +done + +if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then + { + echo "## Memory tester results" + echo + echo "Iterations: ${iterations}" + echo "Warmup: ${warmup}" + echo "Threshold: RSS growth ≤ ${max_rss_growth_mib} MiB; thread growth ≤ ${max_thread_growth}." + echo + echo "| Scenario | RSS 0→${warmup} | RSS ${warmup}→${iterations} | Threads 0→${warmup} | Threads ${warmup}→${iterations} | Status |" + echo "|---|---:|---:|---:|---:|---|" + } >> "${GITHUB_STEP_SUMMARY}" +fi + +report_regex='memory tester: scenario=([a-z-]+), RSS 0 ([0-9]+\.[0-9]+) -> warmup ([0-9]+\.[0-9]+) MiB \((-?[0-9]+\.[0-9]+) MiB\) -> final ([0-9]+\.[0-9]+) MiB \((-?[0-9]+\.[0-9]+) MiB\), threads 0 ([0-9]+) -> warmup ([0-9]+) \((-?[0-9]+)\) -> final ([0-9]+) \((-?[0-9]+)\), verdict=(PASS|FAIL)' + +format_growth() { + if [[ "$1" == "0" || "$1" == "0.00" || "$1" == -* ]]; then + printf '%s' "$1" + else + printf '+%s' "$1" + fi +} + +overall_status=0 +for current_scenario in "${scenarios[@]}"; do + set +e + output="$( + MALLOC_ARENA_MAX="${MALLOC_ARENA_MAX:-1}" "${tester}" \ + --scenario "${current_scenario}" \ + --iterations "${iterations}" \ + --warmup "${warmup}" \ + --max-rss-growth-mib "${max_rss_growth_mib}" \ + --max-thread-growth "${max_thread_growth}" 2>&1 + )" + status=$? + set -e + printf '%s\n' "${output}" + + if [[ "${output}" =~ ${report_regex} ]]; then + if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then + if [[ "${BASH_REMATCH[12]}" == "PASS" ]]; then + verdict=":white_check_mark: PASS" + else + verdict=":x: FAIL" + fi + rss_warmup_growth="$(format_growth "${BASH_REMATCH[4]}")" + rss_final_growth="$(format_growth "${BASH_REMATCH[6]}")" + thread_warmup_growth="$(format_growth "${BASH_REMATCH[9]}")" + thread_final_growth="$(format_growth "${BASH_REMATCH[11]}")" + echo "| ${BASH_REMATCH[1]} | ${rss_warmup_growth} MiB | ${rss_final_growth} MiB | ${thread_warmup_growth} | ${thread_final_growth} | ${verdict} |" \ + >> "${GITHUB_STEP_SUMMARY}" + fi + else + echo "ERROR: ${current_scenario} did not produce a memory report." >&2 + if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then + echo "| ${current_scenario} | — | — | — | — | :x: ERROR |" >> "${GITHUB_STEP_SUMMARY}" + fi + status=1 + fi + + if [[ "${status}" -ne 0 ]]; then + overall_status=1 + fi +done + +exit "${overall_status}" diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index 9583af73..23d5fc65 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -348,3 +348,22 @@ endif() if(TARGET livekit_stress_tests) add_dependencies(run_all_tests livekit_stress_tests) endif() + +if(CMAKE_SYSTEM_NAME STREQUAL "Linux") + add_executable(livekit_memory_tester + "${CMAKE_CURRENT_SOURCE_DIR}/memory/memory_tester.cpp" + ) + target_link_libraries(livekit_memory_tester PRIVATE livekit) + target_include_directories(livekit_memory_tester PRIVATE + ${LIVEKIT_ROOT_DIR}/include + ) + add_custom_command(TARGET livekit_memory_tester POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ + $ + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "$/liblivekit_ffi.so" + $ + COMMENT "Copying shared libraries next to livekit_memory_tester" + ) +endif() diff --git a/src/tests/memory/memory_tester.cpp b/src/tests/memory/memory_tester.cpp new file mode 100644 index 00000000..54db826a --- /dev/null +++ b/src/tests/memory/memory_tester.cpp @@ -0,0 +1,660 @@ +/* + * Copyright 2026 LiveKit + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +using namespace std::chrono_literals; + +constexpr std::string_view kLocalTestLiveKitUrl = "ws://localhost:7880"; +constexpr int kFramesPerCycle = 3; +constexpr auto kThreadQuiescencePeriod = 500ms; +constexpr auto kThreadQuiescencePollInterval = 50ms; +constexpr auto kThreadQuiescenceTimeout = 2s; + +enum class Scenario { + AudioTrack, + VideoTrack, + DataTrack, + RoomReuse, + RoomClientLeave, + RoomServerDelete, +}; + +struct Options { + Scenario scenario{Scenario::AudioTrack}; + std::uint64_t iterations{100}; + std::uint64_t warmup{20}; + std::uint64_t max_rss_growth_mib{8}; + std::uint64_t max_thread_growth{0}; +}; + +struct ProcessSample { + std::uint64_t rss_kib{}; + std::uint64_t threads{}; +}; + +using ThreadNameCounts = std::map; +using ThreadIdentities = std::map; + +struct ProcessSnapshot { + ProcessSample process; + ThreadIdentities threads; +}; + +struct RoomCredentials { + std::string url; + std::string token; +}; + +class SdkLifetime { +public: + SdkLifetime() { + if (!livekit::initialize(livekit::LogLevel::Warn)) { + throw std::runtime_error("LiveKit initialization failed"); + } + } + + ~SdkLifetime() { livekit::shutdown(); } + + SdkLifetime(const SdkLifetime&) = delete; + SdkLifetime& operator=(const SdkLifetime&) = delete; +}; + +class DisconnectTrackingDelegate final : public livekit::RoomDelegate { +public: + void onDisconnected(livekit::Room&, const livekit::DisconnectedEvent& event) override { + { + const std::scoped_lock lock(mutex_); + reason_ = event.reason; + ++disconnect_count_; + } + condition_.notify_all(); + } + + void onRoomEos(livekit::Room&, const livekit::RoomEosEvent&) override { + { + const std::scoped_lock lock(mutex_); + ++eos_count_; + } + condition_.notify_all(); + } + + bool waitForTeardown(std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + return condition_.wait_for(lock, timeout, [this]() { return disconnect_count_ > 0 && eos_count_ > 0; }); + } + + void requireDisconnect(livekit::DisconnectReason expected_reason) const { + const std::scoped_lock lock(mutex_); + if (disconnect_count_ != 1 || reason_ != expected_reason) { + throw std::runtime_error("unexpected room disconnect event"); + } + } + + void requireServerTeardown() const { + const std::scoped_lock lock(mutex_); + if (disconnect_count_ != 1 || eos_count_ != 1 || reason_ != livekit::DisconnectReason::RoomDeleted) { + throw std::runtime_error("unexpected server room teardown events"); + } + } + +private: + mutable std::mutex mutex_; + std::condition_variable condition_; + livekit::DisconnectReason reason_{livekit::DisconnectReason::Unknown}; + int disconnect_count_{0}; + int eos_count_{0}; +}; + +std::string_view scenarioName(Scenario scenario) { + switch (scenario) { + case Scenario::AudioTrack: + return "audio-track"; + case Scenario::VideoTrack: + return "video-track"; + case Scenario::DataTrack: + return "data-track"; + case Scenario::RoomReuse: + return "room-reuse"; + case Scenario::RoomClientLeave: + return "room-client-leave"; + case Scenario::RoomServerDelete: + return "room-server-delete"; + } + throw std::runtime_error("unknown scenario"); +} + +Scenario parseScenario(std::string_view value) { + if (value == "audio-track") { + return Scenario::AudioTrack; + } + if (value == "video-track") { + return Scenario::VideoTrack; + } + if (value == "data-track") { + return Scenario::DataTrack; + } + if (value == "room-reuse") { + return Scenario::RoomReuse; + } + if (value == "room-client-leave") { + return Scenario::RoomClientLeave; + } + if (value == "room-server-delete") { + return Scenario::RoomServerDelete; + } + throw std::runtime_error("invalid value for --scenario"); +} + +std::uint64_t parseUnsigned(std::string_view value, std::string_view option) { + std::uint64_t parsed = 0; + const auto result = std::from_chars(value.data(), value.data() + value.size(), parsed); + if (result.ec != std::errc{} || result.ptr != value.data() + value.size()) { + throw std::runtime_error("invalid value for " + std::string(option)); + } + return parsed; +} + +Options parseOptions(int argc, char* argv[]) { + Options options; + for (int index = 1; index < argc; ++index) { + const std::string_view option(argv[index]); + if (index + 1 >= argc) { + throw std::runtime_error("missing value for " + std::string(option)); + } + const std::string_view value(argv[++index]); + if (option == "--scenario") { + options.scenario = parseScenario(value); + } else if (option == "--iterations") { + options.iterations = parseUnsigned(value, option); + } else if (option == "--warmup") { + options.warmup = parseUnsigned(value, option); + } else if (option == "--max-rss-growth-mib") { + options.max_rss_growth_mib = parseUnsigned(value, option); + } else if (option == "--max-thread-growth") { + options.max_thread_growth = parseUnsigned(value, option); + } else { + throw std::runtime_error("unknown option: " + std::string(option)); + } + } + if (options.warmup == 0 || options.warmup >= options.iterations) { + throw std::runtime_error("--warmup must be between 1 and --iterations - 1"); + } + return options; +} + +std::uint64_t statusValue(std::string_view line) { + const auto separator = line.find(':'); + if (separator == std::string_view::npos) { + throw std::runtime_error("malformed /proc/self/status value"); + } + line.remove_prefix(separator + 1); + while (!line.empty() && std::isspace(static_cast(line.front())) != 0) { + line.remove_prefix(1); + } + const auto end = line.find_first_not_of("0123456789"); + return parseUnsigned(line.substr(0, end), "/proc/self/status"); +} + +ProcessSample sampleProcess() { + ProcessSample sample; + std::ifstream status("/proc/self/status"); + std::string line; + while (std::getline(status, line)) { + if (line.rfind("VmRSS:", 0) == 0) { + sample.rss_kib = statusValue(line); + } else if (line.rfind("Threads:", 0) == 0) { + sample.threads = statusValue(line); + } + } + if (sample.rss_kib == 0 || sample.threads == 0) { + throw std::runtime_error("failed to read RSS or thread count from /proc/self/status"); + } + return sample; +} + +ThreadIdentities sampleThreads() { + ThreadIdentities threads; + std::error_code error; + std::filesystem::directory_iterator iterator("/proc/self/task", error); + const std::filesystem::directory_iterator end; + while (!error && iterator != end) { + std::ifstream comm(iterator->path() / "comm"); + std::string name; + if (std::getline(comm, name) && !name.empty()) { + threads.emplace(iterator->path().filename().string(), std::move(name)); + } + iterator.increment(error); + } + return threads; +} + +ProcessSnapshot sampleProcessSnapshot() { return {sampleProcess(), sampleThreads()}; } + +ProcessSnapshot sampleQuiescentProcess() { + auto snapshot = sampleProcessSnapshot(); + auto stable_since = std::chrono::steady_clock::now(); + const auto deadline = stable_since + kThreadQuiescenceTimeout; + while (std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(kThreadQuiescencePollInterval); + auto next = sampleProcessSnapshot(); + if (next.threads != snapshot.threads) { + stable_since = std::chrono::steady_clock::now(); + } else if (std::chrono::steady_clock::now() - stable_since >= kThreadQuiescencePeriod) { + return next; + } + snapshot = std::move(next); + } + std::cerr << "memory tester: thread count did not quiesce within " + << std::chrono::duration_cast(kThreadQuiescenceTimeout).count() + << " ms; using the latest sample\n"; + return snapshot; +} + +ThreadNameCounts countThreadNames(const ThreadIdentities& threads) { + ThreadNameCounts counts; + for (const auto& [id, name] : threads) { + static_cast(id); + ++counts[name]; + } + return counts; +} + +std::string formatThreadNames(const ThreadIdentities& threads) { + std::ostringstream output; + bool first = true; + for (const auto& [name, count] : countThreadNames(threads)) { + if (!first) { + output << ", "; + } + output << name << '=' << count; + first = false; + } + return output.str(); +} + +std::string formatThreadNameDelta(const ThreadIdentities& before_threads, const ThreadIdentities& after_threads) { + const auto before = countThreadNames(before_threads); + const auto after = countThreadNames(after_threads); + ThreadNameCounts names = before; + names.insert(after.begin(), after.end()); + + std::ostringstream output; + bool first = true; + for (const auto& [name, unused] : names) { + static_cast(unused); + const auto before_entry = before.find(name); + const auto after_entry = after.find(name); + const auto before_count = before_entry == before.end() ? 0 : before_entry->second; + const auto after_count = after_entry == after.end() ? 0 : after_entry->second; + const auto delta = static_cast(after_count) - static_cast(before_count); + if (delta == 0) { + continue; + } + if (!first) { + output << ", "; + } + output << name << '=' << (delta > 0 ? "+" : "") << delta; + first = false; + } + return output.str(); +} + +double kibToMib(std::uint64_t kib) { return static_cast(kib) / 1024.0; } + +double kibToMib(std::int64_t kib) { return static_cast(kib) / 1024.0; } + +RoomCredentials roomCredentials() { + const char* url = std::getenv("LIVEKIT_URL"); + const char* token = std::getenv("LIVEKIT_TOKEN_A"); + if (url == nullptr || token == nullptr || *url == '\0' || *token == '\0') { + throw std::runtime_error("memory scenarios require LIVEKIT_URL and LIVEKIT_TOKEN_A"); + } + return {url, token}; +} + +void connectRoom(livekit::Room& room, const RoomCredentials& credentials) { + if (!room.connect(credentials.url, credentials.token, livekit::RoomOptions{})) { + throw std::runtime_error("room connection failed"); + } + if (room.connectionState() != livekit::ConnectionState::Connected || room.localParticipant().expired()) { + throw std::runtime_error("room did not reach a connected state"); + } +} + +// Exercises SDK startup, room connection, audio publication and capture, +// explicit unpublication, client disconnect, and complete process-side teardown. +void runAudioTrackCycle(const RoomCredentials& credentials) { + SdkLifetime sdk; + DisconnectTrackingDelegate delegate; + livekit::Room room; + room.setDelegate(&delegate); + connectRoom(room, credentials); + + auto audio_source = std::make_shared(48'000, 1); + auto audio_track = livekit::LocalAudioTrack::createLocalAudioTrack("memory-audio", audio_source); + if (!audio_track) { + throw std::runtime_error("failed to create local audio track"); + } + + auto participant = room.localParticipant().lock(); + if (!participant) { + throw std::runtime_error("local participant expired before publishing audio track"); + } + livekit::TrackPublishOptions publish_options; + publish_options.source = livekit::TrackSource::SOURCE_MICROPHONE; + participant->publishTrack(audio_track, publish_options); + const auto publication = audio_track->publication(); + if (!publication) { + throw std::runtime_error("failed to publish local audio track"); + } + + auto frame = livekit::AudioFrame::create(48'000, 1, 480); + std::fill(frame.data().begin(), frame.data().end(), 100); + for (int frame_index = 0; frame_index < kFramesPerCycle; ++frame_index) { + audio_source->captureFrame(frame, 1'000); + } + + participant->unpublishTrack(publication->sid()); + if (!room.disconnect()) { + throw std::runtime_error("client room disconnect failed"); + } + delegate.requireDisconnect(livekit::DisconnectReason::ClientInitiated); +} + +// Exercises SDK startup, room connection, video publication and capture, +// encoder/resource teardown, explicit unpublication, and client disconnect. +void runVideoTrackCycle(const RoomCredentials& credentials) { + SdkLifetime sdk; + DisconnectTrackingDelegate delegate; + livekit::Room room; + room.setDelegate(&delegate); + connectRoom(room, credentials); + + auto video_source = std::make_shared(640, 360); + auto video_track = livekit::LocalVideoTrack::createLocalVideoTrack("memory-video", video_source); + if (!video_track) { + throw std::runtime_error("failed to create local video track"); + } + + auto participant = room.localParticipant().lock(); + if (!participant) { + throw std::runtime_error("local participant expired before publishing video track"); + } + livekit::TrackPublishOptions publish_options; + publish_options.source = livekit::TrackSource::SOURCE_CAMERA; + publish_options.simulcast = false; + participant->publishTrack(video_track, publish_options); + const auto publication = video_track->publication(); + if (!publication) { + throw std::runtime_error("failed to publish local video track"); + } + + auto frame = livekit::VideoFrame::create(640, 360, livekit::VideoBufferType::I420); + std::fill(frame.data(), frame.data() + frame.dataSize(), 0x7f); + for (int frame_index = 0; frame_index < kFramesPerCycle; ++frame_index) { + video_source->captureFrame(frame); + } + + participant->unpublishTrack(publication->sid()); + if (!room.disconnect()) { + throw std::runtime_error("client room disconnect failed"); + } + delegate.requireDisconnect(livekit::DisconnectReason::ClientInitiated); +} + +// Exercises data-track publication, payload enqueueing, explicit unpublication, +// and room teardown while the SDK remains initialized across cycles. +void runDataTrackCycle(const RoomCredentials& credentials) { + DisconnectTrackingDelegate delegate; + livekit::Room room; + room.setDelegate(&delegate); + connectRoom(room, credentials); + + auto participant = room.localParticipant().lock(); + if (!participant) { + throw std::runtime_error("local participant expired before publishing data track"); + } + auto publish_result = participant->publishDataTrack("memory-data"); + if (!publish_result) { + throw std::runtime_error("failed to publish local data track"); + } + + const auto& track = publish_result.value(); + livekit::DataTrackFrame frame; + frame.payload.assign(256, 0xFA); + for (int frame_index = 0; frame_index < kFramesPerCycle; ++frame_index) { + frame.user_timestamp = static_cast(frame_index); + if (!track->tryPush(frame)) { + throw std::runtime_error("failed to push local data track frame"); + } + } + + track->unpublishDataTrack(); + if (track->isPublished()) { + throw std::runtime_error("local data track remained published"); + } + if (!room.disconnect()) { + throw std::runtime_error("client room disconnect failed"); + } + delegate.requireDisconnect(livekit::DisconnectReason::ClientInitiated); +} + +// Exercises repeated connect/disconnect cycles on one Room instance to detect +// connection state, participant, listener, or FFI resources retained by reuse. +void runRoomReuseCycle(livekit::Room& room, const RoomCredentials& credentials) { + connectRoom(room, credentials); + if (!room.disconnect()) { + throw std::runtime_error("client room disconnect failed"); + } + if (room.connectionState() != livekit::ConnectionState::Disconnected || !room.localParticipant().expired()) { + throw std::runtime_error("reused room disconnect did not tear down local state"); + } +} + +// Exercises application-initiated disconnect and destruction of a fresh Room, +// including local-participant cleanup and the expected disconnect callback. +void runRoomClientLeaveCycle(const RoomCredentials& credentials) { + DisconnectTrackingDelegate delegate; + livekit::Room room; + room.setDelegate(&delegate); + connectRoom(room, credentials); + + if (!room.disconnect()) { + throw std::runtime_error("client room disconnect failed"); + } + if (room.connectionState() != livekit::ConnectionState::Disconnected || !room.localParticipant().expired()) { + throw std::runtime_error("client room disconnect did not tear down local state"); + } + delegate.requireDisconnect(livekit::DisconnectReason::ClientInitiated); +} + +// Exercises server-initiated room deletion while one published video track is +// still attached. Teardown must release that publication after the peer +// connection has already closed; unpublishing first avoids the leak this +// scenario targets. The synthetic track uses the default software-capable +// encoder so the scenario can run in CI. +void runRoomServerDeleteCycle(const RoomCredentials& credentials) { + if (std::string_view(credentials.url) != kLocalTestLiveKitUrl) { + throw std::runtime_error("room-server-delete requires LIVEKIT_URL=ws://localhost:7880"); + } + + DisconnectTrackingDelegate delegate; + livekit::Room room; + room.setDelegate(&delegate); + connectRoom(room, credentials); + + auto participant = room.localParticipant().lock(); + if (!participant) { + throw std::runtime_error("local participant expired before publishing video track"); + } + + auto video_source = std::make_shared(640, 360); + auto video_track = livekit::LocalVideoTrack::createLocalVideoTrack("memory-video", video_source); + if (!video_track) { + throw std::runtime_error("failed to create local video track"); + } + livekit::TrackPublishOptions publish_options; + publish_options.source = livekit::TrackSource::SOURCE_CAMERA; + publish_options.simulcast = false; + participant->publishTrack(video_track, publish_options); + if (!video_track->publication()) { + throw std::runtime_error("failed to publish local video track"); + } + + auto frame = livekit::VideoFrame::create(640, 360, livekit::VideoBufferType::I420); + std::fill(frame.data(), frame.data() + frame.dataSize(), 0x7f); + for (int frame_index = 0; frame_index < kFramesPerCycle; ++frame_index) { + video_source->captureFrame(frame); + } + + // The room name is fixed rather than interpolating untrusted environment data into the command. + if (std::system("lk --dev room delete --yes cpp_data_track_test") != 0) { + throw std::runtime_error("failed to delete local test room"); + } + if (!delegate.waitForTeardown(10s)) { + throw std::runtime_error("timed out waiting for server room teardown"); + } + if (room.connectionState() != livekit::ConnectionState::Disconnected || !room.localParticipant().expired()) { + throw std::runtime_error("server room deletion did not tear down local state"); + } + delegate.requireServerTeardown(); +} + +void runCycle(Scenario scenario, const RoomCredentials& credentials, livekit::Room* reusable_room) { + switch (scenario) { + case Scenario::AudioTrack: + runAudioTrackCycle(credentials); + return; + case Scenario::VideoTrack: + runVideoTrackCycle(credentials); + return; + case Scenario::DataTrack: + runDataTrackCycle(credentials); + return; + case Scenario::RoomReuse: + if (reusable_room == nullptr) { + throw std::runtime_error("room-reuse scenario is missing its persistent room"); + } + runRoomReuseCycle(*reusable_room, credentials); + return; + case Scenario::RoomClientLeave: + runRoomClientLeaveCycle(credentials); + return; + case Scenario::RoomServerDelete: + runRoomServerDeleteCycle(credentials); + return; + } + throw std::runtime_error("unknown scenario"); +} + +} // namespace + +int main(int argc, char* argv[]) { + try { + const Options options = parseOptions(argc, argv); + const RoomCredentials credentials = roomCredentials(); + const bool per_cycle_sdk = options.scenario == Scenario::AudioTrack || options.scenario == Scenario::VideoTrack; + std::unique_ptr room_sdk; + if (!per_cycle_sdk) { + room_sdk = std::make_unique(); + } + std::unique_ptr reusable_room; + if (options.scenario == Scenario::RoomReuse) { + reusable_room = std::make_unique(); + } + + const ProcessSnapshot initial = sampleProcessSnapshot(); + ProcessSnapshot warmup; + ProcessSnapshot final; + std::uint64_t sustained_thread_growth_iteration = 0; + ProcessSnapshot sustained_thread_growth; + for (std::uint64_t iteration = 1; iteration <= options.iterations; ++iteration) { + runCycle(options.scenario, credentials, reusable_room.get()); + std::this_thread::sleep_for(100ms); + if (iteration == options.warmup) { + warmup = sampleQuiescentProcess(); + } else if (iteration == options.iterations) { + final = sampleQuiescentProcess(); + } else if (iteration > options.warmup) { + auto current = sampleProcessSnapshot(); + if (current.process.threads <= warmup.process.threads) { + sustained_thread_growth_iteration = 0; + sustained_thread_growth = {}; + } else if (sustained_thread_growth_iteration == 0) { + sustained_thread_growth_iteration = iteration; + sustained_thread_growth = std::move(current); + } + } + } + + const auto warmup_rss_growth = + static_cast(warmup.process.rss_kib) - static_cast(initial.process.rss_kib); + const auto final_rss_growth = + static_cast(final.process.rss_kib) - static_cast(warmup.process.rss_kib); + const auto warmup_thread_growth = + static_cast(warmup.process.threads) - static_cast(initial.process.threads); + const auto final_thread_growth = + static_cast(final.process.threads) - static_cast(warmup.process.threads); + const bool passed = kibToMib(final_rss_growth) <= static_cast(options.max_rss_growth_mib) && + final_thread_growth <= static_cast(options.max_thread_growth); + if (final_thread_growth > 0) { + if (sustained_thread_growth_iteration == 0) { + sustained_thread_growth_iteration = options.iterations; + sustained_thread_growth = final; + } + std::cerr << "memory tester: final thread growth sustained since cycle " << sustained_thread_growth_iteration + << "; warmup={" << formatThreadNames(warmup.threads) << "}; growth-start={" + << formatThreadNames(sustained_thread_growth.threads) << "}; final={" + << formatThreadNames(final.threads) << "}; warmup-to-final delta={" + << formatThreadNameDelta(warmup.threads, final.threads) << "}\n"; + } + std::cout << std::fixed << std::setprecision(2) << "memory tester: scenario=" << scenarioName(options.scenario) + << ", RSS 0 " << kibToMib(initial.process.rss_kib) << " -> warmup " << kibToMib(warmup.process.rss_kib) + << " MiB (" << kibToMib(warmup_rss_growth) << " MiB) -> final " << kibToMib(final.process.rss_kib) + << " MiB (" << kibToMib(final_rss_growth) << " MiB), threads 0 " << initial.process.threads + << " -> warmup " << warmup.process.threads << " (" << warmup_thread_growth << ") -> final " + << final.process.threads << " (" << final_thread_growth << "), verdict=" << (passed ? "PASS" : "FAIL") + << '\n'; + return passed ? 0 : 1; + } catch (const std::exception& error) { + std::cerr << "memory tester failed: " << error.what() << '\n'; + return 1; + } +}