|
@@ -0,0 +1,333 @@
|
|
|
|
|
+// Test 2 — Mixed read/write workload under memory pressure
|
|
|
|
|
+//
|
|
|
|
|
+// Verifies that reads can progress while writes + eviction are running.
|
|
|
|
|
+// This mirrors the shadowman-cpp agent loop: it continuously reads chat
|
|
|
|
|
+// history while chunks of the new assistant turn stream in. If WAL-fallback
|
|
|
|
|
+// pages serialize behind write locks (pre-T8 behaviour), read p99 spikes.
|
|
|
|
|
+//
|
|
|
|
|
+// PASS criteria:
|
|
|
|
|
+// - writer ops/s > 500
|
|
|
|
|
+// - reader ops/s > 1000
|
|
|
|
|
+// - reader p99 latency < 1000 ms
|
|
|
|
|
+// - zero hard reader failures (RESOURCE_EXHAUSTED / timeout)
|
|
|
|
|
+// - not-found is tolerated — it's a cold cache miss, not a bug
|
|
|
|
|
+
|
|
|
|
|
+#include <smartbotic/database/client.hpp>
|
|
|
|
|
+
|
|
|
|
|
+#include <algorithm>
|
|
|
|
|
+#include <atomic>
|
|
|
|
|
+#include <chrono>
|
|
|
|
|
+#include <iomanip>
|
|
|
|
|
+#include <iostream>
|
|
|
|
|
+#include <mutex>
|
|
|
|
|
+#include <random>
|
|
|
|
|
+#include <string>
|
|
|
|
|
+#include <thread>
|
|
|
|
|
+#include <vector>
|
|
|
|
|
+
|
|
|
|
|
+using namespace smartbotic::database;
|
|
|
|
|
+using Clock = std::chrono::steady_clock;
|
|
|
|
|
+
|
|
|
|
|
+// Histogram covers 0..10000 ms in 100 ms buckets (100 buckets) + overflow.
|
|
|
|
|
+static constexpr size_t HIST_BUCKETS = 100; // 0..9999 ms in 100 ms slices
|
|
|
|
|
+static constexpr uint64_t HIST_BUCKET_MS = 100;
|
|
|
|
|
+
|
|
|
|
|
+struct Counters {
|
|
|
|
|
+ std::atomic<uint64_t> writerInserts{0};
|
|
|
|
|
+ std::atomic<uint64_t> writerFails{0};
|
|
|
|
|
+
|
|
|
|
|
+ std::atomic<uint64_t> readerOps{0};
|
|
|
|
|
+ std::atomic<uint64_t> readerNotFound{0};
|
|
|
|
|
+ std::atomic<uint64_t> readerTimeouts{0};
|
|
|
|
|
+ std::atomic<uint64_t> readerResourceExhausted{0};
|
|
|
|
|
+ std::atomic<uint64_t> readerOtherErrors{0};
|
|
|
|
|
+
|
|
|
|
|
+ std::atomic<uint64_t> latencyHist[HIST_BUCKETS + 1] = {}; // [HIST_BUCKETS] = overflow
|
|
|
|
|
+
|
|
|
|
|
+ std::atomic<bool> stop{false};
|
|
|
|
|
+};
|
|
|
|
|
+
|
|
|
|
|
+// Recently-inserted IDs, capped at 1000. Shared between writers and readers.
|
|
|
|
|
+struct RecentIds {
|
|
|
|
|
+ std::mutex mu;
|
|
|
|
|
+ std::vector<std::string> ids;
|
|
|
|
|
+ size_t maxSize = 1000;
|
|
|
|
|
+
|
|
|
|
|
+ void push(const std::string& id) {
|
|
|
|
|
+ std::lock_guard<std::mutex> lk(mu);
|
|
|
|
|
+ if (ids.size() >= maxSize) {
|
|
|
|
|
+ // Drop oldest (simple O(n) shift — cheap relative to gRPC).
|
|
|
|
|
+ ids.erase(ids.begin());
|
|
|
|
|
+ }
|
|
|
|
|
+ ids.push_back(id);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::optional<std::string> sample(std::mt19937& rng) {
|
|
|
|
|
+ std::lock_guard<std::mutex> lk(mu);
|
|
|
|
|
+ if (ids.empty()) return std::nullopt;
|
|
|
|
|
+ std::uniform_int_distribution<size_t> dist(0, ids.size() - 1);
|
|
|
|
|
+ return ids[dist(rng)];
|
|
|
|
|
+ }
|
|
|
|
|
+};
|
|
|
|
|
+
|
|
|
|
|
+static void recordLatency(Counters& c, int64_t micros) {
|
|
|
|
|
+ int64_t ms = micros / 1000;
|
|
|
|
|
+ size_t bucket = ms < 0 ? 0 : static_cast<size_t>(ms) / HIST_BUCKET_MS;
|
|
|
|
|
+ if (bucket >= HIST_BUCKETS) bucket = HIST_BUCKETS; // overflow slot
|
|
|
|
|
+ c.latencyHist[bucket].fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+static uint64_t percentile(const Counters& c, double p) {
|
|
|
|
|
+ uint64_t total = 0;
|
|
|
|
|
+ for (size_t i = 0; i <= HIST_BUCKETS; i++) {
|
|
|
|
|
+ total += c.latencyHist[i].load(std::memory_order_relaxed);
|
|
|
|
|
+ }
|
|
|
|
|
+ if (total == 0) return 0;
|
|
|
|
|
+ uint64_t target = static_cast<uint64_t>(std::ceil(total * p));
|
|
|
|
|
+ uint64_t running = 0;
|
|
|
|
|
+ for (size_t i = 0; i <= HIST_BUCKETS; i++) {
|
|
|
|
|
+ running += c.latencyHist[i].load(std::memory_order_relaxed);
|
|
|
|
|
+ if (running >= target) {
|
|
|
|
|
+ if (i == HIST_BUCKETS) return HIST_BUCKETS * HIST_BUCKET_MS; // >=10s
|
|
|
|
|
+ return (i + 1) * HIST_BUCKET_MS; // upper bound of this bucket
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return HIST_BUCKETS * HIST_BUCKET_MS;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+int main(int argc, char* argv[]) {
|
|
|
|
|
+ int durationSec = 30;
|
|
|
|
|
+ int writers = 3;
|
|
|
|
|
+ int readers = 5;
|
|
|
|
|
+ int docBytes = 2048;
|
|
|
|
|
+ if (argc > 1) durationSec = std::stoi(argv[1]);
|
|
|
|
|
+ if (argc > 2) writers = std::stoi(argv[2]);
|
|
|
|
|
+ if (argc > 3) readers = std::stoi(argv[3]);
|
|
|
|
|
+ if (argc > 4) docBytes = std::stoi(argv[4]);
|
|
|
|
|
+
|
|
|
|
|
+ std::cout << "=== smartbotic-database v1.7.0 load test — mixed read/write ===\n"
|
|
|
|
|
+ << "Server: localhost:9005\n"
|
|
|
|
|
+ << "Duration: " << durationSec << "s\n"
|
|
|
|
|
+ << "Writers: " << writers << "\n"
|
|
|
|
|
+ << "Readers: " << readers << "\n"
|
|
|
|
|
+ << "Doc size: " << docBytes << " bytes\n"
|
|
|
|
|
+ << "Memory cap: 32 MB (configured in server)\n"
|
|
|
|
|
+ << "\n";
|
|
|
|
|
+
|
|
|
|
|
+ Client::Config cfg;
|
|
|
|
|
+ cfg.address = "localhost:9005";
|
|
|
|
|
+ cfg.timeoutMs = 3000;
|
|
|
|
|
+ cfg.writeRetries = 5;
|
|
|
|
|
+ cfg.writeRetryBackoffMs = 50;
|
|
|
|
|
+ cfg.writeRetryMaxBackoffMs = 1000;
|
|
|
|
|
+
|
|
|
|
|
+ Client db(cfg);
|
|
|
|
|
+ if (!db.connect()) {
|
|
|
|
|
+ std::cerr << "connect failed\n";
|
|
|
|
|
+ return 1;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if (!db.createCollection("mixed")) {
|
|
|
|
|
+ std::cout << "(collection mixed already exists, continuing)\n";
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ Counters counters;
|
|
|
|
|
+ RecentIds recent;
|
|
|
|
|
+ recent.maxSize = 1000;
|
|
|
|
|
+
|
|
|
|
|
+ auto startTime = Clock::now();
|
|
|
|
|
+ std::string pad(docBytes, 'x');
|
|
|
|
|
+
|
|
|
|
|
+ // Writers
|
|
|
|
|
+ std::vector<std::thread> writerThreads;
|
|
|
|
|
+ for (int w = 0; w < writers; w++) {
|
|
|
|
|
+ writerThreads.emplace_back([&, w]() {
|
|
|
|
|
+ Client::Config wcfg = cfg;
|
|
|
|
|
+ Client wdb(wcfg);
|
|
|
|
|
+ if (!wdb.connect()) return;
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t seq = 0;
|
|
|
|
|
+ while (!counters.stop.load(std::memory_order_relaxed)) {
|
|
|
|
|
+ std::string id = "w" + std::to_string(w) + "-" + std::to_string(seq);
|
|
|
|
|
+ try {
|
|
|
|
|
+ nlohmann::json doc{
|
|
|
|
|
+ {"worker", w},
|
|
|
|
|
+ {"seq", seq},
|
|
|
|
|
+ {"pad", pad}
|
|
|
|
|
+ };
|
|
|
|
|
+ wdb.insert("mixed", doc, id);
|
|
|
|
|
+ counters.writerInserts.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ recent.push(id);
|
|
|
|
|
+ } catch (const std::exception&) {
|
|
|
|
|
+ counters.writerFails.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ }
|
|
|
|
|
+ seq++;
|
|
|
|
|
+ }
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Readers
|
|
|
|
|
+ std::vector<std::thread> readerThreads;
|
|
|
|
|
+ for (int r = 0; r < readers; r++) {
|
|
|
|
|
+ readerThreads.emplace_back([&, r]() {
|
|
|
|
|
+ Client::Config rcfg = cfg;
|
|
|
|
|
+ Client rdb(rcfg);
|
|
|
|
|
+ if (!rdb.connect()) return;
|
|
|
|
|
+
|
|
|
|
|
+ std::mt19937 rng(std::random_device{}() + r);
|
|
|
|
|
+
|
|
|
|
|
+ while (!counters.stop.load(std::memory_order_relaxed)) {
|
|
|
|
|
+ auto idOpt = recent.sample(rng);
|
|
|
|
|
+ if (!idOpt) {
|
|
|
|
|
+ // Writers haven't produced anything yet — spin briefly.
|
|
|
|
|
+ std::this_thread::sleep_for(std::chrono::milliseconds(1));
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ auto t0 = Clock::now();
|
|
|
|
|
+ try {
|
|
|
|
|
+ auto doc = rdb.get("mixed", *idOpt);
|
|
|
|
|
+ auto dt = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
|
|
|
+ Clock::now() - t0).count();
|
|
|
|
|
+ recordLatency(counters, dt);
|
|
|
|
|
+
|
|
|
|
|
+ if (!doc) {
|
|
|
|
|
+ // Cold miss — eviction may have dropped the body; ID is still
|
|
|
|
|
+ // in our recent list. Not a failure.
|
|
|
|
|
+ counters.readerNotFound.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ }
|
|
|
|
|
+ counters.readerOps.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ } catch (const std::exception& e) {
|
|
|
|
|
+ auto dt = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
|
|
|
+ Clock::now() - t0).count();
|
|
|
|
|
+ recordLatency(counters, dt);
|
|
|
|
|
+ std::string msg = e.what();
|
|
|
|
|
+ if (msg.find("RESOURCE_EXHAUSTED") != std::string::npos) {
|
|
|
|
|
+ counters.readerResourceExhausted.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ } else if (msg.find("Deadline") != std::string::npos ||
|
|
|
|
|
+ msg.find("DEADLINE") != std::string::npos) {
|
|
|
|
|
+ counters.readerTimeouts.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ counters.readerOtherErrors.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ });
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Stats sampler
|
|
|
|
|
+ std::thread statsThread([&]() {
|
|
|
|
|
+ uint64_t lastWriterInserts = 0;
|
|
|
|
|
+ uint64_t lastReaderOps = 0;
|
|
|
|
|
+ auto lastT = Clock::now();
|
|
|
|
|
+ while (!counters.stop.load(std::memory_order_relaxed)) {
|
|
|
|
|
+ std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
|
|
|
|
+ auto s = db.getMemoryStats();
|
|
|
|
|
+ auto now = Clock::now();
|
|
|
|
|
+ auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
|
|
|
+ now - startTime).count();
|
|
|
|
|
+ auto delta = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
|
|
|
+ now - lastT).count();
|
|
|
|
|
+ lastT = now;
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t curInserts = counters.writerInserts.load();
|
|
|
|
|
+ uint64_t curReaderOps = counters.readerOps.load();
|
|
|
|
|
+ uint64_t insertRate = delta > 0 ? (curInserts - lastWriterInserts) * 1000 / delta : 0;
|
|
|
|
|
+ uint64_t readRate = delta > 0 ? (curReaderOps - lastReaderOps) * 1000 / delta : 0;
|
|
|
|
|
+ lastWriterInserts = curInserts;
|
|
|
|
|
+ lastReaderOps = curReaderOps;
|
|
|
|
|
+
|
|
|
|
|
+ std::cout << "[t=" << std::setw(5) << elapsed << "ms] "
|
|
|
|
|
+ << "pressure=" << std::setw(9) << s.pressureLevel
|
|
|
|
|
+ << " " << std::setw(3) << s.pressurePercent << "%"
|
|
|
|
|
+ << " inserts=" << std::setw(7) << curInserts
|
|
|
|
|
+ << " (" << std::setw(5) << insertRate << "/s)"
|
|
|
|
|
+ << " reads=" << std::setw(7) << curReaderOps
|
|
|
|
|
+ << " (" << std::setw(5) << readRate << "/s)"
|
|
|
|
|
+ << " not-found=" << counters.readerNotFound.load()
|
|
|
|
|
+ << "\n";
|
|
|
|
|
+ std::cout.flush();
|
|
|
|
|
+ }
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ std::this_thread::sleep_for(std::chrono::seconds(durationSec));
|
|
|
|
|
+ counters.stop.store(true);
|
|
|
|
|
+ for (auto& t : writerThreads) t.join();
|
|
|
|
|
+ for (auto& t : readerThreads) t.join();
|
|
|
|
|
+ statsThread.join();
|
|
|
|
|
+
|
|
|
|
|
+ auto endTime = Clock::now();
|
|
|
|
|
+ auto totalMs = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime).count();
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t totalInserts = counters.writerInserts.load();
|
|
|
|
|
+ uint64_t totalWriteFails = counters.writerFails.load();
|
|
|
|
|
+ uint64_t totalReads = counters.readerOps.load();
|
|
|
|
|
+ uint64_t notFound = counters.readerNotFound.load();
|
|
|
|
|
+ uint64_t rTimeouts = counters.readerTimeouts.load();
|
|
|
|
|
+ uint64_t rResourceExh = counters.readerResourceExhausted.load();
|
|
|
|
|
+ uint64_t rOther = counters.readerOtherErrors.load();
|
|
|
|
|
+ uint64_t rHardFails = rTimeouts + rResourceExh + rOther;
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t p50 = percentile(counters, 0.50);
|
|
|
|
|
+ uint64_t p99 = percentile(counters, 0.99);
|
|
|
|
|
+
|
|
|
|
|
+ uint64_t writerRate = totalMs > 0 ? (totalInserts * 1000 / totalMs) : 0;
|
|
|
|
|
+ uint64_t readerRate = totalMs > 0 ? (totalReads * 1000 / totalMs) : 0;
|
|
|
|
|
+
|
|
|
|
|
+ auto finalStats = db.getMemoryStats();
|
|
|
|
|
+ uint64_t finalDocs = 0, finalEvicted = 0;
|
|
|
|
|
+ for (const auto& c : finalStats.collections) {
|
|
|
|
|
+ if (c.collection == "mixed") {
|
|
|
|
|
+ finalDocs = c.documentCount;
|
|
|
|
|
+ finalEvicted = c.evictedStubCount;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::cout << "\n=== Summary ===\n";
|
|
|
|
|
+ std::cout << "Duration: " << totalMs << "ms\n";
|
|
|
|
|
+ std::cout << "\n-- Writers --\n";
|
|
|
|
|
+ std::cout << "Successful inserts: " << totalInserts
|
|
|
|
|
+ << " (" << writerRate << " ops/s)\n";
|
|
|
|
|
+ std::cout << "Failed inserts: " << totalWriteFails << "\n";
|
|
|
|
|
+ std::cout << "\n-- Readers --\n";
|
|
|
|
|
+ std::cout << "Total reads: " << totalReads
|
|
|
|
|
+ << " (" << readerRate << " ops/s)\n";
|
|
|
|
|
+ std::cout << "Not-found (cold miss): " << notFound
|
|
|
|
|
+ << " (" << (totalReads > 0 ? notFound * 100 / totalReads : 0) << "%)\n";
|
|
|
|
|
+ std::cout << "Hard failures: " << rHardFails << "\n";
|
|
|
|
|
+ std::cout << " RESOURCE_EXHAUSTED: " << rResourceExh << "\n";
|
|
|
|
|
+ std::cout << " DEADLINE_EXCEEDED: " << rTimeouts << "\n";
|
|
|
|
|
+ std::cout << " Other: " << rOther << "\n";
|
|
|
|
|
+ std::cout << "Latency p50: ~" << p50 << " ms\n";
|
|
|
|
|
+ std::cout << "Latency p99: ~" << p99 << " ms\n";
|
|
|
|
|
+ std::cout << "\n-- Final state --\n";
|
|
|
|
|
+ std::cout << "Pressure: " << finalStats.pressureLevel
|
|
|
|
|
+ << " (" << finalStats.pressurePercent << "%)\n";
|
|
|
|
|
+ std::cout << "Live docs: " << finalDocs << "\n";
|
|
|
|
|
+ std::cout << "Evicted stubs: " << finalEvicted << "\n";
|
|
|
|
|
+ std::cout << "\n";
|
|
|
|
|
+
|
|
|
|
|
+ // Verdict
|
|
|
|
|
+ bool passWriter = writerRate > 500;
|
|
|
|
|
+ bool passReader = readerRate > 1000;
|
|
|
|
|
+ bool passLatency = p99 < 1000;
|
|
|
|
|
+ bool passFails = rHardFails == 0;
|
|
|
|
|
+
|
|
|
|
|
+ std::cout << "PASS criteria:\n";
|
|
|
|
|
+ std::cout << " writer ops/s > 500: " << (passWriter ? "PASS" : "FAIL")
|
|
|
|
|
+ << " (actual " << writerRate << ")\n";
|
|
|
|
|
+ std::cout << " reader ops/s > 1000: " << (passReader ? "PASS" : "FAIL")
|
|
|
|
|
+ << " (actual " << readerRate << ")\n";
|
|
|
|
|
+ std::cout << " reader p99 < 1000ms: " << (passLatency ? "PASS" : "FAIL")
|
|
|
|
|
+ << " (actual ~" << p99 << "ms)\n";
|
|
|
|
|
+ std::cout << " zero reader hard fails: " << (passFails ? "PASS" : "FAIL")
|
|
|
|
|
+ << " (actual " << rHardFails << ")\n";
|
|
|
|
|
+ std::cout << "\n";
|
|
|
|
|
+ std::cout << "Verdict: "
|
|
|
|
|
+ << (passWriter && passReader && passLatency && passFails
|
|
|
|
|
+ ? "PASS — reads stay responsive while writes + eviction run"
|
|
|
|
|
+ : "FAIL — see criteria above")
|
|
|
|
|
+ << "\n";
|
|
|
|
|
+
|
|
|
|
|
+ return (passWriter && passReader && passLatency && passFails) ? 0 : 2;
|
|
|
|
|
+}
|