test_stress.cpp File Reference

Scenario 4: Multi-Association Stress Tests. More...

#include "test_fixtures.h"
#include <catch2/catch_test_macros.hpp>
#include "kcenon/pacs/network/dimse/dimse_message.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 <future>
#include <latch>
#include <mutex>
#include <random>
#include <thread>
#include <vector>
#include <set>

Include dependency graph for test_stress.cpp:

Go to the source code of this file.

Functions
	TEST_CASE ("Concurrent storage from multiple SCUs", "[stress][concurrent]")
	TEST_CASE ("Rapid sequential connections", "[stress][sequential]")
	TEST_CASE ("Large dataset storage", "[stress][large]")
	TEST_CASE ("Connection pool exhaustion recovery", "[stress][exhaustion]")
	TEST_CASE ("Mixed operations stress test", "[stress][mixed]")

Detailed Description

Scenario 4: Multi-Association Stress Tests.

Tests system behavior under load:

1. Start Storage SCP
2. Launch multiple concurrent Storage SCUs
3. Each SCU sends multiple files
4. Verify all files stored
5. Verify database consistency
6. Stop Storage SCP

See also

Issue #111 - Integration Test Suite

Definition in file test_stress.cpp.

Function Documentation

TEST_CASE() [1/5]

TEST_CASE	(	"Concurrent storage from multiple SCUs"	,
		""	[stress][concurrent] )

Definition at line 338 of file test_stress.cpp.

                                                                           {
    auto port = find_available_port();
    stress_test_server server(port, "STRESS_SCP");
 
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    // Test parameters - reduced for CI/testing
    constexpr int num_workers = 5;
    constexpr int files_per_worker = 10;
    constexpr int total_expected = num_workers * files_per_worker;
 
    std::latch start_latch(num_workers + 1);  // +1 for main thread
    std::vector<std::future<worker_result>> futures;
 
    // Launch workers
    for (int i = 0; i < num_workers; ++i) {
        futures.push_back(std::async(
            std::launch::async,
            run_storage_worker,
            port,
            server.ae_title(),
            std::to_string(i),
            files_per_worker,
            std::ref(start_latch)
        ));
    }
 
    // Release all workers simultaneously
    start_latch.arrive_and_wait();
 
    // Collect results
    size_t total_success = 0;
    size_t total_failure = 0;
    std::chrono::milliseconds max_duration{0};
 
    for (auto& future : futures) {
        auto result = future.get();
        total_success += result.success_count;
        total_failure += result.failure_count;
        max_duration = (std::max)(max_duration, result.duration);
 
        if (!result.error_message.empty()) {
            INFO("Worker error: " << result.error_message);
        }
    }
 
    INFO("Total success: " << total_success);
    INFO("Total failure: " << total_failure);
    INFO("Max duration: " << max_duration.count() << " ms");
    INFO("Server stored: " << server.stored_count());
 
    // Verify results
    REQUIRE(total_success == total_expected);
    REQUIRE(total_failure == 0);
    REQUIRE(server.stored_count() == total_expected);
    REQUIRE(server.verify_consistency());
 
    server.stop();
}

References kcenon::pacs::integration_test::find_available_port().

Here is the call graph for this function:

TEST_CASE() [2/5]

TEST_CASE	(	"Connection pool exhaustion recovery"	,
		""	[stress][exhaustion] )

Definition at line 515 of file test_stress.cpp.

                                                                         {
    auto port = find_available_port();
    stress_test_server server(port, "STRESS_SCP");
 
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    // Hold connections open
    constexpr int num_held = 10;
    std::vector<std::optional<association>> held_connections;
 
    for (int i = 0; i < num_held; ++i) {
        association_config config;
        config.calling_ae_title = "HOLD_" + std::to_string(i);
        config.called_ae_title = server.ae_title();
        config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.11";
        config.proposed_contexts.push_back({
            1,
            std::string(verification_sop_class_uid),
            {"1.2.840.10008.1.2.1"}
        });
 
        auto connect_result = association::connect(
            "localhost", port, config, default_timeout());
        if (connect_result.is_ok()) {
            held_connections.push_back(std::move(connect_result.value()));
        }
    }
 
    REQUIRE(held_connections.size() == num_held);
 
    // Try more connections (should still work due to max_associations = 50)
    constexpr int num_additional = 5;
    size_t additional_success = 0;
 
    for (int i = 0; i < num_additional; ++i) {
        association_config config;
        config.calling_ae_title = "EXTRA_" + std::to_string(i);
        config.called_ae_title = server.ae_title();
        config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.12";
        config.proposed_contexts.push_back({
            1,
            std::string(verification_sop_class_uid),
            {"1.2.840.10008.1.2.1"}
        });
 
        auto connect_result = association::connect(
            "localhost", port, config, default_timeout());
        if (connect_result.is_ok()) {
            auto& assoc = connect_result.value();
            (void)assoc.release(std::chrono::milliseconds{500});
            ++additional_success;
        }
    }
 
    REQUIRE(additional_success == num_additional);
 
    // Release held connections
    for (auto& opt_assoc : held_connections) {
        if (opt_assoc) {
            (void)opt_assoc->release(std::chrono::milliseconds{500});
        }
    }
    held_connections.clear();
 
    // Verify new connections work after release
    association_config config;
    config.calling_ae_title = "AFTER_RELEASE";
    config.called_ae_title = server.ae_title();
    config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.13";
    config.proposed_contexts.push_back({
        1,
        std::string(verification_sop_class_uid),
        {"1.2.840.10008.1.2.1"}
    });
 
    auto final_connect = association::connect(
        "localhost", port, config, default_timeout());
    REQUIRE(final_connect.is_ok());
    (void)final_connect.value().release(default_timeout());
 
    server.stop();
}

