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;
}