|
@@ -0,0 +1,187 @@
|
|
|
|
|
+// Test 5a — Crash-recovery insert phase
|
|
|
|
|
+//
|
|
|
|
|
+// Inserts deterministic docs as fast as possible into collection "crash"
|
|
|
|
|
+// across `writers` threads. Every successful insert is appended to an
|
|
|
|
|
+// on-disk log so the verifier (which runs after kill -9 + restart) can
|
|
|
|
|
+// check every "logged" ID independently of whether the inserter was
|
|
|
|
|
+// killed mid-flight.
|
|
|
|
|
+//
|
|
|
|
|
+// Usage: ./load_test_crash_insert [num_docs=20000] [writers=3] [doc_bytes_inner=1500]
|
|
|
|
|
+//
|
|
|
|
|
+// Designed to be killed by the orchestrator (run_crash_test.sh) once the
|
|
|
|
|
+// server has started evicting. Also exits 0 on its own if all N inserts
|
|
|
|
|
+// complete; exit 2 if a write fails permanently after retries.
|
|
|
|
|
+
|
|
|
|
|
+#include <smartbotic/database/client.hpp>
|
|
|
|
|
+
|
|
|
|
|
+#include <atomic>
|
|
|
|
|
+#include <chrono>
|
|
|
|
|
+#include <cstdio>
|
|
|
|
|
+#include <cstdlib>
|
|
|
|
|
+#include <filesystem>
|
|
|
|
|
+#include <fstream>
|
|
|
|
|
+#include <iomanip>
|
|
|
|
|
+#include <iostream>
|
|
|
|
|
+#include <mutex>
|
|
|
|
|
+#include <string>
|
|
|
|
|
+#include <thread>
|
|
|
|
|
+#include <vector>
|
|
|
|
|
+
|
|
|
|
|
+using namespace smartbotic::database;
|
|
|
|
|
+using Clock = std::chrono::steady_clock;
|
|
|
|
|
+
|
|
|
|
|
+// Deterministic content per-id. Includes the id inside the payload so
|
|
|
|
|
+// misdirected reads after recovery would fail the content check.
|
|
|
|
|
+// `doc_bytes_inner` controls roughly how much filler padding to generate.
|
|
|
|
|
+static std::string makeContent(int i, int innerBytes) {
|
|
|
|
|
+ std::string tag = "doc-" + std::to_string(i) + "-crash";
|
|
|
|
|
+ // Each padded line is ~9 bytes; compute how many lines we need.
|
|
|
|
|
+ int lines = innerBytes / 9;
|
|
|
|
|
+ if (lines < 1) lines = 1;
|
|
|
|
|
+ std::string pad;
|
|
|
|
|
+ pad.reserve(lines * 9);
|
|
|
|
|
+ for (int k = 0; k < lines; ++k) {
|
|
|
|
|
+ char buf[32];
|
|
|
|
|
+ std::snprintf(buf, sizeof(buf), "%08d-", i * 1000 + k);
|
|
|
|
|
+ pad += buf;
|
|
|
|
|
+ }
|
|
|
|
|
+ return tag + "-" + pad;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+int main(int argc, char* argv[]) {
|
|
|
|
|
+ int numDocs = 20000;
|
|
|
|
|
+ int writers = 3;
|
|
|
|
|
+ int docBytesInner = 1500;
|
|
|
|
|
+ if (argc > 1) numDocs = std::stoi(argv[1]);
|
|
|
|
|
+ if (argc > 2) writers = std::stoi(argv[2]);
|
|
|
|
|
+ if (argc > 3) docBytesInner = std::stoi(argv[3]);
|
|
|
|
|
+
|
|
|
|
|
+ if (writers < 1) writers = 1;
|
|
|
|
|
+
|
|
|
|
|
+ const std::string logDir = "/tmp/smartbotic-loadtest-crash";
|
|
|
|
|
+ const std::string logPath = logDir + "/inserted.log";
|
|
|
|
|
+
|
|
|
|
|
+ std::filesystem::create_directories(logDir);
|
|
|
|
|
+
|
|
|
|
|
+ // Append mode so multiple runs don't clobber. The orchestrator truncates
|
|
|
|
|
+ // this between runs.
|
|
|
|
|
+ std::ofstream logFile(logPath, std::ios::app);
|
|
|
|
|
+ if (!logFile) {
|
|
|
|
|
+ std::cerr << "ERROR: could not open log file " << logPath << "\n";
|
|
|
|
|
+ return 1;
|
|
|
|
|
+ }
|
|
|
|
|
+ std::mutex logMutex;
|
|
|
|
|
+
|
|
|
|
|
+ std::cout << "=== smartbotic-database v1.7.0 load test — crash insert ===\n"
|
|
|
|
|
+ << "Server: localhost:9005\n"
|
|
|
|
|
+ << "Target docs: " << numDocs << "\n"
|
|
|
|
|
+ << "Writers: " << writers << "\n"
|
|
|
|
|
+ << "Doc bytes: ~" << docBytesInner << "\n"
|
|
|
|
|
+ << "Log file: " << logPath << "\n"
|
|
|
|
|
+ << "\n";
|
|
|
|
|
+
|
|
|
|
|
+ Client::Config cfg;
|
|
|
|
|
+ cfg.address = "localhost:9005";
|
|
|
|
|
+ cfg.timeoutMs = 5000;
|
|
|
|
|
+ cfg.writeRetries = 8;
|
|
|
|
|
+ cfg.writeRetryBackoffMs = 50;
|
|
|
|
|
+ cfg.writeRetryMaxBackoffMs = 2000;
|
|
|
|
|
+
|
|
|
|
|
+ // Bootstrap connection for collection creation + progress stats.
|
|
|
|
|
+ Client boot(cfg);
|
|
|
|
|
+ if (!boot.connect()) {
|
|
|
|
|
+ std::cerr << "ERROR: initial connect failed\n";
|
|
|
|
|
+ return 1;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!boot.createCollection("crash")) {
|
|
|
|
|
+ std::cout << "(collection 'crash' already exists — continuing)\n";
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::atomic<uint64_t> inserted{0};
|
|
|
|
|
+ std::atomic<uint64_t> permanentFails{0};
|
|
|
|
|
+ std::atomic<bool> done{false};
|
|
|
|
|
+ auto t0 = Clock::now();
|
|
|
|
|
+
|
|
|
|
|
+ std::thread progress([&]() {
|
|
|
|
|
+ uint64_t lastInserted = 0;
|
|
|
|
|
+ auto lastTick = Clock::now();
|
|
|
|
|
+ while (!done.load()) {
|
|
|
|
|
+ std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
|
|
|
|
+ if (done.load()) break;
|
|
|
|
|
+ auto now = Clock::now();
|
|
|
|
|
+ auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
|
|
|
+ now - t0).count();
|
|
|
|
|
+ uint64_t cur = inserted.load();
|
|
|
|
|
+ auto dt = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
|
|
|
+ now - lastTick).count();
|
|
|
|
|
+ double rate = (dt > 0) ? double(cur - lastInserted) * 1000.0 / double(dt) : 0.0;
|
|
|
|
|
+ lastInserted = cur;
|
|
|
|
|
+ lastTick = now;
|
|
|
|
|
+
|
|
|
|
|
+ std::cout << " [t=" << std::setw(6) << elapsed << "ms] "
|
|
|
|
|
+ << "inserted=" << std::setw(7) << cur
|
|
|
|
|
+ << "/" << numDocs
|
|
|
|
|
+ << " fails=" << std::setw(4) << permanentFails.load()
|
|
|
|
|
+ << " rate=" << std::fixed << std::setprecision(0) << rate << " ops/s"
|
|
|
|
|
+ << "\n";
|
|
|
|
|
+ std::cout.flush();
|
|
|
|
|
+ }
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ auto worker = [&](int threadIndex) {
|
|
|
|
|
+ Client wdb(cfg);
|
|
|
|
|
+ if (!wdb.connect()) {
|
|
|
|
|
+ std::cerr << "[writer " << threadIndex << "] connect failed\n";
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ for (int i = threadIndex; i < numDocs; i += writers) {
|
|
|
|
|
+ std::string id = "crash-" + std::to_string(i);
|
|
|
|
|
+ nlohmann::json doc{
|
|
|
|
|
+ {"i", i},
|
|
|
|
|
+ {"content", makeContent(i, docBytesInner)}
|
|
|
|
|
+ };
|
|
|
|
|
+ try {
|
|
|
|
|
+ wdb.insert("crash", doc, id);
|
|
|
|
|
+ // Successfully persisted (past client retry loop). Log it to
|
|
|
|
|
+ // disk BEFORE moving on so the verifier sees only acked IDs.
|
|
|
|
|
+ {
|
|
|
|
|
+ std::lock_guard<std::mutex> lg(logMutex);
|
|
|
|
|
+ logFile << id << "\n";
|
|
|
|
|
+ logFile.flush();
|
|
|
|
|
+ }
|
|
|
|
|
+ inserted.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ } catch (const std::exception& e) {
|
|
|
|
|
+ // Permanently failed after client-side retries. Count it but
|
|
|
|
|
+ // do NOT log the id, so the verifier correctly doesn't look
|
|
|
|
|
+ // for it post-recovery.
|
|
|
|
|
+ permanentFails.fetch_add(1, std::memory_order_relaxed);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ };
|
|
|
|
|
+
|
|
|
|
|
+ std::vector<std::thread> threads;
|
|
|
|
|
+ threads.reserve(writers);
|
|
|
|
|
+ for (int t = 0; t < writers; ++t) {
|
|
|
|
|
+ threads.emplace_back(worker, t);
|
|
|
|
|
+ }
|
|
|
|
|
+ for (auto& th : threads) th.join();
|
|
|
|
|
+
|
|
|
|
|
+ done.store(true);
|
|
|
|
|
+ progress.join();
|
|
|
|
|
+
|
|
|
|
|
+ auto tEnd = Clock::now();
|
|
|
|
|
+ auto elapsedMs = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
|
|
|
+ tEnd - t0).count();
|
|
|
|
|
+
|
|
|
|
|
+ std::cout << "\n=== Insert phase summary ===\n"
|
|
|
|
|
+ << "Elapsed: " << elapsedMs << " ms\n"
|
|
|
|
|
+ << "Inserted (logged): " << inserted.load() << "\n"
|
|
|
|
|
+ << "Permanent fails: " << permanentFails.load() << "\n";
|
|
|
|
|
+
|
|
|
|
|
+ if (permanentFails.load() > 0) {
|
|
|
|
|
+ std::cerr << "WARN: some inserts failed permanently after retry — those "
|
|
|
|
|
+ << "IDs are NOT in the log and will not be verified\n";
|
|
|
|
|
+ return 2;
|
|
|
|
|
+ }
|
|
|
|
|
+ return 0;
|
|
|
|
|
+}
|