References kcenon::pacs::network::association_config::called_ae_title, kcenon::pacs::network::association_config::calling_ae_title, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::network::association_config::implementation_class_uid, kcenon::pacs::network::association_config::proposed_contexts, and kcenon::pacs::services::verification_sop_class_uid.

Here is the call graph for this function:

TEST_CASE() [3/5]

TEST_CASE	(	"Large dataset storage"	,
		""	[stress][large] )

Definition at line 435 of file test_stress.cpp.

                                                      {
    auto port = find_available_port();
    stress_test_server server(port, "STRESS_SCP");
 
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    // Create larger dataset (512x512 instead of 64x64)
    dicom_dataset large_ds;
 
    // Standard patient/study info
    large_ds.set_string(tags::patient_name, vr_type::PN, "LARGE^DATASET");
    large_ds.set_string(tags::patient_id, vr_type::LO, "LARGE001");
    large_ds.set_string(tags::study_instance_uid, vr_type::UI, generate_uid());
    large_ds.set_string(tags::series_instance_uid, vr_type::UI, generate_uid());
    large_ds.set_string(tags::sop_class_uid, vr_type::UI, "1.2.840.10008.5.1.4.1.1.2");
    large_ds.set_string(tags::sop_instance_uid, vr_type::UI, generate_uid());
    large_ds.set_string(tags::modality, vr_type::CS, "CT");
 
    // Large image (512x512 16-bit = 512KB pixel data)
    constexpr int rows = 512;
    constexpr int cols = 512;
    large_ds.set_numeric<uint16_t>(tags::rows, vr_type::US, rows);
    large_ds.set_numeric<uint16_t>(tags::columns, vr_type::US, cols);
    large_ds.set_numeric<uint16_t>(tags::bits_allocated, vr_type::US, 16);
    large_ds.set_numeric<uint16_t>(tags::bits_stored, vr_type::US, 12);
    large_ds.set_numeric<uint16_t>(tags::high_bit, vr_type::US, 11);
    large_ds.set_numeric<uint16_t>(tags::pixel_representation, vr_type::US, 0);
    large_ds.set_numeric<uint16_t>(tags::samples_per_pixel, vr_type::US, 1);
    large_ds.set_string(tags::photometric_interpretation, vr_type::CS, "MONOCHROME2");
 
    // Generate pixel data
    std::vector<uint16_t> pixel_data(rows * cols);
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_int_distribution<uint16_t> dist(0, 4095);
    for (auto& pixel : pixel_data) {
        pixel = dist(gen);
    }
    large_ds.insert(dicom_element(
        tags::pixel_data,
        vr_type::OW,
        std::span<const uint8_t>(
            reinterpret_cast<const uint8_t*>(pixel_data.data()),
            pixel_data.size() * sizeof(uint16_t))));
 
    // Store the large dataset
    association_config config;
    config.calling_ae_title = "LARGE_SCU";
    config.called_ae_title = server.ae_title();
    config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.10";
    config.proposed_contexts.push_back({
        1,
        "1.2.840.10008.5.1.4.1.1.2",
        {"1.2.840.10008.1.2.1"}
    });
 
    auto connect_result = association::connect(
        "localhost", port, config, std::chrono::milliseconds{10000});
    REQUIRE(connect_result.is_ok());
 
    auto& assoc = connect_result.value();
    storage_scu_config scu_config;
    scu_config.response_timeout = std::chrono::milliseconds{10000};
    storage_scu scu{scu_config};
 
    auto start = std::chrono::steady_clock::now();
    auto result = scu.store(assoc, large_ds);
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start);
 
    INFO("Large dataset storage took: " << duration.count() << " ms");
 
    REQUIRE(result.is_ok());
    REQUIRE(result.value().is_success());
 
    (void)assoc.release(default_timeout());
    server.stop();
}

References kcenon::pacs::network::association_config::called_ae_title, kcenon::pacs::network::association_config::calling_ae_title, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::integration_test::generate_uid(), kcenon::pacs::network::association_config::implementation_class_uid, kcenon::pacs::core::dicom_dataset::insert(), kcenon::pacs::network::association_config::proposed_contexts, kcenon::pacs::services::storage_scu_config::response_timeout, kcenon::pacs::core::dicom_dataset::set_numeric(), and kcenon::pacs::core::dicom_dataset::set_string().

