test_stability.cpp

Go to the documentation of this file.

 
#include "test_fixtures.h"
#include "test_data_generator.h"
 
#include <catch2/catch_test_macros.hpp>
 
#include "kcenon/pacs/network/dimse/dimse_message.h"
#include "kcenon/pacs/services/query_scp.h"
#include "kcenon/pacs/services/storage_scp.h"
#include "kcenon/pacs/services/storage_scu.h"
#include "kcenon/pacs/services/verification_scp.h"
#include "kcenon/pacs/storage/file_storage.h"
#include "kcenon/pacs/storage/index_database.h"
 
#include <atomic>
#include <chrono>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <future>
#include <iomanip>
#include <iostream>
#include <mutex>
#include <random>
#include <set>
#include <sstream>
#include <thread>
#include <vector>
 
#ifdef __linux__
#include <fstream>
#include <unistd.h>
#elif defined(__APPLE__)
#include <mach/mach.h>
#elif defined(_WIN32)
#include <windows.h>
#include <psapi.h>
#endif
 
using namespace kcenon::pacs::integration_test;
using namespace kcenon::pacs::network;
using namespace kcenon::pacs::services;
using namespace kcenon::pacs::storage;
using namespace kcenon::pacs::core;
using namespace kcenon::pacs::encoding;
using namespace kcenon::pacs::network::dimse;
 
namespace {
 
// =============================================================================
// Stability Test Configuration
// =============================================================================
 
struct stability_config {
    std::chrono::minutes duration{60};  // Default 1 hour
    double store_rate{5.0};             // Stores per second
    double query_rate{1.0};             // Queries per second
    size_t store_workers{4};            // Number of store worker threads
    size_t query_workers{2};            // Number of query worker threads
    size_t max_memory_growth_mb{100};   // Maximum allowed memory growth
    size_t max_associations{100};       // Max concurrent associations
 
    static stability_config from_environment() {
        stability_config config;
 
        if (const char* duration = std::getenv("PACS_STABILITY_TEST_DURATION")) {
            config.duration = std::chrono::minutes{std::atoi(duration)};
        }
        if (const char* store_rate = std::getenv("PACS_STABILITY_STORE_RATE")) {
            config.store_rate = std::atof(store_rate);
        }
        if (const char* query_rate = std::getenv("PACS_STABILITY_QUERY_RATE")) {
            config.query_rate = std::atof(query_rate);
        }
        if (const char* store_workers = std::getenv("PACS_STABILITY_STORE_WORKERS")) {
            config.store_workers = static_cast<size_t>(std::atoi(store_workers));
        }
        if (const char* query_workers = std::getenv("PACS_STABILITY_QUERY_WORKERS")) {
            config.query_workers = static_cast<size_t>(std::atoi(query_workers));
        }
 
        return config;
    }
};
 
// =============================================================================
// Stability Metrics
// =============================================================================
 
struct stability_metrics {
    std::atomic<size_t> stores_completed{0};
    std::atomic<size_t> stores_failed{0};
    std::atomic<size_t> queries_completed{0};
    std::atomic<size_t> queries_failed{0};
    std::atomic<size_t> connections_opened{0};
    std::atomic<size_t> connections_closed{0};
    std::atomic<size_t> connection_errors{0};
 
    size_t initial_memory_kb{0};
    std::atomic<size_t> peak_memory_kb{0};
    std::chrono::steady_clock::time_point start_time;
 
    void reset() {
        stores_completed = 0;
        stores_failed = 0;
        queries_completed = 0;
        queries_failed = 0;
        connections_opened = 0;
        connections_closed = 0;
        connection_errors = 0;
        initial_memory_kb = 0;
        peak_memory_kb = 0;
        start_time = std::chrono::steady_clock::now();
    }
 
    [[nodiscard]] std::chrono::seconds elapsed() const {
        return std::chrono::duration_cast<std::chrono::seconds>(
            std::chrono::steady_clock::now() - start_time);
    }
 
