test_runner.cpp

#include "TCMallocator.h"
#include <iostream>
#include <thread>
#include <vector>
#include <atomic>
#include <chrono>
#include <random>
#include <iomanip>

// 每个线程的统计结构体
struct ThreadStats {
    uint64_t ops = 0;
    uint64_t total_latency_ns = 0; // 累计采样延迟（纳秒）
    uint64_t sample_count = 0;      // 采样次数
};

// 工作线程函数：持续进行随机大小的内存申请与释放，并带延迟抽样
void stress_worker(std::atomic<uint64_t>& global_ops, 
                   std::atomic<uint64_t>& global_latency_ns, 
                   std::atomic<uint64_t>& global_samples, 
                   std::atomic<bool>& running) {
    auto& allocator = TCMallocator::instance();
    
    std::random_device rd;
    std::mt19937 gen(rd() + std::hash<std::thread::id>{}(std::this_thread::get_id()));
    std::uniform_int_distribution<std::size_t> size_dist(16, 256);

    uint64_t local_ops = 0;
    uint64_t local_latency_ns = 0;
    uint64_t local_samples = 0;

    uint32_t counter = 0;

    while (running.load(std::memory_order_relaxed)) {
        std::size_t size = size_dist(gen);

        // 每 1024 次分配抽样一次延迟，避免高频系统时间调用影响性能
        bool measure = (++counter % 1024 == 0);
        
        auto start = measure ? std::chrono::steady_clock::now() : std::chrono::steady_clock::time_point{};
        
        void* ptr = allocator.allocate(size);
        
        if (measure) {
            auto end = std::chrono::steady_clock::now();
            local_latency_ns += std::chrono::duration_cast<std::chrono::nanoseconds>(end - start).count();
            local_samples++;
        }

        if (ptr) {
            allocator.deallocate(ptr);
            local_ops++;
        }

        // 每完成 1000 次操作同步一次到全局原子变量，减少总线风暴，同时保证实时性
        if (local_ops >= 1000) {
            global_ops.fetch_add(local_ops, std::memory_order_relaxed);
            global_latency_ns.fetch_add(local_latency_ns, std::memory_order_relaxed);
            global_samples.fetch_add(local_samples, std::memory_order_relaxed);
            
            local_ops = 0;
            local_latency_ns = 0;
            local_samples = 0;
        }
    }

    // 循环结束后把剩余的累加进去
    if (local_ops > 0) {
        global_ops.fetch_add(local_ops, std::memory_order_relaxed);
        global_latency_ns.fetch_add(local_latency_ns, std::memory_order_relaxed);
        global_samples.fetch_add(local_samples, std::memory_order_relaxed);
    }
}

int main() {
    constexpr int num_threads = 16;
    constexpr int test_duration_minutes = 30; // 运行 30 分钟

    std::atomic<bool> running{true};
    std::atomic<uint64_t> global_ops{0};
    std::atomic<uint64_t> global_latency_ns{0};
    std::atomic<uint64_t> global_samples{0};

    std::cout << "==================================================\n";
    std::cout << " TCMallocator 30-Minute Real-time Stress Test     \n";
    std::cout << " Threads: " << num_threads << " | Target Duration: " << test_duration_minutes << " minutes\n";
    std::cout << "==================================================\n";
    std::cout << "Test started. Real-time statistics per minute:\n\n";

    auto start_time = std::chrono::steady_clock::now();

    // 启动 16 个高并发工作线程
    std::vector<std::thread> workers;
    workers.reserve(num_threads);
    for (int i = 0; i < num_threads; ++i) {
        workers.emplace_back(stress_worker, 
                             std::ref(global_ops), 
                             std::ref(global_latency_ns), 
                             std::ref(global_samples), 
                             std::ref(running));
    }

    uint64_t last_ops = 0;
    uint64_t last_latency_ns = 0;
    uint64_t last_samples = 0;

    // 每分钟监控并输出一次数据
    for (int minute = 1; minute <= test_duration_minutes; ++minute) {
        for (int sec = 0; sec < 60; ++sec) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
            if (!running.load()) break;
        }

        uint64_t current_ops = global_ops.load(std::memory_order_relaxed);
        uint64_t current_latency = global_latency_ns.load(std::memory_order_relaxed);
        uint64_t current_samples = global_samples.load(std::memory_order_relaxed);

        uint64_t ops_this_minute = current_ops - last_ops;
        uint64_t latency_diff = current_latency - last_latency_ns;
        uint64_t samples_diff = current_samples - last_samples;

        double qps_this_minute = static_cast<double>(ops_this_minute) / 60.0;
        double avg_latency_ns = (samples_diff > 0) ? static_cast<double>(latency_diff) / samples_diff : 0.0;

        std::cout << "[Minute " << std::setw(2) << minute << "/" << test_duration_minutes 
                  << "] Ops: " << std::setw(10) << ops_this_minute 
                  << " | QPS: " << std::setw(10) << std::fixed << std::setprecision(2) << qps_this_minute 
                  << " | Avg Latency: " << std::setw(6) << std::setprecision(2) << avg_latency_ns << " ns"
                  << std::endl;

        last_ops = current_ops;
        last_latency_ns = current_latency;
        last_samples = current_samples;
    }

    // 时间到，通知所有线程停止
    running.store(false, std::memory_order_relaxed);

    for (auto& t : workers) {
        t.join();
    }

    auto end_time = std::chrono::steady_clock::now();
    std::chrono::duration<double> elapsed = end_time - start_time;

    uint64_t final_ops = global_ops.load();
    double overall_qps = static_cast<double>(final_ops) / elapsed.count();

    std::cout << "\n==================================================\n";
    std::cout << " 30-Minute Stress Test Completed Summary          \n";
    std::cout << "==================================================\n";
    std::cout << "Total Alloc/Dealloc pairs: " << final_ops << " ops\n";
    std::cout << "Actual elapsed time:       " << elapsed.count() << " seconds\n";
    std::cout << ">>> Overall Average QPS:   " << overall_qps << " ops/sec <<<\n";
    std::cout << "==================================================\n";

    return 0;
}