Here is the call graph for this function:

TEST_CASE() [4/5]

TEST_CASE	(	"Mixed operations stress test"	,
		""	[stress][mixed] )

Definition at line 599 of file test_stress.cpp.

                                                             {
    auto port = find_available_port();
    stress_test_server server(port, "STRESS_SCP");
 
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    constexpr int num_iterations = 10;
    std::atomic<int> echo_success{0};
    std::atomic<int> store_success{0};
 
    std::vector<std::thread> threads;
 
    // Echo workers
    for (int i = 0; i < 3; ++i) {
        threads.emplace_back([&, i]() {
            for (int j = 0; j < num_iterations; ++j) {
                association_config config;
                config.calling_ae_title = "ECHO_" + std::to_string(i);
                config.called_ae_title = server.ae_title();
                config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.14";
                config.proposed_contexts.push_back({
                    1,
                    std::string(verification_sop_class_uid),
                    {"1.2.840.10008.1.2.1"}
                });
 
                auto connect = association::connect(
                    "localhost", port, config, default_timeout());
                if (connect.is_ok()) {
                    auto& assoc = connect.value();
                    auto ctx = assoc.accepted_context_id(verification_sop_class_uid);
                    if (ctx) {
                        auto echo_rq = make_c_echo_rq(1, verification_sop_class_uid);
                        if (assoc.send_dimse(*ctx, echo_rq).is_ok()) {
                            auto recv = assoc.receive_dimse(default_timeout());
                            if (recv.is_ok() && recv.value().second.status() == status_success) {
                                ++echo_success;
                            }
                        }
                    }
                    (void)assoc.release(std::chrono::milliseconds{500});
                }
            }
        });
    }
 
    // Store workers
    for (int i = 0; i < 2; ++i) {
        threads.emplace_back([&, i]() {
            for (int j = 0; j < num_iterations; ++j) {
                association_config config;
                config.calling_ae_title = "STORE_" + std::to_string(i);
                config.called_ae_title = server.ae_title();
                config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.15";
                config.proposed_contexts.push_back({
                    1,
                    "1.2.840.10008.5.1.4.1.1.2",
                    {"1.2.840.10008.1.2.1"}
                });
 
                auto connect = association::connect(
                    "localhost", port, config, default_timeout());
                if (connect.is_ok()) {
                    auto& assoc = connect.value();
                    storage_scu scu;
                    auto ds = generate_ct_dataset();
                    auto result = scu.store(assoc, ds);
                    if (result.is_ok() && result.value().is_success()) {
                        ++store_success;
                    }
                    (void)assoc.release(std::chrono::milliseconds{500});
                }
            }
        });
    }
 
    for (auto& t : threads) {
        t.join();
    }
 
    INFO("Echo success: " << echo_success.load());
    INFO("Store success: " << store_success.load());
 
    REQUIRE(echo_success == 3 * num_iterations);
    REQUIRE(store_success == 2 * num_iterations);
 
    server.stop();
}

References kcenon::pacs::network::association_config::called_ae_title, kcenon::pacs::network::association_config::calling_ae_title, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::integration_test::generate_ct_dataset(), kcenon::pacs::network::association_config::implementation_class_uid, kcenon::pacs::network::dimse::make_c_echo_rq(), kcenon::pacs::network::association_config::proposed_contexts, kcenon::pacs::services::storage_scu::store(), and kcenon::pacs::services::verification_sop_class_uid.

Here is the call graph for this function:

TEST_CASE() [5/5]

TEST_CASE	(	"Rapid sequential connections"	,
		""	[stress][sequential] )

Definition at line 399 of file test_stress.cpp.

                                                                  {
    auto port = find_available_port();
    stress_test_server server(port, "STRESS_SCP");
 
    REQUIRE(server.initialize());
    REQUIRE(server.start());
 
    constexpr int num_connections = 20;
    size_t success_count = 0;
 
    for (int i = 0; i < num_connections; ++i) {
        association_config config;
        config.calling_ae_title = "RAPID_" + std::to_string(i);
        config.called_ae_title = server.ae_title();
        config.implementation_class_uid = "1.2.826.0.1.3680043.9.9999.9";
        config.proposed_contexts.push_back({
            1,
            std::string(verification_sop_class_uid),
            {"1.2.840.10008.1.2.1"}
        });
 
        auto connect_result = association::connect(
            "localhost", port, config, default_timeout());
 
        if (connect_result.is_ok()) {
            auto& assoc = connect_result.value();
            (void)assoc.release(std::chrono::milliseconds{500});
            ++success_count;
        }
    }
 
    REQUIRE(success_count == num_connections);
 
    server.stop();
}

References kcenon::pacs::network::association_config::called_ae_title, kcenon::pacs::network::association_config::calling_ae_title, kcenon::pacs::integration_test::default_timeout(), kcenon::pacs::integration_test::find_available_port(), kcenon::pacs::network::association_config::implementation_class_uid, kcenon::pacs::network::association_config::proposed_contexts, and kcenon::pacs::services::verification_sop_class_uid.

Here is the call graph for this function: