test_soak.cpp

#include "memory.hpp"
#include <iostream>
#include <vector>
#include <thread>
#include <chrono>
#include <atomic>
#include <random>
#include <cassert>
#include <fstream>
#include <string>
#include <iomanip>

#include <unistd.h>

// 魔法数，用于检测内存踩踏（Corruption）
constexpr uint32_t MAGIC_NUMBER = 0xDEADBEEF;

struct Payload {
    uint32_t magic;
    uint32_t thread_id;
    uint64_t seq;
    char dummy[16]; // 填充数据，保证 Payload 体积
};

// 获取当前进程物理内存占用 (RSS, 单位: MB)
double get_process_rss_mb() {
    std::ifstream statm("/proc/self/statm");
    if (!statm.is_open()) return 0.0;
    long pages = 0;
    long rss_pages = 0;
    statm >> pages >> rss_pages;
    long page_size_kb = sysconf(_SC_PAGESIZE) / 1024;
    return (rss_pages * page_size_kb) / 1024.0;
}

// 模拟真实高压业务的工作线程
void worker_thread(FixedSizeMemory& pool, int thread_id, std::atomic<bool>& stop_flag, std::atomic<uint64_t>& total_ops) {
    std::mt19937 rng(1337 + thread_id);

    constexpr std::size_t MAX_HOLD = 1024;
    std::vector<Payload*> held_ptrs;
    held_ptrs.reserve(MAX_HOLD);

    uint64_t local_seq = 0;

    while (!stop_flag.load(std::memory_order_relaxed)) {
        // 1. 批量分配内存 (1 ~ 32 块)
        std::size_t batch_size = (rng() % 32) + 1;
        for (std::size_t i = 0; i < batch_size; ++i) {
            void* p = pool.allocate();
            if (p != nullptr) {
                Payload* payload = new (p) Payload(); // placement new
                payload->magic = MAGIC_NUMBER;
                payload->thread_id = static_cast<uint32_t>(thread_id);
                payload->seq = ++local_seq;
                held_ptrs.push_back(payload);
            }
        }

        total_ops.fetch_add(batch_size, std::memory_order_relaxed);

        // 2. 随机决定是“释放部分内存”还是“继续持有”（制造动态水位波动）
        if (!held_ptrs.empty()) {
            // 如果手里的内存堆积太多（超过 512），优先大比例释放
            std::size_t release_count = (held_ptrs.size() > 512) ? 
                                         held_ptrs.size() / 2 : 
                                         (rng() % (held_ptrs.size() + 1));

            for (std::size_t i = 0; i < release_count; ++i) {
                Payload* payload = held_ptrs.back();
                held_ptrs.pop_back();

                // 踩内存校验：如果 Magic 被篡改，直接崩溃
                assert(payload->magic == MAGIC_NUMBER && "内存发生数据越界/踩内存！");
                
                payload->~Payload();
                pool.deallocate(payload);
            }
        }

        // 3. 极短让出 CPU，模拟真实业务微小停顿
        if (rng() % 10 == 0) {
            std::this_thread::yield();
        }
    }

    // 4. 线程退出前，把手里剩下的全部释放干净
    for (Payload* payload : held_ptrs) {
        assert(payload->magic == MAGIC_NUMBER && "内存发生数据越界/踩内存！");
        payload->~Payload();
        pool.deallocate(payload);
    }
    held_ptrs.clear();

    // 手动刷新该线程的 TLS 缓存到 Page 中，确保退出时不残留
    pool.flush_thread_cache();
}

int main(int argc, char* argv[]) {
    // 默认跑 600 秒 (10分钟)，命令行传参可指定跑多久
    int duration_seconds = 600;
    if (argc > 1) {
        duration_seconds = std::atoi(argv[1]);
    }

    std::cout << "==========================================================================" << std::endl;
    std::cout << "        内存池长跑稳定性 + 分段 QPS 性能衰退监测 (Soak Testing) " << std::endl;
    std::cout << " 压测时长: " << duration_seconds << " 秒" << std::endl;
    std::cout << "==========================================================================" << std::endl;

    // 创建单块 64 字节，每页 1024 块的内存池
    FixedSizeMemory pool(64, 1024);

    unsigned int num_threads = std::thread::hardware_concurrency();
    if (num_threads == 0) num_threads = 4;
    std::cout << "启动压测线程数: " << num_threads << std::endl;

    std::atomic<bool> stop_flag{false};
    std::atomic<uint64_t> total_ops{0};
    std::vector<std::thread> workers;

    // 启动后台监控线程：每 10 秒监控一次分段 QPS 与内存占用
    std::thread monitor_thread([&]() {
        auto start_time = std::chrono::steady_clock::now();
        uint64_t last_ops = 0;

        while (!stop_flag.load(std::memory_order_relaxed)) {
            // 每 10 秒统计一次区间 QPS
            std::this_thread::sleep_for(std::chrono::seconds(10));
            auto now = std::chrono::steady_clock::now();
            auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(now - start_time).count();

            uint64_t current_ops = total_ops.load(std::memory_order_relaxed);
            uint64_t interval_ops = current_ops - last_ops;
            double current_qps = interval_ops / 10.0; // 过去 10 秒内的平均 QPS
            last_ops = current_ops;

            double rss_mb = get_process_rss_mb();
            std::size_t active_allocs = pool.allocated_count();
            std::size_t pages = pool.page_count();

            std::cout << "[" << std::setw(4) << elapsed << "s] "
                      << "区间QPS: " << std::setw(12) << static_cast<uint64_t>(current_qps) << " ops/s | "
                      << "累计操作: " << std::setw(12) << current_ops << " | "
                      << "活动块: " << std::setw(5) << active_allocs << " | "
                      << "Page数: " << std::setw(3) << pages << " | "
                      << "RSS内存: " << std::fixed << std::setprecision(2) << rss_mb << " MB"
                      << std::endl;
        }
    });

    // 启动所有工作线程
    for (unsigned int i = 0; i < num_threads; ++i) {
        workers.emplace_back(worker_thread, std::ref(pool), i, std::ref(stop_flag), std::ref(total_ops));
    }

    // 主线程等待压测倒计时结束
    std::this_thread::sleep_for(std::chrono::seconds(duration_seconds));

    // 通知所有线程平滑退出
    std::cout << "\n压测时间到，正在通知所有线程平滑退出并刷新 TLS..." << std::endl;
    stop_flag.store(true, std::memory_order_relaxed);

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

    if (monitor_thread.joinable()) {
        monitor_thread.join();
    }

    std::cout << "\n================ 最终收尾与内存泄漏验证 ================" << std::endl;
    std::cout << "总完成操作次数  : " << total_ops.load() << std::endl;
    std::cout << "当前未归还内存块: " << pool.allocated_count() << std::endl;
    std::cout << "最终申请 Page 数: " << pool.page_count() << std::endl;

    // 终极断言：所有线程退出并 flush TLS 后，未归还分配数必须严格归零！
    if (pool.allocated_count() == 0) {
        std::cout << " SUCCESS: 长跑压测通过！0 内存泄漏，0 数据损坏，性能平稳！" << std::endl;
    } else {
        std::cerr << " ERROR: 压测失败！检测到内存泄漏！未归还块数: " << pool.allocated_count() << std::endl;
        return 1;
    }

    return 0;
}