    void report(std::ostream& out) const {
        auto duration = elapsed();
        double hours = static_cast<double>(duration.count()) / 3600.0;
 
        out << "\n";
        out << "=================================================\n";
        out << " Stability Test Results\n";
        out << "=================================================\n";
        out << " Duration:             " << duration.count() << " seconds ("
            << std::fixed << std::setprecision(2) << hours << " hours)\n";
        out << "\n";
        out << " Store Operations:\n";
        out << "   Completed:          " << stores_completed.load() << "\n";
        out << "   Failed:             " << stores_failed.load() << "\n";
        out << "   Rate:               " << std::fixed << std::setprecision(2)
            << (static_cast<double>(stores_completed.load()) / duration.count()) << "/sec\n";
        out << "\n";
        out << " Query Operations:\n";
        out << "   Completed:          " << queries_completed.load() << "\n";
        out << "   Failed:             " << queries_failed.load() << "\n";
        out << "   Rate:               " << std::fixed << std::setprecision(2)
            << (static_cast<double>(queries_completed.load()) / duration.count()) << "/sec\n";
        out << "\n";
        out << " Connections:\n";
        out << "   Opened:             " << connections_opened.load() << "\n";
        out << "   Closed:             " << connections_closed.load() << "\n";
        out << "   Errors:             " << connection_errors.load() << "\n";
        out << "\n";
        out << " Memory:\n";
        out << "   Initial:            " << initial_memory_kb / 1024 << " MB\n";
        out << "   Peak:               " << peak_memory_kb / 1024 << " MB\n";
        out << "   Growth:             " << (peak_memory_kb - initial_memory_kb) / 1024 << " MB\n";
        out << "=================================================\n";
    }
 
    void save_to_file(const std::filesystem::path& path) const {
        std::ofstream file(path);
        if (file) {
            report(file);
        }
    }
};
 
// =============================================================================
// Memory Monitoring
// =============================================================================
 
size_t get_process_memory_kb() {
#ifdef __linux__
    std::ifstream status("/proc/self/status");
    std::string line;
    while (std::getline(status, line)) {
        if (line.compare(0, 6, "VmRSS:") == 0) {
            std::istringstream iss(line.substr(6));
            size_t kb;
            iss >> kb;
            return kb;
        }
    }
    return 0;
#elif defined(__APPLE__)
    mach_task_basic_info info;
    mach_msg_type_number_t count = MACH_TASK_BASIC_INFO_COUNT;
    if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO,
                  reinterpret_cast<task_info_t>(&info), &count) == KERN_SUCCESS) {
        return info.resident_size / 1024;
    }
    return 0;
#elif defined(_WIN32)
    PROCESS_MEMORY_COUNTERS pmc;
    if (GetProcessMemoryInfo(GetCurrentProcess(), &pmc, sizeof(pmc))) {
        return pmc.WorkingSetSize / 1024;
    }
    return 0;
#else
    return 0;
#endif
}
 
// =============================================================================
// Stability Test Server
// =============================================================================
 
class stability_test_server {
public:
    explicit stability_test_server(uint16_t port, const std::string& ae_title = "STABILITY_SCP")
        : port_(port)
        , ae_title_(ae_title)
        , test_dir_("stability_test_")
        , storage_dir_(test_dir_.path() / "archive")
        , db_path_(test_dir_.path() / "index.db") {
 
        std::filesystem::create_directories(storage_dir_);
 
        server_config config;
        config.ae_title = ae_title_;
        config.port = port_;
        config.max_associations = 100;
        config.idle_timeout = std::chrono::seconds{300};
        config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.8";
        config.implementation_version_name = "STABILITY_SCP";
 
        server_ = std::make_unique<dicom_server>(config);
 
        file_storage_config fs_conf;
        fs_conf.root_path = storage_dir_;
        file_storage_ = std::make_unique<file_storage>(fs_conf);
 
        auto db_result = index_database::open(db_path_.string());
        if (db_result.is_ok()) {
            database_ = std::move(db_result.value());
        }
    }
 
    bool initialize() {
        server_->register_service(std::make_shared<verification_scp>());
 
        auto storage_scp_ptr = std::make_shared<storage_scp>();
        storage_scp_ptr->set_handler([this](
            const dicom_dataset& dataset,
            const std::string& calling_ae,
            const std::string& sop_class_uid,
            const std::string& sop_instance_uid) {
            return handle_store(dataset, calling_ae, sop_class_uid, sop_instance_uid);
        });
        server_->register_service(storage_scp_ptr);
 
        return true;
    }
 
    bool start() {
        auto result = server_->start();
        if (result.is_ok()) {
            std::this_thread::sleep_for(std::chrono::milliseconds{200});
            return true;
        }
        return false;
    }
 
    void stop() {
        server_->stop();
    }
 
    uint16_t port() const { return port_; }
    const std::string& ae_title() const { return ae_title_; }
 
    size_t stored_count() const { return stored_count_.load(); }
    size_t error_count() const { return error_count_.load(); }
 
    index_database* database() { return database_.get(); }
 
