651e8c8132* fix: improving write paths and write coordinator enqueue logic * perf(metadata): batch write-path updates * perf(metadata): profile write-path batching * fix(build): gate stacktrace diagnostics * fix(write-path): harden update and shutdown edges Signed-off-by: trvon <git@trevon.dev> * fix(metadata): address review nits Signed-off-by: trvon <git@trevon.dev> --------- Signed-off-by: trvon <git@trevon.dev>
1293 lines
68 KiB
C++
1293 lines
68 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 - checkpoint cadence, 0 disables (default: 500)
|
|
// YAMS_BENCH_CHECKPOINT_MODE - truncate|passive checkpoint type (default: truncate)
|
|
// 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 "bench_utils.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};
|
|
std::string checkpointModeName{"truncate"};
|
|
bool checkpointTruncate{true};
|
|
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 CheckpointContextStats {
|
|
uint64_t samples{0};
|
|
std::uintmax_t maxWalBytesBefore{0};
|
|
std::uintmax_t maxWalBytesAfter{0};
|
|
uint64_t maxWalFramesBefore{0};
|
|
uint64_t maxWalFramesAfter{0};
|
|
uint64_t maxWriteSlowHoldersBefore{0};
|
|
uint64_t maxWriteSlowHoldersAfter{0};
|
|
uint64_t maxReadSlowHoldersBefore{0};
|
|
uint64_t maxReadSlowHoldersAfter{0};
|
|
uint64_t maxWriteHolderMicrosBefore{0};
|
|
uint64_t maxWriteHolderMicrosAfter{0};
|
|
uint64_t maxReadHolderMicrosBefore{0};
|
|
uint64_t maxReadHolderMicrosAfter{0};
|
|
uint64_t maxWriteWaitingBefore{0};
|
|
uint64_t maxWriteWaitingAfter{0};
|
|
uint64_t maxReadWaitingBefore{0};
|
|
uint64_t maxReadWaitingAfter{0};
|
|
};
|
|
|
|
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;
|
|
CheckpointContextStats checkpointContext;
|
|
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};
|
|
};
|
|
|
|
using yams::bench::isoTimestamp;
|
|
using yams::bench::readBoolEnv;
|
|
using yams::bench::readIntEnv;
|
|
|
|
std::pair<bool, std::string> readCheckpointModeEnv() {
|
|
const char* val = std::getenv("YAMS_BENCH_CHECKPOINT_MODE");
|
|
if (!val || !*val) {
|
|
return {true, "truncate"};
|
|
}
|
|
std::string s(val);
|
|
std::transform(s.begin(), s.end(), s.begin(),
|
|
[](unsigned char ch) { return static_cast<char>(std::tolower(ch)); });
|
|
if (s == "passive") {
|
|
return {false, "passive"};
|
|
}
|
|
if (s == "truncate") {
|
|
return {true, "truncate"};
|
|
}
|
|
spdlog::warn("Unknown YAMS_BENCH_CHECKPOINT_MODE='{}'; using truncate", s);
|
|
return {true, "truncate"};
|
|
}
|
|
|
|
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);
|
|
auto [checkpointTruncate, checkpointModeName] = readCheckpointModeEnv();
|
|
cfg.checkpointTruncate = checkpointTruncate;
|
|
cfg.checkpointModeName = std::move(checkpointModeName);
|
|
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;
|
|
}
|
|
|
|
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);
|
|
}
|
|
|
|
uint64_t estimateWalFrames(std::uintmax_t bytes, uint64_t pageSize) {
|
|
constexpr uint64_t kWalHeaderBytes = 32;
|
|
constexpr uint64_t kWalFrameHeaderBytes = 24;
|
|
if (pageSize == 0 || bytes <= kWalHeaderBytes) {
|
|
return 0;
|
|
}
|
|
return static_cast<uint64_t>((bytes - kWalHeaderBytes) / (pageSize + kWalFrameHeaderBytes));
|
|
}
|
|
|
|
uint64_t queryPageSize(Database& db) {
|
|
auto stmtResult = db.prepare("PRAGMA page_size");
|
|
if (!stmtResult) {
|
|
return 4096;
|
|
}
|
|
auto& stmt = stmtResult.value();
|
|
auto step = stmt.step();
|
|
if (!step || !step.value()) {
|
|
return 4096;
|
|
}
|
|
const auto value = stmt.getInt64(0);
|
|
return value > 0 ? static_cast<uint64_t>(value) : 4096;
|
|
}
|
|
|
|
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 mergeCheckpointContext(CheckpointContextStats& dst, const CheckpointContextStats& src) {
|
|
dst.samples += src.samples;
|
|
dst.maxWalBytesBefore = std::max(dst.maxWalBytesBefore, src.maxWalBytesBefore);
|
|
dst.maxWalBytesAfter = std::max(dst.maxWalBytesAfter, src.maxWalBytesAfter);
|
|
dst.maxWalFramesBefore = std::max(dst.maxWalFramesBefore, src.maxWalFramesBefore);
|
|
dst.maxWalFramesAfter = std::max(dst.maxWalFramesAfter, src.maxWalFramesAfter);
|
|
dst.maxWriteSlowHoldersBefore =
|
|
std::max(dst.maxWriteSlowHoldersBefore, src.maxWriteSlowHoldersBefore);
|
|
dst.maxWriteSlowHoldersAfter =
|
|
std::max(dst.maxWriteSlowHoldersAfter, src.maxWriteSlowHoldersAfter);
|
|
dst.maxReadSlowHoldersBefore =
|
|
std::max(dst.maxReadSlowHoldersBefore, src.maxReadSlowHoldersBefore);
|
|
dst.maxReadSlowHoldersAfter =
|
|
std::max(dst.maxReadSlowHoldersAfter, src.maxReadSlowHoldersAfter);
|
|
dst.maxWriteHolderMicrosBefore =
|
|
std::max(dst.maxWriteHolderMicrosBefore, src.maxWriteHolderMicrosBefore);
|
|
dst.maxWriteHolderMicrosAfter =
|
|
std::max(dst.maxWriteHolderMicrosAfter, src.maxWriteHolderMicrosAfter);
|
|
dst.maxReadHolderMicrosBefore =
|
|
std::max(dst.maxReadHolderMicrosBefore, src.maxReadHolderMicrosBefore);
|
|
dst.maxReadHolderMicrosAfter =
|
|
std::max(dst.maxReadHolderMicrosAfter, src.maxReadHolderMicrosAfter);
|
|
dst.maxWriteWaitingBefore = std::max(dst.maxWriteWaitingBefore, src.maxWriteWaitingBefore);
|
|
dst.maxWriteWaitingAfter = std::max(dst.maxWriteWaitingAfter, src.maxWriteWaitingAfter);
|
|
dst.maxReadWaitingBefore = std::max(dst.maxReadWaitingBefore, src.maxReadWaitingBefore);
|
|
dst.maxReadWaitingAfter = std::max(dst.maxReadWaitingAfter, src.maxReadWaitingAfter);
|
|
}
|
|
|
|
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);
|
|
mergeCheckpointContext(dst.checkpointContext, src.checkpointContext);
|
|
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}};
|
|
}
|
|
|
|
json checkpointContextJson(const CheckpointContextStats& stats) {
|
|
return json{{"samples", stats.samples},
|
|
{"max_wal_bytes_before", stats.maxWalBytesBefore},
|
|
{"max_wal_bytes_after", stats.maxWalBytesAfter},
|
|
{"max_wal_frames_before_est", stats.maxWalFramesBefore},
|
|
{"max_wal_frames_after_est", stats.maxWalFramesAfter},
|
|
{"max_write_slow_holder_count_before", stats.maxWriteSlowHoldersBefore},
|
|
{"max_write_slow_holder_count_after", stats.maxWriteSlowHoldersAfter},
|
|
{"max_read_slow_holder_count_before", stats.maxReadSlowHoldersBefore},
|
|
{"max_read_slow_holder_count_after", stats.maxReadSlowHoldersAfter},
|
|
{"max_write_holder_micros_before", stats.maxWriteHolderMicrosBefore},
|
|
{"max_write_holder_micros_after", stats.maxWriteHolderMicrosAfter},
|
|
{"max_read_holder_micros_before", stats.maxReadHolderMicrosBefore},
|
|
{"max_read_holder_micros_after", stats.maxReadHolderMicrosAfter},
|
|
{"max_write_waiting_before", stats.maxWriteWaitingBefore},
|
|
{"max_write_waiting_after", stats.maxWriteWaitingAfter},
|
|
{"max_read_waiting_before", stats.maxReadWaitingBefore},
|
|
{"max_read_waiting_after", stats.maxReadWaitingAfter}};
|
|
}
|
|
|
|
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(" checkpointMode: {}", cfg.checkpointModeName);
|
|
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 walPageSize = writePool.withConnection(
|
|
[&](Database& db) -> yams::Result<uint64_t> { return queryPageSize(db); });
|
|
REQUIRE(walPageSize.has_value());
|
|
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 walBytesBefore = walBytes(dbPath);
|
|
const auto writeStatsBefore = writePool.getStats();
|
|
const auto readStatsBefore =
|
|
readPool ? std::optional<ConnectionPool::Stats>(readPool->getStats())
|
|
: std::nullopt;
|
|
const auto t0 = std::chrono::steady_clock::now();
|
|
MetadataOpScope op(cfg.checkpointTruncate ? "bench_checkpoint_truncate"
|
|
: "bench_checkpoint_passive");
|
|
auto result = cfg.checkpointTruncate ? repo.checkpointWalTruncate()
|
|
: repo.checkpointWal();
|
|
const auto us = std::chrono::duration_cast<std::chrono::microseconds>(
|
|
std::chrono::steady_clock::now() - t0)
|
|
.count();
|
|
const auto walBytesAfter = walBytes(dbPath);
|
|
const auto writeStatsAfter = writePool.getStats();
|
|
const auto readStatsAfter =
|
|
readPool ? std::optional<ConnectionPool::Stats>(readPool->getStats())
|
|
: std::nullopt;
|
|
local.checkpointContext.samples++;
|
|
local.checkpointContext.maxWalBytesBefore =
|
|
std::max(local.checkpointContext.maxWalBytesBefore, walBytesBefore);
|
|
local.checkpointContext.maxWalBytesAfter =
|
|
std::max(local.checkpointContext.maxWalBytesAfter, walBytesAfter);
|
|
local.checkpointContext.maxWalFramesBefore =
|
|
std::max(local.checkpointContext.maxWalFramesBefore,
|
|
estimateWalFrames(walBytesBefore, walPageSize.value()));
|
|
local.checkpointContext.maxWalFramesAfter =
|
|
std::max(local.checkpointContext.maxWalFramesAfter,
|
|
estimateWalFrames(walBytesAfter, walPageSize.value()));
|
|
local.checkpointContext.maxWriteSlowHoldersBefore =
|
|
std::max(local.checkpointContext.maxWriteSlowHoldersBefore,
|
|
writeStatsBefore.slowHolderCount);
|
|
local.checkpointContext.maxWriteSlowHoldersAfter =
|
|
std::max(local.checkpointContext.maxWriteSlowHoldersAfter,
|
|
writeStatsAfter.slowHolderCount);
|
|
local.checkpointContext.maxWriteHolderMicrosBefore =
|
|
std::max(local.checkpointContext.maxWriteHolderMicrosBefore,
|
|
writeStatsBefore.maxHolderMicros);
|
|
local.checkpointContext.maxWriteHolderMicrosAfter =
|
|
std::max(local.checkpointContext.maxWriteHolderMicrosAfter,
|
|
writeStatsAfter.maxHolderMicros);
|
|
local.checkpointContext.maxWriteWaitingBefore =
|
|
std::max(local.checkpointContext.maxWriteWaitingBefore,
|
|
static_cast<uint64_t>(writeStatsBefore.waitingRequests));
|
|
local.checkpointContext.maxWriteWaitingAfter =
|
|
std::max(local.checkpointContext.maxWriteWaitingAfter,
|
|
static_cast<uint64_t>(writeStatsAfter.waitingRequests));
|
|
if (readStatsBefore && readStatsAfter) {
|
|
local.checkpointContext.maxReadSlowHoldersBefore =
|
|
std::max(local.checkpointContext.maxReadSlowHoldersBefore,
|
|
readStatsBefore->slowHolderCount);
|
|
local.checkpointContext.maxReadSlowHoldersAfter =
|
|
std::max(local.checkpointContext.maxReadSlowHoldersAfter,
|
|
readStatsAfter->slowHolderCount);
|
|
local.checkpointContext.maxReadHolderMicrosBefore =
|
|
std::max(local.checkpointContext.maxReadHolderMicrosBefore,
|
|
readStatsBefore->maxHolderMicros);
|
|
local.checkpointContext.maxReadHolderMicrosAfter =
|
|
std::max(local.checkpointContext.maxReadHolderMicrosAfter,
|
|
readStatsAfter->maxHolderMicros);
|
|
local.checkpointContext.maxReadWaitingBefore =
|
|
std::max(local.checkpointContext.maxReadWaitingBefore,
|
|
static_cast<uint64_t>(readStatsBefore->waitingRequests));
|
|
local.checkpointContext.maxReadWaitingAfter =
|
|
std::max(local.checkpointContext.maxReadWaitingAfter,
|
|
static_cast<uint64_t>(readStatsAfter->waitingRequests));
|
|
}
|
|
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},
|
|
{"checkpoint_mode", cfg.checkpointModeName},
|
|
{"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},
|
|
{"wal_page_size", walPageSize.value()},
|
|
{"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)},
|
|
{"checkpoint_context", checkpointContextJson(aggregate.checkpointContext)},
|
|
{"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);
|
|
}
|
|
}
|