7076e4b283* perf(daemon): coalesce metadata writes * refactor(metadata): split repository ops * fix(metadata): harden repository invariants * feat: test coverage and continued refactor work in metadata code * fix(graph): handle path-like explore queries - prefer file path symbol lookup for graph explore\n- sanitize alias FTS queries before MATCH * fix(daemon): quiesce shutdown background work - Interrupt pending DB acquires before worker shutdown - Wait for tracked background coroutines and timers to drain * fix(daemon): cancel repair work on shutdown Blocked embed and post-ingest queues could keep salvage and\nstuck-doc repair alive during shutdown. Consume cancellation in\nrepair retry loops and lock the integrated ServiceManager path\nwith regressions so shutdown fails fast instead of drifting. * perf(metadata): reduce sqlite lock contention * perf(metadata): flatten batch content retries * fix(daemon): align shutdown wait budgets A low shutdown timeout override could expire before ServiceManager\nfinished its 5s+30s WorkCoordinator join path. Share the join\nbudget constants, clamp graceful lifecycle waits to that floor,\nand lock the behavior with a daemon regression. * perf(metadata): bound feedback write waits * perf(metadata): prep batch content before tx * fix(daemon): honor no-embeddings tasks Post-ingest queue tasks dropped the per-task noEmbeddings flag,\nso focused extraction tests still emitted embed jobs and Windows\nCI exercised extra async paths. Preserve the flag through the bus,\nskip embed dispatch when requested, and wake the extraction poller\nimmediately after successful queueing. * perf(daemon): batch fallback metadata writes * feat(perf,docs): optimizations for metadata writes, prompt updates and test improvements * perf(metadata): flatten insert retries * fix(metadata): clear CI portability blockers Use the shared portable ScopedEnvVar in metadata search-query helper tests, add mutators for test reuse, and normalize Statement::bindAll integral dispatch so long long arguments bind consistently across platforms. Add focused regression coverage for the bindAll long long path. * perf(metadata): batch writes and steady bench * fix(daemon): skip inactive post-ingest pollers * perf(metadata): tighten tag and path caches * perf(metadata): cache small batch upserts * fix(daemon): harden embedded host teardown * feat(docs): updating docs policies and language * perf(metadata): tune doc update retries * perf(metadata): bound download retries * perf(metadata): bound download metadata waits * bench(metadata): scope post-ingest content path * fix(daemon): export embedded test hooks * fix(daemon): harden test helper linkage * fix(build): add missing standard includes
1143 lines
58 KiB
C++
1143 lines
58 KiB
C++
// Metadata/SQLite contention benchmark
|
|
//
|
|
// Profiles concurrent metadata reads/writes against the SQLite metadata DB,
|
|
// with optional TRUNCATE checkpoints and long-lived read transactions to force
|
|
// realistic lock pressure. This fills the gap between:
|
|
// - write_coordinator_bench (daemon single-writer throughput)
|
|
// - multi_client_ingestion_bench (end-to-end daemon contention)
|
|
// by directly stressing the metadata repository + SQLite connection pools.
|
|
//
|
|
// Environment variables:
|
|
// YAMS_BENCH_SEED_DOCS - Initial seeded documents (default: 2000)
|
|
// YAMS_BENCH_WRITER_THREADS - Concurrent metadata writer threads (default: 2)
|
|
// YAMS_BENCH_READER_THREADS - Concurrent metadata reader threads (default: 8)
|
|
// YAMS_BENCH_LONG_READERS - Long-lived read transaction threads (default: 1)
|
|
// YAMS_BENCH_LONG_READ_HOLD_MS - How long each long reader holds tx (default: 100)
|
|
// YAMS_BENCH_RUNTIME_S - Timed run duration per repeat (default: 8)
|
|
// YAMS_BENCH_REPEAT - Full repeats (default: 1)
|
|
// YAMS_BENCH_METADATA_BATCH - setMetadataBatch entries per write op (default: 8)
|
|
// YAMS_BENCH_STATUS_BATCH - repair-status hashes per write op (default: 64)
|
|
// YAMS_BENCH_READ_BATCH - getMetadataForDocuments ids per read op (default: 16)
|
|
// YAMS_BENCH_CHECKPOINT_EVERY_MS - TRUNCATE checkpoint cadence, 0 disables (default: 500)
|
|
// YAMS_BENCH_DB_BUSY_TIMEOUT_MS - SQLite busy timeout (default: 100)
|
|
// YAMS_BENCH_DUAL_POOL - 1 => separate read-only pool, 0 => single pool (default: 1)
|
|
// YAMS_BENCH_EXTERNAL_LOCKER - 1 => separate writer periodically holds BEGIN IMMEDIATE
|
|
// long enough to trigger metadata retry exhaustion (default:
|
|
// 0)
|
|
// YAMS_BENCH_EXTERNAL_LOCK_HOLD_MS - Hold time for injected writer lock (default: 2500)
|
|
// YAMS_BENCH_EXTERNAL_LOCK_EVERY_MS- Period between injected lock attempts (default: 250)
|
|
// YAMS_BENCH_OUTPUT - JSONL output path (default:
|
|
// bench_results/metadata_contention.jsonl)
|
|
// YAMS_BENCH_WRITE_MODE - mixed|insert_document|metadata_single|metadata_batch|
|
|
// batch_content|feedback_event|repair_status|
|
|
// extraction_status|download_metadata_burst|
|
|
// document_update_burst|post_ingest_batch_content
|
|
// (default: mixed)
|
|
|
|
#include <catch2/catch_test_macros.hpp>
|
|
|
|
#include <algorithm>
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstdint>
|
|
#include <cstdlib>
|
|
#include <filesystem>
|
|
#include <fstream>
|
|
#include <iomanip>
|
|
#include <iostream>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <numeric>
|
|
#include <optional>
|
|
#include <sstream>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <tuple>
|
|
#include <vector>
|
|
|
|
#include <nlohmann/json.hpp>
|
|
#include <spdlog/spdlog.h>
|
|
#include "../common/test_helpers_catch2.h"
|
|
#include <yams/app/services/download_metadata_entries.hpp>
|
|
#include <yams/metadata/connection_pool.h>
|
|
#include <yams/metadata/document_metadata.h>
|
|
#include <yams/metadata/metadata_repository.h>
|
|
#include <yams/metadata/path_utils.h>
|
|
#include <yams/storage/sqlite_retry.h>
|
|
|
|
namespace fs = std::filesystem;
|
|
using json = nlohmann::json;
|
|
using namespace std::chrono_literals;
|
|
using yams::metadata::BatchContentEntry;
|
|
using yams::metadata::ConnectionPool;
|
|
using yams::metadata::ConnectionPoolConfig;
|
|
using yams::metadata::ConnectionPriority;
|
|
using yams::metadata::Database;
|
|
using yams::metadata::DocumentInfo;
|
|
using yams::metadata::FeedbackEvent;
|
|
using yams::metadata::MetadataOpScope;
|
|
using yams::metadata::MetadataRepository;
|
|
using yams::metadata::MetadataValue;
|
|
using yams::storage::sqlite_retry::isBusyOrLockedMessage;
|
|
|
|
namespace {
|
|
|
|
struct BenchConfig {
|
|
int seedDocs{2000};
|
|
int writerThreads{2};
|
|
int readerThreads{8};
|
|
int longReaders{1};
|
|
int longReadHoldMs{100};
|
|
int runtimeSeconds{8};
|
|
int repeat{1};
|
|
int metadataBatchSize{8};
|
|
int statusBatchSize{64};
|
|
int readBatchSize{16};
|
|
int checkpointEveryMs{500};
|
|
int dbBusyTimeoutMs{100};
|
|
bool dualPool{true};
|
|
bool externalLocker{false};
|
|
int externalLockHoldMs{2500};
|
|
int externalLockEveryMs{250};
|
|
std::string outputPath{"bench_results/metadata_contention.jsonl"};
|
|
int writeMode{-1};
|
|
std::string writeModeName{"mixed"};
|
|
};
|
|
|
|
struct OpStats {
|
|
uint64_t attempts{0};
|
|
uint64_t success{0};
|
|
uint64_t errors{0};
|
|
uint64_t lockErrors{0};
|
|
std::vector<int64_t> latenciesUs;
|
|
};
|
|
|
|
struct WorkerStats {
|
|
OpStats writes;
|
|
OpStats insertWrites;
|
|
OpStats metadataSingleWrites;
|
|
OpStats metadataBatchWrites;
|
|
OpStats downloadMetadataWrites;
|
|
OpStats documentUpdateWrites;
|
|
OpStats batchContentWrites;
|
|
OpStats feedbackWrites;
|
|
OpStats repairWrites;
|
|
OpStats extractionWrites;
|
|
OpStats reads;
|
|
OpStats checkpoints;
|
|
OpStats longReads;
|
|
OpStats externalLocks;
|
|
};
|
|
|
|
struct LatencySummary {
|
|
size_t count{0};
|
|
double meanUs{0.0};
|
|
double p50Us{0.0};
|
|
double p95Us{0.0};
|
|
double p99Us{0.0};
|
|
double maxUs{0.0};
|
|
};
|
|
|
|
int readIntEnv(const char* name, int fallback, int minVal, int maxVal) {
|
|
const char* val = std::getenv(name);
|
|
if (!val || !*val) {
|
|
return fallback;
|
|
}
|
|
char* end = nullptr;
|
|
long parsed = std::strtol(val, &end, 10);
|
|
if (end == nullptr || end == val || *end != '\0') {
|
|
return fallback;
|
|
}
|
|
return static_cast<int>(std::clamp<long>(parsed, minVal, maxVal));
|
|
}
|
|
|
|
bool readBoolEnv(const char* name, bool fallback) {
|
|
const char* val = std::getenv(name);
|
|
if (!val || !*val) {
|
|
return fallback;
|
|
}
|
|
std::string s(val);
|
|
std::transform(s.begin(), s.end(), s.begin(),
|
|
[](unsigned char ch) { return static_cast<char>(std::tolower(ch)); });
|
|
if (s == "1" || s == "true" || s == "yes" || s == "on") {
|
|
return true;
|
|
}
|
|
if (s == "0" || s == "false" || s == "no" || s == "off") {
|
|
return false;
|
|
}
|
|
return fallback;
|
|
}
|
|
|
|
std::pair<int, std::string> readWriteModeEnv() {
|
|
const char* val = std::getenv("YAMS_BENCH_WRITE_MODE");
|
|
if (!val || !*val) {
|
|
return {-1, "mixed"};
|
|
}
|
|
std::string s(val);
|
|
std::transform(s.begin(), s.end(), s.begin(),
|
|
[](unsigned char ch) { return static_cast<char>(std::tolower(ch)); });
|
|
if (s == "mixed")
|
|
return {-1, "mixed"};
|
|
if (s == "insert_document")
|
|
return {0, s};
|
|
if (s == "metadata_batch")
|
|
return {1, s};
|
|
if (s == "batch_content")
|
|
return {2, s};
|
|
if (s == "feedback_event")
|
|
return {3, s};
|
|
if (s == "repair_status")
|
|
return {4, s};
|
|
if (s == "extraction_status")
|
|
return {5, s};
|
|
if (s == "metadata_single")
|
|
return {6, s};
|
|
if (s == "download_metadata_burst")
|
|
return {7, s};
|
|
if (s == "document_update_burst")
|
|
return {8, s};
|
|
if (s == "post_ingest_batch_content")
|
|
return {9, s};
|
|
return {-1, "mixed"};
|
|
}
|
|
|
|
BenchConfig loadConfig() {
|
|
BenchConfig cfg;
|
|
cfg.seedDocs = readIntEnv("YAMS_BENCH_SEED_DOCS", 2000, 100, 200000);
|
|
cfg.writerThreads = readIntEnv("YAMS_BENCH_WRITER_THREADS", 2, 1, 64);
|
|
cfg.readerThreads = readIntEnv("YAMS_BENCH_READER_THREADS", 8, 1, 128);
|
|
cfg.longReaders = readIntEnv("YAMS_BENCH_LONG_READERS", 1, 0, 32);
|
|
cfg.longReadHoldMs = readIntEnv("YAMS_BENCH_LONG_READ_HOLD_MS", 100, 0, 5000);
|
|
cfg.runtimeSeconds = readIntEnv("YAMS_BENCH_RUNTIME_S", 8, 1, 300);
|
|
cfg.repeat = readIntEnv("YAMS_BENCH_REPEAT", 1, 1, 20);
|
|
cfg.metadataBatchSize = readIntEnv("YAMS_BENCH_METADATA_BATCH", 8, 1, 128);
|
|
cfg.statusBatchSize = readIntEnv("YAMS_BENCH_STATUS_BATCH", 64, 1, 1024);
|
|
cfg.readBatchSize = readIntEnv("YAMS_BENCH_READ_BATCH", 16, 1, 256);
|
|
cfg.checkpointEveryMs = readIntEnv("YAMS_BENCH_CHECKPOINT_EVERY_MS", 500, 0, 60000);
|
|
cfg.dbBusyTimeoutMs = readIntEnv("YAMS_BENCH_DB_BUSY_TIMEOUT_MS", 100, 0, 30000);
|
|
cfg.dualPool = readBoolEnv("YAMS_BENCH_DUAL_POOL", true);
|
|
cfg.externalLocker = readBoolEnv("YAMS_BENCH_EXTERNAL_LOCKER", false);
|
|
cfg.externalLockHoldMs = readIntEnv("YAMS_BENCH_EXTERNAL_LOCK_HOLD_MS", 2500, 1, 60000);
|
|
cfg.externalLockEveryMs = readIntEnv("YAMS_BENCH_EXTERNAL_LOCK_EVERY_MS", 250, 0, 60000);
|
|
if (const char* out = std::getenv("YAMS_BENCH_OUTPUT"); out && *out) {
|
|
cfg.outputPath = out;
|
|
}
|
|
auto [writeMode, writeModeName] = readWriteModeEnv();
|
|
cfg.writeMode = writeMode;
|
|
cfg.writeModeName = std::move(writeModeName);
|
|
return cfg;
|
|
}
|
|
|
|
std::string isoTimestamp() {
|
|
const auto now = std::chrono::system_clock::now();
|
|
const auto tt = std::chrono::system_clock::to_time_t(now);
|
|
std::tm tm{};
|
|
gmtime_r(&tt, &tm);
|
|
std::ostringstream oss;
|
|
oss << std::put_time(&tm, "%Y-%m-%dT%H:%M:%SZ");
|
|
return oss.str();
|
|
}
|
|
|
|
void ensureOutputDir(const fs::path& path) {
|
|
if (!path.parent_path().empty()) {
|
|
fs::create_directories(path.parent_path());
|
|
}
|
|
}
|
|
|
|
std::uintmax_t walBytes(const fs::path& dbPath) {
|
|
const fs::path walPath = dbPath.string() + "-wal";
|
|
std::error_code ec;
|
|
if (!fs::exists(walPath, ec)) {
|
|
return 0;
|
|
}
|
|
return fs::file_size(walPath, ec);
|
|
}
|
|
|
|
DocumentInfo makeDocument(const fs::path& root, int writerId, uint64_t seq) {
|
|
DocumentInfo doc;
|
|
doc.fileName = "writer_" + std::to_string(writerId) + "_" + std::to_string(seq) + ".txt";
|
|
doc.filePath = (root / "docs" / doc.fileName).string();
|
|
doc.fileExtension = ".txt";
|
|
doc.fileSize = 1024 + static_cast<int64_t>(seq % 2048);
|
|
doc.sha256Hash = "hash_writer_" + std::to_string(writerId) + "_" + std::to_string(seq);
|
|
doc.mimeType = "text/plain";
|
|
const auto now =
|
|
std::chrono::time_point_cast<std::chrono::seconds>(std::chrono::system_clock::now());
|
|
doc.createdTime = now;
|
|
doc.modifiedTime = now;
|
|
doc.indexedTime = now;
|
|
doc.contentExtracted = true;
|
|
doc.extractionStatus = yams::metadata::ExtractionStatus::Success;
|
|
yams::metadata::populatePathDerivedFields(doc);
|
|
return doc;
|
|
}
|
|
|
|
std::vector<std::pair<std::string, MetadataValue>> makeInsertTags(int writerId, uint64_t seq) {
|
|
std::vector<std::pair<std::string, MetadataValue>> tags;
|
|
tags.emplace_back("owner", MetadataValue("bench_writer_" + std::to_string(writerId)));
|
|
tags.emplace_back("seq", MetadataValue(std::to_string(seq)));
|
|
tags.emplace_back("phase", MetadataValue("insert"));
|
|
return tags;
|
|
}
|
|
|
|
bool isContentionError(std::string_view message) {
|
|
return isBusyOrLockedMessage(message) ||
|
|
message.find("max retries exceeded") != std::string_view::npos;
|
|
}
|
|
|
|
void noteResult(OpStats& stats, bool ok, std::optional<std::string_view> errorMessage,
|
|
int64_t latencyUs) {
|
|
stats.attempts++;
|
|
if (ok) {
|
|
stats.success++;
|
|
} else {
|
|
stats.errors++;
|
|
if (errorMessage && isContentionError(*errorMessage)) {
|
|
stats.lockErrors++;
|
|
}
|
|
}
|
|
stats.latenciesUs.push_back(latencyUs);
|
|
}
|
|
|
|
void mergeOpStats(OpStats& dst, const OpStats& src) {
|
|
dst.attempts += src.attempts;
|
|
dst.success += src.success;
|
|
dst.errors += src.errors;
|
|
dst.lockErrors += src.lockErrors;
|
|
if (!src.latenciesUs.empty()) {
|
|
dst.latenciesUs.insert(dst.latenciesUs.end(), src.latenciesUs.begin(),
|
|
src.latenciesUs.end());
|
|
}
|
|
}
|
|
|
|
void mergeWorkerStats(WorkerStats& dst, const WorkerStats& src) {
|
|
mergeOpStats(dst.writes, src.writes);
|
|
mergeOpStats(dst.insertWrites, src.insertWrites);
|
|
mergeOpStats(dst.metadataSingleWrites, src.metadataSingleWrites);
|
|
mergeOpStats(dst.metadataBatchWrites, src.metadataBatchWrites);
|
|
mergeOpStats(dst.downloadMetadataWrites, src.downloadMetadataWrites);
|
|
mergeOpStats(dst.documentUpdateWrites, src.documentUpdateWrites);
|
|
mergeOpStats(dst.batchContentWrites, src.batchContentWrites);
|
|
mergeOpStats(dst.feedbackWrites, src.feedbackWrites);
|
|
mergeOpStats(dst.repairWrites, src.repairWrites);
|
|
mergeOpStats(dst.extractionWrites, src.extractionWrites);
|
|
mergeOpStats(dst.reads, src.reads);
|
|
mergeOpStats(dst.checkpoints, src.checkpoints);
|
|
mergeOpStats(dst.longReads, src.longReads);
|
|
mergeOpStats(dst.externalLocks, src.externalLocks);
|
|
}
|
|
|
|
LatencySummary summarizeLatencies(std::vector<int64_t> values) {
|
|
LatencySummary out;
|
|
if (values.empty()) {
|
|
return out;
|
|
}
|
|
std::sort(values.begin(), values.end());
|
|
out.count = values.size();
|
|
const auto sum = std::accumulate(values.begin(), values.end(), 0.0);
|
|
out.meanUs = sum / static_cast<double>(values.size());
|
|
auto pct = [&](double q) {
|
|
const size_t idx = std::min(
|
|
values.size() - 1, static_cast<size_t>(q * static_cast<double>(values.size() - 1)));
|
|
return static_cast<double>(values[idx]);
|
|
};
|
|
out.p50Us = pct(0.50);
|
|
out.p95Us = pct(0.95);
|
|
out.p99Us = pct(0.99);
|
|
out.maxUs = static_cast<double>(values.back());
|
|
return out;
|
|
}
|
|
|
|
json latencyJson(const OpStats& stats) {
|
|
const auto summary = summarizeLatencies(stats.latenciesUs);
|
|
return json{{"attempts", stats.attempts}, {"success", stats.success},
|
|
{"errors", stats.errors}, {"lock_errors", stats.lockErrors},
|
|
{"latency_count", summary.count}, {"latency_mean_us", summary.meanUs},
|
|
{"latency_p50_us", summary.p50Us}, {"latency_p95_us", summary.p95Us},
|
|
{"latency_p99_us", summary.p99Us}, {"latency_max_us", summary.maxUs}};
|
|
}
|
|
|
|
json poolStatsJson(const ConnectionPool::Stats& stats) {
|
|
return json{{"total_connections", stats.totalConnections},
|
|
{"available_connections", stats.availableConnections},
|
|
{"active_connections", stats.activeConnections},
|
|
{"waiting_requests", stats.waitingRequests},
|
|
{"max_observed_waiting", stats.maxObservedWaiting},
|
|
{"total_wait_micros", stats.totalWaitMicros},
|
|
{"timeout_count", stats.timeoutCount},
|
|
{"total_acquired", stats.totalAcquired},
|
|
{"total_released", stats.totalReleased},
|
|
{"failed_acquisitions", stats.failedAcquisitions},
|
|
{"slow_holder_count", stats.slowHolderCount},
|
|
{"max_holder_micros", stats.maxHolderMicros}};
|
|
}
|
|
|
|
void appendJsonLine(const fs::path& outputPath, const json& record) {
|
|
ensureOutputDir(outputPath);
|
|
std::ofstream out(outputPath, std::ios::app);
|
|
out << record.dump() << '\n';
|
|
out.flush();
|
|
}
|
|
|
|
} // namespace
|
|
|
|
TEST_CASE("MetadataRepository SQLite contention profile", "[bench][metadata][sqlite][contention]") {
|
|
const BenchConfig cfg = loadConfig();
|
|
|
|
spdlog::info("=== Metadata/SQLite Contention Benchmark ===");
|
|
spdlog::info(" seedDocs: {}", cfg.seedDocs);
|
|
spdlog::info(" writerThreads: {}", cfg.writerThreads);
|
|
spdlog::info(" readerThreads: {}", cfg.readerThreads);
|
|
spdlog::info(" longReaders: {}", cfg.longReaders);
|
|
spdlog::info(" longReadHoldMs: {}", cfg.longReadHoldMs);
|
|
spdlog::info(" runtimeSeconds: {}", cfg.runtimeSeconds);
|
|
spdlog::info(" repeat: {}", cfg.repeat);
|
|
spdlog::info(" metadataBatchSize: {}", cfg.metadataBatchSize);
|
|
spdlog::info(" statusBatchSize: {}", cfg.statusBatchSize);
|
|
spdlog::info(" readBatchSize: {}", cfg.readBatchSize);
|
|
spdlog::info(" checkpointEveryMs: {}", cfg.checkpointEveryMs);
|
|
spdlog::info(" dbBusyTimeoutMs: {}", cfg.dbBusyTimeoutMs);
|
|
spdlog::info(" dualPool: {}", cfg.dualPool);
|
|
spdlog::info(" externalLocker: {}", cfg.externalLocker);
|
|
spdlog::info(" externalLockHoldMs: {}", cfg.externalLockHoldMs);
|
|
spdlog::info(" externalLockEveryMs:{}", cfg.externalLockEveryMs);
|
|
spdlog::info(" writeMode: {}", cfg.writeModeName);
|
|
spdlog::info(" output: {}", cfg.outputPath);
|
|
|
|
for (int rep = 0; rep < cfg.repeat; ++rep) {
|
|
const fs::path root = yams::test::make_temp_dir("yams_bench_metadata_contention_");
|
|
const fs::path dbPath = root / "metadata.db";
|
|
|
|
ConnectionPoolConfig writeCfg;
|
|
writeCfg.minConnections = 1;
|
|
writeCfg.maxConnections = 1;
|
|
writeCfg.busyTimeout = std::chrono::milliseconds(cfg.dbBusyTimeoutMs);
|
|
writeCfg.enableWAL = true;
|
|
|
|
ConnectionPool writePool(dbPath.string(), writeCfg);
|
|
REQUIRE(writePool.initialize().has_value());
|
|
writePool.setSlowHolderThreshold(std::chrono::milliseconds(250));
|
|
|
|
std::unique_ptr<ConnectionPool> readPool;
|
|
if (cfg.dualPool) {
|
|
ConnectionPoolConfig readCfg;
|
|
readCfg.minConnections = std::min<size_t>(std::max(2, cfg.readerThreads), 8);
|
|
readCfg.maxConnections =
|
|
std::max<size_t>(readCfg.minConnections,
|
|
static_cast<size_t>(cfg.readerThreads + cfg.longReaders + 2));
|
|
readCfg.busyTimeout = std::chrono::milliseconds(cfg.dbBusyTimeoutMs);
|
|
readCfg.enableWAL = true;
|
|
readCfg.readOnly = true;
|
|
readPool = std::make_unique<ConnectionPool>(dbPath.string(), readCfg);
|
|
REQUIRE(readPool->initialize().has_value());
|
|
}
|
|
|
|
MetadataRepository repo(writePool, readPool ? readPool.get() : nullptr);
|
|
if (readPool) {
|
|
readPool->refreshAll();
|
|
}
|
|
|
|
std::unique_ptr<ConnectionPool> externalLockPool;
|
|
if (cfg.externalLocker) {
|
|
ConnectionPoolConfig externalCfg;
|
|
externalCfg.minConnections = 1;
|
|
externalCfg.maxConnections = 1;
|
|
externalCfg.busyTimeout = std::chrono::milliseconds(cfg.dbBusyTimeoutMs);
|
|
externalCfg.enableWAL = true;
|
|
externalLockPool = std::make_unique<ConnectionPool>(dbPath.string(), externalCfg);
|
|
REQUIRE(externalLockPool->initialize().has_value());
|
|
externalLockPool->setSlowHolderThreshold(std::chrono::milliseconds(250));
|
|
}
|
|
|
|
std::vector<int64_t> docIds;
|
|
std::vector<std::string> hashes;
|
|
std::vector<std::string> paths;
|
|
docIds.reserve(static_cast<size_t>(cfg.seedDocs));
|
|
hashes.reserve(static_cast<size_t>(cfg.seedDocs));
|
|
paths.reserve(static_cast<size_t>(cfg.seedDocs));
|
|
|
|
for (int i = 0; i < cfg.seedDocs; ++i) {
|
|
auto doc = makeDocument(root, -1, static_cast<uint64_t>(i));
|
|
doc.fileName = "seed_" + std::to_string(i) + ".txt";
|
|
doc.filePath = (root / "seed" / doc.fileName).string();
|
|
doc.sha256Hash = "seed_hash_" + std::to_string(i);
|
|
yams::metadata::populatePathDerivedFields(doc);
|
|
|
|
std::vector<std::pair<std::string, MetadataValue>> tags;
|
|
tags.emplace_back("bucket", MetadataValue("seed"));
|
|
tags.emplace_back("ordinal", MetadataValue(std::to_string(i)));
|
|
tags.emplace_back("hot_key", MetadataValue("v0"));
|
|
auto inserted = repo.insertDocumentWithMetadata(doc, tags);
|
|
REQUIRE(inserted.has_value());
|
|
docIds.push_back(inserted.value());
|
|
hashes.push_back(doc.sha256Hash);
|
|
paths.push_back(doc.filePath);
|
|
}
|
|
|
|
const auto walBefore = walBytes(dbPath);
|
|
std::chrono::steady_clock::time_point start;
|
|
std::chrono::steady_clock::time_point deadline;
|
|
std::atomic<bool> stop{false};
|
|
std::atomic<bool> runStarted{false};
|
|
std::atomic<int> readyWorkers{0};
|
|
|
|
WorkerStats aggregate;
|
|
std::mutex aggregateMutex;
|
|
std::vector<std::thread> threads;
|
|
const size_t reservedThreads =
|
|
static_cast<size_t>(cfg.writerThreads) + static_cast<size_t>(cfg.readerThreads) +
|
|
static_cast<size_t>(cfg.longReaders) + (cfg.checkpointEveryMs > 0 ? 1U : 0U) +
|
|
(cfg.externalLocker ? 1U : 0U);
|
|
threads.reserve(reservedThreads);
|
|
const int expectedWorkers = static_cast<int>(reservedThreads);
|
|
auto waitForRunStart = [&]() {
|
|
readyWorkers.fetch_add(1, std::memory_order_release);
|
|
while (!runStarted.load(std::memory_order_acquire)) {
|
|
std::this_thread::sleep_for(1ms);
|
|
}
|
|
};
|
|
|
|
for (int writerId = 0; writerId < cfg.writerThreads; ++writerId) {
|
|
threads.emplace_back([&, writerId]() {
|
|
WorkerStats local;
|
|
uint64_t seq = 0;
|
|
waitForRunStart();
|
|
while (!stop.load(std::memory_order_relaxed) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
const auto t0 = std::chrono::steady_clock::now();
|
|
const auto mode =
|
|
cfg.writeMode >= 0 ? cfg.writeMode : static_cast<int>(seq % 6);
|
|
if (mode == 0) {
|
|
MetadataOpScope op("bench_insert_document_with_metadata");
|
|
auto doc = makeDocument(root, writerId, seq);
|
|
auto result =
|
|
repo.insertDocumentWithMetadata(doc, makeInsertTags(writerId, seq));
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.insertWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 1) {
|
|
MetadataOpScope op("bench_set_metadata_batch");
|
|
std::vector<std::tuple<int64_t, std::string, MetadataValue>> entries;
|
|
entries.reserve(static_cast<size_t>(cfg.metadataBatchSize));
|
|
for (int j = 0; j < cfg.metadataBatchSize; ++j) {
|
|
const auto idx =
|
|
static_cast<size_t>((seq + static_cast<uint64_t>(j) +
|
|
static_cast<uint64_t>(writerId) * 17) %
|
|
docIds.size());
|
|
entries.emplace_back(docIds[idx], "hot_key_" + std::to_string(j % 4),
|
|
MetadataValue("w" + std::to_string(writerId) +
|
|
"_" + std::to_string(seq) + "_" +
|
|
std::to_string(j)));
|
|
}
|
|
auto result = repo.setMetadataBatch(entries);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.metadataBatchWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 2) {
|
|
MetadataOpScope op("bench_batch_insert_content_and_index");
|
|
std::vector<BatchContentEntry> entries;
|
|
entries.reserve(static_cast<size_t>(cfg.metadataBatchSize));
|
|
for (int j = 0; j < cfg.metadataBatchSize; ++j) {
|
|
const auto idx =
|
|
static_cast<size_t>((seq + static_cast<uint64_t>(j) * 13 +
|
|
static_cast<uint64_t>(writerId) * 23) %
|
|
docIds.size());
|
|
BatchContentEntry entry;
|
|
entry.documentId = docIds[idx];
|
|
entry.title = "bench title " + std::to_string(writerId) + "_" +
|
|
std::to_string(seq) + "_" + std::to_string(j);
|
|
entry.contentText = "bench content body " + std::to_string(writerId) +
|
|
"_" + std::to_string(seq) + "_" + std::to_string(j);
|
|
entry.mimeType = "text/plain";
|
|
entry.extractionMethod = "bench";
|
|
entry.language = "en";
|
|
entry.priorStateKnown = true;
|
|
entry.priorContentExtracted = true;
|
|
entry.priorExtractionStatus = yams::metadata::ExtractionStatus::Success;
|
|
entries.push_back(std::move(entry));
|
|
}
|
|
auto result = repo.batchInsertContentAndIndex(entries);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.batchContentWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 3) {
|
|
MetadataOpScope op("bench_insert_feedback_event");
|
|
FeedbackEvent event;
|
|
event.eventId =
|
|
"bench_event_" + std::to_string(writerId) + "_" + std::to_string(seq);
|
|
event.traceId = "bench_trace_" + std::to_string(writerId % 3);
|
|
event.createdAt = std::chrono::floor<std::chrono::seconds>(
|
|
std::chrono::system_clock::now());
|
|
event.source = "bench";
|
|
event.eventType = "retrieval_served";
|
|
event.payloadJson = std::string{"{\"writer\":"} + std::to_string(writerId) +
|
|
",\"seq\":" + std::to_string(seq) + "}";
|
|
auto result = repo.insertFeedbackEvent(event);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.feedbackWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 4) {
|
|
MetadataOpScope op("bench_batch_update_repair_statuses");
|
|
std::vector<std::string> batchHashes;
|
|
batchHashes.reserve(static_cast<size_t>(cfg.statusBatchSize));
|
|
for (int j = 0; j < cfg.statusBatchSize; ++j) {
|
|
const auto idx =
|
|
static_cast<size_t>((seq + static_cast<uint64_t>(j) * 11 +
|
|
static_cast<uint64_t>(writerId) * 29) %
|
|
hashes.size());
|
|
batchHashes.push_back(hashes[idx]);
|
|
}
|
|
auto result = repo.batchUpdateDocumentRepairStatuses(
|
|
batchHashes, yams::metadata::RepairStatus::Processing);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.repairWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 6) {
|
|
MetadataOpScope op("bench_set_metadata_single");
|
|
const auto idx = static_cast<size_t>(
|
|
(seq + static_cast<uint64_t>(writerId) * 37) % docIds.size());
|
|
auto result =
|
|
repo.setMetadata(docIds[idx], "hot_single_" + std::to_string(seq % 4),
|
|
MetadataValue("single_" + std::to_string(writerId) +
|
|
"_" + std::to_string(seq)));
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.metadataSingleWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 7) {
|
|
MetadataOpScope op("bench_download_metadata_burst");
|
|
const auto idx = static_cast<size_t>(
|
|
(seq + static_cast<uint64_t>(writerId) * 41) % docIds.size());
|
|
|
|
yams::app::services::DownloadServiceRequest request;
|
|
request.tags = {"bench-user-tag", "bench-user-tag-2"};
|
|
request.metadata = {{"custom_key", "bench_value_" + std::to_string(seq)}};
|
|
|
|
yams::downloader::FinalResult downloadResult;
|
|
downloadResult.url = "https://bench.example.test/files/" +
|
|
std::to_string(writerId) + "/" + std::to_string(seq) +
|
|
".txt";
|
|
downloadResult.httpStatus = 200;
|
|
downloadResult.etag =
|
|
"etag_" + std::to_string(writerId) + "_" + std::to_string(seq);
|
|
downloadResult.lastModified = "Mon, 01 Jan 2024 00:00:00 GMT";
|
|
downloadResult.checksumOk = true;
|
|
downloadResult.contentType = "text/plain";
|
|
downloadResult.suggestedName = "bench-file.txt";
|
|
|
|
auto result = repo.setMetadataBatch(
|
|
buildDownloadMetadataEntries(docIds[idx], request, downloadResult));
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.downloadMetadataWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 8) {
|
|
MetadataOpScope op("bench_document_update_burst");
|
|
const auto idx = static_cast<size_t>(
|
|
(seq + static_cast<uint64_t>(writerId) * 43) % docIds.size());
|
|
std::vector<std::tuple<int64_t, std::string, MetadataValue>> entries;
|
|
entries.reserve(5);
|
|
entries.emplace_back(docIds[idx], "author",
|
|
MetadataValue("author_" + std::to_string(writerId)));
|
|
entries.emplace_back(docIds[idx], "topic",
|
|
MetadataValue("topic_" + std::to_string(seq)));
|
|
entries.emplace_back(docIds[idx], "topic",
|
|
MetadataValue("topic_final_" + std::to_string(seq)));
|
|
entries.emplace_back(docIds[idx], "tag:bench-tag-a",
|
|
MetadataValue("bench-tag-a"));
|
|
entries.emplace_back(docIds[idx], "tag:bench-tag-b",
|
|
MetadataValue("bench-tag-b"));
|
|
auto result = repo.setMetadataBatch(entries);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.documentUpdateWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 9) {
|
|
MetadataOpScope op("daemon_post_ingest_batch_content");
|
|
std::vector<BatchContentEntry> entries;
|
|
entries.reserve(static_cast<size_t>(cfg.metadataBatchSize));
|
|
for (int j = 0; j < cfg.metadataBatchSize; ++j) {
|
|
const auto idx =
|
|
static_cast<size_t>((seq + static_cast<uint64_t>(j) * 13 +
|
|
static_cast<uint64_t>(writerId) * 23) %
|
|
docIds.size());
|
|
BatchContentEntry entry;
|
|
entry.documentId = docIds[idx];
|
|
entry.title = "bench title " + std::to_string(writerId) + "_" +
|
|
std::to_string(seq) + "_" + std::to_string(j);
|
|
entry.contentText = "bench content body " + std::to_string(writerId) +
|
|
"_" + std::to_string(seq) + "_" + std::to_string(j);
|
|
entry.mimeType = "text/plain";
|
|
entry.extractionMethod = "bench";
|
|
entry.language = "en";
|
|
entry.priorStateKnown = true;
|
|
entry.priorContentExtracted = true;
|
|
entry.priorExtractionStatus = yams::metadata::ExtractionStatus::Success;
|
|
entries.push_back(std::move(entry));
|
|
}
|
|
auto result = repo.batchInsertContentAndIndex(entries);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.batchContentWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else {
|
|
MetadataOpScope op("bench_update_extraction_status");
|
|
const auto idx = static_cast<size_t>(
|
|
(seq + static_cast<uint64_t>(writerId) * 31) % docIds.size());
|
|
const bool extracted = (seq % 8) != 3;
|
|
const auto status = extracted ? yams::metadata::ExtractionStatus::Success
|
|
: yams::metadata::ExtractionStatus::Failed;
|
|
const std::string error = extracted ? std::string{} : "bench_lock_probe";
|
|
auto result = repo.updateDocumentExtractionStatus(docIds[idx], extracted,
|
|
status, error);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.writes, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
noteResult(local.extractionWrites, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
}
|
|
++seq;
|
|
}
|
|
std::lock_guard<std::mutex> lock(aggregateMutex);
|
|
mergeWorkerStats(aggregate, local);
|
|
});
|
|
}
|
|
|
|
for (int readerId = 0; readerId < cfg.readerThreads; ++readerId) {
|
|
threads.emplace_back([&, readerId]() {
|
|
WorkerStats local;
|
|
uint64_t seq = static_cast<uint64_t>(readerId) * 101;
|
|
waitForRunStart();
|
|
while (!stop.load(std::memory_order_relaxed) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
const auto t0 = std::chrono::steady_clock::now();
|
|
const auto mode = seq % 3;
|
|
if (mode == 0) {
|
|
MetadataOpScope op("bench_find_document_by_path");
|
|
auto result = repo.findDocumentByExactPath(paths[seq % paths.size()]);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
const bool ok = result.has_value() && result.value().has_value();
|
|
noteResult(local.reads, ok,
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else if (mode == 1) {
|
|
MetadataOpScope op("bench_get_document_by_hash");
|
|
auto result = repo.getDocumentByHash(hashes[seq % hashes.size()]);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
const bool ok = result.has_value() && result.value().has_value();
|
|
noteResult(local.reads, ok,
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
} else {
|
|
MetadataOpScope op("bench_get_metadata_batch");
|
|
std::vector<int64_t> batch;
|
|
batch.reserve(static_cast<size_t>(cfg.readBatchSize));
|
|
for (int j = 0; j < cfg.readBatchSize; ++j) {
|
|
const auto idx = static_cast<size_t>(
|
|
(seq + static_cast<uint64_t>(j) * 7) % docIds.size());
|
|
batch.push_back(docIds[idx]);
|
|
}
|
|
auto result = repo.getMetadataForDocuments(batch);
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
const bool ok = result.has_value();
|
|
noteResult(local.reads, ok,
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
}
|
|
++seq;
|
|
}
|
|
std::lock_guard<std::mutex> lock(aggregateMutex);
|
|
mergeWorkerStats(aggregate, local);
|
|
});
|
|
}
|
|
|
|
ConnectionPool* longReadPool = readPool ? readPool.get() : &writePool;
|
|
for (int longReaderId = 0; longReaderId < cfg.longReaders; ++longReaderId) {
|
|
threads.emplace_back([&, longReaderId]() {
|
|
WorkerStats local;
|
|
waitForRunStart();
|
|
while (!stop.load(std::memory_order_relaxed) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
const auto t0 = std::chrono::steady_clock::now();
|
|
MetadataOpScope op("bench_long_read_tx");
|
|
auto connResult =
|
|
longReadPool->acquire(5s, ConnectionPriority::High, "bench_long_read_tx");
|
|
if (!connResult) {
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.longReads, false,
|
|
std::optional<std::string_view>(connResult.error().message), us);
|
|
continue;
|
|
}
|
|
|
|
std::unique_ptr<yams::metadata::PooledConnection> conn =
|
|
std::move(connResult).value();
|
|
Database& db = **conn;
|
|
auto begin = db.execute("BEGIN");
|
|
if (!begin) {
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.longReads, false,
|
|
std::optional<std::string_view>(begin.error().message), us);
|
|
continue;
|
|
}
|
|
|
|
bool ok = true;
|
|
std::optional<std::string_view> errorMessage;
|
|
auto stmtResult =
|
|
db.prepare("SELECT d.id, d.file_path, m.key, m.value FROM documents d "
|
|
"LEFT JOIN metadata m ON m.document_id = d.id "
|
|
"ORDER BY d.id LIMIT 64");
|
|
if (!stmtResult) {
|
|
ok = false;
|
|
errorMessage = stmtResult.error().message;
|
|
} else {
|
|
yams::metadata::Statement stmt = std::move(stmtResult).value();
|
|
while (true) {
|
|
auto step = stmt.step();
|
|
if (!step) {
|
|
ok = false;
|
|
errorMessage = step.error().message;
|
|
break;
|
|
}
|
|
if (!step.value()) {
|
|
break;
|
|
}
|
|
const auto rowId = stmt.getInt64(0);
|
|
(void)rowId;
|
|
}
|
|
}
|
|
|
|
if (cfg.longReadHoldMs > 0) {
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(cfg.longReadHoldMs));
|
|
}
|
|
|
|
auto commit = db.execute("COMMIT");
|
|
if (!commit && ok) {
|
|
ok = false;
|
|
errorMessage = commit.error().message;
|
|
}
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.longReads, ok, errorMessage, us);
|
|
|
|
if (!ok) {
|
|
db.execute("ROLLBACK");
|
|
}
|
|
(void)longReaderId;
|
|
}
|
|
std::lock_guard<std::mutex> lock(aggregateMutex);
|
|
mergeWorkerStats(aggregate, local);
|
|
});
|
|
}
|
|
|
|
if (cfg.externalLocker) {
|
|
threads.emplace_back([&]() {
|
|
WorkerStats local;
|
|
waitForRunStart();
|
|
auto nextLockAttempt = start + std::chrono::milliseconds(cfg.externalLockEveryMs);
|
|
while (!stop.load(std::memory_order_relaxed) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
if (cfg.externalLockEveryMs > 0) {
|
|
std::this_thread::sleep_until(nextLockAttempt);
|
|
nextLockAttempt += std::chrono::milliseconds(cfg.externalLockEveryMs);
|
|
}
|
|
if (stop.load(std::memory_order_relaxed) ||
|
|
std::chrono::steady_clock::now() >= deadline) {
|
|
break;
|
|
}
|
|
|
|
const auto t0 = std::chrono::steady_clock::now();
|
|
auto connResult = externalLockPool->acquire(5s, ConnectionPriority::Normal,
|
|
"bench_external_locker");
|
|
if (!connResult) {
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.externalLocks, false,
|
|
std::optional<std::string_view>(connResult.error().message), us);
|
|
continue;
|
|
}
|
|
|
|
std::unique_ptr<yams::metadata::PooledConnection> conn =
|
|
std::move(connResult).value();
|
|
Database& db = **conn;
|
|
bool ok = true;
|
|
bool transactionStarted = false;
|
|
std::optional<std::string_view> errorMessage;
|
|
auto begin = db.execute("BEGIN IMMEDIATE");
|
|
if (!begin) {
|
|
ok = false;
|
|
errorMessage = begin.error().message;
|
|
} else {
|
|
transactionStarted = true;
|
|
}
|
|
|
|
if (ok) {
|
|
std::this_thread::sleep_for(
|
|
std::chrono::milliseconds(cfg.externalLockHoldMs));
|
|
auto commit = db.execute("COMMIT");
|
|
if (!commit) {
|
|
ok = false;
|
|
errorMessage = commit.error().message;
|
|
} else {
|
|
transactionStarted = false;
|
|
}
|
|
}
|
|
|
|
if (!ok && transactionStarted) {
|
|
db.execute("ROLLBACK");
|
|
}
|
|
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.externalLocks, ok, errorMessage, us);
|
|
}
|
|
std::lock_guard<std::mutex> lock(aggregateMutex);
|
|
mergeWorkerStats(aggregate, local);
|
|
});
|
|
}
|
|
|
|
if (cfg.checkpointEveryMs > 0) {
|
|
threads.emplace_back([&]() {
|
|
WorkerStats local;
|
|
waitForRunStart();
|
|
auto nextCheckpoint = start + std::chrono::milliseconds(cfg.checkpointEveryMs);
|
|
while (!stop.load(std::memory_order_relaxed) &&
|
|
std::chrono::steady_clock::now() < deadline) {
|
|
std::this_thread::sleep_until(nextCheckpoint);
|
|
nextCheckpoint += std::chrono::milliseconds(cfg.checkpointEveryMs);
|
|
if (stop.load(std::memory_order_relaxed) ||
|
|
std::chrono::steady_clock::now() >= deadline) {
|
|
break;
|
|
}
|
|
const auto t0 = std::chrono::steady_clock::now();
|
|
MetadataOpScope op("bench_checkpoint_truncate");
|
|
auto result = repo.checkpointWalTruncate();
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
noteResult(local.checkpoints, result.has_value(),
|
|
result ? std::nullopt
|
|
: std::optional<std::string_view>(result.error().message),
|
|
us);
|
|
}
|
|
std::lock_guard<std::mutex> lock(aggregateMutex);
|
|
mergeWorkerStats(aggregate, local);
|
|
});
|
|
}
|
|
|
|
while (readyWorkers.load(std::memory_order_acquire) < expectedWorkers) {
|
|
std::this_thread::sleep_for(1ms);
|
|
}
|
|
start = std::chrono::steady_clock::now();
|
|
deadline = start + std::chrono::seconds(cfg.runtimeSeconds);
|
|
runStarted.store(true, std::memory_order_release);
|
|
|
|
for (auto& thread : threads) {
|
|
thread.join();
|
|
}
|
|
stop.store(true, std::memory_order_relaxed);
|
|
|
|
const auto totalWallSeconds =
|
|
std::chrono::duration<double>(std::chrono::steady_clock::now() - start).count();
|
|
const auto activeElapsedSeconds = std::chrono::duration<double>(deadline - start).count();
|
|
const auto cleanupTailSeconds = std::max(0.0, totalWallSeconds - activeElapsedSeconds);
|
|
const auto walAfter = walBytes(dbPath);
|
|
const auto writePoolStats = writePool.getStats();
|
|
const auto readPoolStats =
|
|
readPool ? std::optional<ConnectionPool::Stats>(readPool->getStats()) : std::nullopt;
|
|
|
|
INFO("write p95(us): " << latencyJson(aggregate.writes)["latency_p95_us"]);
|
|
INFO("read p95(us): " << latencyJson(aggregate.reads)["latency_p95_us"]);
|
|
INFO("metadata-single lock errors: " << aggregate.metadataSingleWrites.lockErrors);
|
|
INFO("metadata-batch lock errors: " << aggregate.metadataBatchWrites.lockErrors);
|
|
INFO("download-metadata lock errors: " << aggregate.downloadMetadataWrites.lockErrors);
|
|
INFO("document-update lock errors: " << aggregate.documentUpdateWrites.lockErrors);
|
|
INFO("content-index lock errors: " << aggregate.batchContentWrites.lockErrors);
|
|
INFO("feedback lock errors: " << aggregate.feedbackWrites.lockErrors);
|
|
INFO("repair lock errors: " << aggregate.repairWrites.lockErrors);
|
|
INFO("extraction lock errors: " << aggregate.extractionWrites.lockErrors);
|
|
CHECK((aggregate.writes.success > 0));
|
|
CHECK((aggregate.reads.success > 0));
|
|
if (cfg.externalLocker) {
|
|
CHECK((aggregate.externalLocks.success > 0));
|
|
CHECK(
|
|
(aggregate.writes.lockErrors > 0 || aggregate.metadataSingleWrites.lockErrors > 0 ||
|
|
aggregate.metadataBatchWrites.lockErrors > 0 ||
|
|
aggregate.downloadMetadataWrites.lockErrors > 0 ||
|
|
aggregate.documentUpdateWrites.lockErrors > 0 ||
|
|
aggregate.batchContentWrites.lockErrors > 0 ||
|
|
aggregate.feedbackWrites.lockErrors > 0 || aggregate.repairWrites.lockErrors > 0 ||
|
|
aggregate.extractionWrites.lockErrors > 0));
|
|
}
|
|
|
|
json record{{"timestamp", isoTimestamp()},
|
|
{"repeat", rep + 1},
|
|
{"duration_seconds", activeElapsedSeconds},
|
|
{"total_wall_seconds", totalWallSeconds},
|
|
{"cleanup_tail_seconds", cleanupTailSeconds},
|
|
{"db_path", dbPath.string()},
|
|
{"dual_pool", cfg.dualPool},
|
|
{"seed_docs", cfg.seedDocs},
|
|
{"writer_threads", cfg.writerThreads},
|
|
{"reader_threads", cfg.readerThreads},
|
|
{"long_readers", cfg.longReaders},
|
|
{"long_read_hold_ms", cfg.longReadHoldMs},
|
|
{"metadata_batch_size", cfg.metadataBatchSize},
|
|
{"status_batch_size", cfg.statusBatchSize},
|
|
{"read_batch_size", cfg.readBatchSize},
|
|
{"checkpoint_every_ms", cfg.checkpointEveryMs},
|
|
{"db_busy_timeout_ms", cfg.dbBusyTimeoutMs},
|
|
{"external_locker", cfg.externalLocker},
|
|
{"external_lock_hold_ms", cfg.externalLockHoldMs},
|
|
{"external_lock_every_ms", cfg.externalLockEveryMs},
|
|
{"worker_start_barrier", true},
|
|
{"write_mode", cfg.writeModeName},
|
|
{"wal_bytes_before", walBefore},
|
|
{"wal_bytes_after", walAfter},
|
|
{"writes", latencyJson(aggregate.writes)},
|
|
{"insert_document_writes", latencyJson(aggregate.insertWrites)},
|
|
{"metadata_single_writes", latencyJson(aggregate.metadataSingleWrites)},
|
|
{"metadata_batch_writes", latencyJson(aggregate.metadataBatchWrites)},
|
|
{"download_metadata_writes", latencyJson(aggregate.downloadMetadataWrites)},
|
|
{"document_update_writes", latencyJson(aggregate.documentUpdateWrites)},
|
|
{"batch_content_writes", latencyJson(aggregate.batchContentWrites)},
|
|
{"feedback_event_writes", latencyJson(aggregate.feedbackWrites)},
|
|
{"repair_status_writes", latencyJson(aggregate.repairWrites)},
|
|
{"extraction_status_writes", latencyJson(aggregate.extractionWrites)},
|
|
{"reads", latencyJson(aggregate.reads)},
|
|
{"checkpoints", latencyJson(aggregate.checkpoints)},
|
|
{"long_reads", latencyJson(aggregate.longReads)},
|
|
{"external_locks", latencyJson(aggregate.externalLocks)},
|
|
{"write_pool", poolStatsJson(writePoolStats)}};
|
|
|
|
if (readPoolStats.has_value()) {
|
|
record["read_pool"] = poolStatsJson(*readPoolStats);
|
|
}
|
|
|
|
appendJsonLine(cfg.outputPath, record);
|
|
|
|
std::cout << "[metadata-contention] repeat=" << (rep + 1) << " writes/s="
|
|
<< (activeElapsedSeconds > 0.0
|
|
? static_cast<double>(aggregate.writes.success) / activeElapsedSeconds
|
|
: 0.0)
|
|
<< " reads/s="
|
|
<< (activeElapsedSeconds > 0.0
|
|
? static_cast<double>(aggregate.reads.success) / activeElapsedSeconds
|
|
: 0.0)
|
|
<< " write_p95_us="
|
|
<< latencyJson(aggregate.writes)["latency_p95_us"].get<double>()
|
|
<< " insert_lock_errors=" << aggregate.insertWrites.lockErrors
|
|
<< " metadata_single_lock_errors=" << aggregate.metadataSingleWrites.lockErrors
|
|
<< " metadata_lock_errors=" << aggregate.metadataBatchWrites.lockErrors
|
|
<< " download_metadata_lock_errors="
|
|
<< aggregate.downloadMetadataWrites.lockErrors
|
|
<< " document_update_lock_errors=" << aggregate.documentUpdateWrites.lockErrors
|
|
<< " batch_content_lock_errors=" << aggregate.batchContentWrites.lockErrors
|
|
<< " feedback_lock_errors=" << aggregate.feedbackWrites.lockErrors
|
|
<< " repair_lock_errors=" << aggregate.repairWrites.lockErrors
|
|
<< " extraction_lock_errors=" << aggregate.extractionWrites.lockErrors
|
|
<< " read_p95_us=" << latencyJson(aggregate.reads)["latency_p95_us"].get<double>()
|
|
<< " checkpoint_p95_us="
|
|
<< latencyJson(aggregate.checkpoints)["latency_p95_us"].get<double>()
|
|
<< " cleanup_tail_s=" << cleanupTailSeconds << std::endl;
|
|
|
|
if (externalLockPool) {
|
|
externalLockPool->shutdown();
|
|
}
|
|
if (readPool) {
|
|
readPool->shutdown();
|
|
}
|
|
writePool.shutdown();
|
|
std::error_code ec;
|
|
fs::remove_all(root, ec);
|
|
}
|
|
}
|