    std::vector<std::string> stored_instance_uids() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return stored_instance_uids_;
    }
 
    bool verify_consistency() const {
        std::lock_guard<std::mutex> lock(mutex_);
        std::set<std::string> unique_uids(
            stored_instance_uids_.begin(),
            stored_instance_uids_.end());
        return unique_uids.size() == stored_count_.load();
    }
 
private:
    storage_status handle_store(
        const dicom_dataset& dataset,
        const std::string& /* calling_ae */,
        const std::string& /* sop_class_uid */,
        const std::string& sop_instance_uid) {
 
        auto store_result = file_storage_->store(dataset);
        if (store_result.is_err()) {
            ++error_count_;
            return storage_status::storage_error;
        }
 
        // Note: Database indexing is handled by the storage layer
        // For stability testing, we only track file storage success
 
        {
            std::lock_guard<std::mutex> lock(mutex_);
            stored_instance_uids_.push_back(sop_instance_uid);
        }
        ++stored_count_;
 
        return storage_status::success;
    }
 
    uint16_t port_;
    std::string ae_title_;
    test_directory test_dir_;
    std::filesystem::path storage_dir_;
    std::filesystem::path db_path_;
    std::unique_ptr<dicom_server> server_;
    std::unique_ptr<file_storage> file_storage_;
    std::unique_ptr<index_database> database_;
 
    std::atomic<size_t> stored_count_{0};
    std::atomic<size_t> error_count_{0};
    mutable std::mutex mutex_;
    std::vector<std::string> stored_instance_uids_;
};
 
// =============================================================================
// Random Dataset Generator
// =============================================================================
 
dicom_dataset generate_random_dataset() {
    static std::mt19937 gen(std::random_device{}());
    std::uniform_int_distribution<int> modality_dist(0, 3);
 
    switch (modality_dist(gen)) {
        case 0: return test_data_generator::ct();
        case 1: return test_data_generator::mr();
        case 2: return test_data_generator::xa();
        case 3: return test_data_generator::us();
        default: return test_data_generator::ct();
    }
}
 
}  // anonymous namespace
 
// =============================================================================
// Test Cases
// =============================================================================
 
TEST_CASE("Continuous store/query operation", "[stability][.slow]") {
    auto config = stability_config::from_environment();
    stability_metrics metrics;
    metrics.reset();
 
    // Record initial memory
    metrics.initial_memory_kb = get_process_memory_kb();
    metrics.peak_memory_kb = metrics.initial_memory_kb;
 
    INFO("Starting stability test for " << config.duration.count() << " minutes");
 
    uint16_t port = find_available_port();
    stability_test_server server(port);
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    std::atomic<bool> running{true};
    std::vector<std::thread> workers;
 
    // Store workers
    for (size_t i = 0; i < config.store_workers; ++i) {
        workers.emplace_back([&, i]() {
            auto interval = std::chrono::milliseconds{
                static_cast<long>(1000.0 / config.store_rate * config.store_workers)};
 
            while (running.load()) {
                try {
                    auto ds = generate_random_dataset();
 
                    association_config assoc_config;
                    assoc_config.calling_ae_title = "STORE_SCU_" + std::to_string(i);
                    assoc_config.called_ae_title = server.ae_title();
                    assoc_config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.9";
                    assoc_config.proposed_contexts.push_back({
                        1,
                        ds.get_string(tags::sop_class_uid),
                        {"1.2.840.10008.1.2.1", "1.2.840.10008.1.2"}
                    });
 
                    auto assoc_result = association::connect(
                        "127.0.0.1", port, assoc_config, std::chrono::seconds{30});
 
                    if (assoc_result.is_ok()) {
                        ++metrics.connections_opened;
 
                        auto& assoc = assoc_result.value();
                        storage_scu scu;
                        auto store_result = scu.store(assoc, ds);
 
                        if (store_result.is_ok()) {
                            ++metrics.stores_completed;
                        } else {
                            ++metrics.stores_failed;
                        }
 
                        (void)assoc.release(std::chrono::seconds{5});
                        ++metrics.connections_closed;
                    } else {
                        ++metrics.connection_errors;
                    }
                } catch (...) {
                    ++metrics.stores_failed;
                }
 
                std::this_thread::sleep_for(interval);
            }
        });
    }
 
    // Memory monitoring thread
    workers.emplace_back([&]() {
        while (running.load()) {
            size_t current = get_process_memory_kb();
            size_t peak = metrics.peak_memory_kb.load();
            while (current > peak &&
                   !metrics.peak_memory_kb.compare_exchange_weak(peak, current)) {
                // Retry until successful
            }
            std::this_thread::sleep_for(std::chrono::seconds{5});
        }
    });
 
    // Run for configured duration
    std::this_thread::sleep_for(config.duration);
 
    running = false;
    for (auto& w : workers) {
        if (w.joinable()) {
            w.join();
        }
    }
 
    server.stop();
 
    // Report results
    metrics.report(std::cout);
 
    // Save to file for CI analysis
    auto report_path = std::filesystem::temp_directory_path() / "stability_test_report.txt";
    metrics.save_to_file(report_path);
    INFO("Report saved to: " << report_path.string());
 
    // Verify no critical errors
    REQUIRE(metrics.stores_failed.load() == 0);
    REQUIRE(metrics.connection_errors.load() == 0);
    REQUIRE(server.error_count() == 0);
 
    // Verify memory growth is bounded
    size_t memory_growth = (metrics.peak_memory_kb - metrics.initial_memory_kb) / 1024;
    REQUIRE(memory_growth < config.max_memory_growth_mb);
}
 
TEST_CASE("Memory stability over iterations", "[stability][memory]") {
    uint16_t port = find_available_port();
    stability_test_server server(port);
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    size_t initial_memory = get_process_memory_kb();
    constexpr size_t num_iterations = 100;
    constexpr size_t max_growth_kb = 50 * 1024;  // 50 MB max growth
 
    for (size_t i = 0; i < num_iterations; ++i) {
        auto ds = generate_random_dataset();
 
        association_config assoc_config;
        assoc_config.calling_ae_title = "MEM_TEST_SCU";
        assoc_config.called_ae_title = server.ae_title();
        assoc_config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.10";
        assoc_config.proposed_contexts.push_back({
            1,
            ds.get_string(tags::sop_class_uid),
            {"1.2.840.10008.1.2.1", "1.2.840.10008.1.2"}
        });
 
        auto assoc_result = association::connect(
            "127.0.0.1", port, assoc_config, std::chrono::seconds{30});
 
        REQUIRE(assoc_result.is_ok());
 
        auto& assoc = assoc_result.value();
        storage_scu scu;
        auto store_result = scu.store(assoc, ds);
        REQUIRE(store_result.is_ok());
 
        (void)assoc.release(std::chrono::seconds{5});
 
        // Check memory growth periodically
        if ((i + 1) % 20 == 0) {
            size_t current_memory = get_process_memory_kb();
            size_t growth = current_memory - initial_memory;
 
            INFO("Iteration " << (i + 1) << ": Memory growth = "
                 << growth / 1024 << " MB");
 
            REQUIRE(growth < max_growth_kb);
        }
    }
 
    server.stop();
    REQUIRE(server.stored_count() == num_iterations);
}
 
TEST_CASE("Connection pool exhaustion recovery", "[stability][network]") {
    uint16_t port = find_available_port();
    stability_test_server server(port);
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    constexpr size_t max_concurrent = 20;
    constexpr size_t cycles = 5;
 
    for (size_t cycle = 0; cycle < cycles; ++cycle) {
        INFO("Cycle " << (cycle + 1) << " of " << cycles);
 
        std::vector<association> associations;
        associations.reserve(max_concurrent);
 
        // Open many associations
        for (size_t i = 0; i < max_concurrent; ++i) {
            association_config assoc_config;
            assoc_config.calling_ae_title = "POOL_TEST_" + std::to_string(i);
            assoc_config.called_ae_title = server.ae_title();
            assoc_config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.11";
            assoc_config.proposed_contexts.push_back({1, "1.2.840.10008.1.1", {"1.2.840.10008.1.2"}});
 
            auto assoc_result = association::connect(
                "127.0.0.1", port, assoc_config, std::chrono::seconds{30});
 
            REQUIRE(assoc_result.is_ok());
            associations.push_back(std::move(assoc_result.value()));
        }
 
        // Release all associations
        for (auto& assoc : associations) {
            (void)assoc.release(std::chrono::seconds{5});
        }
        associations.clear();
 
        // Allow server to clean up
        std::this_thread::sleep_for(std::chrono::milliseconds{500});
 
        // Verify server can still accept new connections
        association_config verify_config;
        verify_config.calling_ae_title = "VERIFY_SCU";
        verify_config.called_ae_title = server.ae_title();
        verify_config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.12";
        verify_config.proposed_contexts.push_back({1, "1.2.840.10008.1.1", {"1.2.840.10008.1.2"}});
 
        auto test_result = association::connect(
            "127.0.0.1", port, verify_config, std::chrono::seconds{30});
 
        REQUIRE(test_result.is_ok());
        (void)test_result.value().release(std::chrono::seconds{5});
    }
 
    server.stop();
}
 
TEST_CASE("Database integrity under concurrent load", "[stability][database]") {
    uint16_t port = find_available_port();
    stability_test_server server(port);
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    constexpr size_t num_workers = 4;
    constexpr size_t images_per_worker = 25;
    constexpr size_t total_images = num_workers * images_per_worker;
 
    std::atomic<size_t> completed{0};
    std::atomic<size_t> failed{0};
    std::vector<std::thread> workers;
    std::vector<std::string> all_instance_uids;
    std::mutex uid_mutex;
 
    for (size_t w = 0; w < num_workers; ++w) {
        workers.emplace_back([&, w]() {
            for (size_t i = 0; i < images_per_worker; ++i) {
                auto ds = generate_random_dataset();
                std::string instance_uid = ds.get_string(tags::sop_instance_uid);
 
                {
                    std::lock_guard<std::mutex> lock(uid_mutex);
                    all_instance_uids.push_back(instance_uid);
                }
 
                association_config assoc_config;
                assoc_config.calling_ae_title = "DB_TEST_" + std::to_string(w);
                assoc_config.called_ae_title = server.ae_title();
                assoc_config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.13";
                assoc_config.proposed_contexts.push_back({
                    1,
                    ds.get_string(tags::sop_class_uid),
                    {"1.2.840.10008.1.2.1", "1.2.840.10008.1.2"}
                });
 
                auto assoc_result = association::connect(
                    "127.0.0.1", port, assoc_config, std::chrono::seconds{30});
 
                if (assoc_result.is_ok()) {
                    auto& assoc = assoc_result.value();
                    storage_scu scu;
                    auto store_result = scu.store(assoc, ds);
 
                    if (store_result.is_ok()) {
                        ++completed;
                    } else {
                        ++failed;
                    }
 
                    (void)assoc.release(std::chrono::seconds{5});
                } else {
                    ++failed;
                }
            }
        });
    }
 
    for (auto& w : workers) {
        w.join();
    }
 
    server.stop();
 
    // Verify counts
    REQUIRE(completed.load() == total_images);
    REQUIRE(failed.load() == 0);
    REQUIRE(server.stored_count() == total_images);
 
    // Verify no duplicate UIDs were generated
    std::set<std::string> unique_uids(all_instance_uids.begin(), all_instance_uids.end());
    REQUIRE(unique_uids.size() == total_images);
 
    // Verify database consistency
    REQUIRE(server.verify_consistency());
}
 
TEST_CASE("Short stability smoke test", "[stability][smoke]") {
    // Quick 10-second test for CI validation
    auto config = stability_config::from_environment();
    config.duration = std::chrono::minutes{0};  // Override for smoke test
 
    stability_metrics metrics;
    metrics.reset();
    metrics.initial_memory_kb = get_process_memory_kb();
 
    uint16_t port = find_available_port();
    stability_test_server server(port);
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    std::atomic<bool> running{true};
    std::thread store_worker([&]() {
        while (running.load()) {
            auto ds = test_data_generator::ct();
 
            association_config assoc_config;
            assoc_config.calling_ae_title = "SMOKE_SCU";
            assoc_config.called_ae_title = server.ae_title();
            assoc_config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.14";
            assoc_config.proposed_contexts.push_back({
                1,
                ds.get_string(tags::sop_class_uid),
                {"1.2.840.10008.1.2.1", "1.2.840.10008.1.2"}
            });
 
            auto assoc_result = association::connect(
                "127.0.0.1", port, assoc_config, std::chrono::seconds{10});
 
            if (assoc_result.is_ok()) {
                auto& assoc = assoc_result.value();
                storage_scu scu;
                if (scu.store(assoc, ds).is_ok()) {
                    ++metrics.stores_completed;
                } else {
                    ++metrics.stores_failed;
                }
                (void)assoc.release(std::chrono::seconds{5});
            } else {
                ++metrics.connection_errors;
            }
 
            std::this_thread::sleep_for(std::chrono::milliseconds{100});
        }
    });
 
    // Run for 10 seconds
    std::this_thread::sleep_for(std::chrono::seconds{10});
 
    running = false;
    store_worker.join();
    server.stop();
 
    // Verify basic functionality
    REQUIRE(metrics.stores_completed.load() > 0);
    REQUIRE(metrics.stores_failed.load() == 0);
    REQUIRE(metrics.connection_errors.load() == 0);
 
    INFO("Smoke test completed: " << metrics.stores_completed.load() << " stores in 10 seconds");
}