// Multi-client ingestion and concurrent access benchmark // Benchmarks the daemon's ability to handle: // 1. Concurrent ingestion from multiple clients // 2. Mixed read/write workloads (ingest + search + list) // 3. Connection fairness and latency under contention // 4. Per-client throughput scaling vs total throughput // 5. Backpressure and queue depth behavior under multi-client load // // Designed for recursive tuning: run repeatedly, adjusting TuneAdvisor knobs, // connection pool sizes, and I/O thread counts to optimize multi-client performance. // // Environment variables: // YAMS_BENCH_NUM_CLIENTS - Number of concurrent clients (default: 4) // YAMS_BENCH_DOCS_PER_CLIENT - Documents each client ingests (default: 100) // YAMS_BENCH_DOC_SIZE_BYTES - Average document size in bytes (default: 2048) // YAMS_BENCH_SEARCH_RATIO - Fraction of ops that are search (default: 0.2) // YAMS_BENCH_MIXED_OPS_PER_CLIENT - Fixed operation budget per mixed client (default: derived) // YAMS_BENCH_WARMUP_DOCS - Docs to pre-ingest for search queries (default: 50) // YAMS_BENCH_OUTPUT - JSONL output path (default: bench_results/multi_client.jsonl) // YAMS_BENCH_DRAIN_TIMEOUT_S - Seconds to wait for queue drain (default: 120) // YAMS_BENCH_SCALING_MAX_CLIENTS - Max clients in scaling sweep (default: 32) // YAMS_TUNING_PROFILE - TuneAdvisor profile (Quiet/Balanced/Performance) // YAMS_BENCH_DATA_DIR - Existing YAMS data dir (for large-corpus bench) // YAMS_BENCH_ENABLE_PLUGINS - Enable plugin loading: 1/true/yes (default: off) // YAMS_BENCH_PLUGIN_DIR - Directory containing plugin .dylib/.so files // YAMS_BENCH_TRUSTED_PLUGIN_PATHS - Colon-separated trusted plugin search paths // YAMS_BENCH_REAL_MODEL_PROVIDER - Use real ONNX model provider: 1/true/yes // YAMS_BENCH_SEARCH_QUERY - Daemon search query for mixed-load phases // YAMS_BENCH_SEARCH_TYPE - Daemon search type: keyword|semantic|hybrid // YAMS_BENCH_SEARCH_LIMIT - Daemon search result limit // YAMS_BENCH_SEARCH_PATHS_ONLY - Skip result hydration and return paths only // YAMS_BENCH_SNIPPET_HYDRATION_TIMEOUT_MS - Search snippet hydration budget // YAMS_BENCH_PROFILE_HYDRATION_SURFACES - Add gated grep/graph render probes // YAMS_BENCH_CLI_SEARCH_QUERY - CLI search query for full subprocess probes // YAMS_BENCH_CLI_SEARCH_TYPE - CLI search type: keyword|semantic|hybrid // YAMS_BENCH_CLI_SEARCH_LIMIT - CLI search result limit // YAMS_BENCH_MCP_SESSION_CHURN_EVERY_N - MCP only: every Nth request uses a fresh // MCP session (default: 0; profile may override) // YAMS_BENCH_IDLE_PROBE - Enable post-run idle transition probe (default: on) // YAMS_BENCH_IDLE_PROBE_BASELINE_MS - Idle baseline sampling window (default: 1500) // YAMS_BENCH_IDLE_PROBE_SAMPLE_MS - Idle probe sampling interval (default: 50) // YAMS_BENCH_IDLE_PROBE_DETECT_TIMEOUT_MS - Max wake detection window (default: 5000) // YAMS_BENCH_MIXED_WRITERS - Mixed-ops writer client count (default: 4) // YAMS_BENCH_MIXED_SEARCHERS - Mixed-ops search client count (default: 4) // YAMS_BENCH_MIXED_LISTERS - Mixed-ops list client count (default: 4) // YAMS_BENCH_MIXED_STATUS - Mixed-ops status/get client count (default: 4) // YAMS_BENCH_MIXED_READER_OPS - Mixed-ops ops per reader client (default: 200) // YAMS_BENCH_MIXED_MAX_FAIL_RATE - Mixed-ops maximum tolerated fail rate (default: 0.0) #define CATCH_CONFIG_MAIN #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #ifdef _WIN32 #include #define getpid _getpid #else #include #include #include #include #include #include #include #include extern char** environ; #endif #include #include #include "../common/test_helpers_catch2.h" #include "../integration/daemon/test_async_helpers.h" #include "../integration/daemon/test_daemon_harness.h" #include "bench_utils.h" #include #include #include #include #include #include #include #include #include #include namespace fs = std::filesystem; using json = nlohmann::json; using namespace std::chrono_literals; using namespace yams::daemon; using namespace yams::test; // ───────────────────────────────────────────────────────────────────────────── // Configuration // ───────────────────────────────────────────────────────────────────────────── namespace { constexpr int kMaxBenchClients = 512; constexpr int kMaxBenchDocsPerClient = 1'000'000; constexpr int kMaxBenchWarmupDocs = 1'000'000; constexpr int kMaxBenchSeconds = 24 * 60 * 60; constexpr int kMaxBenchMilliseconds = 60 * 60 * 1000; constexpr int kMaxBenchStressRounds = 1000; constexpr int kMaxBenchOpsPerClient = 1'000'000; constexpr int kMaxBenchPageSize = 1'000'000; constexpr int kMaxBenchHashLimit = 10'000'000; constexpr std::size_t kMaxBenchDocSizeBytes = 1024ULL * 1024ULL * 1024ULL; constexpr std::size_t kMaxBenchSearchLimit = 1'000'000; constexpr std::int64_t kMaxBenchmarkLatencyUs = 30LL * 24 * 60 * 60 * 1'000'000; constexpr std::int64_t kMaxBenchmarkWallMs = 30LL * 24 * 60 * 60 * 1000; int clampInt64ToInt(std::int64_t value, int minValue, int maxValue) noexcept { if (value < static_cast(minValue)) { return minValue; } if (value > static_cast(maxValue)) { return maxValue; } return static_cast(value); } int parseEnvIntClamped(const char* name, int defaultValue, int minValue, int maxValue) noexcept { const char* raw = std::getenv(name); if (raw == nullptr || *raw == '\0') { return std::clamp(defaultValue, minValue, maxValue); } char* end = nullptr; errno = 0; const long long parsed = std::strtoll(raw, &end, 10); if (end == raw || (end != nullptr && *end != '\0')) { return std::clamp(defaultValue, minValue, maxValue); } if (errno == ERANGE) { return parsed < 0 ? minValue : maxValue; } return clampInt64ToInt(static_cast(parsed), minValue, maxValue); } bool parseEnvBool(const char* name, bool defaultValue = false) { const char* raw = std::getenv(name); if (raw == nullptr) { return defaultValue; } const std::string value(raw); return value == "1" || value == "true" || value == "yes" || value == "on"; } std::size_t parseEnvSizeClamped(const char* name, std::size_t defaultValue, std::size_t minValue, std::size_t maxValue) noexcept { const char* raw = std::getenv(name); if (raw == nullptr || *raw == '\0' || *raw == '-') { return std::clamp(defaultValue, minValue, maxValue); } char* end = nullptr; errno = 0; const unsigned long long parsed = std::strtoull(raw, &end, 10); if (end == raw || (end != nullptr && *end != '\0')) { return std::clamp(defaultValue, minValue, maxValue); } if (errno == ERANGE) { return maxValue; } return std::clamp(static_cast(parsed), minValue, maxValue); } double parseEnvDoubleClamped(const char* name, double defaultValue, double minValue, double maxValue) noexcept { const char* raw = std::getenv(name); if (raw == nullptr || *raw == '\0') { return std::clamp(defaultValue, minValue, maxValue); } char* end = nullptr; errno = 0; const double parsed = std::strtod(raw, &end); if (end == raw || (end != nullptr && *end != '\0') || errno == ERANGE || !std::isfinite(parsed)) { return std::clamp(defaultValue, minValue, maxValue); } return std::clamp(parsed, minValue, maxValue); } int checkedMulToInt(int lhs, int rhs, int maxValue = std::numeric_limits::max()) noexcept { const auto product = static_cast(lhs) * static_cast(rhs); return clampInt64ToInt(product, 0, maxValue); } int checkedAddToInt(std::initializer_list values, int maxValue = std::numeric_limits::max()) noexcept { std::int64_t total = 0; for (int value : values) { total += static_cast(value); if (total > static_cast(maxValue)) { return maxValue; } } return clampInt64ToInt(total, 0, maxValue); } std::int64_t clampMetricSample(std::int64_t value, std::int64_t maxValue = kMaxBenchmarkLatencyUs) noexcept { if (value <= 0) { return 0; } return std::min(value, maxValue); } std::int64_t clampUnsignedMetricSample(std::uint64_t value, std::int64_t maxValue = kMaxBenchmarkLatencyUs) noexcept { const auto unsignedMax = static_cast(maxValue); return value > unsignedMax ? maxValue : static_cast(value); } void pushMetricSample(std::vector& values, std::int64_t value, std::int64_t maxValue = kMaxBenchmarkLatencyUs) { if (value >= 0) { values.push_back(clampMetricSample(value, maxValue)); } } std::int64_t checkedMetricAdd(std::int64_t lhs, std::int64_t rhs, std::int64_t maxValue = std::numeric_limits::max()) { lhs = clampMetricSample(lhs, maxValue); rhs = clampMetricSample(rhs, maxValue); if (lhs > maxValue - rhs) { return maxValue; } return lhs + rhs; } std::int64_t checkedMetricMul(std::int64_t lhs, std::int64_t rhs, std::int64_t maxValue = std::numeric_limits::max()) { lhs = clampMetricSample(lhs, maxValue); rhs = clampMetricSample(rhs, maxValue); if (lhs != 0 && rhs > maxValue / lhs) { return maxValue; } return lhs * rhs; } struct BenchConfig { int numClients{4}; int docsPerClient{100}; size_t docSizeBytes{2048}; double searchRatio{0.2}; int warmupDocs{50}; fs::path outputPath{"bench_results/multi_client.jsonl"}; int drainTimeoutSecs{120}; int scalingMaxClients{32}; int stressRounds{1}; int cliProbeOpsPerCommand{3}; int cliProbeTimeoutSecs{10}; std::string tuningProfile; std::optional cliBinary; std::string searchQuery{"architecture performance tuning daemon"}; std::string searchType{"hybrid"}; size_t searchLimit{5}; bool searchPathsOnly{false}; int snippetHydrationTimeoutMs{5000}; bool profileHydrationSurfaces{false}; int resourceSampleMs{500}; std::string cliSearchQuery{"architecture performance tuning daemon"}; std::string cliSearchType{"hybrid"}; size_t cliSearchLimit{5}; bool idleProbeEnabled{true}; int idleProbeBaselineMs{1500}; int idleProbeSampleMs{50}; int idleProbeDetectTimeoutMs{5000}; int mixedWriterClients{4}; int mixedSearchClients{4}; int mixedListClients{4}; int mixedStatusClients{4}; int mixedOpsPerReader{200}; int mixedOpsPerClient{0}; double mixedMaxFailRate{0.0}; // --- Storage / corpus overrides --- // Point at an existing YAMS data directory (e.g. /Volumes/picaso/yams) // When set, the daemon serves this corpus instead of creating an empty temp db. std::optional dataDir; // Optional explicit config path used for daemon startup/config lookup. std::optional configPath; // --- Plugin / embedding config --- // Enable real plugin loading (ONNX embeddings, PDF extraction, etc.) bool enablePlugins{false}; // Directory containing built plugin .dylib/.so files std::optional pluginDir; // Trusted plugin search paths (from plugins.trust) std::vector trustedPluginPaths; // Use real model provider instead of mock (requires ONNX runtime + models) bool useRealModelProvider{false}; bool useMcpPath{false}; std::string usageProfile{"mixed"}; std::string backend{"unknown"}; std::string authMode{""}; static BenchConfig fromEnv() { BenchConfig cfg; cfg.numClients = parseEnvIntClamped("YAMS_BENCH_NUM_CLIENTS", cfg.numClients, 1, kMaxBenchClients); cfg.docsPerClient = parseEnvIntClamped("YAMS_BENCH_DOCS_PER_CLIENT", cfg.docsPerClient, 1, kMaxBenchDocsPerClient); cfg.docSizeBytes = parseEnvSizeClamped("YAMS_BENCH_DOC_SIZE_BYTES", cfg.docSizeBytes, 64, kMaxBenchDocSizeBytes); cfg.searchRatio = parseEnvDoubleClamped("YAMS_BENCH_SEARCH_RATIO", cfg.searchRatio, 0.0, 1.0); cfg.warmupDocs = parseEnvIntClamped("YAMS_BENCH_WARMUP_DOCS", cfg.warmupDocs, 0, kMaxBenchWarmupDocs); if (auto* v = std::getenv("YAMS_BENCH_OUTPUT")) cfg.outputPath = v; cfg.drainTimeoutSecs = parseEnvIntClamped("YAMS_BENCH_DRAIN_TIMEOUT_S", cfg.drainTimeoutSecs, 10, kMaxBenchSeconds); cfg.scalingMaxClients = parseEnvIntClamped("YAMS_BENCH_SCALING_MAX_CLIENTS", cfg.scalingMaxClients, 1, kMaxBenchClients); cfg.stressRounds = parseEnvIntClamped("YAMS_BENCH_STRESS_ROUNDS", cfg.stressRounds, 1, kMaxBenchStressRounds); cfg.cliProbeOpsPerCommand = parseEnvIntClamped( "YAMS_BENCH_CLI_PROBE_OPS", cfg.cliProbeOpsPerCommand, 0, kMaxBenchOpsPerClient); cfg.cliProbeTimeoutSecs = parseEnvIntClamped("YAMS_BENCH_CLI_PROBE_TIMEOUT_S", cfg.cliProbeTimeoutSecs, 1, kMaxBenchSeconds); if (auto* v = std::getenv("YAMS_BENCH_CLI_BIN")) cfg.cliBinary = fs::path(v); if (auto* v = std::getenv("YAMS_BENCH_SEARCH_QUERY")) cfg.searchQuery = v; if (auto* v = std::getenv("YAMS_BENCH_SEARCH_TYPE")) cfg.searchType = v; cfg.searchLimit = parseEnvSizeClamped("YAMS_BENCH_SEARCH_LIMIT", cfg.searchLimit, 1, kMaxBenchSearchLimit); cfg.searchPathsOnly = parseEnvBool("YAMS_BENCH_SEARCH_PATHS_ONLY"); cfg.snippetHydrationTimeoutMs = parseEnvIntClamped("YAMS_BENCH_SNIPPET_HYDRATION_TIMEOUT_MS", cfg.snippetHydrationTimeoutMs, 0, kMaxBenchMilliseconds); cfg.profileHydrationSurfaces = parseEnvBool("YAMS_BENCH_PROFILE_HYDRATION_SURFACES"); cfg.resourceSampleMs = parseEnvIntClamped("YAMS_BENCH_RESOURCE_SAMPLE_MS", cfg.resourceSampleMs, 10, kMaxBenchMilliseconds); if (auto* v = std::getenv("YAMS_BENCH_CLI_SEARCH_QUERY")) cfg.cliSearchQuery = v; if (auto* v = std::getenv("YAMS_BENCH_CLI_SEARCH_TYPE")) cfg.cliSearchType = v; cfg.cliSearchLimit = parseEnvSizeClamped("YAMS_BENCH_CLI_SEARCH_LIMIT", cfg.cliSearchLimit, 1, kMaxBenchSearchLimit); if (auto* v = std::getenv("YAMS_BENCH_IDLE_PROBE")) { std::string s(v); cfg.idleProbeEnabled = !(s == "0" || s == "false" || s == "no" || s == "off"); } cfg.idleProbeBaselineMs = parseEnvIntClamped( "YAMS_BENCH_IDLE_PROBE_BASELINE_MS", cfg.idleProbeBaselineMs, 0, kMaxBenchMilliseconds); cfg.idleProbeSampleMs = parseEnvIntClamped( "YAMS_BENCH_IDLE_PROBE_SAMPLE_MS", cfg.idleProbeSampleMs, 10, kMaxBenchMilliseconds); cfg.idleProbeDetectTimeoutMs = parseEnvIntClamped("YAMS_BENCH_IDLE_PROBE_DETECT_TIMEOUT_MS", cfg.idleProbeDetectTimeoutMs, 250, kMaxBenchMilliseconds); cfg.mixedWriterClients = parseEnvIntClamped("YAMS_BENCH_MIXED_WRITERS", cfg.mixedWriterClients, 1, kMaxBenchClients); cfg.mixedSearchClients = parseEnvIntClamped("YAMS_BENCH_MIXED_SEARCHERS", cfg.mixedSearchClients, 1, kMaxBenchClients); cfg.mixedListClients = parseEnvIntClamped("YAMS_BENCH_MIXED_LISTERS", cfg.mixedListClients, 1, kMaxBenchClients); cfg.mixedStatusClients = parseEnvIntClamped("YAMS_BENCH_MIXED_STATUS", cfg.mixedStatusClients, 1, kMaxBenchClients); cfg.mixedOpsPerReader = parseEnvIntClamped("YAMS_BENCH_MIXED_READER_OPS", cfg.mixedOpsPerReader, 1, kMaxBenchOpsPerClient); cfg.mixedOpsPerClient = parseEnvIntClamped("YAMS_BENCH_MIXED_OPS_PER_CLIENT", cfg.mixedOpsPerClient, 0, kMaxBenchOpsPerClient); cfg.mixedMaxFailRate = parseEnvDoubleClamped("YAMS_BENCH_MIXED_MAX_FAIL_RATE", cfg.mixedMaxFailRate, 0.0, 1.0); cfg.scalingMaxClients = std::max(cfg.scalingMaxClients, checkedMulToInt(cfg.numClients, 2, kMaxBenchClients)); cfg.cliProbeOpsPerCommand = std::max(cfg.cliProbeOpsPerCommand, cfg.stressRounds); if (cfg.cliSearchQuery.empty()) cfg.cliSearchQuery = cfg.searchQuery; if (cfg.cliSearchType.empty()) cfg.cliSearchType = cfg.searchType; if (cfg.cliSearchLimit == 0) cfg.cliSearchLimit = cfg.searchLimit; if (auto* v = std::getenv("YAMS_TUNING_PROFILE")) cfg.tuningProfile = v; // Storage path override: point at an existing corpus if (auto* v = std::getenv("YAMS_BENCH_DATA_DIR")) cfg.dataDir = fs::path(v); if (auto* v = std::getenv("YAMS_BENCH_CONFIG")) cfg.configPath = fs::path(v); // Plugin / embedding configuration if (auto* v = std::getenv("YAMS_BENCH_ENABLE_PLUGINS")) { std::string s(v); cfg.enablePlugins = (s == "1" || s == "true" || s == "yes"); } if (auto* v = std::getenv("YAMS_BENCH_PLUGIN_DIR")) cfg.pluginDir = fs::path(v); if (auto* v = std::getenv("YAMS_BENCH_TRUSTED_PLUGIN_PATHS")) { // Colon-separated list of paths std::string paths(v); std::istringstream iss(paths); std::string path; while (std::getline(iss, path, ':')) { if (!path.empty()) cfg.trustedPluginPaths.push_back(fs::path(path)); } } if (auto* v = std::getenv("YAMS_BENCH_REAL_MODEL_PROVIDER")) { std::string s(v); cfg.useRealModelProvider = (s == "1" || s == "true" || s == "yes"); } if (auto* v = std::getenv("YAMS_BENCH_USE_MCP")) { std::string s(v); cfg.useMcpPath = (s == "1" || s == "true" || s == "yes"); } if (auto* v = std::getenv("YAMS_BENCH_USAGE_PROFILE")) { cfg.usageProfile = v; } if (auto* v = std::getenv("YAMS_BENCH_BACKEND")) { cfg.backend = v; } if (auto* v = std::getenv("YAMS_BENCH_AUTH_MODE")) { cfg.authMode = v; } return cfg; } }; struct OutputLivenessCounters { std::atomic attempts{0}; std::atomic nonemptySuccesses{0}; std::atomic emptySuccesses{0}; std::atomic errors{0}; void recordSuccess(bool nonempty) { attempts.fetch_add(1, std::memory_order_relaxed); (nonempty ? nonemptySuccesses : emptySuccesses).fetch_add(1, std::memory_order_relaxed); } void recordError() { attempts.fetch_add(1, std::memory_order_relaxed); errors.fetch_add(1, std::memory_order_relaxed); } [[nodiscard]] json toJson() const { return { {"attempts", attempts.load(std::memory_order_relaxed)}, {"nonempty_successes", nonemptySuccesses.load(std::memory_order_relaxed)}, {"empty_successes", emptySuccesses.load(std::memory_order_relaxed)}, {"errors", errors.load(std::memory_order_relaxed)}, }; } }; TEST_CASE("Multi-client benchmark env parsing clamps extreme numeric values", "[multi-client][config]") { ScopedEnvVar clients("YAMS_BENCH_NUM_CLIENTS", std::string("999999999999999999999")); ScopedEnvVar docs("YAMS_BENCH_DOCS_PER_CLIENT", std::string("999999999999999999999")); ScopedEnvVar docSize("YAMS_BENCH_DOC_SIZE_BYTES", std::string("999999999999999999999")); ScopedEnvVar ratio("YAMS_BENCH_SEARCH_RATIO", std::string("inf")); ScopedEnvVar scaling("YAMS_BENCH_SCALING_MAX_CLIENTS", std::string("999999999999999999999")); ScopedEnvVar mixedReaders("YAMS_BENCH_MIXED_READER_OPS", std::string("999999999999999999999")); const auto cfg = BenchConfig::fromEnv(); CHECK(cfg.numClients == kMaxBenchClients); CHECK(cfg.docsPerClient == kMaxBenchDocsPerClient); CHECK(cfg.docSizeBytes == kMaxBenchDocSizeBytes); CHECK(cfg.searchRatio == 0.2); CHECK(cfg.scalingMaxClients == kMaxBenchClients); CHECK(cfg.mixedOpsPerReader == kMaxBenchOpsPerClient); } std::vector buildCliSearchArgs(const BenchConfig& cfg) { std::vector args{ "search", cfg.cliSearchQuery, "--limit", std::to_string(cfg.cliSearchLimit), }; if (!cfg.cliSearchType.empty()) { args.push_back("--type"); args.push_back(cfg.cliSearchType); } return args; } class NullTransport : public yams::mcp::ITransport { public: void send(const nlohmann::json&) override {} yams::mcp::MessageResult receive() override { return yams::Error{yams::ErrorCode::NotImplemented, "Null transport"}; } bool isConnected() const override { return false; } void close() override {} yams::mcp::TransportState getState() const override { return yams::mcp::TransportState::Disconnected; } }; class MCPPipelineClient { public: MCPPipelineClient() { server_ = std::make_shared(std::make_unique()); initialize(); } yams::Result queryStep(const std::string& op, const nlohmann::json& params) { std::lock_guard lock(mu_); return queryStepUnlocked(op, params); } private: static bool isQueryCompositeReadOp(const std::string& op) { return op == "search" || op == "grep" || op == "list" || op == "list_collections" || op == "list_snapshots" || op == "graph" || op == "get" || op == "status"; } yams::Result queryStepUnlocked(const std::string& op, const nlohmann::json& params) { nlohmann::json request = nlohmann::json::object(); request["jsonrpc"] = "2.0"; request["id"] = requestId_++; request["method"] = "tools/call"; request["params"] = nlohmann::json::object(); const bool useQueryTool = hasQueryTool_ && isQueryCompositeReadOp(op); request["params"]["name"] = useQueryTool ? "query" : op; if (useQueryTool) { request["params"]["arguments"] = nlohmann::json::object(); request["params"]["arguments"]["steps"] = nlohmann::json::array({{{"op", op}, {"params", params}}}); } else { request["params"]["arguments"] = params.is_object() ? params : nlohmann::json::object(); } auto responseResult = server_->processMessage(request); if (!responseResult) { return yams::Error{responseResult.error()}; } const auto& response = responseResult.value(); if (response.contains("error")) { const std::string msg = response["error"].value("message", "MCP tools/call error"); if (hasQueryTool_ && msg.rfind("Unknown tool: query", 0) == 0) { hasQueryTool_ = false; return queryStepUnlocked(op, params); } return yams::Error{yams::ErrorCode::InvalidState, msg}; } if (!response.contains("result") || !response["result"].is_object()) { return yams::Error{yams::ErrorCode::InvalidState, "MCP tools/call missing result object"}; } const auto& result = response["result"]; if (result.value("isError", false)) { std::string msg{"MCP tool returned error"}; if (result.contains("content") && result["content"].is_array() && !result["content"].empty() && result["content"][0].contains("text")) { msg = result["content"][0]["text"].get(); } return yams::Error{yams::ErrorCode::InvalidState, msg}; } if (result.contains("structuredContent") && result["structuredContent"].is_object()) { const auto& sc = result["structuredContent"]; if (sc.contains("data")) { return sc["data"]; } } if (result.contains("content") && result["content"].is_array() && !result["content"].empty() && result["content"][0].contains("text")) { try { return nlohmann::json::parse(result["content"][0]["text"].get()); } catch (...) { return nlohmann::json{{"text", result["content"][0]["text"]}}; } } return yams::Error{yams::ErrorCode::InvalidState, "MCP tool result had no data"}; } void initialize() { nlohmann::json initReq = nlohmann::json::object(); initReq["jsonrpc"] = "2.0"; initReq["id"] = requestId_++; initReq["method"] = "initialize"; initReq["params"] = nlohmann::json::object(); initReq["params"]["protocolVersion"] = "2025-06-18"; initReq["params"]["capabilities"] = nlohmann::json::object(); initReq["params"]["clientInfo"] = nlohmann::json{{"name", "multi_client_ingestion_bench"}, {"version", "1.0"}}; auto initRes = server_->processMessage(initReq); if (!initRes) { throw std::runtime_error("MCP initialize failed: " + initRes.error().message); } nlohmann::json initializedNotif = {{"jsonrpc", "2.0"}, {"method", "notifications/initialized"}, {"params", nlohmann::json::object()}}; (void)server_->processMessage(initializedNotif); nlohmann::json toolsReq = nlohmann::json::object(); toolsReq["jsonrpc"] = "2.0"; toolsReq["id"] = requestId_++; toolsReq["method"] = "tools/list"; toolsReq["params"] = nlohmann::json::object(); auto toolsRes = server_->processMessage(toolsReq); if (!toolsRes) { throw std::runtime_error("MCP tools/list failed: " + toolsRes.error().message); } const auto& toolsResp = toolsRes.value(); if (!toolsResp.contains("result") || !toolsResp["result"].is_object() || !toolsResp["result"].contains("tools") || !toolsResp["result"]["tools"].is_array()) { throw std::runtime_error("MCP tools/list returned invalid shape"); } hasQueryTool_ = false; for (const auto& tool : toolsResp["result"]["tools"]) { if (!tool.is_object() || !tool.contains("name") || !tool["name"].is_string()) { continue; } if (tool["name"].get() == "query") { hasQueryTool_ = true; break; } } } std::shared_ptr server_; std::mutex mu_; bool hasQueryTool_{false}; std::uint64_t requestId_{1}; }; class MCPPipelineClientPool { public: MCPPipelineClientPool(const fs::path& socketPath, size_t poolSize) : socketEnvironment_("YAMS_DAEMON_SOCKET_PATH", socketPath.string()), autoStartEnvironment_("YAMS_CLI_DISABLE_DAEMON_AUTOSTART", std::string{"1"}) { const size_t size = std::max(1, poolSize); clients_.reserve(size); for (size_t i = 0; i < size; ++i) { clients_.push_back(std::make_unique()); } } yams::Result queryStepForSlot(size_t slot, const std::string& op, const nlohmann::json& params) { const size_t idx = slot % clients_.size(); return clients_[idx]->queryStep(op, params); } private: yams::test::ScopedEnvVar socketEnvironment_; yams::test::ScopedEnvVar autoStartEnvironment_; std::vector> clients_; }; // ───────────────────────────────────────────────────────────────────────────── // Latency & throughput tracking // ───────────────────────────────────────────────────────────────────────────── struct OpLatency { std::string opType; // "add", "search", "list", "status" int clientId{0}; int64_t latencyUs{0}; bool success{false}; std::string errorMsg; }; struct ClientResult { int clientId{0}; int docsIngested{0}; int searchesExecuted{0}; int listsExecuted{0}; int statusChecks{0}; int failures{0}; double elapsedSeconds{0.0}; double addThroughputDocsPerSec{0.0}; std::vector latencies; std::vector snippetHydrationUs; std::vector snippetLookupUs; std::vector snippetContentFetchUs; std::vector snippetRenderUs; int64_t snippetTimeoutHits{0}; int64_t snippetHydrationCandidates{0}; int64_t snippetHydrationMissing{0}; int64_t snippetHydratedResults{0}; }; double safeRate(double numerator, double elapsedSeconds) { return (numerator > 0.0 && elapsedSeconds > 0.0) ? numerator / elapsedSeconds : 0.0; } struct PercentileStats { int64_t min{0}; int64_t p50{0}; int64_t p90{0}; int64_t p95{0}; int64_t p99{0}; int64_t max{0}; double mean{0.0}; size_t count{0}; static PercentileStats compute(std::vector& values) { values.erase( std::remove_if(values.begin(), values.end(), [](int64_t value) { return value < 0; }), values.end()); for (auto& value : values) { value = clampMetricSample(value); } PercentileStats stats; stats.count = values.size(); if (values.empty()) return stats; std::sort(values.begin(), values.end()); const auto percentileIndex = [&](double percentile) { return std::min(values.size() - 1, static_cast(percentile * static_cast(values.size()))); }; stats.min = values.front(); stats.max = values.back(); stats.p50 = values[percentileIndex(0.50)]; stats.p90 = values[percentileIndex(0.90)]; stats.p95 = values[percentileIndex(0.95)]; stats.p99 = values[percentileIndex(0.99)]; const long double total = std::accumulate(values.begin(), values.end(), 0.0L, [](long double sum, int64_t value) { return sum + static_cast(value); }); stats.mean = static_cast(total / static_cast(values.size())); return stats; } json toJson() const { return json{{"count", count}, {"min_us", min}, {"p50_us", p50}, {"p90_us", p90}, {"p95_us", p95}, {"p99_us", p99}, {"max_us", max}, {"mean_us", mean}}; } }; std::optional findExecutableInPath(const std::string& name) { if (const char* pathEnv = std::getenv("PATH")) { std::stringstream ss(pathEnv); std::string entry; while (std::getline(ss, entry, ':')) { if (entry.empty()) { continue; } fs::path candidate = fs::path(entry) / name; std::error_code ec; if (fs::exists(candidate, ec) && !ec) { return candidate; } } } return std::nullopt; } struct CliBinarySelection { std::optional path; std::string source{"missing"}; std::string note; bool explicitSelection{false}; bool pathFallback{false}; bool usableForStress{false}; json toJson() const { return json{{"path", path ? path->string() : std::string()}, {"source", source}, {"note", note}, {"explicit_selection", explicitSelection}, {"path_fallback", pathFallback}, {"usable_for_stress", usableForStress}}; } }; bool isExecutableBinary(const fs::path& candidate) { std::error_code ec; if (!fs::exists(candidate, ec) || ec) { return false; } #ifdef _WIN32 return true; #else return ::access(candidate.c_str(), X_OK) == 0; #endif } CliBinarySelection findYamsBinary(const BenchConfig& cfg) { auto resolveCandidate = [](const fs::path& candidate) -> std::optional { if (!isExecutableBinary(candidate)) { return std::nullopt; } return fs::absolute(candidate); }; if (cfg.cliBinary && fs::exists(cfg.cliBinary.value())) { CliBinarySelection selection; selection.path = fs::absolute(cfg.cliBinary.value()); selection.source = "explicit"; selection.explicitSelection = true; selection.usableForStress = true; return selection; } if (const char* buildRoot = std::getenv("MESON_BUILD_ROOT")) { for (const auto& relative : {fs::path("tools/yams-cli/yams-cli"), fs::path("tools/yams-cli/yams"), fs::path("src/cli/yams")}) { #ifdef _WIN32 const fs::path candidate = fs::path(buildRoot) / relative.replace_extension(".exe"); #else const fs::path candidate = fs::path(buildRoot) / relative; #endif if (auto resolved = resolveCandidate(candidate)) { CliBinarySelection selection; selection.path = resolved.value(); selection.source = "meson_build_root"; selection.usableForStress = true; return selection; } } } for (const auto& relative : {fs::path("builddir/tools/yams-cli/yams-cli"), fs::path("build/tools/yams-cli/yams-cli"), fs::path("build/debug/tools/yams-cli/yams-cli"), fs::path("build/release/tools/yams-cli/yams-cli"), fs::path("build/asan/tools/yams-cli/yams-cli"), fs::path("build/opencode-tests/tools/yams-cli/yams-cli"), fs::path("builddir/src/cli/yams"), fs::path("build/src/cli/yams"), fs::path("build/debug/src/cli/yams"), fs::path("build/release/src/cli/yams"), fs::path("build/asan/src/cli/yams"), fs::path("build/opencode-tests/src/cli/yams")}) { #ifdef _WIN32 const fs::path candidate = relative.replace_extension(".exe"); #else const fs::path candidate = relative; #endif if (auto resolved = resolveCandidate(candidate)) { CliBinarySelection selection; selection.path = resolved.value(); selection.source = "local_build_tree"; selection.usableForStress = true; return selection; } } #ifdef _WIN32 if (auto candidate = findExecutableInPath("yams.exe")) { CliBinarySelection selection; selection.path = candidate.value(); selection.source = "path_fallback"; selection.note = "PATH fallback may not match the build under test; set YAMS_BENCH_CLI_BIN for " "deterministic CLI stress results"; selection.pathFallback = true; selection.usableForStress = false; return selection; } #else if (auto candidate = findExecutableInPath("yams")) { CliBinarySelection selection; selection.path = candidate.value(); selection.source = "path_fallback"; selection.note = "PATH fallback may not match the build under test; set YAMS_BENCH_CLI_BIN for " "deterministic CLI stress results"; selection.pathFallback = true; selection.usableForStress = false; return selection; } #endif return CliBinarySelection{.source = "missing", .note = "No in-tree CLI binary found; set YAMS_BENCH_CLI_BIN or build " "tools/yams-cli/yams-cli to enable CLI stress probes", .usableForStress = false}; } struct ProbeDaemonState { bool captured{false}; size_t activeConnections{0}; size_t maxConnections{0}; size_t connectionSlotsFree{0}; int pressureLevel{0}; uint32_t retryAfterMs{0}; std::string error; bool underStress() const { if (captured) { return (maxConnections > 0 && activeConnections >= maxConnections) || connectionSlotsFree == 0 || retryAfterMs > 0 || pressureLevel >= 1; } if (error.empty()) { return false; } std::string lower; lower.reserve(error.size()); for (unsigned char ch : error) { lower.push_back(static_cast(std::tolower(ch))); } auto has = [&lower](std::string_view needle) { return lower.find(needle) != std::string::npos; }; return has("ipc:eof") || has("eof") || has("end of file") || has("timed out") || has("timeout") || has("connection reset") || has("broken pipe") || has("resource exhausted") || has("retry after") || has("server busy") || has("bad file descriptor"); } json toJson() const { return json{{"captured", captured}, {"active_connections", activeConnections}, {"max_connections", maxConnections}, {"connection_slots_free", connectionSlotsFree}, {"pressure_level", pressureLevel}, {"retry_after_ms", retryAfterMs}, {"error", error}}; } }; struct CliProbeResult { bool attempted{false}; bool success{false}; int exitCode{-1}; int signal{-1}; int64_t latencyUs{0}; bool timedOut{false}; std::string opType; std::string error; std::string stderrText; std::string failureBucket; std::string failureMode; ProbeDaemonState beforeState; ProbeDaemonState afterState; std::unordered_map stageElapsedUs; json toJson() const { return json{{"attempted", attempted}, {"success", success}, {"op_type", opType}, {"exit_code", exitCode}, {"signal", signal}, {"latency_us", latencyUs}, {"timed_out", timedOut}, {"error", error}, {"stderr", stderrText}, {"failure_bucket", failureBucket}, {"failure_mode", failureMode}, {"before_state", beforeState.toJson()}, {"after_state", afterState.toJson()}, {"stage_elapsed_us", stageElapsedUs}}; } }; #ifndef _WIN32 namespace {} // namespace CliProbeResult runCliProbe(const fs::path& yamsBinary, const fs::path& socketPath, const std::vector& args, std::chrono::milliseconds timeout = 10s) { CliProbeResult result; result.attempted = true; int sinkFd = ::open("/dev/null", O_WRONLY); if (sinkFd < 0) { result.error = "open(/dev/null) failed"; return result; } int tracePipe[2] = {-1, -1}; if (::pipe(tracePipe) != 0) { ::close(sinkFd); result.error = "pipe() failed"; return result; } std::vector owned; owned.reserve(args.size() + 1); owned.push_back(yamsBinary.string()); owned.insert(owned.end(), args.begin(), args.end()); std::vector argv; argv.reserve(owned.size() + 1); for (auto& arg : owned) { argv.push_back(arg.data()); } argv.push_back(nullptr); std::vector envOwned; if (environ != nullptr) { for (char** env = environ; *env != nullptr; ++env) { envOwned.emplace_back(*env); } } auto setOrReplaceEnv = [&](std::string key, std::string value) { const std::string prefix = key + "="; for (auto& entry : envOwned) { if (entry.rfind(prefix, 0) == 0) { entry = prefix + value; return; } } envOwned.push_back(prefix + value); }; setOrReplaceEnv("YAMS_DAEMON_SOCKET_PATH", socketPath.string()); setOrReplaceEnv("YAMS_CLI_DISABLE_DAEMON_AUTOSTART", "1"); setOrReplaceEnv("YAMS_CLI_ONE_SHOT", "1"); setOrReplaceEnv("YAMS_CLI_PERF_TRACE", "1"); setOrReplaceEnv("YAMS_LOG_LEVEL", "error"); std::vector envp; envp.reserve(envOwned.size() + 1); for (auto& entry : envOwned) { envp.push_back(entry.data()); } envp.push_back(nullptr); posix_spawn_file_actions_t actions; if (::posix_spawn_file_actions_init(&actions) != 0) { ::close(sinkFd); ::close(tracePipe[0]); ::close(tracePipe[1]); result.error = "posix_spawn file actions init failed"; return result; } (void)::posix_spawn_file_actions_adddup2(&actions, sinkFd, STDOUT_FILENO); (void)::posix_spawn_file_actions_adddup2(&actions, tracePipe[1], STDERR_FILENO); (void)::posix_spawn_file_actions_addclose(&actions, sinkFd); (void)::posix_spawn_file_actions_addclose(&actions, tracePipe[0]); (void)::posix_spawn_file_actions_addclose(&actions, tracePipe[1]); posix_spawnattr_t attr; (void)::posix_spawnattr_init(&attr); #ifdef POSIX_SPAWN_CLOEXEC_DEFAULT short spawnFlags = 0; spawnFlags |= POSIX_SPAWN_CLOEXEC_DEFAULT; (void)::posix_spawnattr_setflags(&attr, spawnFlags); #endif const auto start = std::chrono::steady_clock::now(); pid_t pid = -1; const int spawnErr = ::posix_spawn(&pid, yamsBinary.c_str(), &actions, &attr, argv.data(), envp.data()); (void)::posix_spawn_file_actions_destroy(&actions); (void)::posix_spawnattr_destroy(&attr); if (spawnErr != 0) { ::close(sinkFd); ::close(tracePipe[0]); ::close(tracePipe[1]); result.error = "posix_spawn() failed: " + std::string(std::strerror(spawnErr)); return result; } ::close(sinkFd); ::close(tracePipe[1]); int status = 0; const auto deadline = std::chrono::steady_clock::now() + timeout; for (;;) { pid_t waited = ::waitpid(pid, &status, WNOHANG); if (waited == pid) { break; } if (waited < 0) { if (errno == EINTR) { continue; } result.error = "waitpid() failed"; break; } if (std::chrono::steady_clock::now() >= deadline) { ::kill(pid, SIGKILL); while (::waitpid(pid, &status, 0) < 0 && errno == EINTR) { } result.timedOut = true; result.error = "cli probe timed out"; result.exitCode = 124; result.latencyUs = std::chrono::duration_cast( std::chrono::steady_clock::now() - start) .count(); break; } std::this_thread::sleep_for(10ms); } result.latencyUs = std::chrono::duration_cast( std::chrono::steady_clock::now() - start) .count(); if (WIFSIGNALED(status)) { result.signal = WTERMSIG(status); result.error = "cli probe terminated by signal " + std::to_string(result.signal); } else if (WIFEXITED(status)) { result.exitCode = WEXITSTATUS(status); result.success = (result.exitCode == 0); } else { result.error = "child terminated abnormally"; } std::string trace; std::array buffer{}; for (;;) { ssize_t n = ::read(tracePipe[0], buffer.data(), buffer.size()); if (n > 0) { trace.append(buffer.data(), static_cast(n)); continue; } break; } ::close(tracePipe[0]); std::stringstream traceStream(trace); std::string line; std::vector stderrLines; while (std::getline(traceStream, line)) { constexpr std::string_view kPrefix = "[yams-cli-perf] stage="; auto prefixPos = line.find(kPrefix); if (prefixPos == std::string::npos) { if (!line.empty()) { stderrLines.push_back(line); } continue; } auto stageStart = prefixPos + kPrefix.size(); auto elapsedPos = line.find(" elapsed_us=", stageStart); if (elapsedPos == std::string::npos) { continue; } auto stage = line.substr(stageStart, elapsedPos - stageStart); auto valueStart = elapsedPos + std::string_view(" elapsed_us=").size(); auto valueEnd = line.find(' ', valueStart); auto valueText = line.substr( valueStart, valueEnd == std::string::npos ? std::string::npos : valueEnd - valueStart); try { result.stageElapsedUs[stage] = std::stoll(valueText); } catch (...) { continue; } } if (!stderrLines.empty()) { std::ostringstream stderrBuf; for (size_t i = 0; i < stderrLines.size(); ++i) { if (i > 0) { stderrBuf << '\n'; } stderrBuf << stderrLines[i]; } result.stderrText = stderrBuf.str(); } if (!result.success && result.error.empty()) { if (!result.stderrText.empty()) { result.error = result.stderrText; } else if (result.exitCode >= 0) { result.error = "cli probe exited with code " + std::to_string(result.exitCode); } } return result; } #endif // ───────────────────────────────────────────────────────────────────────────── // Daemon status snapshot helper // ───────────────────────────────────────────────────────────────────────────── struct FdCategoryCounts { uint64_t total{0}; uint64_t socket{0}; uint64_t pipe{0}; uint64_t file{0}; uint64_t device{0}; uint64_t kqueue{0}; uint64_t other{0}; }; FdCategoryCounts countOpenFileDescriptorCategories() { FdCategoryCounts counts; #ifdef _WIN32 return counts; #else DIR* dir = opendir("/dev/fd"); if (!dir) { return counts; } while (auto* entry = readdir(dir)) { const std::string_view name(entry->d_name); if (name != "." && name != "..") { ++counts.total; char* end = nullptr; const auto fdNumber = std::strtol(entry->d_name, &end, 10); if (end && *end == '\0' && fdNumber >= 0) { struct stat st{}; if (::fstat(static_cast(fdNumber), &st) == 0) { if (S_ISSOCK(st.st_mode)) { ++counts.socket; } else if (S_ISFIFO(st.st_mode)) { ++counts.pipe; } else if (S_ISREG(st.st_mode) || S_ISDIR(st.st_mode)) { ++counts.file; } else if (S_ISCHR(st.st_mode) || S_ISBLK(st.st_mode)) { ++counts.device; } else { ++counts.other; } continue; } } const std::string fdPath = std::string{"/dev/fd/"} + std::string{name}; std::array target{}; const auto len = ::readlink(fdPath.c_str(), target.data(), PATH_MAX); if (len < 0) { ++counts.other; continue; } target[static_cast(len)] = '\0'; const std::string_view t(target.data(), static_cast(len)); if (t.find("socket") != std::string_view::npos) { ++counts.socket; } else if (t.find("pipe") != std::string_view::npos) { ++counts.pipe; } else if (t.find("kqueue") != std::string_view::npos) { ++counts.kqueue; } else if (t.starts_with("/dev/")) { ++counts.device; } else if (t.starts_with("/") || t.starts_with(".")) { ++counts.file; } else { ++counts.other; } } } closedir(dir); return counts; #endif } constexpr uint64_t backgroundWorkerActive(uint64_t observedWorkerActive) { // StatusRequest is itself tracked as an active worker before its snapshot is captured. return observedWorkerActive > 0 ? observedWorkerActive - 1 : 0; } struct DaemonSnapshot { size_t activeConnections{0}; size_t maxConnections{0}; size_t connectionSlotsFree{0}; uint64_t oldestConnectionAge{0}; uint64_t forcedCloseCount{0}; uint64_t requestsProcessed{0}; double memoryUsageMb{0.0}; double cpuUsagePercent{0.0}; uint64_t rssBytes{0}; uint64_t postIngestQueued{0}; uint64_t postIngestInflight{0}; uint64_t workerActive{0}; uint64_t workerQueued{0}; uint64_t deferredQueueDepth{0}; int pressureLevel{0}; uint64_t documentsTotal{0}; uint64_t searchActive{0}; uint64_t searchQueued{0}; uint64_t dbWritePoolWaiting{0}; uint64_t dbReadPoolWaiting{0}; uint64_t dbWritePoolMaxObservedWaiting{0}; uint64_t dbReadPoolMaxObservedWaiting{0}; uint64_t dbWritePoolTotalWaitMicros{0}; uint64_t dbReadPoolTotalWaitMicros{0}; uint64_t dbWritePoolTimeoutCount{0}; uint64_t dbReadPoolTimeoutCount{0}; uint64_t dbWritePoolFailedAcquisitions{0}; uint64_t dbReadPoolFailedAcquisitions{0}; uint64_t dbWritePoolSlowHolders{0}; uint64_t dbReadPoolSlowHolders{0}; uint64_t dbWritePoolMaxHolderMicros{0}; uint64_t dbReadPoolMaxHolderMicros{0}; uint64_t muxWriterBudgetBytes{0}; uint64_t writeQueueDepth{0}; uint64_t writeQueueCapacity{0}; uint64_t writeQueueDepthMax{0}; uint64_t writeQueueCapacityRejections{0}; uint64_t writeQueueForcedOverCapacity{0}; uint64_t writeInFlight{0}; uint64_t writeMaxBatchApplyMs{0}; uint64_t writeMaxBatchQueueWaitMs{0}; uint64_t writeMaxBatchExcessQueueWaitMs{0}; uint64_t processFdCount{0}; uint64_t processFdSockets{0}; uint64_t processFdPipes{0}; uint64_t processFdFiles{0}; uint64_t processFdDevices{0}; uint64_t processFdKqueues{0}; uint64_t processFdOther{0}; uint64_t postCommitContentIndexCalls{0}; uint64_t postCommitContentIndexTotalMs{0}; uint64_t postCommitContentIndexMaxMs{0}; uint64_t addDispatchSamples{0}; uint64_t addDispatchTotalUs{0}; uint64_t addDispatchMaxUs{0}; uint64_t addFingerprintTotalUs{0}; uint64_t addFingerprintMaxUs{0}; uint64_t addEnqueueTotalUs{0}; uint64_t addEnqueueMaxUs{0}; uint64_t clientPoolOpenSockets{0}; uint64_t clientPoolAliveConnections{0}; uint64_t clientPoolInUseConnections{0}; std::string writeHotSources; uint32_t ipcPoolSize{0}; uint32_t ioPoolSize{0}; uint32_t retryAfterMs{0}; uint64_t metadataWalBytes{0}; static DaemonSnapshot capture(DaemonClient& client, const fs::path& dataDir = {}) { DaemonSnapshot snap; if (!dataDir.empty()) { std::error_code ec; const auto walPath = dataDir / "yams.db-wal"; if (fs::exists(walPath, ec) && !ec) { snap.metadataWalBytes = static_cast(fs::file_size(walPath, ec)); } } auto st = yams::cli::run_sync(client.status(), 5s); if (!st) return snap; const auto fdCounts = countOpenFileDescriptorCategories(); snap.processFdCount = fdCounts.total; snap.processFdSockets = fdCounts.socket; snap.processFdPipes = fdCounts.pipe; snap.processFdFiles = fdCounts.file; snap.processFdDevices = fdCounts.device; snap.processFdKqueues = fdCounts.kqueue; snap.processFdOther = fdCounts.other; const auto poolStats = AsioConnectionPool::registry_stats(); snap.clientPoolOpenSockets = poolStats.openSockets; snap.clientPoolAliveConnections = poolStats.aliveConnections; snap.clientPoolInUseConnections = poolStats.inUseConnections; const auto& s = st.value(); snap.activeConnections = s.activeConnections; snap.maxConnections = s.maxConnections; snap.connectionSlotsFree = s.connectionSlotsFree; snap.oldestConnectionAge = s.oldestConnectionAge; snap.forcedCloseCount = s.forcedCloseCount; snap.requestsProcessed = s.requestsProcessed; snap.memoryUsageMb = s.memoryUsageMb; snap.cpuUsagePercent = s.cpuUsagePercent; snap.muxWriterBudgetBytes = s.muxWriterBudgetBytes; snap.ipcPoolSize = s.ipcPoolSize; snap.ioPoolSize = s.ioPoolSize; snap.retryAfterMs = s.retryAfterMs; auto getCount = [&](std::string_view key) -> uint64_t { auto it = s.requestCounts.find(std::string(key)); return (it != s.requestCounts.end()) ? it->second : 0; }; if (snap.maxConnections == 0) { snap.maxConnections = static_cast(getCount("status_max_connections")); } if (snap.connectionSlotsFree == 0) { snap.connectionSlotsFree = static_cast(getCount("status_connection_slots_free")); } if (snap.oldestConnectionAge == 0) { snap.oldestConnectionAge = getCount("status_oldest_connection_age_s"); } if (snap.forcedCloseCount == 0) { snap.forcedCloseCount = getCount("status_forced_close_count"); } if (snap.muxWriterBudgetBytes == 0) { snap.muxWriterBudgetBytes = getCount("status_mux_writer_budget_bytes"); } if (snap.ipcPoolSize == 0) { snap.ipcPoolSize = static_cast(getCount("status_ipc_pool_size")); } if (snap.ioPoolSize == 0) { snap.ioPoolSize = static_cast(getCount("status_io_pool_size")); } if (snap.retryAfterMs == 0) { snap.retryAfterMs = static_cast(getCount("status_retry_after_ms")); } snap.rssBytes = getCount("status_rss_bytes"); snap.postIngestQueued = getCount(metrics::kPostIngestQueued); snap.postIngestInflight = getCount(metrics::kPostIngestInflight); snap.workerActive = backgroundWorkerActive(getCount(metrics::kWorkerActive)); snap.workerQueued = getCount(metrics::kWorkerQueued); snap.deferredQueueDepth = getCount(metrics::kDeferredQueueDepth); snap.pressureLevel = static_cast(getCount(metrics::kPressureLevel)); snap.documentsTotal = getCount(metrics::kDocumentsTotal); snap.searchActive = getCount(metrics::kSearchActive); snap.searchQueued = getCount(metrics::kSearchQueued); snap.dbWritePoolWaiting = getCount(metrics::kDbWritePoolWaitingRequests); snap.dbReadPoolWaiting = getCount(metrics::kDbReadPoolWaitingRequests); snap.addDispatchSamples = getCount("add_phase_dispatch_samples"); snap.addDispatchTotalUs = getCount("add_phase_dispatch_total_us"); snap.addDispatchMaxUs = getCount("add_phase_dispatch_max_us"); snap.addFingerprintTotalUs = getCount("add_phase_fingerprint_total_us"); snap.addFingerprintMaxUs = getCount("add_phase_fingerprint_max_us"); snap.addEnqueueTotalUs = getCount("add_phase_enqueue_total_us"); snap.addEnqueueMaxUs = getCount("add_phase_enqueue_max_us"); GetStatsRequest statsReq; auto statsRes = yams::cli::run_sync(client.getStats(statsReq), 3s); const auto* additionalStats = statsRes ? &statsRes.value().additionalStats : nullptr; auto getAdditionalUint = [&](std::string_view key) -> uint64_t { if (!additionalStats) return 0; auto it = additionalStats->find(std::string(key)); if (it == additionalStats->end() || it->second.empty()) return 0; try { return static_cast(std::stoull(it->second)); } catch (...) { return 0; } }; snap.writeMaxBatchApplyMs = getAdditionalUint("write_max_batch_apply_ms"); snap.writeQueueDepth = getAdditionalUint("write_queue_depth"); snap.writeInFlight = getAdditionalUint("write_in_flight"); snap.writeQueueCapacity = getAdditionalUint("write_queue_capacity"); snap.writeQueueDepthMax = getAdditionalUint("write_queue_depth_max"); snap.writeQueueCapacityRejections = getAdditionalUint("write_queue_capacity_rejections"); snap.writeQueueForcedOverCapacity = getAdditionalUint("write_queue_forced_over_capacity"); snap.dbWritePoolMaxObservedWaiting = getAdditionalUint(metrics::kDbWritePoolMaxObservedWaiting); snap.dbReadPoolMaxObservedWaiting = getAdditionalUint(metrics::kDbReadPoolMaxObservedWaiting); snap.dbWritePoolTotalWaitMicros = getAdditionalUint(metrics::kDbWritePoolTotalWaitMicros); snap.dbReadPoolTotalWaitMicros = getAdditionalUint(metrics::kDbReadPoolTotalWaitMicros); snap.dbWritePoolTimeoutCount = getAdditionalUint(metrics::kDbWritePoolTimeoutCount); snap.dbReadPoolTimeoutCount = getAdditionalUint(metrics::kDbReadPoolTimeoutCount); snap.dbWritePoolFailedAcquisitions = getAdditionalUint(metrics::kDbWritePoolFailedAcquisitions); snap.dbReadPoolFailedAcquisitions = getAdditionalUint(metrics::kDbReadPoolFailedAcquisitions); snap.dbWritePoolSlowHolders = getAdditionalUint(metrics::kDbWritePoolSlowHolders); snap.dbReadPoolSlowHolders = getAdditionalUint(metrics::kDbReadPoolSlowHolders); snap.dbWritePoolMaxHolderMicros = getAdditionalUint(metrics::kDbWritePoolMaxHolderMicros); snap.dbReadPoolMaxHolderMicros = getAdditionalUint(metrics::kDbReadPoolMaxHolderMicros); snap.writeMaxBatchQueueWaitMs = getAdditionalUint("write_max_batch_queue_wait_ms"); snap.writeMaxBatchExcessQueueWaitMs = getAdditionalUint("write_max_batch_excess_queue_wait_ms"); snap.postCommitContentIndexCalls = getAdditionalUint("post_timing_commit_content_index_calls"); snap.postCommitContentIndexTotalMs = getAdditionalUint("post_timing_commit_content_index_total_ms"); snap.postCommitContentIndexMaxMs = getAdditionalUint("post_timing_commit_content_index_max_ms"); if (additionalStats) { if (auto it = additionalStats->find("write_hot_sources"); it != additionalStats->end()) { snap.writeHotSources = it->second; } } return snap; } json toJson() const { return json{{"active_connections", activeConnections}, {"max_connections", maxConnections}, {"connection_slots_free", connectionSlotsFree}, {"oldest_connection_age_s", oldestConnectionAge}, {"forced_close_count", forcedCloseCount}, {"requests_processed", requestsProcessed}, {"memory_mb", memoryUsageMb}, {"cpu_pct", cpuUsagePercent}, {"rss_bytes", rssBytes}, {"post_ingest_queued", postIngestQueued}, {"post_ingest_inflight", postIngestInflight}, {"worker_active", workerActive}, {"worker_queued", workerQueued}, {"deferred_queue_depth", deferredQueueDepth}, {"pressure_level", pressureLevel}, {"documents_total", documentsTotal}, {"search_active", searchActive}, {"search_queued", searchQueued}, {"db_write_pool_waiting", dbWritePoolWaiting}, {"db_read_pool_waiting", dbReadPoolWaiting}, {"db_write_pool_max_observed_waiting", dbWritePoolMaxObservedWaiting}, {"db_read_pool_max_observed_waiting", dbReadPoolMaxObservedWaiting}, {"db_write_pool_total_wait_us", dbWritePoolTotalWaitMicros}, {"db_read_pool_total_wait_us", dbReadPoolTotalWaitMicros}, {"db_write_pool_timeouts", dbWritePoolTimeoutCount}, {"db_read_pool_timeouts", dbReadPoolTimeoutCount}, {"db_write_pool_failed_acquisitions", dbWritePoolFailedAcquisitions}, {"db_read_pool_failed_acquisitions", dbReadPoolFailedAcquisitions}, {"db_write_pool_slow_holders", dbWritePoolSlowHolders}, {"db_read_pool_slow_holders", dbReadPoolSlowHolders}, {"db_write_pool_max_holder_us", dbWritePoolMaxHolderMicros}, {"db_read_pool_max_holder_us", dbReadPoolMaxHolderMicros}, {"write_queue_depth", writeQueueDepth}, {"write_queue_capacity", writeQueueCapacity}, {"write_queue_depth_max", writeQueueDepthMax}, {"write_queue_capacity_rejections", writeQueueCapacityRejections}, {"write_queue_forced_over_capacity", writeQueueForcedOverCapacity}, {"write_in_flight", writeInFlight}, {"write_max_batch_apply_ms", writeMaxBatchApplyMs}, {"write_max_batch_queue_wait_ms", writeMaxBatchQueueWaitMs}, {"write_max_batch_excess_queue_wait_ms", writeMaxBatchExcessQueueWaitMs}, {"process_fd_count", processFdCount}, {"process_fd_sockets", processFdSockets}, {"process_fd_pipes", processFdPipes}, {"process_fd_files", processFdFiles}, {"process_fd_devices", processFdDevices}, {"process_fd_kqueues", processFdKqueues}, {"process_fd_other", processFdOther}, {"post_commit_content_index_calls", postCommitContentIndexCalls}, {"post_commit_content_index_total_ms", postCommitContentIndexTotalMs}, {"post_commit_content_index_max_ms", postCommitContentIndexMaxMs}, {"add_dispatch_samples", addDispatchSamples}, {"add_dispatch_total_us", addDispatchTotalUs}, {"add_dispatch_max_us", addDispatchMaxUs}, {"add_fingerprint_total_us", addFingerprintTotalUs}, {"add_fingerprint_max_us", addFingerprintMaxUs}, {"add_enqueue_total_us", addEnqueueTotalUs}, {"add_enqueue_max_us", addEnqueueMaxUs}, {"client_pool_open_sockets", clientPoolOpenSockets}, {"client_pool_alive_connections", clientPoolAliveConnections}, {"client_pool_in_use_connections", clientPoolInUseConnections}, {"write_hot_sources", writeHotSources}, {"mux_writer_budget_bytes", muxWriterBudgetBytes}, {"ipc_pool_size", ipcPoolSize}, {"io_pool_size", ioPoolSize}, {"retry_after_ms", retryAfterMs}, {"metadata_wal_bytes", metadataWalBytes}}; } }; ProbeDaemonState captureProbeDaemonState(const fs::path& socketPath, std::chrono::milliseconds timeout = 3s) { ProbeDaemonState state; ClientConfig cfg; cfg.socketPath = socketPath; cfg.autoStart = false; cfg.singleUseConnections = true; cfg.requestTimeout = timeout; DaemonClient client(cfg); auto st = yams::cli::run_sync(client.status(), timeout); if (!st) { state.error = st.error().message; return state; } state.captured = true; state.activeConnections = st.value().activeConnections; state.maxConnections = st.value().maxConnections; state.connectionSlotsFree = st.value().connectionSlotsFree; state.pressureLevel = 0; auto getCount = [&](std::string_view key) -> uint64_t { auto it = st.value().requestCounts.find(std::string(key)); return (it != st.value().requestCounts.end()) ? it->second : 0ULL; }; if (state.maxConnections == 0) { state.maxConnections = static_cast(getCount("status_max_connections")); } if (state.connectionSlotsFree == 0) { state.connectionSlotsFree = static_cast(getCount("status_connection_slots_free")); } auto it = st.value().requestCounts.find(std::string(metrics::kPressureLevel)); if (it != st.value().requestCounts.end()) { state.pressureLevel = static_cast(it->second); } state.retryAfterMs = st.value().retryAfterMs; if (state.retryAfterMs == 0) { state.retryAfterMs = static_cast(getCount("status_retry_after_ms")); } return state; } std::string classifyFailureMessage(const std::string& errorMsg); std::string classifyCliProbeBucket(const CliProbeResult& probe) { if (probe.success) { return "success"; } if (probe.timedOut) { return "timeout"; } if (probe.signal > 0) { return "signal"; } if (probe.exitCode == 127) { return "exec_failed"; } return classifyFailureMessage(!probe.stderrText.empty() ? probe.stderrText : probe.error); } std::string classifyCliProbeMode(const CliProbeResult& probe) { if (probe.success) { return "success"; } const bool underStress = probe.beforeState.underStress() || probe.afterState.underStress(); const std::string bucket = probe.failureBucket.empty() ? classifyCliProbeBucket(probe) : probe.failureBucket; if (bucket == "exec_failed") { return "benchmark_wiring"; } if (bucket == "signal") { return "unexpected_signal"; } if (bucket == "protocol") { return "protocol_or_cli_bug"; } if (bucket == "connect_refused") { return underStress ? "stress_expected" : "ambient_daemon_down"; } if (bucket == "backpressure") { return underStress ? "overload_under_load" : "ambient_failure"; } if (bucket == "connection_drop") { return underStress ? "stress_expected" : "ambient_failure"; } if (bucket == "timeout") { return underStress ? "timeout_under_load" : "ambient_failure"; } return underStress ? "stress_expected" : "unexpected_cli_failure"; } struct RecoveryCheck { bool statusOk{false}; bool listOk{false}; std::string statusError; std::string listError; DaemonSnapshot snapshot; bool ok() const { return statusOk && listOk; } json toJson() const { return json{{"status_ok", statusOk}, {"list_ok", listOk}, {"status_error", statusError}, {"list_error", listError}, {"snapshot", snapshot.toJson()}}; } }; RecoveryCheck runRecoveryCheck(const fs::path& socketPath) { RecoveryCheck check; ClientConfig cfg; cfg.socketPath = socketPath; cfg.autoStart = false; cfg.singleUseConnections = true; cfg.requestTimeout = 10s; DaemonClient client(cfg); auto statusRes = yams::cli::run_sync(client.status(), 10s); if (statusRes) { check.statusOk = true; check.snapshot = DaemonSnapshot::capture(client); } else { check.statusError = statusRes.error().message; } ListRequest listReq; listReq.limit = 10; auto listRes = yams::cli::run_sync(client.list(listReq), 10s); if (listRes) { check.listOk = true; } else { check.listError = listRes.error().message; } return check; } void printRecoverySummary(const RecoveryCheck& recovery) { std::cout << " Recovery: status=" << (recovery.statusOk ? "ok" : "FAIL") << " list=" << (recovery.listOk ? "ok" : "FAIL") << "\n"; if (!recovery.statusOk && !recovery.statusError.empty()) { std::cout << " status error: " << recovery.statusError << "\n"; } if (!recovery.listOk && !recovery.listError.empty()) { std::cout << " list error: " << recovery.listError << "\n"; } } CliProbeResult runObservedCliProbe(const std::string& opType, const fs::path& cliBinary, const fs::path& socketPath, const std::vector& args, std::chrono::milliseconds timeout = 10s) { auto beforeState = captureProbeDaemonState(socketPath, std::min(timeout, 1500ms)); auto probe = runCliProbe(cliBinary, socketPath, args, timeout); probe.opType = opType; probe.beforeState = beforeState; probe.afterState = captureProbeDaemonState(socketPath, std::min(timeout, 1500ms)); probe.failureBucket = classifyCliProbeBucket(probe); probe.failureMode = classifyCliProbeMode(probe); return probe; } void printCliProbeFailureSummary(const std::map& failureModes, const std::map& failureSamples) { if (failureModes.empty()) { return; } std::cout << " CLI failure modes:"; for (const auto& [mode, count] : failureModes) { std::cout << " " << mode << "=" << count; } std::cout << "\n"; for (const auto& [mode, sample] : failureSamples) { std::cout << " sample[" << mode << "]: " << sample << "\n"; } } // ───────────────────────────────────────────────────────────────────────────── // Time-series sampler (runs in background during benchmark) // ───────────────────────────────────────────────────────────────────────────── class TimeSeriesSampler { public: struct Sample { uint64_t elapsedMs{0}; DaemonSnapshot snap; }; ~TimeSeriesSampler() { stop(); } void start(DaemonClient& client, std::chrono::milliseconds interval = 500ms, fs::path dataDir = {}) { stop(); stop_ = false; startTime_ = std::chrono::steady_clock::now(); { std::lock_guard lock(mutex_); samples_.clear(); } thread_ = std::thread([this, &client, interval, dataDir = std::move(dataDir)]() { while (!stop_) { Sample sample; sample.elapsedMs = std::chrono::duration_cast( std::chrono::steady_clock::now() - startTime_) .count(); sample.snap = DaemonSnapshot::capture(client, dataDir); { std::lock_guard lock(mutex_); samples_.push_back(sample); } std::this_thread::sleep_for(interval); } }); } void stop() { stop_ = true; if (thread_.joinable()) thread_.join(); } std::vector getSamples() const { std::lock_guard lock(mutex_); return samples_; } private: std::atomic stop_{false}; std::chrono::steady_clock::time_point startTime_; std::thread thread_; mutable std::mutex mutex_; std::vector samples_; }; struct SlotScalingSummary { size_t initialLimit{0}; size_t peakLimit{0}; size_t finalLimit{0}; size_t minFreeSlots{0}; size_t peakActiveConnections{0}; double peakUtilization{0.0}; int growEvents{0}; int shrinkEvents{0}; int retryAfterSamples{0}; }; SlotScalingSummary summarizeSlotScaling(const std::vector& samples) { SlotScalingSummary summary; if (samples.empty()) { return summary; } summary.initialLimit = samples.front().snap.maxConnections; summary.peakLimit = samples.front().snap.maxConnections; summary.finalLimit = samples.back().snap.maxConnections; summary.minFreeSlots = samples.front().snap.connectionSlotsFree; size_t previousLimit = samples.front().snap.maxConnections; for (const auto& sample : samples) { const auto& snap = sample.snap; summary.peakLimit = std::max(summary.peakLimit, snap.maxConnections); summary.minFreeSlots = std::min(summary.minFreeSlots, snap.connectionSlotsFree); summary.peakActiveConnections = std::max(summary.peakActiveConnections, snap.activeConnections); if (snap.maxConnections > 0) { summary.peakUtilization = std::max(summary.peakUtilization, static_cast(snap.activeConnections) / static_cast(snap.maxConnections)); } if (snap.retryAfterMs > 0) { summary.retryAfterSamples++; } if (snap.maxConnections > previousLimit) { summary.growEvents++; } else if (snap.maxConnections < previousLimit) { summary.shrinkEvents++; } previousLimit = snap.maxConnections; } return summary; } struct TimedLatencySample { std::string opType; uint64_t wallClockMs{0}; int64_t latencyUs{0}; bool success{false}; }; std::optional parseInt64(std::string_view value) { try { return static_cast(std::stoll(std::string(value))); } catch (...) { return std::nullopt; } } std::optional metricInt(const std::map& stats, std::string_view key) { const auto it = stats.find(std::string(key)); if (it == stats.end()) { return std::nullopt; } return parseInt64(it->second); } void collectSearchHydrationTelemetry(const SearchResponse& response, ClientResult& result) { const auto appendTiming = [&](std::string_view key, std::vector& samples) { if (auto value = metricInt(response.searchStats, key)) { samples.push_back(*value); } }; const auto addCounter = [&](std::string_view key, int64_t& counter) { if (auto value = metricInt(response.searchStats, key)) { counter = checkedMetricAdd(counter, *value); } }; appendTiming("timing_snippet_hydration_us", result.snippetHydrationUs); appendTiming("timing_snippet_lookup_us", result.snippetLookupUs); appendTiming("timing_snippet_content_fetch_us", result.snippetContentFetchUs); appendTiming("timing_snippet_render_us", result.snippetRenderUs); addCounter("snippet_hydration_candidates", result.snippetHydrationCandidates); addCounter("snippet_hydration_missing", result.snippetHydrationMissing); addCounter("snippet_hydrated_results", result.snippetHydratedResults); const auto timeout = response.searchStats.find("snippet_timeout_hit"); if (timeout != response.searchStats.end() && timeout->second == "true") { ++result.snippetTimeoutHits; } } void appendStringTimingStats(std::map>& dest, std::string_view op, const std::map& stats, int64_t wallUs) { std::optional dispatchTotalUs; for (const auto& [key, value] : stats) { if (!key.ends_with("_us")) { continue; } auto parsed = parseInt64(value); if (!parsed) { continue; } dest[std::string(op) + ":" + key].push_back(*parsed); if (key == "phase_dispatch_total_us") { dispatchTotalUs = *parsed; } } if (dispatchTotalUs) { dest[std::string(op) + ":client_transport_overhead_us"].push_back( std::max(0, wallUs - *dispatchTotalUs)); } } void appendStatusTimingStats(std::map>& dest, const std::map& stats, int64_t wallUs) { auto it = stats.find("phase_dispatch_total_us"); if (it == stats.end()) { return; } const auto dispatchTotalUs = static_cast(it->second); dest["status:phase_dispatch_total_us"].push_back(dispatchTotalUs); dest["status:client_transport_overhead_us"].push_back( std::max(0, wallUs - dispatchTotalUs)); } json summarizeTimingStats(std::map> stats) { json out = json::object(); for (auto& [key, values] : stats) { out[key] = PercentileStats::compute(values).toJson(); } return out; } struct ResourcePeaks { double peakMemoryMb{0.0}; double peakCpuPercent{0.0}; uint64_t peakRssBytes{0}; size_t peakActiveConnections{0}; size_t peakMaxConnections{0}; uint64_t peakPostIngestQueued{0}; uint64_t peakPostIngestInflight{0}; uint64_t peakSearchActive{0}; uint64_t peakSearchQueued{0}; uint64_t peakDbWritePoolWaiting{0}; uint64_t peakDbReadPoolWaiting{0}; uint64_t peakDbWritePoolMaxObservedWaiting{0}; uint64_t peakDbReadPoolMaxObservedWaiting{0}; uint64_t peakDbWritePoolTotalWaitMicros{0}; uint64_t peakDbReadPoolTotalWaitMicros{0}; uint64_t peakDbWritePoolTimeoutCount{0}; uint64_t peakDbReadPoolTimeoutCount{0}; uint64_t peakDbWritePoolFailedAcquisitions{0}; uint64_t peakDbReadPoolFailedAcquisitions{0}; uint64_t peakDbWritePoolSlowHolders{0}; uint64_t peakDbReadPoolSlowHolders{0}; uint64_t peakDbWritePoolMaxHolderMicros{0}; uint64_t peakDbReadPoolMaxHolderMicros{0}; uint64_t peakMuxWriterBudgetBytes{0}; uint64_t peakWriteQueueDepth{0}; uint64_t writeQueueCapacity{0}; uint64_t peakWriteQueueDepthMax{0}; uint64_t peakWriteQueueCapacityRejections{0}; uint64_t peakWriteQueueForcedOverCapacity{0}; uint64_t peakWriteInFlight{0}; uint64_t peakWriteMaxBatchApplyMs{0}; uint64_t peakWriteMaxBatchQueueWaitMs{0}; uint64_t peakWriteMaxBatchExcessQueueWaitMs{0}; uint64_t peakProcessFdCount{0}; uint64_t peakProcessFdSockets{0}; uint64_t peakProcessFdPipes{0}; uint64_t peakProcessFdFiles{0}; uint64_t peakProcessFdDevices{0}; uint64_t peakProcessFdKqueues{0}; uint64_t peakProcessFdOther{0}; uint64_t peakPostCommitContentIndexCalls{0}; uint64_t peakPostCommitContentIndexTotalMs{0}; uint64_t peakPostCommitContentIndexMaxMs{0}; uint64_t peakClientPoolOpenSockets{0}; uint64_t peakClientPoolAliveConnections{0}; uint64_t peakClientPoolInUseConnections{0}; uint64_t peakMetadataWalBytes{0}; std::string lastWriteHotSources; uint32_t maxPressureLevel{0}; json toJson() const { return json{ {"peak_memory_mb", peakMemoryMb}, {"peak_cpu_pct", peakCpuPercent}, {"peak_rss_bytes", peakRssBytes}, {"peak_active_connections", peakActiveConnections}, {"peak_max_connections", peakMaxConnections}, {"peak_post_ingest_queued", peakPostIngestQueued}, {"peak_post_ingest_inflight", peakPostIngestInflight}, {"peak_search_active", peakSearchActive}, {"peak_search_queued", peakSearchQueued}, {"peak_db_write_pool_waiting", peakDbWritePoolWaiting}, {"peak_db_read_pool_waiting", peakDbReadPoolWaiting}, {"peak_db_write_pool_max_observed_waiting", peakDbWritePoolMaxObservedWaiting}, {"peak_db_read_pool_max_observed_waiting", peakDbReadPoolMaxObservedWaiting}, {"peak_db_write_pool_total_wait_us", peakDbWritePoolTotalWaitMicros}, {"peak_db_read_pool_total_wait_us", peakDbReadPoolTotalWaitMicros}, {"peak_db_write_pool_timeouts", peakDbWritePoolTimeoutCount}, {"peak_db_read_pool_timeouts", peakDbReadPoolTimeoutCount}, {"peak_db_write_pool_failed_acquisitions", peakDbWritePoolFailedAcquisitions}, {"peak_db_read_pool_failed_acquisitions", peakDbReadPoolFailedAcquisitions}, {"peak_db_write_pool_slow_holders", peakDbWritePoolSlowHolders}, {"peak_db_read_pool_slow_holders", peakDbReadPoolSlowHolders}, {"peak_db_write_pool_max_holder_us", peakDbWritePoolMaxHolderMicros}, {"peak_db_read_pool_max_holder_us", peakDbReadPoolMaxHolderMicros}, {"peak_mux_writer_budget_bytes", peakMuxWriterBudgetBytes}, {"peak_write_queue_depth", peakWriteQueueDepth}, {"write_queue_capacity", writeQueueCapacity}, {"peak_write_queue_depth_max", peakWriteQueueDepthMax}, {"peak_write_queue_capacity_rejections", peakWriteQueueCapacityRejections}, {"peak_write_queue_forced_over_capacity", peakWriteQueueForcedOverCapacity}, {"peak_write_in_flight", peakWriteInFlight}, {"peak_write_max_batch_apply_ms", peakWriteMaxBatchApplyMs}, {"peak_write_max_batch_queue_wait_ms", peakWriteMaxBatchQueueWaitMs}, {"peak_write_max_batch_excess_queue_wait_ms", peakWriteMaxBatchExcessQueueWaitMs}, {"peak_process_fd_count", peakProcessFdCount}, {"peak_process_fd_sockets", peakProcessFdSockets}, {"peak_process_fd_pipes", peakProcessFdPipes}, {"peak_process_fd_files", peakProcessFdFiles}, {"peak_process_fd_devices", peakProcessFdDevices}, {"peak_process_fd_kqueues", peakProcessFdKqueues}, {"peak_process_fd_other", peakProcessFdOther}, {"peak_post_commit_content_index_calls", peakPostCommitContentIndexCalls}, {"peak_post_commit_content_index_total_ms", peakPostCommitContentIndexTotalMs}, {"peak_post_commit_content_index_max_ms", peakPostCommitContentIndexMaxMs}, {"peak_client_pool_open_sockets", peakClientPoolOpenSockets}, {"peak_client_pool_alive_connections", peakClientPoolAliveConnections}, {"peak_client_pool_in_use_connections", peakClientPoolInUseConnections}, {"peak_metadata_wal_bytes", peakMetadataWalBytes}, {"last_write_hot_sources", lastWriteHotSources}, {"max_pressure_level", maxPressureLevel}}; } }; ResourcePeaks summarizeResourcePeaks(const std::vector& samples) { ResourcePeaks peaks; for (const auto& sample : samples) { const auto& snap = sample.snap; peaks.peakMemoryMb = std::max(peaks.peakMemoryMb, snap.memoryUsageMb); peaks.peakCpuPercent = std::max(peaks.peakCpuPercent, snap.cpuUsagePercent); peaks.peakRssBytes = std::max(peaks.peakRssBytes, snap.rssBytes); peaks.peakActiveConnections = std::max(peaks.peakActiveConnections, snap.activeConnections); peaks.peakMaxConnections = std::max(peaks.peakMaxConnections, snap.maxConnections); peaks.peakPostIngestQueued = std::max(peaks.peakPostIngestQueued, snap.postIngestQueued); peaks.peakPostIngestInflight = std::max(peaks.peakPostIngestInflight, snap.postIngestInflight); peaks.peakSearchActive = std::max(peaks.peakSearchActive, snap.searchActive); peaks.peakSearchQueued = std::max(peaks.peakSearchQueued, snap.searchQueued); peaks.peakDbWritePoolWaiting = std::max(peaks.peakDbWritePoolWaiting, snap.dbWritePoolWaiting); peaks.peakDbReadPoolWaiting = std::max(peaks.peakDbReadPoolWaiting, snap.dbReadPoolWaiting); peaks.peakDbWritePoolMaxObservedWaiting = std::max(peaks.peakDbWritePoolMaxObservedWaiting, snap.dbWritePoolMaxObservedWaiting); peaks.peakDbReadPoolMaxObservedWaiting = std::max(peaks.peakDbReadPoolMaxObservedWaiting, snap.dbReadPoolMaxObservedWaiting); peaks.peakDbWritePoolTotalWaitMicros = std::max(peaks.peakDbWritePoolTotalWaitMicros, snap.dbWritePoolTotalWaitMicros); peaks.peakDbReadPoolTotalWaitMicros = std::max(peaks.peakDbReadPoolTotalWaitMicros, snap.dbReadPoolTotalWaitMicros); peaks.peakDbWritePoolTimeoutCount = std::max(peaks.peakDbWritePoolTimeoutCount, snap.dbWritePoolTimeoutCount); peaks.peakDbReadPoolTimeoutCount = std::max(peaks.peakDbReadPoolTimeoutCount, snap.dbReadPoolTimeoutCount); peaks.peakDbWritePoolFailedAcquisitions = std::max(peaks.peakDbWritePoolFailedAcquisitions, snap.dbWritePoolFailedAcquisitions); peaks.peakDbReadPoolFailedAcquisitions = std::max(peaks.peakDbReadPoolFailedAcquisitions, snap.dbReadPoolFailedAcquisitions); peaks.peakDbWritePoolSlowHolders = std::max(peaks.peakDbWritePoolSlowHolders, snap.dbWritePoolSlowHolders); peaks.peakDbReadPoolSlowHolders = std::max(peaks.peakDbReadPoolSlowHolders, snap.dbReadPoolSlowHolders); peaks.peakDbWritePoolMaxHolderMicros = std::max(peaks.peakDbWritePoolMaxHolderMicros, snap.dbWritePoolMaxHolderMicros); peaks.peakDbReadPoolMaxHolderMicros = std::max(peaks.peakDbReadPoolMaxHolderMicros, snap.dbReadPoolMaxHolderMicros); peaks.peakMuxWriterBudgetBytes = std::max(peaks.peakMuxWriterBudgetBytes, snap.muxWriterBudgetBytes); peaks.peakWriteQueueDepth = std::max(peaks.peakWriteQueueDepth, snap.writeQueueDepth); peaks.writeQueueCapacity = std::max(peaks.writeQueueCapacity, snap.writeQueueCapacity); peaks.peakWriteQueueDepthMax = std::max(peaks.peakWriteQueueDepthMax, snap.writeQueueDepthMax); peaks.peakWriteQueueCapacityRejections = std::max(peaks.peakWriteQueueCapacityRejections, snap.writeQueueCapacityRejections); peaks.peakWriteQueueForcedOverCapacity = std::max(peaks.peakWriteQueueForcedOverCapacity, snap.writeQueueForcedOverCapacity); peaks.peakWriteInFlight = std::max(peaks.peakWriteInFlight, snap.writeInFlight); peaks.peakWriteMaxBatchApplyMs = std::max(peaks.peakWriteMaxBatchApplyMs, snap.writeMaxBatchApplyMs); peaks.peakWriteMaxBatchQueueWaitMs = std::max(peaks.peakWriteMaxBatchQueueWaitMs, snap.writeMaxBatchQueueWaitMs); peaks.peakWriteMaxBatchExcessQueueWaitMs = std::max(peaks.peakWriteMaxBatchExcessQueueWaitMs, snap.writeMaxBatchExcessQueueWaitMs); peaks.peakProcessFdCount = std::max(peaks.peakProcessFdCount, snap.processFdCount); peaks.peakProcessFdSockets = std::max(peaks.peakProcessFdSockets, snap.processFdSockets); peaks.peakProcessFdPipes = std::max(peaks.peakProcessFdPipes, snap.processFdPipes); peaks.peakProcessFdFiles = std::max(peaks.peakProcessFdFiles, snap.processFdFiles); peaks.peakProcessFdDevices = std::max(peaks.peakProcessFdDevices, snap.processFdDevices); peaks.peakProcessFdKqueues = std::max(peaks.peakProcessFdKqueues, snap.processFdKqueues); peaks.peakProcessFdOther = std::max(peaks.peakProcessFdOther, snap.processFdOther); peaks.peakPostCommitContentIndexCalls = std::max(peaks.peakPostCommitContentIndexCalls, snap.postCommitContentIndexCalls); peaks.peakPostCommitContentIndexTotalMs = std::max(peaks.peakPostCommitContentIndexTotalMs, snap.postCommitContentIndexTotalMs); peaks.peakPostCommitContentIndexMaxMs = std::max(peaks.peakPostCommitContentIndexMaxMs, snap.postCommitContentIndexMaxMs); peaks.peakClientPoolOpenSockets = std::max(peaks.peakClientPoolOpenSockets, snap.clientPoolOpenSockets); peaks.peakClientPoolAliveConnections = std::max(peaks.peakClientPoolAliveConnections, snap.clientPoolAliveConnections); peaks.peakClientPoolInUseConnections = std::max(peaks.peakClientPoolInUseConnections, snap.clientPoolInUseConnections); peaks.peakMetadataWalBytes = std::max(peaks.peakMetadataWalBytes, snap.metadataWalBytes); if (!snap.writeHotSources.empty()) { peaks.lastWriteHotSources = snap.writeHotSources; } peaks.maxPressureLevel = std::max(peaks.maxPressureLevel, static_cast(std::max(0, snap.pressureLevel))); } return peaks; } json summarizeLatencyHotspots(const std::vector& samples, uint64_t windowMs = 500) { json out = json{{"window_ms", windowMs}, {"slowest_ops", json::array()}, {"worst_window", json::object()}, {"per_op_p95_rank", json::array()}}; if (samples.empty()) { return out; } auto sorted = samples; std::sort(sorted.begin(), sorted.end(), [](const auto& a, const auto& b) { return a.latencyUs > b.latencyUs; }); for (size_t i = 0; i < std::min(5, sorted.size()); ++i) { out["slowest_ops"].push_back(json{{"op_type", sorted[i].opType}, {"wall_clock_ms", sorted[i].wallClockMs}, {"latency_us", sorted[i].latencyUs}, {"success", sorted[i].success}}); } std::unordered_map> byOp; for (const auto& sample : samples) { if (sample.success) { pushMetricSample(byOp[sample.opType], sample.latencyUs); } } std::vector> ranked; ranked.reserve(byOp.size()); for (auto& [op, values] : byOp) { ranked.emplace_back(op, PercentileStats::compute(values)); } std::sort(ranked.begin(), ranked.end(), [](const auto& a, const auto& b) { return a.second.p95 > b.second.p95; }); for (const auto& [op, stats] : ranked) { out["per_op_p95_rank"].push_back(json{{"op_type", op}, {"p95_us", stats.p95}, {"mean_us", stats.mean}, {"count", stats.count}}); } size_t left = 0; int64_t currentSumUs = 0; size_t bestCount = 0; int64_t bestSumUs = 0; uint64_t bestStartMs = static_cast( clampUnsignedMetricSample(samples.front().wallClockMs, kMaxBenchmarkWallMs)); uint64_t bestEndMs = bestStartMs; auto byTime = samples; std::sort(byTime.begin(), byTime.end(), [](const auto& a, const auto& b) { return a.wallClockMs < b.wallClockMs; }); const uint64_t boundedWindowMs = std::max(1, std::min(windowMs, kMaxBenchmarkWallMs)); for (size_t right = 0; right < byTime.size(); ++right) { const auto rightLatency = clampMetricSample(byTime[right].latencyUs); currentSumUs = checkedMetricAdd(currentSumUs, rightLatency); auto rightWallMs = static_cast( clampUnsignedMetricSample(byTime[right].wallClockMs, kMaxBenchmarkWallMs)); auto leftWallMs = static_cast( clampUnsignedMetricSample(byTime[left].wallClockMs, kMaxBenchmarkWallMs)); while (rightWallMs >= leftWallMs && rightWallMs - leftWallMs > boundedWindowMs) { currentSumUs = std::max(0, currentSumUs - clampMetricSample(byTime[left].latencyUs)); ++left; leftWallMs = static_cast( clampUnsignedMetricSample(byTime[left].wallClockMs, kMaxBenchmarkWallMs)); } const size_t count = right - left + 1; if (count > 0 && (currentSumUs > bestSumUs || (currentSumUs == bestSumUs && count > bestCount))) { bestSumUs = currentSumUs; bestCount = count; bestStartMs = leftWallMs; bestEndMs = rightWallMs; } } out["worst_window"] = json{ {"start_ms", bestStartMs}, {"end_ms", bestEndMs}, {"sample_count", bestCount}, {"latency_sum_us", bestSumUs}, {"avg_latency_us", bestCount > 0 ? static_cast(bestSumUs) / static_cast(bestCount) : 0.0}}; return out; } class HeldSocketConnections { public: bool open(const fs::path& socketPath, int count) { lastError_.clear(); lastFailedIndex_ = -1; openedCount_ = 0; sockets_.clear(); sockets_.reserve(static_cast(std::max(0, count))); boost::system::error_code ec; for (int i = 0; i < count; ++i) { auto sock = std::make_unique(io_); sock->connect(boost::asio::local::stream_protocol::endpoint(socketPath.string()), ec); if (ec) { lastFailedIndex_ = i; openedCount_ = static_cast(sockets_.size()); lastError_ = ec.message(); closeAll(); return false; } sockets_.push_back(std::move(sock)); openedCount_ = static_cast(sockets_.size()); } return true; } void closeAll() { for (auto& sock : sockets_) { if (!sock) { continue; } boost::system::error_code ec; sock->close(ec); } sockets_.clear(); } ~HeldSocketConnections() { closeAll(); } [[nodiscard]] const std::string& lastError() const noexcept { return lastError_; } [[nodiscard]] int lastFailedIndex() const noexcept { return lastFailedIndex_; } [[nodiscard]] int openedCount() const noexcept { return openedCount_; } private: boost::asio::io_context io_; std::vector> sockets_; std::string lastError_; int lastFailedIndex_{-1}; int openedCount_{0}; }; class ScopedConnectionSlotOverrides { public: ScopedConnectionSlotOverrides(uint32_t minSlots, uint32_t maxSlots, uint32_t scaleStep) { TuneAdvisor::setConnectionSlotsMin(minSlots); TuneAdvisor::setConnectionSlotsMax(maxSlots); TuneAdvisor::setConnectionSlotsScaleStep(scaleStep); } ~ScopedConnectionSlotOverrides() { TuneAdvisor::setConnectionSlotsMin(0); TuneAdvisor::setConnectionSlotsMax(0); TuneAdvisor::setConnectionSlotsScaleStep(0); } ScopedConnectionSlotOverrides(const ScopedConnectionSlotOverrides&) = delete; ScopedConnectionSlotOverrides& operator=(const ScopedConnectionSlotOverrides&) = delete; }; bool waitForSlotLimit(DaemonClient& client, size_t targetLimit, std::chrono::milliseconds timeout, bool requireAtLeast) { const auto deadline = std::chrono::steady_clock::now() + timeout; while (std::chrono::steady_clock::now() < deadline) { const auto snap = DaemonSnapshot::capture(client); if (requireAtLeast) { if (snap.maxConnections >= targetLimit) { return true; } } else if (snap.maxConnections <= targetLimit) { return true; } std::this_thread::sleep_for(250ms); } return false; } bool waitForActiveConnectionsAtMost(DaemonClient& client, size_t targetActive, std::chrono::milliseconds timeout) { const auto deadline = std::chrono::steady_clock::now() + timeout; while (std::chrono::steady_clock::now() < deadline) { const auto snap = DaemonSnapshot::capture(client); if (snap.activeConnections <= targetActive) { return true; } std::this_thread::sleep_for(250ms); } return false; } // ───────────────────────────────────────────────────────────────────────────── // Document generator // ───────────────────────────────────────────────────────────────────────────── std::string generateDocument(int clientId, int docIndex, size_t targetBytes) { std::ostringstream oss; oss << "# Client " << clientId << " Document " << docIndex << "\n\n"; oss << "Created at: " << std::chrono::system_clock::now().time_since_epoch().count() << "\n\n"; // Fill with pseudo-realistic text to reach target size static const char* paragraphs[] = { "The system architecture follows a layered design pattern with clear separation " "of concerns between the daemon core, service layer, and client interfaces.\n\n", "Performance tuning parameters are exposed through TuneAdvisor knobs allowing " "runtime adjustment of thread pool sizes, queue capacities, and concurrency limits.\n\n", "The ingestion pipeline processes documents through extraction, indexing, knowledge " "graph construction, symbol analysis, entity recognition, and embedding stages.\n\n", "Resource governance monitors memory pressure, CPU utilization, and I/O throughput " "to dynamically scale worker thread counts and admission control thresholds.\n\n", "The IPC transport uses Unix domain sockets with a binary framing protocol supporting " "multiplexed request/response streams with backpressure signaling.\n\n", }; constexpr int numParagraphs = 5; int paraIdx = 0; while (oss.tellp() < static_cast(targetBytes)) { oss << paragraphs[paraIdx % numParagraphs]; paraIdx++; } auto result = oss.str(); if (result.size() > targetBytes) result.resize(targetBytes); return result; } using yams::bench::isoTimestamp; std::string toLowerCopy(std::string value) { std::transform(value.begin(), value.end(), value.begin(), [](unsigned char ch) { return static_cast(std::tolower(ch)); }); return value; } bool containsInsensitive(const std::string& haystackLower, std::string_view needle) { return haystackLower.find(std::string(needle)) != std::string::npos; } std::string classifyFailureMessage(const std::string& errorMsg) { if (errorMsg.empty()) { return "unknown"; } const std::string lower = toLowerCopy(errorMsg); if (containsInsensitive(lower, "document content not found")) { return "content_not_found"; } if (containsInsensitive(lower, "document not found") || containsInsensitive(lower, "no documents found matching hash prefix")) { return "document_not_found"; } if (containsInsensitive(lower, "timed out") || containsInsensitive(lower, "timeout")) { return "timeout"; } if (containsInsensitive(lower, "retry after") || containsInsensitive(lower, "retry in ") || containsInsensitive(lower, "server busy") || containsInsensitive(lower, "resource exhausted") || containsInsensitive(lower, "backpressure")) { return "backpressure"; } if (containsInsensitive(lower, "connection refused") || containsInsensitive(lower, "no such file or directory")) { return "connect_refused"; } if (containsInsensitive(lower, "broken pipe") || containsInsensitive(lower, "eof") || containsInsensitive(lower, "connection reset")) { return "connection_drop"; } if (containsInsensitive(lower, "cancel")) { return "cancelled"; } if (containsInsensitive(lower, "database is locked") || containsInsensitive(lower, "sql_busy")) { return "db_locked"; } if (containsInsensitive(lower, "queue") && (containsInsensitive(lower, "full") || containsInsensitive(lower, "overflow"))) { return "queue_full"; } if (containsInsensitive(lower, "parse") || containsInsensitive(lower, "protocol") || containsInsensitive(lower, "invalid response")) { return "protocol"; } return "other"; } bool isDocumentMissingFailure(const std::string& errorMsg) { if (errorMsg.empty()) { return false; } const std::string lower = toLowerCopy(errorMsg); return containsInsensitive(lower, "document not found") || containsInsensitive(lower, "document content not found") || containsInsensitive(lower, "no documents found matching hash prefix"); } bool isHotspotWindowEligible(int ops, int totalClients) { return ops >= std::max(10, totalClients); } // ───────────────────────────────────────────────────────────────────────────── // Wait for queue drain // ───────────────────────────────────────────────────────────────────────────── struct DrainObservation { bool drained{false}; uint64_t elapsedMs{0}; int polls{0}; int stablePolls{0}; uint64_t minimumDocuments{0}; uint64_t documentsBelowMinimumPolls{0}; uint64_t postIngestQueuedPolls{0}; uint64_t postIngestInflightPolls{0}; uint64_t workerActivePolls{0}; uint64_t workerQueuedPolls{0}; uint64_t deferredQueuePolls{0}; uint64_t coordinatedWritePolls{0}; uint64_t peakPostIngestQueued{0}; uint64_t peakPostIngestInflight{0}; uint64_t peakWorkerActive{0}; uint64_t peakWorkerQueued{0}; uint64_t peakDeferredQueueDepth{0}; double peakCpuPercent{0.0}; double peakMemoryMb{0.0}; uint32_t maxPressureLevel{0}; DaemonSnapshot finalSnapshot; json toJson() const { return json{{"drained", drained}, {"elapsed_ms", elapsedMs}, {"polls", polls}, {"stable_polls", stablePolls}, {"minimum_documents", minimumDocuments}, {"blocker_polls", {{"documents_below_minimum", documentsBelowMinimumPolls}, {"post_ingest_queued", postIngestQueuedPolls}, {"post_ingest_inflight", postIngestInflightPolls}, {"worker_active", workerActivePolls}, {"worker_queued", workerQueuedPolls}, {"deferred_queue", deferredQueuePolls}, {"coordinated_write", coordinatedWritePolls}}}, {"peak_post_ingest_queued", peakPostIngestQueued}, {"peak_post_ingest_inflight", peakPostIngestInflight}, {"peak_worker_active", peakWorkerActive}, {"peak_worker_queued", peakWorkerQueued}, {"peak_deferred_queue_depth", peakDeferredQueueDepth}, {"peak_cpu_pct", peakCpuPercent}, {"peak_memory_mb", peakMemoryMb}, {"max_pressure_level", maxPressureLevel}, {"final_snapshot", finalSnapshot.toJson()}}; } }; struct IdleBaselineObservation { uint64_t durationMs{0}; int samples{0}; double meanCpuPercent{0.0}; double peakCpuPercent{0.0}; double meanMemoryMb{0.0}; double peakMemoryMb{0.0}; uint64_t maxPostIngestQueued{0}; uint64_t maxPostIngestInflight{0}; size_t maxActiveConnections{0}; uint32_t maxPressureLevel{0}; json toJson() const { return json{{"duration_ms", durationMs}, {"samples", samples}, {"mean_cpu_pct", meanCpuPercent}, {"peak_cpu_pct", peakCpuPercent}, {"mean_memory_mb", meanMemoryMb}, {"peak_memory_mb", peakMemoryMb}, {"max_post_ingest_queued", maxPostIngestQueued}, {"max_post_ingest_inflight", maxPostIngestInflight}, {"max_active_connections", maxActiveConnections}, {"max_pressure_level", maxPressureLevel}}; } }; struct IdleWakeProbeObservation { bool enabled{false}; bool attempted{false}; bool addCompleted{false}; bool addSucceeded{false}; bool workVisible{false}; uint64_t addLatencyUs{0}; uint64_t workVisibleUs{0}; std::string addError; DaemonSnapshot beforeSnapshot; DaemonSnapshot visibleSnapshot; DrainObservation preProbeDrain; IdleBaselineObservation idleBaseline; DrainObservation postProbeDrain; json toJson() const { return json{{"enabled", enabled}, {"attempted", attempted}, {"add_completed", addCompleted}, {"add_succeeded", addSucceeded}, {"work_visible", workVisible}, {"add_latency_us", addLatencyUs}, {"work_visible_us", workVisibleUs}, {"add_error", addError}, {"before_snapshot", beforeSnapshot.toJson()}, {"visible_snapshot", visibleSnapshot.toJson()}, {"pre_probe_drain", preProbeDrain.toJson()}, {"idle_baseline", idleBaseline.toJson()}, {"post_probe_drain", postProbeDrain.toJson()}}; } }; bool writePipelineDrained(const DaemonSnapshot& snapshot, uint64_t minimumDocuments = 0) { return snapshot.postIngestQueued == 0 && snapshot.postIngestInflight == 0 && snapshot.workerActive == 0 && snapshot.workerQueued == 0 && snapshot.deferredQueueDepth == 0 && snapshot.writeQueueDepth == 0 && snapshot.writeInFlight == 0 && snapshot.documentsTotal >= minimumDocuments; } TEST_CASE("Drain contract includes post-ingest and coordinated writes", "[benchmark-contract][drain]") { CHECK(backgroundWorkerActive(0) == 0); CHECK(backgroundWorkerActive(1) == 0); CHECK(backgroundWorkerActive(3) == 2); DaemonSnapshot snapshot; CHECK(writePipelineDrained(snapshot)); CHECK(writePipelineDrained(snapshot, 0)); snapshot.postIngestQueued = 1; CHECK_FALSE(writePipelineDrained(snapshot)); snapshot.postIngestQueued = 0; snapshot.postIngestInflight = 1; CHECK_FALSE(writePipelineDrained(snapshot)); snapshot.postIngestInflight = 0; snapshot.writeQueueDepth = 1; CHECK_FALSE(writePipelineDrained(snapshot)); snapshot.writeQueueDepth = 0; snapshot.writeInFlight = 1; CHECK_FALSE(writePipelineDrained(snapshot)); snapshot.writeInFlight = 0; snapshot.workerActive = 1; CHECK_FALSE(writePipelineDrained(snapshot)); snapshot.workerActive = 0; snapshot.workerQueued = 1; CHECK_FALSE(writePipelineDrained(snapshot)); snapshot.workerQueued = 0; snapshot.deferredQueueDepth = 1; CHECK_FALSE(writePipelineDrained(snapshot)); snapshot.deferredQueueDepth = 0; snapshot.documentsTotal = 41; CHECK_FALSE(writePipelineDrained(snapshot, 42)); snapshot.documentsTotal = 42; CHECK(writePipelineDrained(snapshot, 42)); } DrainObservation waitForDrainObservation(DaemonClient& client, std::chrono::milliseconds timeout, std::chrono::milliseconds sampleInterval = 500ms, uint64_t minimumDocuments = 0) { DrainObservation observation; observation.minimumDocuments = minimumDocuments; const auto start = std::chrono::steady_clock::now(); const auto deadline = start + timeout; constexpr int stableRequired = 5; int stableCount = 0; while (std::chrono::steady_clock::now() < deadline) { auto snap = DaemonSnapshot::capture(client); observation.polls++; observation.peakPostIngestQueued = std::max(observation.peakPostIngestQueued, snap.postIngestQueued); observation.peakPostIngestInflight = std::max(observation.peakPostIngestInflight, snap.postIngestInflight); observation.peakWorkerActive = std::max(observation.peakWorkerActive, snap.workerActive); observation.peakWorkerQueued = std::max(observation.peakWorkerQueued, snap.workerQueued); observation.peakDeferredQueueDepth = std::max(observation.peakDeferredQueueDepth, snap.deferredQueueDepth); observation.peakCpuPercent = std::max(observation.peakCpuPercent, snap.cpuUsagePercent); observation.peakMemoryMb = std::max(observation.peakMemoryMb, snap.memoryUsageMb); observation.maxPressureLevel = std::max( observation.maxPressureLevel, static_cast(std::max(0, snap.pressureLevel))); observation.documentsBelowMinimumPolls += snap.documentsTotal < minimumDocuments ? 1u : 0u; observation.postIngestQueuedPolls += snap.postIngestQueued > 0 ? 1u : 0u; observation.postIngestInflightPolls += snap.postIngestInflight > 0 ? 1u : 0u; observation.workerActivePolls += snap.workerActive > 0 ? 1u : 0u; observation.workerQueuedPolls += snap.workerQueued > 0 ? 1u : 0u; observation.deferredQueuePolls += snap.deferredQueueDepth > 0 ? 1u : 0u; observation.coordinatedWritePolls += (snap.writeQueueDepth > 0 || snap.writeInFlight > 0) ? 1u : 0u; const bool drained = writePipelineDrained(snap, minimumDocuments); if (drained) { ++stableCount; if (stableCount >= stableRequired) { observation.drained = true; observation.stablePolls = stableCount; observation.finalSnapshot = snap; observation.elapsedMs = std::chrono::duration_cast( std::chrono::steady_clock::now() - start) .count(); return observation; } } else { stableCount = 0; } std::this_thread::sleep_for(sampleInterval); } observation.stablePolls = stableCount; observation.elapsedMs = std::chrono::duration_cast( std::chrono::steady_clock::now() - start) .count(); observation.finalSnapshot = DaemonSnapshot::capture(client); return observation; } IdleBaselineObservation sampleIdleBaseline(DaemonClient& client, std::chrono::milliseconds duration, std::chrono::milliseconds sampleInterval) { IdleBaselineObservation observation; observation.durationMs = static_cast(std::max(0, duration.count())); if (duration <= 0ms) { return observation; } const auto start = std::chrono::steady_clock::now(); const auto deadline = start + duration; double cpuSum = 0.0; double memorySum = 0.0; while (true) { auto snap = DaemonSnapshot::capture(client); observation.samples++; cpuSum += snap.cpuUsagePercent; memorySum += snap.memoryUsageMb; observation.peakCpuPercent = std::max(observation.peakCpuPercent, snap.cpuUsagePercent); observation.peakMemoryMb = std::max(observation.peakMemoryMb, snap.memoryUsageMb); observation.maxPostIngestQueued = std::max(observation.maxPostIngestQueued, snap.postIngestQueued); observation.maxPostIngestInflight = std::max(observation.maxPostIngestInflight, snap.postIngestInflight); observation.maxActiveConnections = std::max(observation.maxActiveConnections, snap.activeConnections); observation.maxPressureLevel = std::max( observation.maxPressureLevel, static_cast(std::max(0, snap.pressureLevel))); if (std::chrono::steady_clock::now() >= deadline) { break; } std::this_thread::sleep_for(sampleInterval); } if (observation.samples > 0) { observation.meanCpuPercent = cpuSum / static_cast(observation.samples); observation.meanMemoryMb = memorySum / static_cast(observation.samples); } observation.durationMs = std::chrono::duration_cast( std::chrono::steady_clock::now() - start) .count(); return observation; } bool waitForDrain(DaemonClient& client, std::chrono::seconds timeout) { return waitForDrainObservation(client, timeout).drained; } IdleWakeProbeObservation runIdleWakeProbe(const BenchConfig& cfg, const fs::path& socketPath, DaemonClient& monitorClient) { IdleWakeProbeObservation observation; observation.enabled = cfg.idleProbeEnabled; if (!cfg.idleProbeEnabled) { return observation; } observation.attempted = true; const auto sampleInterval = std::chrono::milliseconds(cfg.idleProbeSampleMs); observation.preProbeDrain = waitForDrainObservation( monitorClient, std::chrono::seconds(cfg.drainTimeoutSecs), sampleInterval); if (!observation.preProbeDrain.drained) { return observation; } observation.idleBaseline = sampleIdleBaseline( monitorClient, std::chrono::milliseconds(cfg.idleProbeBaselineMs), sampleInterval); observation.beforeSnapshot = DaemonSnapshot::capture(monitorClient); struct ProbeAddResult { bool success{false}; std::string error; int64_t latencyUs{0}; }; auto addFuture = std::async(std::launch::async, [socketPath, &cfg]() { ProbeAddResult result; ClientConfig ccfg; ccfg.socketPath = socketPath; ccfg.autoStart = false; ccfg.requestTimeout = std::chrono::seconds(15); ccfg.singleUseConnections = true; DaemonClient client(ccfg); AddDocumentRequest req; req.name = "idle_probe_" + std::to_string(::getpid()) + ".md"; req.content = generateDocument( 777, 0, std::max(256, std::min(1024, cfg.docSizeBytes))); req.tags = {"bench", "idle-probe"}; req.collection = "bench_idle_probe"; req.noEmbeddings = true; const auto start = std::chrono::steady_clock::now(); auto addResult = yams::cli::run_sync(client.streamingAddDocument(req), 15s); result.latencyUs = std::chrono::duration_cast( std::chrono::steady_clock::now() - start) .count(); result.success = static_cast(addResult); if (!addResult) { result.error = addResult.error().message; } return result; }); const auto probeStart = std::chrono::steady_clock::now(); const auto detectDeadline = probeStart + std::chrono::milliseconds(cfg.idleProbeDetectTimeoutMs); while (std::chrono::steady_clock::now() < detectDeadline) { if (!observation.addCompleted && addFuture.wait_for(0ms) == std::future_status::ready) { const auto addResult = addFuture.get(); observation.addCompleted = true; observation.addSucceeded = addResult.success; observation.addLatencyUs = static_cast(std::max(0, addResult.latencyUs)); observation.addError = addResult.error; } auto snap = DaemonSnapshot::capture(monitorClient); if (!observation.workVisible && (snap.documentsTotal > observation.beforeSnapshot.documentsTotal || snap.postIngestQueued > 0 || snap.postIngestInflight > 0 || snap.searchQueued > 0 || snap.searchActive > 0)) { observation.workVisible = true; observation.workVisibleUs = std::chrono::duration_cast( std::chrono::steady_clock::now() - probeStart) .count(); observation.visibleSnapshot = snap; } if (observation.workVisible && observation.addCompleted) { break; } std::this_thread::sleep_for(sampleInterval); } if (!observation.addCompleted && addFuture.wait_for(std::chrono::seconds(20)) == std::future_status::ready) { const auto addResult = addFuture.get(); observation.addCompleted = true; observation.addSucceeded = addResult.success; observation.addLatencyUs = static_cast(std::max(0, addResult.latencyUs)); observation.addError = addResult.error; } if (observation.visibleSnapshot.maxConnections == 0) { observation.visibleSnapshot = DaemonSnapshot::capture(monitorClient); } if (observation.addSucceeded) { observation.postProbeDrain = waitForDrainObservation( monitorClient, std::chrono::seconds(cfg.drainTimeoutSecs), sampleInterval); } return observation; } // ───────────────────────────────────────────────────────────────────────────── // Output helpers // ───────────────────────────────────────────────────────────────────────────── void emitJsonl(const fs::path& outputPath, const json& record) { json enriched = record; enriched["schema_version"] = "multi_client_record_v4"; if (const auto* v = std::getenv("YAMS_BENCH_RUN_ID"); v && *v) { enriched["run_id"] = v; } if (const auto* v = std::getenv("YAMS_BENCH_PHASE"); v && *v) { enriched["run_phase"] = v; } if (const auto* v = std::getenv("YAMS_BENCH_CASE"); v && *v) { enriched["run_case"] = v; } if (const auto* v = std::getenv("YAMS_BENCH_TRANSPORT"); v && *v) { enriched["run_transport"] = v; } if (const auto* v = std::getenv("YAMS_BENCH_GIT_SHA"); v && *v) { enriched["git_sha"] = v; } if (const auto* v = std::getenv("YAMS_BENCH_BACKEND"); v && *v) { enriched["run_backend"] = v; } if (const auto* v = std::getenv("YAMS_BENCH_AUTH_MODE"); v && *v) { enriched["run_auth_mode"] = v; } fs::create_directories(outputPath.parent_path()); std::ofstream out(outputPath, std::ios::app); if (out) { out << enriched.dump() << '\n'; } } std::optional jsonToInt64(const nlohmann::json& value) { try { if (value.is_number_integer()) { return value.get(); } if (value.is_number_unsigned()) { return static_cast(value.get()); } if (value.is_number_float()) { return static_cast(value.get()); } if (value.is_string()) { const auto& s = value.get_ref(); if (!s.empty()) { return std::stoll(s); } } } catch (const std::exception&) { return std::nullopt; } return std::nullopt; } std::optional jsonGetInt64AnyKey(const nlohmann::json& object, std::initializer_list keys) { if (!object.is_object()) { return std::nullopt; } for (const auto* key : keys) { if (!key) { continue; } auto it = object.find(key); if (it == object.end()) { continue; } if (auto parsed = jsonToInt64(*it); parsed.has_value()) { return parsed; } } return std::nullopt; } void printPercentiles(const std::string& label, const PercentileStats& stats) { std::cout << " " << label << ": n=" << stats.count << " min=" << stats.min << "us" << " p50=" << stats.p50 << "us" << " p95=" << stats.p95 << "us" << " p99=" << stats.p99 << "us" << " max=" << stats.max << "us" << " mean=" << std::fixed << std::setprecision(0) << stats.mean << "us" << std::endl; } void printDrainAndIdleProbeSummary(const DrainObservation& drainObservation, const IdleWakeProbeObservation& idleProbe) { std::cout << " Drain time: " << drainObservation.elapsedMs << " ms\n"; if (idleProbe.attempted) { std::cout << " Idle probe: baseline_cpu=" << std::fixed << std::setprecision(2) << idleProbe.idleBaseline.meanCpuPercent << "% peak=" << idleProbe.idleBaseline.peakCpuPercent << "% work_visible=" << idleProbe.workVisibleUs << "us add_latency=" << idleProbe.addLatencyUs << "us post_drain=" << idleProbe.postProbeDrain.elapsedMs << "ms\n"; } } // Create harness options tuned for benchmarking: // - Disable model provider (we don't need embeddings for ingestion benchmarks) // - Use mock model provider as fallback // - No auto-loaded plugins (reduce startup variance) // Overloads accept a BenchConfig to enable plugins, real models, and custom data dirs. DaemonHarnessOptions benchHarnessOptions() { return DaemonHarnessOptions{ .enableModelProvider = false, .useMockModelProvider = true, .autoLoadPlugins = false, .enableAutoRepair = true, .isolateState = true, }; } DaemonHarnessOptions benchHarnessOptions(const BenchConfig& cfg) { DaemonHarnessOptions opts{ .enableModelProvider = cfg.enablePlugins || cfg.useRealModelProvider, .useMockModelProvider = !cfg.useRealModelProvider, .autoLoadPlugins = cfg.enablePlugins, .enableAutoRepair = true, .isolateState = true, .pluginDir = cfg.pluginDir, .dataDir = cfg.dataDir, .configPath = cfg.configPath, .trustedPluginPaths = cfg.trustedPluginPaths, }; return opts; } // Startup timeout: 30s to handle TSan/debug builds where init is ~5-10x slower constexpr auto kStartTimeout = std::chrono::seconds(30); constexpr auto kCorpusStartTimeout = std::chrono::seconds(180); std::chrono::seconds benchStartTimeout(const BenchConfig& cfg) { return cfg.dataDir.has_value() ? kCorpusStartTimeout : kStartTimeout; } TEST_CASE("Multi-client benchmark allows explicit corpora a longer startup window", "[multi-client][config]") { BenchConfig synthetic; CHECK(benchStartTimeout(synthetic) == kStartTimeout); BenchConfig corpus; corpus.dataDir = fs::path("symbol-rich-corpus"); CHECK(benchStartTimeout(corpus) == kCorpusStartTimeout); } bool startHarnessWithRetry(DaemonHarness& harness, std::chrono::milliseconds timeout, int maxAttempts = 3) { for (int attempt = 0; attempt < maxAttempts; ++attempt) { if (harness.start(timeout)) { return true; } harness.stop(); std::this_thread::sleep_for(500ms); } return false; } } // namespace // ============================================================================= // TEST CASES // ============================================================================= TEST_CASE("Multi-client ingestion: baseline single client", "[!benchmark][multi-client][ingestion]") { auto cfg = BenchConfig::fromEnv(); cfg.numClients = 1; // Force single client for baseline DaemonHarness harness(benchHarnessOptions(cfg)); REQUIRE(harness.start(benchStartTimeout(cfg))); // Create client ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); // Ingest documents std::vector addLatencies; auto startTime = std::chrono::steady_clock::now(); for (int i = 0; i < cfg.docsPerClient; ++i) { auto doc = generateDocument(0, i, cfg.docSizeBytes); AddDocumentRequest req; req.name = "baseline_doc_" + std::to_string(i) + ".md"; req.content = std::move(doc); req.tags = {"bench", "baseline"}; auto t0 = std::chrono::steady_clock::now(); auto result = yams::cli::run_sync(client.streamingAddDocument(req), 30s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (result) { addLatencies.push_back(latUs); } else { WARN("Add failed at doc " << i << ": " << result.error().message); } } auto elapsed = std::chrono::steady_clock::now() - startTime; double elapsedSec = std::chrono::duration(elapsed).count(); double throughput = static_cast(addLatencies.size()) / elapsedSec; // Wait for pipeline drain and then probe idle->work transition once. const auto drainObservation = waitForDrainObservation(client, std::chrono::seconds(cfg.drainTimeoutSecs)); REQUIRE(drainObservation.drained); const auto idleProbe = runIdleWakeProbe(cfg, harness.socketPath(), client); auto snap = DaemonSnapshot::capture(client); const auto recovery = runRecoveryCheck(harness.socketPath()); auto stats = PercentileStats::compute(addLatencies); std::cout << "\n=== Single Client Baseline ===\n"; std::cout << " Docs ingested: " << addLatencies.size() << "/" << cfg.docsPerClient << "\n"; std::cout << " Elapsed: " << std::fixed << std::setprecision(2) << elapsedSec << "s\n"; std::cout << " Throughput: " << throughput << " docs/s\n"; printPercentiles("Add latency", stats); printDrainAndIdleProbeSummary(drainObservation, idleProbe); printRecoverySummary(recovery); std::cout << " Final docs_total: " << snap.documentsTotal << "\n"; std::cout << " Memory: " << snap.memoryUsageMb << " MB\n\n"; CHECK(static_cast(addLatencies.size()) == cfg.docsPerClient); CHECK(throughput > 0.0); CHECK(recovery.ok()); // Emit JSONL record for regression tracking json record{ {"timestamp", isoTimestamp()}, {"test", "baseline_single_client"}, {"num_clients", 1}, {"docs_per_client", cfg.docsPerClient}, {"doc_size_bytes", cfg.docSizeBytes}, {"elapsed_seconds", elapsedSec}, {"throughput_docs_per_sec", throughput}, {"add_latency", stats.toJson()}, {"drain_metrics", drainObservation.toJson()}, {"idle_probe", idleProbe.toJson()}, {"recovery", recovery.toJson()}, {"daemon_snapshot", snap.toJson()}, {"tuning_profile", cfg.tuningProfile}, }; emitJsonl(cfg.outputPath, record); } TEST_CASE("Multi-client ingestion: concurrent pure ingest", "[!benchmark][multi-client][ingestion]") { auto cfg = BenchConfig::fromEnv(); DaemonHarness harness(benchHarnessOptions(cfg)); REQUIRE(harness.start(kStartTimeout)); // Status-monitoring client (separate from worker clients) ClientConfig monCfg; monCfg.socketPath = harness.socketPath(); monCfg.autoStart = false; monCfg.requestTimeout = 10s; monCfg.singleUseConnections = true; DaemonClient monitorClient(monCfg); // Start time-series sampler TimeSeriesSampler sampler; sampler.start(monitorClient, std::chrono::milliseconds(cfg.resourceSampleMs), harness.dataDir()); // Shared counters std::atomic totalDocsIngested{0}; std::atomic totalFailures{0}; std::vector clientResults(cfg.numClients); auto globalStart = std::chrono::steady_clock::now(); // Launch concurrent client threads std::vector threads; threads.reserve(cfg.numClients); for (int c = 0; c < cfg.numClients; ++c) { threads.emplace_back([&, c]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); ClientResult& result = clientResults[c]; result.clientId = c; auto clientStart = std::chrono::steady_clock::now(); for (int i = 0; i < cfg.docsPerClient; ++i) { auto doc = generateDocument(c, i, cfg.docSizeBytes); AddDocumentRequest req; req.name = "client" + std::to_string(c) + "_doc_" + std::to_string(i) + ".md"; req.content = std::move(doc); req.tags = {"bench", "multi-client", "client-" + std::to_string(c)}; req.collection = "bench_client_" + std::to_string(c); auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.streamingAddDocument(req), 30s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); OpLatency op; op.opType = "add"; op.clientId = c; op.latencyUs = latUs; op.success = static_cast(res); if (!res) op.errorMsg = res.error().message; result.latencies.push_back(op); if (res) { result.docsIngested++; totalDocsIngested.fetch_add(1); } else { result.failures++; totalFailures.fetch_add(1); } } auto clientEnd = std::chrono::steady_clock::now(); result.elapsedSeconds = std::chrono::duration(clientEnd - clientStart).count(); result.addThroughputDocsPerSec = (result.docsIngested > 0) ? static_cast(result.docsIngested) / result.elapsedSeconds : 0.0; }); } for (auto& t : threads) t.join(); sampler.stop(); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); double aggregateThroughput = static_cast(totalDocsIngested.load()) / globalElapsed; // Wait for pipeline drain and then probe idle->work transition once. const auto drainObservation = waitForDrainObservation(monitorClient, std::chrono::seconds(cfg.drainTimeoutSecs)); const bool drained = drainObservation.drained; const auto idleProbe = runIdleWakeProbe(cfg, harness.socketPath(), monitorClient); auto finalSnap = DaemonSnapshot::capture(monitorClient); const auto recovery = runRecoveryCheck(harness.socketPath()); // Compute aggregate latency stats std::vector allAddLatencies; for (auto& cr : clientResults) { for (auto& op : cr.latencies) { if (op.opType == "add" && op.success) allAddLatencies.push_back(op.latencyUs); } } auto aggStats = PercentileStats::compute(allAddLatencies); std::map failureBreakdown; std::map failureSamples; for (const auto& cr : clientResults) { for (const auto& op : cr.latencies) { if (op.opType != "add" || op.success) { continue; } const auto bucket = classifyFailureMessage(op.errorMsg); failureBreakdown[bucket] += 1; if (!op.errorMsg.empty() && !failureSamples.contains(bucket)) { failureSamples[bucket] = op.errorMsg; } } } // Print summary std::cout << "\n=== Concurrent Pure Ingest (N=" << cfg.numClients << ") ===\n"; const int expectedDocs = checkedMulToInt(cfg.numClients, cfg.docsPerClient); std::cout << " Total docs: " << totalDocsIngested.load() << "/" << expectedDocs << "\n"; std::cout << " Total failures: " << totalFailures.load() << "\n"; std::cout << " Wall time: " << std::fixed << std::setprecision(2) << globalElapsed << "s\n"; std::cout << " Aggregate throughput: " << aggregateThroughput << " docs/s\n"; printPercentiles("Add latency (all)", aggStats); if (!failureBreakdown.empty()) { std::cout << "\n Failure breakdown (add requests):\n"; for (const auto& [bucket, count] : failureBreakdown) { std::cout << " " << bucket << ": " << count; auto sampleIt = failureSamples.find(bucket); if (sampleIt != failureSamples.end()) { std::cout << " (sample: " << sampleIt->second << ")"; } std::cout << "\n"; } } std::cout << "\n Per-client breakdown:\n"; for (auto& cr : clientResults) { std::cout << " Client " << cr.clientId << ": " << cr.docsIngested << " docs, " << std::fixed << std::setprecision(1) << cr.addThroughputDocsPerSec << " docs/s, failures=" << cr.failures << "\n"; } // Fairness metric: coefficient of variation of per-client throughput { std::vector perClientThroughput; for (auto& cr : clientResults) { if (cr.addThroughputDocsPerSec > 0) perClientThroughput.push_back(cr.addThroughputDocsPerSec); } if (perClientThroughput.size() > 1) { double mean = std::accumulate(perClientThroughput.begin(), perClientThroughput.end(), 0.0) / static_cast(perClientThroughput.size()); double variance = 0.0; for (auto v : perClientThroughput) variance += (v - mean) * (v - mean); variance /= static_cast(perClientThroughput.size()); double cv = (mean > 0) ? std::sqrt(variance) / mean : 0.0; std::cout << " Fairness (CV of throughput): " << std::fixed << std::setprecision(3) << cv << " (lower = more fair)\n"; } } std::cout << " Queue drained: " << (drained ? "yes" : "NO") << "\n"; printDrainAndIdleProbeSummary(drainObservation, idleProbe); printRecoverySummary(recovery); std::cout << " Final docs_total: " << finalSnap.documentsTotal << "\n"; std::cout << " Final memory: " << finalSnap.memoryUsageMb << " MB\n\n"; CHECK(totalDocsIngested.load() == expectedDocs); CHECK(drained); CHECK(recovery.ok()); // Emit JSONL json perClient = json::array(); for (auto& cr : clientResults) { perClient.push_back(json{ {"client_id", cr.clientId}, {"docs_ingested", cr.docsIngested}, {"throughput_docs_per_sec", cr.addThroughputDocsPerSec}, {"elapsed_seconds", cr.elapsedSeconds}, {"failures", cr.failures}, }); } // Time-series data json timeSeries = json::array(); for (auto& s : sampler.getSamples()) { timeSeries.push_back(json{{"elapsed_ms", s.elapsedMs}, {"snapshot", s.snap.toJson()}}); } json failureBreakdownJson = json::object(); for (const auto& [bucket, count] : failureBreakdown) { failureBreakdownJson[bucket] = count; } json failureSamplesJson = json::object(); for (const auto& [bucket, sample] : failureSamples) { failureSamplesJson[bucket] = sample; } json record{ {"timestamp", isoTimestamp()}, {"test", "concurrent_pure_ingest"}, {"num_clients", cfg.numClients}, {"docs_per_client", cfg.docsPerClient}, {"doc_size_bytes", cfg.docSizeBytes}, {"total_docs_ingested", totalDocsIngested.load()}, {"total_failures", totalFailures.load()}, {"elapsed_seconds", globalElapsed}, {"aggregate_throughput_docs_per_sec", aggregateThroughput}, {"add_latency", aggStats.toJson()}, {"failure_breakdown", failureBreakdownJson}, {"failure_samples", failureSamplesJson}, {"per_client", perClient}, {"daemon_snapshot_final", finalSnap.toJson()}, {"drain_metrics", drainObservation.toJson()}, {"idle_probe", idleProbe.toJson()}, {"recovery", recovery.toJson()}, {"time_series", timeSeries}, {"tuning_profile", cfg.tuningProfile}, {"drained", drained}, }; emitJsonl(cfg.outputPath, record); } TEST_CASE("Multi-client ingestion: mixed read/write workload", "[!benchmark][multi-client][mixed]") { auto cfg = BenchConfig::fromEnv(); DaemonHarness harness(benchHarnessOptions(cfg)); REQUIRE(harness.start(benchStartTimeout(cfg))); // Monitor client ClientConfig monCfg; monCfg.socketPath = harness.socketPath(); monCfg.autoStart = false; monCfg.requestTimeout = 10s; DaemonClient monitorClient(monCfg); // Pre-seed corpus so search queries return results if (cfg.warmupDocs > 0) { ClientConfig warmCfg; warmCfg.socketPath = harness.socketPath(); warmCfg.autoStart = false; warmCfg.requestTimeout = 30s; warmCfg.singleUseConnections = true; DaemonClient warmClient(warmCfg); std::cout << " Pre-seeding " << cfg.warmupDocs << " warmup docs...\n"; for (int i = 0; i < cfg.warmupDocs; ++i) { AddDocumentRequest req; req.name = "warmup_doc_" + std::to_string(i) + ".md"; req.content = generateDocument(99, i, cfg.docSizeBytes); req.tags = {"warmup"}; auto res = yams::cli::run_sync(warmClient.streamingAddDocument(req), 30s); if (!res) { WARN("Warmup doc " << i << " failed: " << res.error().message); } } // Wait for warmup to settle (void)waitForDrain(warmClient, 30s); std::this_thread::sleep_for(1s); } constexpr std::string_view kHydrationGraphSymbol = "benchHydratedSymbol"; if (cfg.profileHydrationSurfaces) { ClientConfig setupCfg; setupCfg.socketPath = harness.socketPath(); setupCfg.autoStart = false; setupCfg.requestTimeout = 30s; setupCfg.singleUseConnections = true; DaemonClient setupClient(setupCfg); const auto sourcePath = harness.dataDir() / "bench_hydration_source.cpp"; const std::string source = "int benchHydratedSymbol() {\n return 42;\n}\n"; { std::ofstream out(sourcePath); REQUIRE(out.good()); out << source; } AddDocumentRequest addReq; addReq.name = sourcePath.string(); addReq.content = source; addReq.tags = {"bench", "hydration"}; auto addResult = yams::cli::run_sync(setupClient.streamingAddDocument(addReq), 30s); REQUIRE(addResult); (void)waitForDrain(setupClient, 30s); auto* serviceManager = harness.daemon()->getServiceManager(); REQUIRE(serviceManager != nullptr); auto metadataRepo = serviceManager->getMetadataRepo(); REQUIRE(metadataRepo != nullptr); auto kgStore = metadataRepo->getKnowledgeGraphStore(); REQUIRE(kgStore != nullptr); yams::metadata::KGNode node; node.nodeKey = "function:bench::benchHydratedSymbol@" + sourcePath.string(); node.label = std::string(kHydrationGraphSymbol); node.type = "function"; node.properties = nlohmann::json{{"file_path", sourcePath.string()}, {"qualified_name", "bench::benchHydratedSymbol"}, {"start_line", 1}, {"end_line", 3}} .dump(); REQUIRE(kgStore->upsertNode(node)); } const auto resourceBaseline = DaemonSnapshot::capture(monitorClient, harness.dataDir()); TimeSeriesSampler sampler; sampler.start(monitorClient, std::chrono::milliseconds(cfg.resourceSampleMs), harness.dataDir()); std::atomic totalAdds{0}; std::atomic totalSearches{0}; std::atomic totalLists{0}; std::atomic totalFailures{0}; OutputLivenessCounters searchOutputLiveness; OutputLivenessCounters grepOutputLiveness; OutputLivenessCounters graphOutputLiveness; std::vector clientResults(cfg.numClients); std::mutex hydrationSurfaceMutex; std::vector grepContentRetrievalMs; std::vector grepMatchRenderUs; std::vector grepWorkerCriticalTotalMs; std::vector graphSnippetRenderUs; std::vector graphSnippetsRendered; auto globalStart = std::chrono::steady_clock::now(); std::vector threads; threads.reserve(cfg.numClients + (cfg.profileHydrationSurfaces ? 1 : 0)); for (int c = 0; c < cfg.numClients; ++c) { threads.emplace_back([&, c]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; ccfg.singleUseConnections = true; DaemonClient client(ccfg); ClientResult& result = clientResults[c]; result.clientId = c; std::mt19937 rng(static_cast(c * 1000 + 42)); std::uniform_real_distribution opDist(0.0, 1.0); auto clientStart = std::chrono::steady_clock::now(); int docIdx = 0; // Total operations = docsPerClient ingestions + proportional reads int totalOps = cfg.mixedOpsPerClient; if (totalOps <= 0 && cfg.searchRatio >= 1.0) { totalOps = cfg.docsPerClient; } else if (totalOps <= 0 && cfg.searchRatio > 0.0) { totalOps = static_cast(static_cast(cfg.docsPerClient) / (1.0 - cfg.searchRatio)); } else if (totalOps <= 0) { totalOps = cfg.docsPerClient; } for (int op = 0; op < totalOps; ++op) { double roll = opDist(rng); bool doSearch = (roll < cfg.searchRatio) && (docIdx > 0 || cfg.warmupDocs > 0); bool doList = (roll >= cfg.searchRatio && roll < cfg.searchRatio + 0.05); if (doSearch) { // Search operation SearchRequest req; req.query = cfg.searchQuery; req.searchType = cfg.searchType; req.limit = cfg.searchLimit; req.pathsOnly = cfg.searchPathsOnly; req.snippetHydrationTimeoutMs = cfg.snippetHydrationTimeoutMs; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.search(req), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); OpLatency lat; lat.opType = "search"; lat.clientId = c; lat.latencyUs = latUs; lat.success = static_cast(res); if (!res) lat.errorMsg = res.error().message; result.latencies.push_back(lat); if (res) { result.searchesExecuted++; totalSearches.fetch_add(1); searchOutputLiveness.recordSuccess(!res.value().results.empty()); collectSearchHydrationTelemetry(res.value(), result); } else { searchOutputLiveness.recordError(); result.failures++; totalFailures.fetch_add(1); } } else if (doList) { // List operation ListRequest req; req.limit = 20; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.list(req), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); OpLatency lat; lat.opType = "list"; lat.clientId = c; lat.latencyUs = latUs; lat.success = static_cast(res); if (!res) lat.errorMsg = res.error().message; result.latencies.push_back(lat); if (res) { result.listsExecuted++; totalLists.fetch_add(1); } else { result.failures++; totalFailures.fetch_add(1); } } else { // Add operation if (docIdx >= cfg.docsPerClient) continue; // Already ingested all docs for this client auto doc = generateDocument(c, docIdx, cfg.docSizeBytes); AddDocumentRequest req; req.name = "mixed_c" + std::to_string(c) + "_d" + std::to_string(docIdx) + ".md"; req.content = std::move(doc); req.tags = {"bench", "mixed", "client-" + std::to_string(c)}; req.collection = "bench_mixed_" + std::to_string(c); auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.streamingAddDocument(req), 30s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); OpLatency lat; lat.opType = "add"; lat.clientId = c; lat.latencyUs = latUs; lat.success = static_cast(res); if (!res) lat.errorMsg = res.error().message; result.latencies.push_back(lat); if (res) { result.docsIngested++; totalAdds.fetch_add(1); } else { result.failures++; totalFailures.fetch_add(1); } docIdx++; } } auto clientEnd = std::chrono::steady_clock::now(); result.elapsedSeconds = std::chrono::duration(clientEnd - clientStart).count(); result.addThroughputDocsPerSec = (result.docsIngested > 0) ? static_cast(result.docsIngested) / result.elapsedSeconds : 0.0; }); } if (cfg.profileHydrationSurfaces) { threads.emplace_back([&]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; ccfg.singleUseConnections = true; DaemonClient client(ccfg); const int probeOps = std::clamp(cfg.mixedOpsPerClient / 4, 4, 25); const auto recordGrepRender = [&](const yams::Result& result) { if (!result) { grepOutputLiveness.recordError(); return; } grepOutputLiveness.recordSuccess(!result.value().matches.empty()); std::lock_guard lock(hydrationSurfaceMutex); if (const auto value = metricInt(result.value().searchStats, "content_retrieval_ms")) { grepContentRetrievalMs.push_back(*value); } if (const auto value = metricInt(result.value().searchStats, "match_render_us")) { grepMatchRenderUs.push_back(*value); } if (const auto value = metricInt(result.value().searchStats, "worker_critical_total_ms")) { grepWorkerCriticalTotalMs.push_back(*value); } }; const auto recordGraphRender = [&](const yams::Result& result) { if (!result) { graphOutputLiveness.recordError(); return; } const auto& response = result.value(); graphOutputLiveness.recordSuccess(!response.entrySymbols.empty() || !response.files.empty() || !response.relationships.empty()); std::lock_guard lock(hydrationSurfaceMutex); graphSnippetRenderUs.push_back(static_cast(response.snippetRenderMicros)); graphSnippetsRendered.push_back(static_cast(response.snippetsRendered)); }; for (int i = 0; i < probeOps; ++i) { GrepRequest grepReq; grepReq.pattern = std::string(kHydrationGraphSymbol); grepReq.literalText = true; grepReq.maxMatches = 20; auto grepResult = yams::cli::run_sync(client.grep(grepReq), 30s); recordGrepRender(grepResult); GraphExploreRequest graphReq; graphReq.query = std::string(kHydrationGraphSymbol); graphReq.maxFiles = 4; auto graphResult = yams::cli::run_sync(client.call(graphReq), 30s); recordGraphRender(graphResult); } }); } for (auto& t : threads) t.join(); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); const uint64_t minimumDocuments = resourceBaseline.documentsTotal + static_cast(totalAdds.load()); const auto drainObservation = waitForDrainObservation( monitorClient, std::chrono::seconds(cfg.drainTimeoutSecs), 500ms, minimumDocuments); sampler.stop(); const auto resourcePeaks = summarizeResourcePeaks(sampler.getSamples()); const bool drained = drainObservation.drained; const auto idleProbe = runIdleWakeProbe(cfg, harness.socketPath(), monitorClient); auto finalSnap = DaemonSnapshot::capture(monitorClient, harness.dataDir()); const auto recovery = runRecoveryCheck(harness.socketPath()); // Compute per-op-type latency stats std::vector addLats, searchLats, listLats; std::vector snippetHydrationUs, snippetLookupUs, snippetContentFetchUs, snippetRenderUs; int64_t snippetTimeoutHits = 0; int64_t snippetHydrationCandidates = 0; int64_t snippetHydrationMissing = 0; int64_t snippetHydratedResults = 0; for (auto& cr : clientResults) { for (auto& op : cr.latencies) { if (!op.success) continue; if (op.opType == "add") addLats.push_back(op.latencyUs); else if (op.opType == "search") searchLats.push_back(op.latencyUs); else if (op.opType == "list") listLats.push_back(op.latencyUs); } snippetHydrationUs.insert(snippetHydrationUs.end(), cr.snippetHydrationUs.begin(), cr.snippetHydrationUs.end()); snippetLookupUs.insert(snippetLookupUs.end(), cr.snippetLookupUs.begin(), cr.snippetLookupUs.end()); snippetContentFetchUs.insert(snippetContentFetchUs.end(), cr.snippetContentFetchUs.begin(), cr.snippetContentFetchUs.end()); snippetRenderUs.insert(snippetRenderUs.end(), cr.snippetRenderUs.begin(), cr.snippetRenderUs.end()); snippetTimeoutHits = checkedMetricAdd(snippetTimeoutHits, cr.snippetTimeoutHits); snippetHydrationCandidates = checkedMetricAdd(snippetHydrationCandidates, cr.snippetHydrationCandidates); snippetHydrationMissing = checkedMetricAdd(snippetHydrationMissing, cr.snippetHydrationMissing); snippetHydratedResults = checkedMetricAdd(snippetHydratedResults, cr.snippetHydratedResults); } auto addStats = PercentileStats::compute(addLats); auto searchStats = PercentileStats::compute(searchLats); auto listStats = PercentileStats::compute(listLats); auto snippetHydrationStats = PercentileStats::compute(snippetHydrationUs); auto snippetLookupStats = PercentileStats::compute(snippetLookupUs); auto snippetContentFetchStats = PercentileStats::compute(snippetContentFetchUs); auto snippetRenderStats = PercentileStats::compute(snippetRenderUs); auto grepContentRetrievalStats = PercentileStats::compute(grepContentRetrievalMs); auto grepMatchRenderStats = PercentileStats::compute(grepMatchRenderUs); auto grepWorkerCriticalTotalStats = PercentileStats::compute(grepWorkerCriticalTotalMs); auto graphSnippetRenderStats = PercentileStats::compute(graphSnippetRenderUs); const auto graphSnippetsRenderedTotal = std::accumulate(graphSnippetsRendered.begin(), graphSnippetsRendered.end(), int64_t{0}); std::cout << "\n=== Mixed Read/Write Workload (N=" << cfg.numClients << ", search_ratio=" << cfg.searchRatio << ") ===\n"; std::cout << " Adds: " << totalAdds.load() << ", Searches: " << totalSearches.load() << ", Lists: " << totalLists.load() << ", Failures: " << totalFailures.load() << "\n"; std::cout << " Wall time: " << std::fixed << std::setprecision(2) << globalElapsed << "s\n"; printPercentiles("Add latency", addStats); printPercentiles("Search latency", searchStats); printPercentiles("List latency", listStats); std::cout << " Queue drained: " << (drained ? "yes" : "NO") << "\n"; printDrainAndIdleProbeSummary(drainObservation, idleProbe); printRecoverySummary(recovery); std::cout << " Final docs_total: " << finalSnap.documentsTotal << "\n\n"; if (cfg.searchRatio < 1.0) { CHECK(totalAdds.load() > 0); } CHECK((totalAdds.load() + totalSearches.load() + totalLists.load()) > 0); if (cfg.profileHydrationSurfaces) { CHECK_FALSE(grepMatchRenderUs.empty()); CHECK_FALSE(graphSnippetRenderUs.empty()); CHECK(grepOutputLiveness.attempts.load(std::memory_order_relaxed) > 0); CHECK(graphOutputLiveness.attempts.load(std::memory_order_relaxed) > 0); } CHECK(drained); CHECK(recovery.ok()); json record{ {"timestamp", isoTimestamp()}, {"test", "mixed_read_write"}, {"num_clients", cfg.numClients}, {"docs_per_client", cfg.docsPerClient}, {"mixed_ops_per_client", cfg.mixedOpsPerClient}, {"search_ratio", cfg.searchRatio}, {"search_paths_only", cfg.searchPathsOnly}, {"snippet_hydration_timeout_ms", cfg.snippetHydrationTimeoutMs}, {"warmup_docs", cfg.warmupDocs}, {"total_adds", totalAdds.load()}, {"total_searches", totalSearches.load()}, {"total_lists", totalLists.load()}, {"total_failures", totalFailures.load()}, {"elapsed_seconds", globalElapsed}, {"add_latency", addStats.toJson()}, {"search_latency", searchStats.toJson()}, {"list_latency", listStats.toJson()}, {"search_server", {{"timing_snippet_hydration_us", snippetHydrationStats.toJson()}, {"timing_snippet_lookup_us", snippetLookupStats.toJson()}, {"timing_snippet_content_fetch_us", snippetContentFetchStats.toJson()}, {"timing_snippet_render_us", snippetRenderStats.toJson()}, {"snippet_timeout_hits", snippetTimeoutHits}, {"snippet_hydration_candidates", snippetHydrationCandidates}, {"snippet_hydration_missing", snippetHydrationMissing}, {"snippet_hydrated_results", snippetHydratedResults}}}, {"grep_server", {{"content_retrieval_ms", grepContentRetrievalStats.toJson()}, {"match_render_us", grepMatchRenderStats.toJson()}, {"worker_critical_total_ms", grepWorkerCriticalTotalStats.toJson()}}}, {"graph_server", {{"snippet_render_us", graphSnippetRenderStats.toJson()}, {"snippets_rendered", graphSnippetsRenderedTotal}}}, {"output_liveness", {{"search", searchOutputLiveness.toJson()}, {"grep", grepOutputLiveness.toJson()}, {"graph", graphOutputLiveness.toJson()}}}, {"daemon_snapshot_final", finalSnap.toJson()}, {"drain_metrics", drainObservation.toJson()}, {"idle_probe", idleProbe.toJson()}, {"recovery", recovery.toJson()}, {"resource_baseline", resourceBaseline.toJson()}, {"resource_peaks", resourcePeaks.toJson()}, {"tuning_profile", cfg.tuningProfile}, {"drained", drained}, }; emitJsonl(cfg.outputPath, record); } TEST_CASE("Multi-client ingestion: socket scaling vs ux latency", "[!benchmark][multi-client][ux-scaling]") { auto cfg = BenchConfig::fromEnv(); // Three phases: // 1. Warmup: seed a small searchable corpus. // 2. Realistic mixed load: measure ingest + UX latency under pressure. // 3. Deterministic control: force slot growth, then force shrink, and assert only on // sampled slot metrics rather than transient active-connection counts. constexpr size_t kInitialSlotLimit = 12; constexpr size_t kSlotMax = 16; constexpr size_t kSlotStep = 4; ScopedConnectionSlotOverrides slotOverrides(static_cast(kInitialSlotLimit), static_cast(kSlotMax), static_cast(kSlotStep)); yams::test::ScopedEnvVar maxActiveConn("YAMS_MAX_ACTIVE_CONN", std::to_string(kInitialSlotLimit)); DaemonHarness harness(benchHarnessOptions()); REQUIRE(startHarnessWithRetry(harness, kStartTimeout)); ClientConfig monCfg; monCfg.socketPath = harness.socketPath(); monCfg.autoStart = false; monCfg.requestTimeout = 10s; DaemonClient monitorClient(monCfg); // Pre-seed a small searchable corpus so UX requests hit real work. { ClientConfig warmCfg; warmCfg.socketPath = harness.socketPath(); warmCfg.autoStart = false; warmCfg.requestTimeout = 30s; DaemonClient warmClient(warmCfg); const int warmDocs = std::max(16, cfg.warmupDocs); for (int i = 0; i < warmDocs; ++i) { AddDocumentRequest req; req.name = "ux_scale_warmup_" + std::to_string(i) + ".md"; req.content = generateDocument(77, i, cfg.docSizeBytes); req.tags = {"warmup", "ux-scaling"}; auto res = yams::cli::run_sync(warmClient.streamingAddDocument(req), 30s); REQUIRE(res); } REQUIRE(waitForDrain(warmClient, 30s)); } TimeSeriesSampler sampler; sampler.start(monitorClient, 100ms); const int writerClients = std::max(2, cfg.numClients); const int uxClients = std::max(4, checkedMulToInt(cfg.numClients, 3, kMaxBenchClients * 3)); const int writerOpsPerClient = checkedMulToInt(cfg.docsPerClient, cfg.stressRounds, kMaxBenchOpsPerClient); const int uxOpsPerClient = checkedMulToInt(std::max(24, cfg.docsPerClient), cfg.stressRounds, kMaxBenchOpsPerClient); const int cliProbeOpsPerCommand = cfg.cliProbeOpsPerCommand; const size_t grownSlotTarget = std::min(kSlotMax, kInitialSlotLimit + kSlotStep); // Allow one lingering request/observer connection beyond the steady monitor connection. const size_t controlLightLoadTarget = std::max(3, std::min(4, static_cast(cfg.stressRounds + 2))); const auto yamsBinary = findYamsBinary(cfg); const bool cliStressProbesEnabled = yamsBinary.path.has_value() && yamsBinary.usableForStress && cliProbeOpsPerCommand > 0; std::atomic addOps{0}; std::atomic addFails{0}; std::atomic uxOps{0}; std::atomic uxFails{0}; std::atomic retryAfterCount{0}; std::atomic cliOps{0}; std::atomic cliFails{0}; std::mutex latMutex; std::vector addLatencies; std::vector searchLatencies; std::vector listLatencies; std::vector statusLatencies; std::vector cliListLatencies; std::vector cliSearchLatencies; std::vector cliStatusLatencies; std::map> cliStageLatencies; std::vector cliProbeResults; std::vector hotspotSamples; std::map> opStageLatencies; std::map uxErrorCategories; std::map uxFailureSamples; HeldSocketConnections heldConnections; const bool heldConnectionsOpened = heldConnections.open(harness.socketPath(), static_cast(kInitialSlotLimit)); INFO("HeldSocketConnections.open initial ramp socket=" << harness.socketPath().string() << " requested=" << kInitialSlotLimit << " opened=" << heldConnections.openedCount() << " failed_index=" << heldConnections.lastFailedIndex() << " error='" << heldConnections.lastError() << "'"); REQUIRE(heldConnectionsOpened); auto globalStart = std::chrono::steady_clock::now(); std::vector threads; threads.reserve(writerClients + uxClients); for (int writer = 0; writer < writerClients; ++writer) { threads.emplace_back([&, writer]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); std::vector localAdds; for (int i = 0; i < writerOpsPerClient; ++i) { AddDocumentRequest req; req.name = "ux_scale_writer_" + std::to_string(writer) + "_" + std::to_string(i) + ".md"; req.content = generateDocument(writer, i, cfg.docSizeBytes); req.tags = {"bench", "ux-scaling", "writer-" + std::to_string(writer)}; req.collection = "ux_scaling_writer_" + std::to_string(writer); const auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.streamingAddDocument(req), 30s); const auto t1 = std::chrono::steady_clock::now(); const auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { addOps.fetch_add(1); localAdds.push_back(latUs); } else { addFails.fetch_add(1); } { std::lock_guard lock(latMutex); hotspotSamples.push_back(TimedLatencySample{ .opType = "add", .wallClockMs = static_cast( std::chrono::duration_cast(t1 - globalStart) .count()), .latencyUs = latUs, .success = static_cast(res), }); } } std::lock_guard lock(latMutex); addLatencies.insert(addLatencies.end(), localAdds.begin(), localAdds.end()); }); } for (int ux = 0; ux < uxClients; ++ux) { threads.emplace_back([&]() { std::vector localSearch; std::vector localList; std::vector localStatus; std::map localErrorCategories; std::map localFailureSamples; ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 10s; DaemonClient client(ccfg); for (int i = 0; i < uxOpsPerClient; ++i) { const auto t0 = std::chrono::steady_clock::now(); bool success = false; std::string errorMsg; std::string opType; switch (i % 3) { case 0: { opType = "status"; auto res = yams::cli::run_sync(client.status(), 10s); success = static_cast(res); if (!res) { errorMsg = res.error().message; } if (res && res.value().retryAfterMs > 0) { retryAfterCount.fetch_add(1); } const auto t1 = std::chrono::steady_clock::now(); const auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (success) { localStatus.push_back(latUs); } { std::lock_guard lock(latMutex); if (res) { appendStatusTimingStats(opStageLatencies, res.value().requestCounts, latUs); } hotspotSamples.push_back(TimedLatencySample{ .opType = "status", .wallClockMs = static_cast( std::chrono::duration_cast( t1 - globalStart) .count()), .latencyUs = latUs, .success = success, }); } break; } case 1: { opType = "list"; ListRequest req; req.limit = 10; auto res = yams::cli::run_sync(client.list(req), 10s); success = static_cast(res); if (!res) { errorMsg = res.error().message; } const auto t1 = std::chrono::steady_clock::now(); const auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (success) { localList.push_back(latUs); } { std::lock_guard lock(latMutex); if (res) { appendStringTimingStats(opStageLatencies, "list", res.value().listStats, latUs); } hotspotSamples.push_back(TimedLatencySample{ .opType = "list", .wallClockMs = static_cast( std::chrono::duration_cast( t1 - globalStart) .count()), .latencyUs = latUs, .success = success, }); } break; } default: { opType = "search"; SearchRequest req; req.query = cfg.searchQuery; req.searchType = cfg.searchType; req.limit = cfg.searchLimit; auto res = yams::cli::run_sync(client.search(req), 10s); success = static_cast(res); if (!res) { errorMsg = res.error().message; } const auto t1 = std::chrono::steady_clock::now(); if (success) { localSearch.push_back( std::chrono::duration_cast(t1 - t0) .count()); } { std::lock_guard lock(latMutex); hotspotSamples.push_back(TimedLatencySample{ .opType = "search", .wallClockMs = static_cast( std::chrono::duration_cast( t1 - globalStart) .count()), .latencyUs = std::chrono::duration_cast(t1 - t0) .count(), .success = success, }); } break; } } uxOps.fetch_add(1); if (!success) { uxFails.fetch_add(1); const auto bucket = classifyFailureMessage(errorMsg); const auto key = opType + ":" + bucket; localErrorCategories[key] += 1; if (!errorMsg.empty() && !localFailureSamples.contains(key)) { localFailureSamples[key] = errorMsg; } } } std::lock_guard lock(latMutex); searchLatencies.insert(searchLatencies.end(), localSearch.begin(), localSearch.end()); listLatencies.insert(listLatencies.end(), localList.begin(), localList.end()); statusLatencies.insert(statusLatencies.end(), localStatus.begin(), localStatus.end()); for (const auto& [key, count] : localErrorCategories) { uxErrorCategories[key] += count; } for (const auto& [key, sample] : localFailureSamples) { if (!uxFailureSamples.contains(key)) { uxFailureSamples[key] = sample; } } }); } #ifndef _WIN32 if (cliStressProbesEnabled) { threads.emplace_back([&, cliBin = yamsBinary.path.value()]() { std::vector localCliList; std::vector localCliSearch; std::vector localCliStatus; std::map> localCliStageLatencies; std::vector localCliResults; const auto cliProbeTimeout = std::chrono::seconds(cfg.cliProbeTimeoutSecs); auto recordProbe = [&](const std::string& opType, const CliProbeResult& probe, std::vector& storage) { cliOps.fetch_add(1); if (probe.success) { storage.push_back(probe.latencyUs); } else { cliFails.fetch_add(1); } std::lock_guard lock(latMutex); hotspotSamples.push_back(TimedLatencySample{ .opType = opType, .wallClockMs = static_cast(std::chrono::duration_cast( std::chrono::steady_clock::now() - globalStart) .count()), .latencyUs = probe.latencyUs, .success = probe.success, }); for (const auto& [stage, elapsedUs] : probe.stageElapsedUs) { localCliStageLatencies[opType + ":" + stage].push_back(elapsedUs); } localCliResults.push_back(probe); }; for (int i = 0; i < cliProbeOpsPerCommand; ++i) { recordProbe("cli_list", runObservedCliProbe("cli_list", cliBin, harness.socketPath(), {"list", "--limit", "10"}, cliProbeTimeout), localCliList); recordProbe("cli_status", runObservedCliProbe("cli_status", cliBin, harness.socketPath(), {"status"}, cliProbeTimeout), localCliStatus); recordProbe("cli_search", runObservedCliProbe("cli_search", cliBin, harness.socketPath(), buildCliSearchArgs(cfg), cliProbeTimeout), localCliSearch); } std::lock_guard lock(latMutex); cliListLatencies.insert(cliListLatencies.end(), localCliList.begin(), localCliList.end()); cliSearchLatencies.insert(cliSearchLatencies.end(), localCliSearch.begin(), localCliSearch.end()); cliStatusLatencies.insert(cliStatusLatencies.end(), localCliStatus.begin(), localCliStatus.end()); for (auto& [stage, values] : localCliStageLatencies) { auto& dest = cliStageLatencies[stage]; dest.insert(dest.end(), values.begin(), values.end()); } cliProbeResults.insert(cliProbeResults.end(), localCliResults.begin(), localCliResults.end()); }); } #endif for (auto& thread : threads) { thread.join(); } sampler.stop(); heldConnections.closeAll(); std::this_thread::sleep_for(500ms); const auto globalEnd = std::chrono::steady_clock::now(); const double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); const auto drainObservation = waitForDrainObservation(monitorClient, std::chrono::seconds(cfg.drainTimeoutSecs)); const bool drained = drainObservation.drained; const auto idleProbe = runIdleWakeProbe(cfg, harness.socketPath(), monitorClient); const bool recoveryQuiesced = waitForActiveConnectionsAtMost(monitorClient, controlLightLoadTarget, 20s); const auto finalSnap = DaemonSnapshot::capture(monitorClient); const auto recovery = runRecoveryCheck(harness.socketPath()); const bool cliRecoveryAttempted = yamsBinary.path.has_value() && yamsBinary.usableForStress; const auto cliRecoveryStatus = cliRecoveryAttempted ? runObservedCliProbe("cli_status_recovery", yamsBinary.path.value(), harness.socketPath(), {"status"}, 10s) : CliProbeResult{}; const auto cliRecoveryList = cliRecoveryAttempted ? runObservedCliProbe("cli_list_recovery", yamsBinary.path.value(), harness.socketPath(), {"list", "--limit", "10"}, 10s) : CliProbeResult{}; const bool cliRecoveryOk = !cliRecoveryAttempted || (cliRecoveryStatus.success && cliRecoveryList.success); const auto samples = sampler.getSamples(); const auto slotSummary = summarizeSlotScaling(samples); const auto resourcePeaks = summarizeResourcePeaks(samples); const auto hotspotSummary = summarizeLatencyHotspots(hotspotSamples); const auto opStageSummary = summarizeTimingStats(opStageLatencies); const auto addStats = PercentileStats::compute(addLatencies); const auto searchStats = PercentileStats::compute(searchLatencies); const auto listStats = PercentileStats::compute(listLatencies); const auto statusStats = PercentileStats::compute(statusLatencies); const auto cliListStats = PercentileStats::compute(cliListLatencies); const auto cliSearchStats = PercentileStats::compute(cliSearchLatencies); const auto cliStatusStats = PercentileStats::compute(cliStatusLatencies); json cliProbeResultsJson = json::array(); for (const auto& probe : cliProbeResults) { cliProbeResultsJson.push_back(probe.toJson()); } std::map cliFailureBuckets; std::map cliFailureModes; std::map cliFailureSamples; for (const auto& probe : cliProbeResults) { if (probe.success) { continue; } cliFailureBuckets[probe.failureBucket] += 1; cliFailureModes[probe.failureMode] += 1; if (!probe.error.empty() && !cliFailureSamples.contains(probe.failureMode)) { cliFailureSamples[probe.failureMode] = probe.error; } } int unacceptableCliFailures = 0; for (const auto& [mode, count] : cliFailureModes) { if (mode != "timeout_under_load" && mode != "overload_under_load" && mode != "stress_expected") { unacceptableCliFailures += count; } } const bool observedGrowth = slotSummary.peakLimit >= grownSlotTarget || slotSummary.growEvents > 0; const bool observedShrink = slotSummary.finalLimit <= kInitialSlotLimit || slotSummary.shrinkEvents > 0; const double addThroughput = (globalElapsed > 0.0) ? static_cast(addOps.load()) / globalElapsed : 0.0; const double uxThroughput = (globalElapsed > 0.0) ? static_cast(uxOps.load()) / globalElapsed : 0.0; const double uxFailureRate = (uxOps.load() > 0) ? static_cast(uxFails.load()) / static_cast(uxOps.load()) : 0.0; std::cout << "\n=== Socket Scaling vs UX Latency ===\n"; std::cout << " Adds: " << addOps.load() << " (fails=" << addFails.load() << ")" << " UX ops: " << uxOps.load() << " (fails=" << uxFails.load() << ")\n"; std::cout << " Throughput: add=" << std::fixed << std::setprecision(1) << addThroughput << " docs/s, ux=" << uxThroughput << " ops/s\n"; std::cout << " Search config: daemon(type=" << cfg.searchType << ", limit=" << cfg.searchLimit << ", query=\"" << cfg.searchQuery << "\")" << " cli(type=" << cfg.cliSearchType << ", limit=" << cfg.cliSearchLimit << ", query=\"" << cfg.cliSearchQuery << "\")\n"; if (cliStressProbesEnabled) { std::cout << " CLI probes: " << cliOps.load() << " (fails=" << cliFails.load() << ") binary=" << yamsBinary.path.value().string() << "\n"; } else { std::cout << " CLI probes: skipped"; if (!yamsBinary.note.empty()) { std::cout << " (" << yamsBinary.note << ")"; } std::cout << "\n"; } if (!cliFailureModes.empty()) { printCliProbeFailureSummary(cliFailureModes, cliFailureSamples); } if (!uxErrorCategories.empty()) { std::cout << " UX failure categories:"; for (const auto& [key, count] : uxErrorCategories) { std::cout << " " << key << "=" << count; } std::cout << "\n"; } std::cout << " Slot scaling: initial=" << slotSummary.initialLimit << " peak=" << slotSummary.peakLimit << " final=" << slotSummary.finalLimit << " grow_events=" << slotSummary.growEvents << " shrink_events=" << slotSummary.shrinkEvents << " min_free=" << slotSummary.minFreeSlots << " peak_active=" << slotSummary.peakActiveConnections << " peak_util=" << std::setprecision(2) << (slotSummary.peakUtilization * 100.0) << "% retry_after_samples=" << slotSummary.retryAfterSamples << " observed_growth=" << (observedGrowth ? "yes" : "NO") << " observed_shrink=" << (observedShrink ? "yes" : "NO") << "\n"; printPercentiles("Add latency", addStats); printPercentiles("Search latency", searchStats); printPercentiles("List latency", listStats); printPercentiles("Status latency", statusStats); if (!opStageSummary.empty()) { auto printStageP95 = [&](std::string_view key) { auto it = opStageSummary.find(std::string(key)); if (it != opStageSummary.end() && it->is_object()) { std::cout << " " << key << " p95=" << it->value("p95_us", 0) << "us mean=" << std::fixed << std::setprecision(1) << it->value("mean_us", 0.0) << "us\n"; } }; std::cout << " Direct op overhead trace:\n"; printStageP95("list:phase_dispatch_total_us"); printStageP95("list:phase_dispatch_service_us"); printStageP95("list:phase_dispatch_map_us"); printStageP95("list:client_transport_overhead_us"); printStageP95("status:phase_dispatch_total_us"); printStageP95("status:client_transport_overhead_us"); if (finalSnap.addDispatchSamples > 0) { const auto meanDispatch = static_cast(finalSnap.addDispatchTotalUs) / static_cast(finalSnap.addDispatchSamples); const auto meanFingerprint = static_cast(finalSnap.addFingerprintTotalUs) / static_cast(finalSnap.addDispatchSamples); const auto meanEnqueue = static_cast(finalSnap.addEnqueueTotalUs) / static_cast(finalSnap.addDispatchSamples); std::cout << " add:server_aggregate samples=" << finalSnap.addDispatchSamples << " mean_dispatch=" << std::fixed << std::setprecision(1) << meanDispatch << "us max_dispatch=" << finalSnap.addDispatchMaxUs << "us mean_fingerprint=" << meanFingerprint << "us max_fingerprint=" << finalSnap.addFingerprintMaxUs << "us mean_enqueue=" << meanEnqueue << "us max_enqueue=" << finalSnap.addEnqueueMaxUs << "us\n"; } } if (cliListStats.count + cliSearchStats.count + cliStatusStats.count > 0) { printPercentiles("CLI list latency", cliListStats); printPercentiles("CLI search latency", cliSearchStats); printPercentiles("CLI status latency", cliStatusStats); for (auto& [stage, values] : cliStageLatencies) { auto stageStats = PercentileStats::compute(values); printPercentiles("CLI stage " + stage, stageStats); } } std::cout << " Resource peaks: memory=" << std::fixed << std::setprecision(1) << resourcePeaks.peakMemoryMb << " MB cpu=" << resourcePeaks.peakCpuPercent << "% rss=" << resourcePeaks.peakRssBytes << "B post_q=" << resourcePeaks.peakPostIngestQueued << " post_inflight=" << resourcePeaks.peakPostIngestInflight << " search_active=" << resourcePeaks.peakSearchActive << " search_q=" << resourcePeaks.peakSearchQueued << " db_write_wait=" << resourcePeaks.peakDbWritePoolWaiting << " db_read_wait=" << resourcePeaks.peakDbReadPoolWaiting << " write_apply_ms=" << resourcePeaks.peakWriteMaxBatchApplyMs << " write_wait_ms=" << resourcePeaks.peakWriteMaxBatchQueueWaitMs << " write_excess_wait_ms=" << resourcePeaks.peakWriteMaxBatchExcessQueueWaitMs << " fds=" << resourcePeaks.peakProcessFdCount << " (socket=" << resourcePeaks.peakProcessFdSockets << " pipe=" << resourcePeaks.peakProcessFdPipes << " file=" << resourcePeaks.peakProcessFdFiles << " dev=" << resourcePeaks.peakProcessFdDevices << " kqueue=" << resourcePeaks.peakProcessFdKqueues << " other=" << resourcePeaks.peakProcessFdOther << ")" << " post_content_index_calls=" << resourcePeaks.peakPostCommitContentIndexCalls << " post_content_index_total_ms=" << resourcePeaks.peakPostCommitContentIndexTotalMs << " post_content_index_max_ms=" << resourcePeaks.peakPostCommitContentIndexMaxMs << " client_pool_open=" << resourcePeaks.peakClientPoolOpenSockets << " client_pool_alive=" << resourcePeaks.peakClientPoolAliveConnections << " client_pool_in_use=" << resourcePeaks.peakClientPoolInUseConnections << "\n"; if (!resourcePeaks.lastWriteHotSources.empty()) { std::cout << " Write hot sources: " << resourcePeaks.lastWriteHotSources << "\n"; } if (hotspotSummary.contains("worst_window")) { const auto& window = hotspotSummary["worst_window"]; std::cout << " Worst latency window: start=" << window.value("start_ms", 0) << "ms end=" << window.value("end_ms", 0) << "ms avg=" << window.value("avg_latency_us", 0.0) << "us samples=" << window.value("sample_count", 0) << "\n"; } std::cout << " Queue drained: " << (drained ? "yes" : "NO") << " UX failure rate: " << std::setprecision(3) << uxFailureRate << " Final active/max: " << finalSnap.activeConnections << "/" << finalSnap.maxConnections << "\n\n"; printDrainAndIdleProbeSummary(drainObservation, idleProbe); std::cout << " Recovery quiesced: " << (recoveryQuiesced ? "yes" : "NO") << "\n"; printRecoverySummary(recovery); if (cliRecoveryAttempted) { std::cout << " CLI post-stress recovery: status=" << (cliRecoveryStatus.success ? "ok" : "FAIL") << " list=" << (cliRecoveryList.success ? "ok" : "FAIL") << "\n"; } std::cout << "\n"; // Phase 3: deterministic control cycle for growth/shrink validation on a fresh daemon // instance so realistic phase-2 idle sessions do not distort shrink behavior. harness.stop(); DaemonHarness controlHarness(benchHarnessOptions(cfg)); REQUIRE(startHarnessWithRetry(controlHarness, kStartTimeout)); ClientConfig controlMonCfg; controlMonCfg.socketPath = controlHarness.socketPath(); controlMonCfg.autoStart = false; controlMonCfg.requestTimeout = 10s; DaemonClient controlMonitorClient(controlMonCfg); TimeSeriesSampler controlSampler; controlSampler.start(controlMonitorClient, 100ms); std::atomic controlAddOps{0}; std::atomic controlUxOps{0}; HeldSocketConnections controlHeldConnections; const size_t controlHeldCount = kInitialSlotLimit - 2; const bool controlHeldConnectionsOpened = controlHeldConnections.open( controlHarness.socketPath(), static_cast(controlHeldCount)); INFO("HeldSocketConnections.open control ramp socket=" << controlHarness.socketPath().string() << " requested=" << controlHeldCount << " opened=" << controlHeldConnections.openedCount() << " failed_index=" << controlHeldConnections.lastFailedIndex() << " error='" << controlHeldConnections.lastError() << "'"); REQUIRE(controlHeldConnectionsOpened); const int controlAddsPerWriter = checkedMulToInt(std::max(8, checkedMulToInt(cfg.docsPerClient, 2, kMaxBenchOpsPerClient)), cfg.stressRounds, kMaxBenchOpsPerClient); const int controlUxOpsPerClient = checkedMulToInt(12, cfg.stressRounds, kMaxBenchOpsPerClient); std::thread controlWriter([&]() { ClientConfig ccfg; ccfg.socketPath = controlHarness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); for (int i = 0; i < controlAddsPerWriter; ++i) { AddDocumentRequest req; req.name = "ux_scale_control_add_" + std::to_string(i) + ".md"; req.content = generateDocument(901, i, cfg.docSizeBytes); req.tags = {"bench", "ux-scaling", "control"}; req.collection = "ux_scaling_control"; auto res = yams::cli::run_sync(client.streamingAddDocument(req), 30s); if (res) { controlAddOps.fetch_add(1); } } }); std::thread controlUx([&]() { ClientConfig ccfg; ccfg.socketPath = controlHarness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 10s; DaemonClient client(ccfg); for (int i = 0; i < controlUxOpsPerClient; ++i) { switch (i % 3) { case 0: { (void)yams::cli::run_sync(client.status(), 10s); break; } case 1: { ListRequest req; req.limit = 10; (void)yams::cli::run_sync(client.list(req), 10s); break; } default: { SearchRequest req; req.query = cfg.searchQuery; req.searchType = cfg.searchType; req.limit = cfg.searchLimit; (void)yams::cli::run_sync(client.search(req), 10s); break; } } controlUxOps.fetch_add(1); } }); controlWriter.join(); controlUx.join(); const bool controlGrew = waitForSlotLimit(controlMonitorClient, grownSlotTarget, 6s, true); controlHeldConnections.closeAll(); const bool controlReachedLightLoad = waitForActiveConnectionsAtMost(controlMonitorClient, controlLightLoadTarget, 12s); const bool controlShrank = waitForSlotLimit(controlMonitorClient, kInitialSlotLimit, 10s, false); controlSampler.stop(); const auto controlSamples = controlSampler.getSamples(); const auto controlSlotSummary = summarizeSlotScaling(controlSamples); const auto controlFinalSnap = DaemonSnapshot::capture(controlMonitorClient); std::cout << "=== Deterministic Control Phase ===\n"; std::cout << " Adds: " << controlAddOps.load() << " UX ops: " << controlUxOps.load() << "\n"; std::cout << " Slot scaling: initial=" << controlSlotSummary.initialLimit << " peak=" << controlSlotSummary.peakLimit << " final=" << controlSlotSummary.finalLimit << " grow_events=" << controlSlotSummary.growEvents << " shrink_events=" << controlSlotSummary.shrinkEvents << " peak_active=" << controlSlotSummary.peakActiveConnections << " final_active=" << controlFinalSnap.activeConnections << " final_fds=" << controlFinalSnap.processFdCount << " grew=" << (controlGrew ? "yes" : "NO") << " light_load=" << (controlReachedLightLoad ? "yes" : "NO") << " shrank=" << (controlShrank ? "yes" : "NO") << "\n\n"; CHECK((addOps.load() + addFails.load()) > 0); CHECK(uxOps.load() > 0); CHECK((searchStats.count + listStats.count + statusStats.count) > 0); CHECK(drained); INFO("recoveryQuiesced=" << recoveryQuiesced); CHECK(recovery.ok()); CHECK(cliRecoveryOk); CHECK(unacceptableCliFailures == 0); CHECK(controlAddOps.load() > 0); CHECK(controlUxOps.load() > 0); CHECK(controlGrew); CHECK(controlSlotSummary.peakLimit >= grownSlotTarget); CHECK(controlSlotSummary.growEvents > 0); if (controlReachedLightLoad) { CHECK(controlShrank); CHECK(controlSlotSummary.finalLimit <= kInitialSlotLimit); CHECK(controlSlotSummary.shrinkEvents > 0); } else { WARN("Skipping shrink assertions because control active connections did not reach " "the light-load target; final_active=" << controlFinalSnap.activeConnections << " target=" << controlLightLoadTarget << " final_limit=" << controlSlotSummary.finalLimit << " final_fd_count=" << controlFinalSnap.processFdCount); } json record{ {"timestamp", isoTimestamp()}, {"test", "socket_scaling_vs_ux_latency"}, {"num_writer_clients", writerClients}, {"num_ux_clients", uxClients}, {"docs_per_writer", writerOpsPerClient}, {"ux_ops_per_client", uxOpsPerClient}, {"stress_rounds", cfg.stressRounds}, {"doc_size_bytes", cfg.docSizeBytes}, {"search_config", json{{"daemon", json{{"query", cfg.searchQuery}, {"type", cfg.searchType}, {"limit", cfg.searchLimit}}}, {"cli", json{{"query", cfg.cliSearchQuery}, {"type", cfg.cliSearchType}, {"limit", cfg.cliSearchLimit}}}}}, {"add", json{{"ops", addOps.load()}, {"fails", addFails.load()}, {"throughput_per_sec", addThroughput}, {"latency", addStats.toJson()}}}, {"search", json{{"latency", searchStats.toJson()}}}, {"list", json{{"latency", listStats.toJson()}}}, {"status", json{{"latency", statusStats.toJson()}}}, {"cli", json{{"enabled", cliStressProbesEnabled}, {"binary", yamsBinary.path ? yamsBinary.path.value().string() : std::string()}, {"selection", yamsBinary.toJson()}, {"ops", cliOps.load()}, {"fails", cliFails.load()}, {"ops_per_command", cliProbeOpsPerCommand}, {"failure_buckets", cliFailureBuckets}, {"failure_modes", cliFailureModes}, {"unacceptable_failures", unacceptableCliFailures}, {"failure_samples", cliFailureSamples}, {"probe_results", cliProbeResultsJson}, {"post_stress_recovery", json{{"attempted", cliRecoveryAttempted}, {"ok", cliRecoveryOk}, {"status_probe", cliRecoveryStatus.toJson()}, {"list_probe", cliRecoveryList.toJson()}}}, {"list_latency", cliListStats.toJson()}, {"search_latency", cliSearchStats.toJson()}, {"status_latency", cliStatusStats.toJson()}}}, {"ux", json{{"ops", uxOps.load()}, {"fails", uxFails.load()}, {"throughput_per_sec", uxThroughput}, {"failure_rate", uxFailureRate}, {"retry_after_count", retryAfterCount.load()}, {"error_categories", uxErrorCategories}, {"failure_samples", uxFailureSamples}}}, {"slot_scaling", json{{"initial_limit", slotSummary.initialLimit}, {"peak_limit", slotSummary.peakLimit}, {"final_limit", slotSummary.finalLimit}, {"observed_growth", observedGrowth}, {"observed_shrink", observedShrink}, {"growth_target", grownSlotTarget}, {"final_floor_target", kInitialSlotLimit}, {"min_free_slots", slotSummary.minFreeSlots}, {"peak_active_connections", slotSummary.peakActiveConnections}, {"peak_utilization", slotSummary.peakUtilization}, {"grow_events", slotSummary.growEvents}, {"shrink_events", slotSummary.shrinkEvents}, {"retry_after_samples", slotSummary.retryAfterSamples}}}, {"deterministic_control", json{{"add_ops", controlAddOps.load()}, {"ux_ops", controlUxOps.load()}, {"grew_to_target", controlGrew}, {"reached_light_load", controlReachedLightLoad}, {"shrank_to_floor", controlShrank}, {"slot_scaling", json{{"initial_limit", controlSlotSummary.initialLimit}, {"peak_limit", controlSlotSummary.peakLimit}, {"final_limit", controlSlotSummary.finalLimit}, {"min_free_slots", controlSlotSummary.minFreeSlots}, {"peak_active_connections", controlSlotSummary.peakActiveConnections}, {"peak_utilization", controlSlotSummary.peakUtilization}, {"grow_events", controlSlotSummary.growEvents}, {"shrink_events", controlSlotSummary.shrinkEvents}, {"retry_after_samples", controlSlotSummary.retryAfterSamples}, {"final_active_connections", controlFinalSnap.activeConnections}}}}}, {"resource_peaks", resourcePeaks.toJson()}, {"hotspots", hotspotSummary}, {"op_stage_latency", opStageSummary}, {"elapsed_seconds", globalElapsed}, {"drained", drained}, {"recovery_quiesced", recoveryQuiesced}, {"drain_metrics", drainObservation.toJson()}, {"idle_probe", idleProbe.toJson()}, {"recovery", recovery.toJson()}, {"daemon_snapshot_final", finalSnap.toJson()}, {"time_series_samples", samples.size()}, {"tuning_profile", cfg.tuningProfile}, }; emitJsonl(cfg.outputPath, record); } TEST_CASE("Multi-client ingestion: scaling curve", "[!benchmark][multi-client][scaling]") { // Run the same workload at increasing powers-of-two clients to find scaling limits. auto baseCfg = BenchConfig::fromEnv(); std::vector clientCounts; for (std::int64_t n = 1; n <= baseCfg.scalingMaxClients;) { clientCounts.push_back(static_cast(n)); n *= 2; } if (!clientCounts.empty() && clientCounts.back() != baseCfg.scalingMaxClients) { clientCounts.push_back(baseCfg.scalingMaxClients); } json scalingResults = json::array(); for (int numClients : clientCounts) { std::cout << "\n--- Scaling test: " << numClients << " client(s) ---\n"; DaemonHarness harness(benchHarnessOptions()); REQUIRE(startHarnessWithRetry(harness, kStartTimeout)); ClientConfig monCfg; monCfg.socketPath = harness.socketPath(); monCfg.autoStart = false; monCfg.requestTimeout = 10s; DaemonClient monitorClient(monCfg); std::atomic totalDocs{0}; std::atomic totalFail{0}; std::vector results(numClients); auto globalStart = std::chrono::steady_clock::now(); std::vector threads; for (int c = 0; c < numClients; ++c) { threads.emplace_back([&, c]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); auto start = std::chrono::steady_clock::now(); for (int i = 0; i < baseCfg.docsPerClient; ++i) { AddDocumentRequest req; req.name = "scale_c" + std::to_string(c) + "_d" + std::to_string(i) + ".md"; req.content = generateDocument(c, i, baseCfg.docSizeBytes); req.tags = {"bench", "scaling"}; auto res = yams::cli::run_sync(client.streamingAddDocument(req), 30s); if (res) { results[c].docsIngested++; totalDocs.fetch_add(1); } else { results[c].failures++; totalFail.fetch_add(1); } } auto end = std::chrono::steady_clock::now(); results[c].elapsedSeconds = std::chrono::duration(end - start).count(); results[c].addThroughputDocsPerSec = results[c].docsIngested / results[c].elapsedSeconds; }); } for (auto& t : threads) t.join(); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); double aggThroughput = static_cast(totalDocs.load()) / globalElapsed; bool drained = waitForDrain(monitorClient, std::chrono::seconds(baseCfg.drainTimeoutSecs)); auto finalSnap = DaemonSnapshot::capture(monitorClient); // Per-client throughput stats std::vector perClient; for (auto& r : results) perClient.push_back(r.addThroughputDocsPerSec); double meanPerClient = std::accumulate(perClient.begin(), perClient.end(), 0.0) / static_cast(perClient.size()); std::cout << " Clients: " << numClients << " Aggregate: " << std::fixed << std::setprecision(1) << aggThroughput << " docs/s" << " Per-client mean: " << meanPerClient << " docs/s" << " Failures: " << totalFail.load() << " Memory: " << finalSnap.memoryUsageMb << " MB" << " Drained: " << (drained ? "yes" : "NO") << "\n"; CHECK(totalDocs.load() == checkedMulToInt(numClients, baseCfg.docsPerClient)); scalingResults.push_back(json{ {"num_clients", numClients}, {"aggregate_throughput", aggThroughput}, {"per_client_mean_throughput", meanPerClient}, {"total_docs", totalDocs.load()}, {"total_failures", totalFail.load()}, {"elapsed_seconds", globalElapsed}, {"memory_mb", finalSnap.memoryUsageMb}, {"drained", drained}, }); // Harness destructor will stop the daemon } // Print scaling summary std::cout << "\n=== Scaling Curve Summary ===\n"; std::cout << std::setw(10) << "Clients" << std::setw(18) << "Agg docs/s" << std::setw(18) << "Per-client docs/s" << std::setw(12) << "Efficiency" << "\n"; double baselineThroughput = 0; for (auto& r : scalingResults) { double agg = r["aggregate_throughput"]; double perClient = r["per_client_mean_throughput"]; int n = r["num_clients"]; if (n == 1) baselineThroughput = agg; double efficiency = (baselineThroughput > 0) ? (agg / (baselineThroughput * n)) * 100.0 : 0.0; std::cout << std::setw(10) << n << std::setw(18) << std::fixed << std::setprecision(1) << agg << std::setw(18) << perClient << std::setw(11) << std::setprecision(1) << efficiency << "%\n"; } std::cout << std::endl; json record{ {"timestamp", isoTimestamp()}, {"test", "scaling_curve"}, {"docs_per_client", baseCfg.docsPerClient}, {"doc_size_bytes", baseCfg.docSizeBytes}, {"scaling_results", scalingResults}, {"tuning_profile", baseCfg.tuningProfile}, }; emitJsonl(baseCfg.outputPath, record); } TEST_CASE("Multi-client ingestion: 16-client concurrent mixed ops", "[!benchmark][multi-client][mixed-16]") { // 16 clients performing list, search, status, and get simultaneously while // background writers are ingesting. Measures per-operation-type latency and // throughput under realistic concurrent access patterns. auto cfg = BenchConfig::fromEnv(); const int kWriterClients = cfg.mixedWriterClients; const int kSearchClients = cfg.mixedSearchClients; const int kListClients = cfg.mixedListClients; const int kStatusClients = cfg.mixedStatusClients; const int kTotalClients = checkedAddToInt( {kWriterClients, kSearchClients, kListClients, kStatusClients}, kMaxBenchClients * 4); // Each reader client runs for a fixed number of operations const int kOpsPerReader = cfg.mixedOpsPerReader; DaemonHarness harness(benchHarnessOptions()); REQUIRE(harness.start(kStartTimeout)); // Pre-seed 100 docs so search/list have data from the start std::vector knownHashes; { ClientConfig warmCfg; warmCfg.socketPath = harness.socketPath(); warmCfg.autoStart = false; warmCfg.requestTimeout = 30s; DaemonClient warmClient(warmCfg); std::cout << " Pre-seeding 100 warmup docs...\n"; for (int i = 0; i < 100; ++i) { AddDocumentRequest req; req.name = "mixed16_warmup_" + std::to_string(i) + ".md"; req.content = generateDocument(99, i, cfg.docSizeBytes); req.tags = {"warmup", "bench"}; auto res = yams::cli::run_sync(warmClient.streamingAddDocument(req), 30s); if (res && !res.value().hash.empty()) { knownHashes.push_back(res.value().hash); } } (void)waitForDrain(warmClient, 60s); std::this_thread::sleep_for(1s); std::cout << " Warmup complete (" << knownHashes.size() << " hashes)\n"; } // Monitor client ClientConfig monCfg; monCfg.socketPath = harness.socketPath(); monCfg.autoStart = false; monCfg.requestTimeout = 10s; DaemonClient monitorClient(monCfg); TimeSeriesSampler sampler; sampler.start(monitorClient, 500ms); // Per-thread latency collectors std::mutex searchLatMutex, listLatMutex, statusLatMutex, getLatMutex, addLatMutex; std::vector searchLatencies, listLatencies, statusLatencies, getLatencies, addLatencies; std::map mixedErrorCategories; std::map mixedFailureSamples; std::atomic searchOps{0}, listOps{0}, statusOps{0}, getOps{0}, addOps{0}; std::atomic searchFails{0}, listFails{0}, statusFails{0}, getFails{0}, addFails{0}; // Barrier: all threads start at the same instant std::atomic go{false}; auto globalStart = std::chrono::steady_clock::now(); std::vector threads; threads.reserve(kTotalClients); // --- Writer threads (4): continuous ingestion --- for (int w = 0; w < kWriterClients; ++w) { threads.emplace_back([&, w]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); std::map localErrorCategories; std::map localFailureSamples; while (!go.load()) std::this_thread::yield(); for (int i = 0; i < cfg.docsPerClient; ++i) { AddDocumentRequest req; req.name = "mixed16_w" + std::to_string(w) + "_" + std::to_string(i) + ".md"; req.content = generateDocument(w, i, cfg.docSizeBytes); req.tags = {"bench", "mixed16", "writer-" + std::to_string(w)}; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.streamingAddDocument(req), 30s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { addOps.fetch_add(1); std::lock_guard lk(addLatMutex); addLatencies.push_back(latUs); } else { addFails.fetch_add(1); const auto bucket = classifyFailureMessage(res.error().message); const auto key = std::string("add:") + bucket; localErrorCategories[key] += 1; if (!res.error().message.empty() && !localFailureSamples.contains(key)) { localFailureSamples[key] = res.error().message; } } } if (!localErrorCategories.empty()) { std::lock_guard lk(addLatMutex); for (const auto& [key, count] : localErrorCategories) { mixedErrorCategories[key] += count; } for (const auto& [key, sample] : localFailureSamples) { if (!mixedFailureSamples.contains(key)) { mixedFailureSamples[key] = sample; } } } }); } // --- Search threads (4): continuous search queries --- for (int s = 0; s < kSearchClients; ++s) { threads.emplace_back([&]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 10s; DaemonClient client(ccfg); std::map localErrorCategories; std::map localFailureSamples; while (!go.load()) std::this_thread::yield(); static const char* queries[] = { "architecture performance tuning", "ingestion pipeline extraction", "resource governance memory pressure", "thread pool concurrency limits", "IPC transport protocol", }; for (int i = 0; i < kOpsPerReader; ++i) { SearchRequest req; req.query = queries[i % 5]; req.limit = 10; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.search(req), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { searchOps.fetch_add(1); std::lock_guard lk(searchLatMutex); searchLatencies.push_back(latUs); } else { searchFails.fetch_add(1); const auto bucket = classifyFailureMessage(res.error().message); const auto key = std::string("search:") + bucket; localErrorCategories[key] += 1; if (!res.error().message.empty() && !localFailureSamples.contains(key)) { localFailureSamples[key] = res.error().message; } } } if (!localErrorCategories.empty()) { std::lock_guard lk(searchLatMutex); for (const auto& [key, count] : localErrorCategories) { mixedErrorCategories[key] += count; } for (const auto& [key, sample] : localFailureSamples) { if (!mixedFailureSamples.contains(key)) { mixedFailureSamples[key] = sample; } } } }); } // --- List threads (4): continuous list queries with varying filters --- for (int l = 0; l < kListClients; ++l) { threads.emplace_back([&]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 10s; DaemonClient client(ccfg); std::map localErrorCategories; std::map localFailureSamples; while (!go.load()) std::this_thread::yield(); for (int i = 0; i < kOpsPerReader; ++i) { ListRequest req; req.limit = (i % 3 == 0) ? 5 : (i % 3 == 1) ? 20 : 50; // Alternate between filtered and unfiltered if (i % 4 == 0) { req.tags = {"warmup"}; } auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.list(req), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { listOps.fetch_add(1); std::lock_guard lk(listLatMutex); listLatencies.push_back(latUs); } else { listFails.fetch_add(1); const auto bucket = classifyFailureMessage(res.error().message); const auto key = std::string("list:") + bucket; localErrorCategories[key] += 1; if (!res.error().message.empty() && !localFailureSamples.contains(key)) { localFailureSamples[key] = res.error().message; } } } if (!localErrorCategories.empty()) { std::lock_guard lk(listLatMutex); for (const auto& [key, count] : localErrorCategories) { mixedErrorCategories[key] += count; } for (const auto& [key, sample] : localFailureSamples) { if (!mixedFailureSamples.contains(key)) { mixedFailureSamples[key] = sample; } } } }); } // --- Status threads (4): continuous status + get queries --- for (int t = 0; t < kStatusClients; ++t) { threads.emplace_back([&, t]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 10s; DaemonClient client(ccfg); std::mt19937 rng(static_cast(t * 997 + 13)); std::map localErrorCategories; std::map localFailureSamples; while (!go.load()) std::this_thread::yield(); for (int i = 0; i < kOpsPerReader; ++i) { if (i % 3 == 0) { // Status query auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.status(), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { statusOps.fetch_add(1); std::lock_guard lk(statusLatMutex); statusLatencies.push_back(latUs); } else { statusFails.fetch_add(1); const auto bucket = classifyFailureMessage(res.error().message); const auto key = std::string("status:") + bucket; localErrorCategories[key] += 1; if (!res.error().message.empty() && !localFailureSamples.contains(key)) { localFailureSamples[key] = res.error().message; } } } else { // Get (metadata) by hash std::string hash; { if (!knownHashes.empty()) { std::uniform_int_distribution dist(0, knownHashes.size() - 1); hash = knownHashes[dist(rng)]; } } if (hash.empty()) { statusOps.fetch_add(1); // count as no-op continue; } GetRequest req; req.hash = hash; req.metadataOnly = true; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.get(req), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { getOps.fetch_add(1); std::lock_guard lk(getLatMutex); getLatencies.push_back(latUs); } else { getFails.fetch_add(1); const auto bucket = classifyFailureMessage(res.error().message); const auto key = std::string("get:") + bucket; localErrorCategories[key] += 1; if (!res.error().message.empty() && !localFailureSamples.contains(key)) { localFailureSamples[key] = res.error().message; } } } } if (!localErrorCategories.empty()) { std::lock_guard lk(statusLatMutex); for (const auto& [key, count] : localErrorCategories) { mixedErrorCategories[key] += count; } for (const auto& [key, sample] : localFailureSamples) { if (!mixedFailureSamples.contains(key)) { mixedFailureSamples[key] = sample; } } } }); } // Release all threads simultaneously go.store(true); for (auto& t : threads) t.join(); sampler.stop(); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); // Drain bool drained = waitForDrain(monitorClient, std::chrono::seconds(cfg.drainTimeoutSecs)); auto finalSnap = DaemonSnapshot::capture(monitorClient); auto resourcePeaks = summarizeResourcePeaks(sampler.getSamples()); // Compute stats auto searchStats = PercentileStats::compute(searchLatencies); auto listStats = PercentileStats::compute(listLatencies); auto statusStats = PercentileStats::compute(statusLatencies); auto getStats = PercentileStats::compute(getLatencies); auto addStats = PercentileStats::compute(addLatencies); int totalOps = searchOps.load() + listOps.load() + statusOps.load() + getOps.load() + addOps.load(); int totalFails = searchFails.load() + listFails.load() + statusFails.load() + getFails.load() + addFails.load(); double opsPerSec = static_cast(totalOps) / globalElapsed; double failRate = (totalOps > 0) ? (static_cast(totalFails) / totalOps) : 0.0; const std::string stressTier = !drained ? "failed" : (totalFails == 0 ? "clean" : "degraded"); // Print summary std::cout << "\n=== " << kTotalClients << "-Client Concurrent Mixed Ops ===\n"; std::cout << " Layout: " << kWriterClients << " writers + " << kSearchClients << " searchers" << " + " << kListClients << " listers + " << kStatusClients << " status/get\n"; std::cout << " Wall time: " << std::fixed << std::setprecision(2) << globalElapsed << "s\n"; std::cout << " Total ops: " << totalOps << " (" << std::setprecision(0) << opsPerSec << " ops/s)" << " Failures: " << totalFails << " (rate=" << std::setprecision(4) << failRate << ")\n\n"; std::cout << " Op breakdown:\n"; std::cout << " Add: " << addOps.load() << " ok, " << addFails.load() << " fail\n"; std::cout << " Search: " << searchOps.load() << " ok, " << searchFails.load() << " fail\n"; std::cout << " List: " << listOps.load() << " ok, " << listFails.load() << " fail\n"; std::cout << " Status: " << statusOps.load() << " ok, " << statusFails.load() << " fail\n"; std::cout << " Get: " << getOps.load() << " ok, " << getFails.load() << " fail\n\n"; if (!mixedErrorCategories.empty()) { std::cout << " Failure categories:"; for (const auto& [key, count] : mixedErrorCategories) { std::cout << " " << key << "=" << count; } std::cout << "\n\n"; } printPercentiles("Add latency ", addStats); printPercentiles("Search latency", searchStats); printPercentiles("List latency ", listStats); printPercentiles("Status latency", statusStats); printPercentiles("Get latency ", getStats); std::cout << "\n Queue drained: " << (drained ? "yes" : "NO") << "\n"; std::cout << " Peak DB read/write waiting: " << resourcePeaks.peakDbReadPoolWaiting << "/" << resourcePeaks.peakDbWritePoolWaiting << "\n"; std::cout << " Final docs_total: " << finalSnap.documentsTotal << "\n"; std::cout << " Final memory: " << finalSnap.memoryUsageMb << " MB\n\n"; // Key assertions: all op types must make progress CHECK(addOps.load() > 0); CHECK(searchOps.load() > 0); CHECK(listOps.load() > 0); CHECK(statusOps.load() > 0); CHECK(getOps.load() > 0); CHECK(drained); CHECK(failRate <= cfg.mixedMaxFailRate); // Emit JSONL json timeSeries = json::array(); for (auto& s : sampler.getSamples()) { timeSeries.push_back(json{{"elapsed_ms", s.elapsedMs}, {"snapshot", s.snap.toJson()}}); } json record{ {"timestamp", isoTimestamp()}, {"test", "16_client_concurrent_mixed_ops"}, {"layout", json{{"writers", kWriterClients}, {"searchers", kSearchClients}, {"listers", kListClients}, {"status_get", kStatusClients}}}, {"ops_per_reader", kOpsPerReader}, {"docs_per_writer", cfg.docsPerClient}, {"doc_size_bytes", cfg.docSizeBytes}, {"total_ops", totalOps}, {"total_failures", totalFails}, {"fail_rate", failRate}, {"stress_tier", stressTier}, {"elapsed_seconds", globalElapsed}, {"ops_per_sec", opsPerSec}, {"add", json{{"ops", addOps.load()}, {"fails", addFails.load()}, {"latency", addStats.toJson()}}}, {"search", json{{"ops", searchOps.load()}, {"fails", searchFails.load()}, {"latency", searchStats.toJson()}}}, {"list", json{{"ops", listOps.load()}, {"fails", listFails.load()}, {"latency", listStats.toJson()}}}, {"status", json{{"ops", statusOps.load()}, {"fails", statusFails.load()}, {"latency", statusStats.toJson()}}}, {"get", json{{"ops", getOps.load()}, {"fails", getFails.load()}, {"latency", getStats.toJson()}}}, {"daemon_snapshot_final", finalSnap.toJson()}, {"resource_peaks", resourcePeaks.toJson()}, {"error_categories", mixedErrorCategories}, {"failure_samples", mixedFailureSamples}, {"time_series", timeSeries}, {"tuning_profile", cfg.tuningProfile}, {"drained", drained}, }; emitJsonl(cfg.outputPath, record); } // ============================================================================= // EXISTING CORPUS BENCHMARK // ============================================================================= TEST_CASE("Multi-client ingestion: large corpus reads", "[!benchmark][multi-client][large-corpus]") { // Instrumented benchmark against an existing YAMS corpus (e.g. /Volumes/picaso/yams // with 48k+ docs). Does NOT ingest — hammers search/list/get/status against the // real corpus to measure read-path performance. Outputs: // - Per-op latency traces (every operation recorded) // - Live progress with rolling throughput // - Memory watermarks (peak RSS, delta over time) // - Error classification breakdown // - Detailed JSONL with all latency traces for post-analysis // // Env overrides: // YAMS_BENCH_DATA_DIR - required: existing corpus path // YAMS_BENCH_SEARCH_CLIENTS - search thread count (default: 4) // YAMS_BENCH_LIST_CLIENTS - list thread count (default: 4) // YAMS_BENCH_GREP_CLIENTS - grep thread count (default: 2) // YAMS_BENCH_STATUS_GET_CLIENTS - status/get thread count (default: 4) // YAMS_BENCH_OPS_PER_CLIENT - ops per thread (default: 50) // YAMS_BENCH_OP_TIMEOUT_S - per-op timeout in seconds (default: 60) // YAMS_BENCH_STATUS_EVERY_N - status cadence in status/get threads (default: 4) // YAMS_BENCH_CAT_EVERY_N - cat cadence among non-status status/get ops // (default: 3) // YAMS_BENCH_LIST_INFLIGHT_CAP - max concurrent list ops (0 = unlimited, default: 0) // YAMS_BENCH_USAGE_PROFILE - mixed|extension_direct|external_agent_churn // (default: mixed) // YAMS_BENCH_MCP_SESSION_CHURN_EVERY_N - MCP only; every Nth request uses a // transient MCP session (default: 6 for // external_agent_churn, otherwise 0) auto cfg = BenchConfig::fromEnv(); if (!cfg.dataDir) { SKIP("Set YAMS_BENCH_DATA_DIR to an existing YAMS corpus to run this benchmark"); } const bool extensionLikeProfile = (cfg.usageProfile == "extension_direct" || cfg.usageProfile == "vscode_blackboard"); const bool externalAgentChurnProfile = (cfg.usageProfile == "external_agent_churn" || cfg.usageProfile == "agent_churn"); // --- Configurable layout --- const int defaultSearchClients = externalAgentChurnProfile ? 4 : (extensionLikeProfile ? 6 : 4); const int defaultListClients = externalAgentChurnProfile ? 4 : (extensionLikeProfile ? 6 : 4); const int defaultGrepClients = externalAgentChurnProfile ? 1 : (extensionLikeProfile ? 1 : 2); const int defaultStatusGetClients = externalAgentChurnProfile ? 12 : (extensionLikeProfile ? 10 : 4); const int defaultOpsPerClient = externalAgentChurnProfile ? 90 : (extensionLikeProfile ? 80 : 50); const int defaultStatusEveryN = externalAgentChurnProfile ? 2 : (extensionLikeProfile ? 8 : 4); const int defaultCatEveryN = externalAgentChurnProfile ? 14 : (extensionLikeProfile ? 10 : 3); const int defaultListInflightCap = externalAgentChurnProfile ? 1 : (extensionLikeProfile ? 2 : 0); const int kSearchClients = parseEnvIntClamped("YAMS_BENCH_SEARCH_CLIENTS", defaultSearchClients, 0, kMaxBenchClients); const int kListClients = parseEnvIntClamped("YAMS_BENCH_LIST_CLIENTS", defaultListClients, 0, kMaxBenchClients); const int kGrepClients = parseEnvIntClamped("YAMS_BENCH_GREP_CLIENTS", defaultGrepClients, 0, kMaxBenchClients); const int kStatusGetClients = parseEnvIntClamped("YAMS_BENCH_STATUS_GET_CLIENTS", defaultStatusGetClients, 0, kMaxBenchClients); const int kTotalClients = checkedAddToInt( {kSearchClients, kListClients, kGrepClients, kStatusGetClients}, kMaxBenchClients * 4); const int kOpsPerClient = parseEnvIntClamped("YAMS_BENCH_OPS_PER_CLIENT", defaultOpsPerClient, 1, kMaxBenchOpsPerClient); const int kOpTimeoutS = parseEnvIntClamped("YAMS_BENCH_OP_TIMEOUT_S", 60, 1, kMaxBenchSeconds); const auto kOpTimeout = std::chrono::seconds(kOpTimeoutS); const int kStatusEveryN = parseEnvIntClamped("YAMS_BENCH_STATUS_EVERY_N", defaultStatusEveryN, 1, kMaxBenchOpsPerClient); const int kCatEveryN = parseEnvIntClamped("YAMS_BENCH_CAT_EVERY_N", defaultCatEveryN, 1, kMaxBenchOpsPerClient); const int kListInflightCap = parseEnvIntClamped("YAMS_BENCH_LIST_INFLIGHT_CAP", defaultListInflightCap, 0, kMaxBenchClients); const int kDiscoverPageSize = parseEnvIntClamped("YAMS_BENCH_DISCOVER_PAGE_SIZE", 500, 1, kMaxBenchPageSize); const int kDiscoverHashLimit = parseEnvIntClamped("YAMS_BENCH_DISCOVER_HASH_LIMIT", 5000, 1, kMaxBenchHashLimit); const int kHotspotWindowMs = parseEnvIntClamped("YAMS_BENCH_HOTSPOT_WINDOW_MS", 1000, 1, kMaxBenchMilliseconds); const bool useMcpPath = cfg.useMcpPath; const bool useSessionGrep = parseEnvIntClamped("YAMS_BENCH_GREP_USE_SESSION", 0, 0, 1) != 0; const int defaultMcpSessionChurnEveryN = (externalAgentChurnProfile && useMcpPath) ? 6 : 0; const int kMcpSessionChurnEveryN = parseEnvIntClamped("YAMS_BENCH_MCP_SESSION_CHURN_EVERY_N", defaultMcpSessionChurnEveryN, 0, kMaxBenchOpsPerClient); const size_t mcpPoolSize = static_cast(parseEnvIntClamped( "YAMS_BENCH_MCP_POOL_SIZE", std::max(1, kTotalClients), 1, kMaxBenchClients * 4)); std::cout << "\n=== Large Corpus Read Benchmark (instrumented) ===\n"; std::cout << " Data dir: " << cfg.dataDir->string() << "\n"; std::cout << " Layout: " << kSearchClients << " search + " << kListClients << " list + " << kGrepClients << " grep + " << kStatusGetClients << " status/get/cat = " << kTotalClients << " threads\n"; std::cout << " Ops/thread: " << kOpsPerClient << "\n"; std::cout << " Op timeout: " << kOpTimeoutS << "s\n"; std::cout << " Status every: " << kStatusEveryN << " op(s) on status/get/cat threads\n"; std::cout << " Cat every: " << kCatEveryN << " non-status op(s) on status/get/cat threads\n"; std::cout << " List cap: " << (kListInflightCap > 0 ? std::to_string(kListInflightCap) : "unlimited") << " in-flight\n"; std::cout << " Discover page: " << kDiscoverPageSize << " (limit " << kDiscoverHashLimit << ")\n"; std::cout << " Hotspot window: " << kHotspotWindowMs << "ms\n"; std::cout << " Usage profile: " << cfg.usageProfile << "\n"; std::cout << " Transport path: " << (useMcpPath ? "mcp(query-or-direct-tools)" : "daemon-ipc") << "\n"; if (useMcpPath) { std::cout << " MCP pool size: " << mcpPoolSize << "\n"; std::cout << " MCP churn: every " << (kMcpSessionChurnEveryN > 0 ? std::to_string(kMcpSessionChurnEveryN) : std::string("disabled")) << " request(s) uses a fresh MCP session\n"; } if (kGrepClients > 0) { std::cout << " Grep mode: " << (useSessionGrep ? "session hot path" : "global cold path") << "\n"; } // --- Server-side timeout configuration --- // Stay within TuneAdvisor range clamps: // ipcTimeoutMs: [500, 600000] // streamChunkTimeoutMs: [1000, 600000] // searchBuildTimeout: unclamped std::string timeoutStr = std::to_string(std::min(checkedMulToInt(kOpTimeoutS, 2000, 600000), 600000)); std::vector timeoutEnvironment; if (!yams::config::getenv_nonempty("YAMS_SEARCH_BUILD_TIMEOUT_MS")) timeoutEnvironment.emplace_back("YAMS_SEARCH_BUILD_TIMEOUT_MS", std::string{"120000"}); if (!yams::config::getenv_nonempty("YAMS_IPC_TIMEOUT_MS")) timeoutEnvironment.emplace_back("YAMS_IPC_TIMEOUT_MS", timeoutStr); if (!yams::config::getenv_nonempty("YAMS_STREAM_CHUNK_TIMEOUT_MS")) timeoutEnvironment.emplace_back("YAMS_STREAM_CHUNK_TIMEOUT_MS", timeoutStr); const auto ipcTimeout = yams::config::getenv_nonempty("YAMS_IPC_TIMEOUT_MS"); const auto streamTimeout = yams::config::getenv_nonempty("YAMS_STREAM_CHUNK_TIMEOUT_MS"); std::cout << " IPC timeout: " << ipcTimeout.value_or("default") << " ms\n"; std::cout << " Stream timeout: " << streamTimeout.value_or("default") << " ms\n"; // --- Start daemon --- auto opts = benchHarnessOptions(cfg); DaemonHarness harness(opts); REQUIRE(harness.start(std::chrono::seconds(300))); std::this_thread::sleep_for(3s); // let async init settle // Helper: create a client config with all timeouts aligned to kOpTimeout. // The default headerTimeout (30s) was causing cascading 30s stalls because // the client would time out waiting for the first response frame header. auto makeClientConfig = [&](std::chrono::milliseconds timeout = std::chrono::milliseconds{0}) { ClientConfig c; c.socketPath = harness.socketPath(); c.autoStart = false; auto t = timeout.count() > 0 ? timeout : kOpTimeout; c.requestTimeout = t; c.headerTimeout = t; c.bodyTimeout = t; return c; }; // --- Discover corpus --- auto discoverCfg = makeClientConfig(); DaemonClient discoverClient(discoverCfg); std::shared_ptr sharedMcpPool; if (useMcpPath) { sharedMcpPool = std::make_shared(harness.socketPath(), mcpPoolSize); } std::atomic mcpQueryCalls{0}; std::atomic mcpTransientSessionCalls{0}; auto queryMcpStepForSlot = [&](size_t slot, const std::string& op, const nlohmann::json& params) -> yams::Result { if (!sharedMcpPool) { return yams::Error{yams::ErrorCode::InvalidState, "MCP pool is not initialized for MCP benchmark path"}; } const auto callNumber = mcpQueryCalls.fetch_add(1, std::memory_order_relaxed) + static_cast(1); if (kMcpSessionChurnEveryN <= 0) { return sharedMcpPool->queryStepForSlot(slot, op, params); } const bool useTransient = (callNumber % static_cast(kMcpSessionChurnEveryN) == 0); if (!useTransient) { return sharedMcpPool->queryStepForSlot(slot, op, params); } mcpTransientSessionCalls.fetch_add(1, std::memory_order_relaxed); try { MCPPipelineClient transientClient; return transientClient.queryStep(op, params); } catch (const std::exception& e) { return yams::Error{yams::ErrorCode::InvalidState, std::string("Transient MCP session failed: ") + e.what()}; } }; auto setupBenchmarkSession = [&](const std::string& sessionName) -> bool { // Large-corpus grep benchmarks use sessions to exercise the hot path, not to scope // results to the YAMS data directory itself. Adding a watched selector rooted at the // storage dir narrows MCP grep to internal corpus files and stops matching the direct // daemon hot-path workload. if (useMcpPath) { auto startRes = queryMcpStepForSlot(0, "session_start", {{"name", sessionName}, {"description", "large corpus benchmark session"}, {"warm", false}, {"limit", 0}}); if (!startRes) { WARN("session_start failed: " << startRes.error().message); return false; } return true; } yams::test::ScopedEnvVar socketEnvironment{"YAMS_DAEMON_SOCKET_PATH", harness.socketPath().string()}; yams::test::ScopedEnvVar autoStartEnvironment{"YAMS_CLI_DISABLE_DAEMON_AUTOSTART", std::string{"1"}}; MCPPipelineClient sessionClient; auto startRes = sessionClient.queryStep("session_start", {{"name", sessionName}, {"description", "large corpus benchmark session"}, {"warm", false}, {"limit", 0}}); if (!startRes) { WARN("session_start failed: " << startRes.error().message); return false; } return true; }; std::vector knownHashes; { auto discoverDirectPage = [&](std::int64_t offset, int& itemsInPage) -> bool { if (offset > std::numeric_limits::max()) { WARN("Discovery list offset exceeds ListRequest range: " << offset); return false; } ListRequest listReq; listReq.limit = kDiscoverPageSize; listReq.offset = static_cast(offset); auto listRes = yams::cli::run_sync(discoverClient.list(listReq), kOpTimeout); if (!listRes) { WARN("Discovery list failed at offset " << offset << ": " << listRes.error().message); return false; } for (auto& item : listRes.value().items) { if (!item.hash.empty()) { knownHashes.push_back(item.hash); ++itemsInPage; if (static_cast(knownHashes.size()) >= kDiscoverHashLimit) { break; } } } return true; }; bool useDirectDiscoveryFallback = false; bool done = false; for (std::int64_t offset = 0; !done && static_cast(knownHashes.size()) < kDiscoverHashLimit; offset += static_cast(kDiscoverPageSize)) { int itemsInPage = 0; if (useMcpPath && !useDirectDiscoveryFallback) { auto listRes = queryMcpStepForSlot( 0, "list", {{"limit", kDiscoverPageSize}, {"offset", offset}, {"use_session", false}}); if (!listRes) { WARN("Discovery list failed at offset " << offset << ": " << listRes.error().message); break; } auto collectHashes = [&](const nlohmann::json& arr) { if (!arr.is_array()) { return; } for (const auto& item : arr) { if (!item.is_object()) { continue; } std::string hash; if (item.contains("hash") && item["hash"].is_string()) { hash = item["hash"].get(); } else if (item.contains("id") && item["id"].is_string()) { hash = item["id"].get(); } else if (item.contains("document_hash") && item["document_hash"].is_string()) { hash = item["document_hash"].get(); } if (!hash.empty()) { knownHashes.push_back(std::move(hash)); ++itemsInPage; if (static_cast(knownHashes.size()) >= kDiscoverHashLimit) { break; } } } }; const auto& payload = listRes.value(); if (payload.contains("items")) { collectHashes(payload["items"]); } // MCP list handler returns "documents" key (not "items") if (itemsInPage == 0 && payload.contains("documents")) { collectHashes(payload["documents"]); } if (itemsInPage == 0 && payload.contains("results")) { collectHashes(payload["results"]); } if (offset == 0 && itemsInPage == 0) { int directItemsInPage = 0; if (discoverDirectPage(offset, directItemsInPage) && directItemsInPage > 0) { useDirectDiscoveryFallback = true; itemsInPage = directItemsInPage; INFO("Discovery fallback: using direct list() pagination because MCP list " "did not return discoverable hashes."); } } } else { if (!discoverDirectPage(offset, itemsInPage)) { break; } } if (itemsInPage < kDiscoverPageSize) { done = true; } } std::sort(knownHashes.begin(), knownHashes.end()); knownHashes.erase(std::unique(knownHashes.begin(), knownHashes.end()), knownHashes.end()); } auto initSnap = DaemonSnapshot::capture(discoverClient); std::cout << " Corpus docs: " << initSnap.documentsTotal << "\n"; std::cout << " Known hashes: " << knownHashes.size() << "\n"; std::cout << " Initial RSS: " << initSnap.memoryUsageMb << " MB\n"; if (knownHashes.empty() && initSnap.documentsTotal == 0) { SKIP("Large-corpus benchmark requires a non-empty corpus: discovery returned 0 hashes or " "daemon reports 0 documents. Verify YAMS_BENCH_DATA_DIR points to populated data."); } const std::string benchmarkSessionName = "bench_large_corpus_" + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()); if (useSessionGrep) { REQUIRE(setupBenchmarkSession(benchmarkSessionName)); } // ───────────────────────────────────────────────────────────────────────── // Phase 1: Single-client warmup — baseline latency without contention // ───────────────────────────────────────────────────────────────────────── std::cout << "\n--- Phase 1: Single-client warmup (5 ops each) ---\n"; { auto warmCfg = makeClientConfig(); DaemonClient warmClient(warmCfg); // Warm search for (int i = 0; i < 5; ++i) { auto t0 = std::chrono::steady_clock::now(); yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot( 0, "search", {{"query", "architecture design"}, {"limit", 10}, {"type", cfg.searchType}}); } else { SearchRequest req; req.query = "architecture design"; req.limit = 10; req.searchType = cfg.searchType; auto direct = yams::cli::run_sync(warmClient.search(req), kOpTimeout); if (direct) { nlohmann::json j; j["results"] = nlohmann::json::array(); for (const auto& r : direct.value().results) { j["results"].push_back({{"id", r.id}, {"path", r.path}}); } res = j; } else { res = yams::Error{direct.error()}; } } auto us = std::chrono::duration_cast( std::chrono::steady_clock::now() - t0) .count(); std::cout << " search[" << i << "]: " << us / 1000 << "ms" << (res ? "" : " FAIL") << "\n"; } // Warm list for (int i = 0; i < 5; ++i) { auto t0 = std::chrono::steady_clock::now(); yams::Result res; int items = 0; if (useMcpPath) { res = queryMcpStepForSlot(0, "list", {{"limit", 50}, {"use_session", false}}); if (res && res.value().contains("items") && res.value()["items"].is_array()) { items = static_cast(res.value()["items"].size()); } else if (res && res.value().contains("documents") && res.value()["documents"].is_array()) { items = static_cast(res.value()["documents"].size()); } } else { ListRequest req; req.limit = 50; auto direct = yams::cli::run_sync(warmClient.list(req), kOpTimeout); if (direct) { items = static_cast(direct.value().items.size()); nlohmann::json j; j["items"] = nlohmann::json::array(); res = j; } else { res = yams::Error{direct.error()}; } } auto us = std::chrono::duration_cast( std::chrono::steady_clock::now() - t0) .count(); std::cout << " list[" << i << "]: " << us / 1000 << "ms (" << items << " items)" << (res ? "" : " FAIL") << "\n"; } // Warm get if (!knownHashes.empty()) { for (int i = 0; i < 5 && i < static_cast(knownHashes.size()); ++i) { GetRequest req; req.hash = knownHashes[i]; req.metadataOnly = true; auto t0 = std::chrono::steady_clock::now(); yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot( 0, "get", {{"hash", knownHashes[i]}, {"include_content", false}}); } else { auto direct = yams::cli::run_sync(warmClient.get(req), kOpTimeout); if (direct) { res = nlohmann::json::object(); } else { res = yams::Error{direct.error()}; } } auto us = std::chrono::duration_cast( std::chrono::steady_clock::now() - t0) .count(); std::cout << " get[" << i << "]: " << us / 1000 << "ms" << (res ? "" : " FAIL") << "\n"; } } } // ───────────────────────────────────────────────────────────────────────── // Phase 2: Concurrent load with per-op tracing // ───────────────────────────────────────────────────────────────────────── std::cout << "\n--- Phase 2: Concurrent load (" << kTotalClients << " threads x " << kOpsPerClient << " ops) ---\n"; // Per-op trace record struct OpTrace { std::string opType; // "search", "list", "grep", "status", "get", "cat" int threadId{0}; int opIndex{0}; int64_t latencyUs{0}; bool success{false}; std::string errorMsg; int resultCount{0}; // items returned (search hits, list items) int64_t wallClockMs{0}; // ms since benchmark start }; // Monitor with sampler auto monCfg = makeClientConfig(10s); DaemonClient monitorClient(monCfg); TimeSeriesSampler sampler; sampler.start(monitorClient, 1s); // Per-op trace collectors (one per thread for lock-free append) std::vector> perThreadTraces(kTotalClients); // Shared counters std::atomic searchOps{0}, listOps{0}, grepOps{0}, statusOps{0}, getOps{0}, catOps{0}; std::atomic searchFails{0}, listFails{0}, grepFails{0}, statusFails{0}, getFails{0}, catFails{0}; std::atomic totalCompleted{0}; std::atomic listInFlight{0}; const int totalExpected = checkedMulToInt(kTotalClients, kOpsPerClient); std::mutex invalidHashMutex; std::unordered_set invalidHashes; auto markHashInvalid = [&](const std::string& hash) { if (hash.empty()) { return; } std::lock_guard lk(invalidHashMutex); invalidHashes.insert(hash); }; auto sampleKnownHash = [&](std::mt19937& rng) -> std::optional { if (knownHashes.empty()) { return std::nullopt; } std::uniform_int_distribution dist(0, knownHashes.size() - 1); constexpr int kPickAttempts = 8; for (int attempt = 0; attempt < kPickAttempts; ++attempt) { std::string candidate = knownHashes[dist(rng)]; bool isInvalid = false; { std::lock_guard lk(invalidHashMutex); isInvalid = invalidHashes.find(candidate) != invalidHashes.end(); } if (!isInvalid) { return candidate; } } return std::nullopt; }; // Grep server-side telemetry sampled from grep responses std::mutex grepServerStatsMutex; std::vector grepExecMsSamples; std::vector grepDocsScannedSamples; std::vector grepLinesScannedSamples; std::vector grepBytesScannedSamples; std::vector grepContentRetrievalMsSamples; std::vector grepMatchRenderUsSamples; std::vector grepWorkerScanMsSamples; std::vector grepRegexSearchCallsSamples; std::vector grepBmhPrefilterSkipsSamples; std::vector grepCandidateDiscoveryMsSamples; std::vector grepPathFilteringMsSamples; std::vector grepCandidateClassificationMsSamples; std::vector grepMetadataBatchFetchMsSamples; std::vector grepMetadataEnumerationMsSamples; std::vector grepRegexScanMsSamples; std::vector grepContentDecodeEstimateMsSamples; std::vector grepWorkerCriticalTotalMsSamples; std::vector grepWorkerCriticalRetrievalMsSamples; std::vector grepWorkerCriticalRegexMsSamples; std::vector grepWorkerCriticalOtherMsSamples; // Search/list server-side telemetry sampled from responses std::vector searchExecMsSamples; std::vector searchDispatchServiceMsSamples; std::vector searchDispatchServiceUsSamples; std::vector searchDispatchMapSortMsSamples; std::vector searchDispatchMapSortUsSamples; std::vector searchDispatchFeedbackMsSamples; std::vector searchDispatchFeedbackUsSamples; std::vector searchDispatchResponseMsSamples; std::vector searchDispatchResponseUsSamples; std::vector searchDispatchTotalMsSamples; std::vector searchDispatchTotalUsSamples; std::vector searchTimingEmbeddingUsSamples; std::vector searchTimingTextUsSamples; std::vector searchTimingPathUsSamples; std::vector searchTimingVectorUsSamples; std::vector searchTimingEntityVectorUsSamples; std::vector searchTimingCompressedAnnUsSamples; std::vector searchTimingKgUsSamples; std::vector searchTimingMetadataUsSamples; std::vector searchTimingTagUsSamples; std::vector searchTimingMultiVectorUsSamples; std::vector searchTimingConceptsUsSamples; std::vector searchTimingRerankerUsSamples; std::vector searchTimingGraphRerankUsSamples; std::vector searchSubphraseFtsHitSamples; std::vector searchAggressiveFtsHitSamples; std::vector searchGraphExpansionFtsHitSamples; std::vector searchGraphDocExpansionFtsHitSamples; std::vector searchMultiVectorRawHitSamples; std::vector searchGraphVectorRawHitSamples; std::vector searchGraphQueryNeighborSeedDocSamples; std::vector listServiceExecMsSamples; std::vector listDispatchServiceMsSamples; std::vector listDispatchServiceUsSamples; std::vector listDispatchMapMsSamples; std::vector listDispatchMapUsSamples; std::vector listDispatchTotalMsSamples; std::vector listDispatchTotalUsSamples; std::vector searchDebugSamples; auto collectSearchServerTelemetry = [&](const nlohmann::json& payload) { const auto searchStats = [&]() -> nlohmann::json { if (payload.contains("search_stats")) { return payload["search_stats"]; } if (payload.contains("searchStats")) { return payload["searchStats"]; } return nlohmann::json::object(); }(); std::lock_guard lk(grepServerStatsMutex); if (searchDebugSamples.size() < 3 && !searchStats.empty()) { searchDebugSamples.push_back(searchStats); } if (auto v = jsonGetInt64AnyKey(payload, {"execution_time_ms", "executionTimeMs"})) { searchExecMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_service_ms"})) { searchDispatchServiceMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_service_us"})) { searchDispatchServiceUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_map_sort_ms"})) { searchDispatchMapSortMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_map_sort_us"})) { searchDispatchMapSortUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_feedback_ms"})) { searchDispatchFeedbackMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_feedback_us"})) { searchDispatchFeedbackUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_response_ms"})) { searchDispatchResponseMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_response_us"})) { searchDispatchResponseUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_total_ms"})) { searchDispatchTotalMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_dispatch_total_us"})) { searchDispatchTotalUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_embedding_us"})) { searchTimingEmbeddingUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_text_us"})) { searchTimingTextUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_path_us"})) { searchTimingPathUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_vector_us"})) { searchTimingVectorUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_entity_vector_us"})) { searchTimingEntityVectorUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_compressed_ann_us"})) { searchTimingCompressedAnnUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_kg_us"})) { searchTimingKgUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_metadata_us"})) { searchTimingMetadataUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_tag_us"})) { searchTimingTagUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_multi_vector_us"})) { searchTimingMultiVectorUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_concepts_us"})) { searchTimingConceptsUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_reranker_us"})) { searchTimingRerankerUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"timing_graph_rerank_us"})) { searchTimingGraphRerankUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"subphrase_fts_hit_count"})) { searchSubphraseFtsHitSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"aggressive_fts_hit_count"})) { searchAggressiveFtsHitSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"graph_expansion_fts_hit_count"})) { searchGraphExpansionFtsHitSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"graph_doc_expansion_fts_hit_count"})) { searchGraphDocExpansionFtsHitSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"multi_vector_raw_hit_count"})) { searchMultiVectorRawHitSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"graph_vector_raw_hit_count"})) { searchGraphVectorRawHitSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"graph_query_neighbor_seed_docs"})) { searchGraphQueryNeighborSeedDocSamples.push_back(*v); } }; auto collectListServerTelemetry = [&](const nlohmann::json& payload) { const auto listStats = [&]() -> nlohmann::json { if (payload.contains("list_stats")) { return payload["list_stats"]; } if (payload.contains("listStats")) { return payload["listStats"]; } return nlohmann::json::object(); }(); std::lock_guard lk(grepServerStatsMutex); if (auto v = jsonGetInt64AnyKey(listStats, {"service_execution_time_ms"})) { listServiceExecMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(listStats, {"phase_dispatch_service_ms"})) { listDispatchServiceMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(listStats, {"phase_dispatch_service_us"})) { listDispatchServiceUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(listStats, {"phase_dispatch_map_ms"})) { listDispatchMapMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(listStats, {"phase_dispatch_map_us"})) { listDispatchMapUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(listStats, {"phase_dispatch_total_ms"})) { listDispatchTotalMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(listStats, {"phase_dispatch_total_us"})) { listDispatchTotalUsSamples.push_back(*v); } }; // Live progress reporter std::atomic progressStop{false}; auto progressStart = std::chrono::steady_clock::now(); std::thread progressThread([&]() { int lastCompleted = 0; while (!progressStop.load()) { std::this_thread::sleep_for(2s); int cur = totalCompleted.load(); auto elapsed = std::chrono::duration(std::chrono::steady_clock::now() - progressStart) .count(); double rate = cur > 0 ? cur / elapsed : 0.0; int delta = cur - lastCompleted; lastCompleted = cur; auto snap = DaemonSnapshot::capture(monitorClient); std::cout << " [" << std::fixed << std::setprecision(1) << elapsed << "s] " << cur << "/" << totalExpected << " ops (" << std::setprecision(1) << rate << " ops/s, +" << delta << ") " << "RSS=" << std::setprecision(0) << snap.memoryUsageMb << "MB " << "conns=" << snap.activeConnections << "/" << snap.maxConnections << " " << "pressure=" << snap.pressureLevel << "\n"; } }); std::atomic go{false}; auto globalStart = std::chrono::steady_clock::now(); std::vector threads; threads.reserve(kTotalClients); // Search queries static const char* corpusQueries[] = { "architecture design patterns", "security vulnerability analysis", "memory management allocation", "network protocol implementation", "database query optimization", "concurrent thread synchronization", "error handling recovery", "configuration file parsing", "build system compilation", "testing framework assertions", }; static const char* corpusGrepPatterns[] = { "TODO|FIXME", "error|exception", "thread|mutex|lock", "config|settings", "benchmark|performance", "daemon|socket", "search|index", "vector|embedding", "plugin|model", "timeout|retry", }; int threadIdx = 0; // --- Search threads --- for (int s = 0; s < kSearchClients; ++s) { int tid = threadIdx++; perThreadTraces[tid].reserve(kOpsPerClient); threads.emplace_back([&, tid, s]() { std::optional client; if (!useMcpPath) { auto ccfg = makeClientConfig(); client.emplace(ccfg); } while (!go.load()) std::this_thread::yield(); for (int i = 0; i < kOpsPerClient; ++i) { SearchRequest req; req.query = corpusQueries[(s * kOpsPerClient + i) % 10]; req.limit = (i % 3 == 0) ? 5 : (i % 3 == 1) ? 10 : 25; req.searchType = cfg.searchType; auto t0 = std::chrono::steady_clock::now(); yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot( static_cast(tid), "search", {{"query", req.query}, {"limit", req.limit}, {"type", req.searchType}}); } else { auto direct = yams::cli::run_sync(client->search(req), kOpTimeout); if (direct) { nlohmann::json j; j["results"] = nlohmann::json::array(); for (const auto& r : direct.value().results) { j["results"].push_back({{"id", r.id}}); } j["execution_time_ms"] = direct.value().elapsed.count(); j["query_info"] = direct.value().queryInfo; j["search_stats"] = direct.value().searchStats; res = j; } else { res = yams::Error{direct.error()}; } } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart).count(); OpTrace trace; trace.opType = "search"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { trace.success = true; const auto& payload = res.value(); if (payload.contains("results") && payload["results"].is_array()) { trace.resultCount = static_cast(payload["results"].size()); } collectSearchServerTelemetry(payload); searchOps.fetch_add(1); } else { trace.success = false; trace.errorMsg = res.error().message; searchFails.fetch_add(1); } perThreadTraces[tid].push_back(std::move(trace)); totalCompleted.fetch_add(1); } }); } // --- List threads --- for (int l = 0; l < kListClients; ++l) { int tid = threadIdx++; perThreadTraces[tid].reserve(kOpsPerClient); threads.emplace_back([&, tid]() { std::optional client; if (!useMcpPath) { auto ccfg = makeClientConfig(); client.emplace(ccfg); } while (!go.load()) std::this_thread::yield(); for (int i = 0; i < kOpsPerClient; ++i) { ListRequest req; req.limit = (i % 4 == 0) ? 10 : (i % 4 == 1) ? 25 : (i % 4 == 2) ? 50 : 100; bool listPermitAcquired = false; if (kListInflightCap > 0) { while (!listPermitAcquired) { int cur = listInFlight.load(std::memory_order_relaxed); if (cur < kListInflightCap) { listPermitAcquired = listInFlight.compare_exchange_weak( cur, cur + 1, std::memory_order_relaxed); } if (!listPermitAcquired) { std::this_thread::yield(); } } } auto t0 = std::chrono::steady_clock::now(); yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot(static_cast(tid), "list", {{"limit", req.limit}, {"use_session", false}}); } else { auto direct = yams::cli::run_sync(client->list(req), kOpTimeout); if (direct) { nlohmann::json j; j["items"] = nlohmann::json::array(); for (const auto& it : direct.value().items) { j["items"].push_back({{"hash", it.hash}}); } j["query_info"] = direct.value().queryInfo; j["list_stats"] = direct.value().listStats; res = j; } else { res = yams::Error{direct.error()}; } } if (kListInflightCap > 0 && listPermitAcquired) { listInFlight.fetch_sub(1, std::memory_order_relaxed); } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart).count(); OpTrace trace; trace.opType = "list"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { const auto& payload = res.value(); trace.success = true; if (payload.contains("items") && payload["items"].is_array()) { trace.resultCount = static_cast(payload["items"].size()); } else if (payload.contains("documents") && payload["documents"].is_array()) { trace.resultCount = static_cast(payload["documents"].size()); } collectListServerTelemetry(payload); listOps.fetch_add(1); } else { trace.success = false; trace.errorMsg = res.error().message; listFails.fetch_add(1); } perThreadTraces[tid].push_back(std::move(trace)); totalCompleted.fetch_add(1); } }); } // --- Status/Get threads --- for (int g = 0; g < kGrepClients; ++g) { int tid = threadIdx++; perThreadTraces[tid].reserve(kOpsPerClient); threads.emplace_back([&, tid, g]() { std::optional client; if (!useMcpPath) { auto ccfg = makeClientConfig(); client.emplace(ccfg); } while (!go.load()) std::this_thread::yield(); for (int i = 0; i < kOpsPerClient; ++i) { const char* pattern = corpusGrepPatterns[(g * kOpsPerClient + i) % 10]; auto t0 = std::chrono::steady_clock::now(); yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot( static_cast(tid), "grep", {{"pattern", pattern}, {"max_count", (i % 2 == 0) ? 25 : 50}, {"ignore_case", (i % 3 == 0)}, {"context", i % 2}, {"use_session", useSessionGrep}, {"session", useSessionGrep ? benchmarkSessionName : ""}}); } else { GrepRequest req; req.pattern = pattern; req.maxMatches = static_cast((i % 2 == 0) ? 25 : 50); req.caseInsensitive = (i % 3 == 0); req.contextLines = i % 2; req.useSession = useSessionGrep; req.sessionName = useSessionGrep ? benchmarkSessionName : std::string{}; auto direct = yams::cli::run_sync(client->grep(req), kOpTimeout); if (direct) { nlohmann::json j; j["match_count"] = direct.value().totalMatches; j["file_count"] = direct.value().filesSearched; j["regex_matches"] = direct.value().regexMatches; j["semantic_matches"] = direct.value().semanticMatches; j["execution_time_ms"] = direct.value().executionTimeMs; j["query_info"] = direct.value().queryInfo; j["search_stats"] = direct.value().searchStats; res = j; } else { res = yams::Error{direct.error()}; } } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart).count(); OpTrace trace; trace.opType = "grep"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { const auto& payload = res.value(); trace.success = true; if (payload.contains("match_count") && payload["match_count"].is_number_integer()) { trace.resultCount = payload["match_count"].get(); } const auto searchStats = [&]() -> nlohmann::json { if (payload.contains("search_stats")) { return payload["search_stats"]; } if (payload.contains("searchStats")) { return payload["searchStats"]; } return nlohmann::json::object(); }(); std::lock_guard lk(grepServerStatsMutex); if (auto v = jsonGetInt64AnyKey(payload, {"execution_time_ms", "executionTimeMs"})) { grepExecMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"docs_scanned"})) { grepDocsScannedSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"lines_scanned"})) { grepLinesScannedSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"bytes_scanned"})) { grepBytesScannedSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"content_retrieval_ms"})) { grepContentRetrievalMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"match_render_us"})) { grepMatchRenderUsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_worker_scan_ms"})) { grepWorkerScanMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_candidate_discovery_ms"})) { grepCandidateDiscoveryMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_path_filtering_ms"})) { grepPathFilteringMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_candidate_classification_ms"})) { grepCandidateClassificationMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_metadata_batch_fetch_ms"})) { grepMetadataBatchFetchMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"regex_scan_ms"})) { grepRegexScanMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"content_decode_ms_estimate"})) { grepContentDecodeEstimateMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"worker_critical_total_ms"})) { grepWorkerCriticalTotalMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"worker_critical_content_retrieval_ms"})) { grepWorkerCriticalRetrievalMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"worker_critical_regex_scan_ms"})) { grepWorkerCriticalRegexMsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"worker_critical_other_ms"})) { grepWorkerCriticalOtherMsSamples.push_back(*v); } { int64_t metadataEnumerationMs = 0; bool hasMetadataEnumeration = false; if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_candidate_discovery_ms"})) { metadataEnumerationMs += *v; hasMetadataEnumeration = true; } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_path_filtering_ms"})) { metadataEnumerationMs += *v; hasMetadataEnumeration = true; } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_candidate_classification_ms"})) { metadataEnumerationMs += *v; hasMetadataEnumeration = true; } if (auto v = jsonGetInt64AnyKey(searchStats, {"phase_metadata_batch_fetch_ms"})) { metadataEnumerationMs += *v; hasMetadataEnumeration = true; } if (auto v = jsonGetInt64AnyKey(searchStats, {"metadata_enumeration_ms"})) { metadataEnumerationMs = *v; hasMetadataEnumeration = true; } if (hasMetadataEnumeration) { grepMetadataEnumerationMsSamples.push_back(metadataEnumerationMs); } } if (auto v = jsonGetInt64AnyKey(searchStats, {"regex_search_calls"})) { grepRegexSearchCallsSamples.push_back(*v); } if (auto v = jsonGetInt64AnyKey(searchStats, {"bmh_prefilter_skips"})) { grepBmhPrefilterSkipsSamples.push_back(*v); } grepOps.fetch_add(1); } else { trace.success = false; trace.errorMsg = res.error().message; grepFails.fetch_add(1); } perThreadTraces[tid].push_back(std::move(trace)); totalCompleted.fetch_add(1); } }); } // --- Status/Get/Cat threads --- for (int g = 0; g < kStatusGetClients; ++g) { int tid = threadIdx++; perThreadTraces[tid].reserve(kOpsPerClient); threads.emplace_back([&, tid, g]() { std::optional client; if (!useMcpPath) { auto ccfg = makeClientConfig(); client.emplace(ccfg); } std::mt19937 rng(static_cast(g * 997 + 17)); while (!go.load()) std::this_thread::yield(); for (int i = 0; i < kOpsPerClient; ++i) { auto t0 = std::chrono::steady_clock::now(); if (extensionLikeProfile) { const int slot = (i + g) % 20; const bool doStatus = (slot == 0) || knownHashes.empty(); const bool doSearch = !doStatus && (slot >= 1 && slot <= 4); const bool doList = !doStatus && !doSearch && (slot >= 5 && slot <= 10); const bool doCat = !doStatus && !doSearch && !doList && (slot == 19); if (doStatus) { yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot(static_cast(tid), "status", nlohmann::json::object()); } else { auto direct = yams::cli::run_sync(client->status(), kOpTimeout); if (direct) { res = nlohmann::json::object(); } else { res = yams::Error{direct.error()}; } } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart) .count(); OpTrace trace; trace.opType = "status"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { trace.success = true; statusOps.fetch_add(1); } else { trace.success = false; trace.errorMsg = res.error().message; statusFails.fetch_add(1); } perThreadTraces[tid].push_back(std::move(trace)); totalCompleted.fetch_add(1); continue; } if (doSearch) { SearchRequest req; req.query = corpusQueries[(g * kOpsPerClient + i) % 10]; req.limit = 5; req.searchType = cfg.searchType; yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot(static_cast(tid), "search", {{"query", req.query}, {"limit", req.limit}, {"type", req.searchType}}); } else { auto direct = yams::cli::run_sync(client->search(req), kOpTimeout); if (direct) { nlohmann::json j; j["results"] = nlohmann::json::array(); for (const auto& r : direct.value().results) { j["results"].push_back({{"id", r.id}}); } j["execution_time_ms"] = direct.value().elapsed.count(); j["query_info"] = direct.value().queryInfo; j["search_stats"] = direct.value().searchStats; res = j; } else { res = yams::Error{direct.error()}; } } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart) .count(); OpTrace trace; trace.opType = "search"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { const auto& payload = res.value(); trace.success = true; if (payload.contains("results") && payload["results"].is_array()) { trace.resultCount = static_cast(payload["results"].size()); } collectSearchServerTelemetry(payload); searchOps.fetch_add(1); } else { trace.success = false; trace.errorMsg = res.error().message; searchFails.fetch_add(1); } perThreadTraces[tid].push_back(std::move(trace)); totalCompleted.fetch_add(1); continue; } if (doList) { ListRequest req; req.limit = ((i % 2) == 0) ? 10 : 25; yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot(static_cast(tid), "list", {{"limit", req.limit}, {"use_session", false}}); } else { auto direct = yams::cli::run_sync(client->list(req), kOpTimeout); if (direct) { nlohmann::json j; j["items"] = nlohmann::json::array(); for (const auto& it : direct.value().items) { j["items"].push_back({{"hash", it.hash}}); } j["query_info"] = direct.value().queryInfo; j["list_stats"] = direct.value().listStats; res = j; } else { res = yams::Error{direct.error()}; } } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart) .count(); OpTrace trace; trace.opType = "list"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { const auto& payload = res.value(); trace.success = true; if (payload.contains("items") && payload["items"].is_array()) { trace.resultCount = static_cast(payload["items"].size()); } else if (payload.contains("documents") && payload["documents"].is_array()) { trace.resultCount = static_cast(payload["documents"].size()); } collectListServerTelemetry(payload); listOps.fetch_add(1); } else { trace.success = false; trace.errorMsg = res.error().message; listFails.fetch_add(1); } perThreadTraces[tid].push_back(std::move(trace)); totalCompleted.fetch_add(1); continue; } auto runGetOrCat = [&](const std::string& hash, bool requestContent) -> yams::Result { if (requestContent) { if (useMcpPath) { return queryMcpStepForSlot( static_cast(tid), "get", {{"hash", hash}, {"include_content", true}}); } CatRequest req; req.hash = hash; auto direct = yams::cli::run_sync(client->cat(req), kOpTimeout); if (direct) { nlohmann::json j; j["has_content"] = direct.value().hasContent; j["size"] = direct.value().size; return j; } return yams::Error{direct.error()}; } if (useMcpPath) { return queryMcpStepForSlot( static_cast(tid), "get", {{"hash", hash}, {"include_content", false}}); } GetRequest req; req.hash = hash; req.metadataOnly = true; auto direct = yams::cli::run_sync(client->get(req), kOpTimeout); if (direct) { return nlohmann::json::object(); } return yams::Error{direct.error()}; }; const bool requestContent = doCat; yams::Result res = yams::Error{yams::ErrorCode::NotFound, "No hash candidates available"}; constexpr int kHashRetryAttempts = 2; for (int attempt = 0; attempt < kHashRetryAttempts; ++attempt) { auto hashOpt = sampleKnownHash(rng); if (!hashOpt.has_value()) { break; } const auto& hash = *hashOpt; res = runGetOrCat(hash, requestContent); if (res) { break; } if (isDocumentMissingFailure(res.error().message)) { markHashInvalid(hash); continue; } break; } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart) .count(); OpTrace trace; trace.opType = doCat ? "cat" : "get"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { trace.success = true; if (doCat) { if (res.value().contains("size") && res.value()["size"].is_number()) { trace.resultCount = static_cast(res.value()["size"].get()); } catOps.fetch_add(1); } else { getOps.fetch_add(1); } } else { trace.success = false; trace.errorMsg = res.error().message; if (doCat) { catFails.fetch_add(1); } else { getFails.fetch_add(1); } } perThreadTraces[tid].push_back(std::move(trace)); totalCompleted.fetch_add(1); continue; } bool isStatus = (i % kStatusEveryN == 0) || knownHashes.empty(); if (isStatus) { yams::Result res; if (useMcpPath) { res = queryMcpStepForSlot(static_cast(tid), "status", nlohmann::json::object()); } else { auto direct = yams::cli::run_sync(client->status(), kOpTimeout); if (direct) { res = nlohmann::json::object(); } else { res = yams::Error{direct.error()}; } } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart) .count(); OpTrace trace; trace.opType = "status"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { trace.success = true; statusOps.fetch_add(1); } else { trace.success = false; trace.errorMsg = res.error().message; statusFails.fetch_add(1); } perThreadTraces[tid].push_back(std::move(trace)); } else { bool doCat = (i % kCatEveryN == 0); auto runGetOrCat = [&](const std::string& hash, bool requestContent) -> yams::Result { if (requestContent) { if (useMcpPath) { return queryMcpStepForSlot( static_cast(tid), "get", {{"hash", hash}, {"include_content", true}}); } CatRequest req; req.hash = hash; auto direct = yams::cli::run_sync(client->cat(req), kOpTimeout); if (direct) { nlohmann::json j; j["has_content"] = direct.value().hasContent; j["size"] = direct.value().size; return j; } return yams::Error{direct.error()}; } if (useMcpPath) { return queryMcpStepForSlot( static_cast(tid), "get", {{"hash", hash}, {"include_content", false}}); } GetRequest req; req.hash = hash; req.metadataOnly = true; auto direct = yams::cli::run_sync(client->get(req), kOpTimeout); if (direct) { return nlohmann::json::object(); } return yams::Error{direct.error()}; }; const bool requestContent = doCat; yams::Result res = yams::Error{yams::ErrorCode::NotFound, "No hash candidates available"}; constexpr int kHashRetryAttempts = 2; for (int attempt = 0; attempt < kHashRetryAttempts; ++attempt) { auto hashOpt = sampleKnownHash(rng); if (!hashOpt.has_value()) { break; } const auto& hash = *hashOpt; res = runGetOrCat(hash, requestContent); if (res) { break; } if (isDocumentMissingFailure(res.error().message)) { markHashInvalid(hash); continue; } break; } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); auto wallMs = std::chrono::duration_cast(t1 - globalStart) .count(); OpTrace trace; trace.opType = doCat ? "cat" : "get"; trace.threadId = tid; trace.opIndex = i; trace.latencyUs = latUs; trace.wallClockMs = wallMs; if (res) { trace.success = true; if (doCat) { if (res.value().contains("size") && res.value()["size"].is_number()) { trace.resultCount = static_cast(res.value()["size"].get()); } catOps.fetch_add(1); } else { getOps.fetch_add(1); } } else { trace.success = false; trace.errorMsg = res.error().message; if (doCat) { catFails.fetch_add(1); } else { getFails.fetch_add(1); } } perThreadTraces[tid].push_back(std::move(trace)); } totalCompleted.fetch_add(1); } }); } go.store(true); for (auto& t : threads) t.join(); progressStop.store(true); progressThread.join(); sampler.stop(); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); // ───────────────────────────────────────────────────────────────────────── // Analysis // ───────────────────────────────────────────────────────────────────────── auto finalSnap = DaemonSnapshot::capture(monitorClient); // Aggregate latency vectors from per-thread traces std::vector allSearchLat, allListLat, allGrepLat, allStatusLat, allGetLat, allCatLat; std::vector allTraces; std::map errorCategories; for (auto& threadTraces : perThreadTraces) { for (auto& tr : threadTraces) { allTraces.push_back(tr); if (tr.success) { if (tr.opType == "search") pushMetricSample(allSearchLat, tr.latencyUs); else if (tr.opType == "list") pushMetricSample(allListLat, tr.latencyUs); else if (tr.opType == "grep") pushMetricSample(allGrepLat, tr.latencyUs); else if (tr.opType == "status") pushMetricSample(allStatusLat, tr.latencyUs); else if (tr.opType == "get") pushMetricSample(allGetLat, tr.latencyUs); else if (tr.opType == "cat") pushMetricSample(allCatLat, tr.latencyUs); } else { errorCategories[classifyFailureMessage(tr.errorMsg)]++; } } } auto searchStats = PercentileStats::compute(allSearchLat); auto listStats = PercentileStats::compute(allListLat); auto grepStats = PercentileStats::compute(allGrepLat); auto statusStats = PercentileStats::compute(allStatusLat); auto getStats = PercentileStats::compute(allGetLat); auto catStats = PercentileStats::compute(allCatLat); auto summarizeSamples = [](const std::vector& values) { json summary{{"count", values.size()}, {"min", 0}, {"p50", 0}, {"p90", 0}, {"p95", 0}, {"p99", 0}, {"max", 0}, {"mean", 0.0}}; if (values.empty()) { return summary; } auto sortedValues = values; std::sort(sortedValues.begin(), sortedValues.end()); const auto atPct = [&](double pct) { const auto idx = std::min(sortedValues.size() - 1, static_cast(pct * static_cast(sortedValues.size()))); return sortedValues[idx]; }; sortedValues.erase(std::remove_if(sortedValues.begin(), sortedValues.end(), [](int64_t value) { return value < 0; }), sortedValues.end()); for (auto& value : sortedValues) { value = clampMetricSample(value); } if (sortedValues.empty()) { return summary; } std::sort(sortedValues.begin(), sortedValues.end()); const long double total = std::accumulate( sortedValues.begin(), sortedValues.end(), 0.0L, [](long double sum, int64_t value) { return sum + static_cast(value); }); const auto mean = static_cast(total / static_cast(sortedValues.size())); summary["min"] = sortedValues.front(); summary["p50"] = atPct(0.50); summary["p90"] = atPct(0.90); summary["p95"] = atPct(0.95); summary["p99"] = atPct(0.99); summary["max"] = sortedValues.back(); summary["mean"] = mean; return summary; }; const json grepExecStats = summarizeSamples(grepExecMsSamples); const json grepDocsScannedStats = summarizeSamples(grepDocsScannedSamples); const json grepLinesScannedStats = summarizeSamples(grepLinesScannedSamples); const json grepBytesScannedStats = summarizeSamples(grepBytesScannedSamples); const json grepContentRetrievalStats = summarizeSamples(grepContentRetrievalMsSamples); const json grepMatchRenderStats = summarizeSamples(grepMatchRenderUsSamples); const json grepWorkerScanStats = summarizeSamples(grepWorkerScanMsSamples); const json grepCandidateDiscoveryStats = summarizeSamples(grepCandidateDiscoveryMsSamples); const json grepPathFilteringStats = summarizeSamples(grepPathFilteringMsSamples); const json grepCandidateClassificationStats = summarizeSamples(grepCandidateClassificationMsSamples); const json grepMetadataBatchFetchStats = summarizeSamples(grepMetadataBatchFetchMsSamples); const json grepMetadataEnumerationStats = summarizeSamples(grepMetadataEnumerationMsSamples); const json grepRegexScanStats = summarizeSamples(grepRegexScanMsSamples); const json grepContentDecodeEstimateStats = summarizeSamples(grepContentDecodeEstimateMsSamples); const json grepWorkerCriticalTotalStats = summarizeSamples(grepWorkerCriticalTotalMsSamples); const json grepWorkerCriticalRetrievalStats = summarizeSamples(grepWorkerCriticalRetrievalMsSamples); const json grepWorkerCriticalRegexStats = summarizeSamples(grepWorkerCriticalRegexMsSamples); const json grepWorkerCriticalOtherStats = summarizeSamples(grepWorkerCriticalOtherMsSamples); const json grepRegexSearchCallsStats = summarizeSamples(grepRegexSearchCallsSamples); const json grepBmhPrefilterSkipsStats = summarizeSamples(grepBmhPrefilterSkipsSamples); const json searchExecStats = summarizeSamples(searchExecMsSamples); const json searchDispatchServiceStats = summarizeSamples(searchDispatchServiceMsSamples); const json searchDispatchServiceUsStats = summarizeSamples(searchDispatchServiceUsSamples); const json searchDispatchMapSortStats = summarizeSamples(searchDispatchMapSortMsSamples); const json searchDispatchMapSortUsStats = summarizeSamples(searchDispatchMapSortUsSamples); const json searchDispatchFeedbackStats = summarizeSamples(searchDispatchFeedbackMsSamples); const json searchDispatchFeedbackUsStats = summarizeSamples(searchDispatchFeedbackUsSamples); const json searchDispatchResponseStats = summarizeSamples(searchDispatchResponseMsSamples); const json searchDispatchResponseUsStats = summarizeSamples(searchDispatchResponseUsSamples); const json searchDispatchTotalStats = summarizeSamples(searchDispatchTotalMsSamples); const json searchDispatchTotalUsStats = summarizeSamples(searchDispatchTotalUsSamples); const json searchTimingEmbeddingUsStats = summarizeSamples(searchTimingEmbeddingUsSamples); const json searchTimingTextUsStats = summarizeSamples(searchTimingTextUsSamples); const json searchTimingPathUsStats = summarizeSamples(searchTimingPathUsSamples); const json searchTimingVectorUsStats = summarizeSamples(searchTimingVectorUsSamples); const json searchTimingEntityVectorUsStats = summarizeSamples(searchTimingEntityVectorUsSamples); const json searchTimingCompressedAnnUsStats = summarizeSamples(searchTimingCompressedAnnUsSamples); const json searchTimingKgUsStats = summarizeSamples(searchTimingKgUsSamples); const json searchTimingMetadataUsStats = summarizeSamples(searchTimingMetadataUsSamples); const json searchTimingTagUsStats = summarizeSamples(searchTimingTagUsSamples); const json searchTimingMultiVectorUsStats = summarizeSamples(searchTimingMultiVectorUsSamples); const json searchTimingConceptsUsStats = summarizeSamples(searchTimingConceptsUsSamples); const json searchTimingRerankerUsStats = summarizeSamples(searchTimingRerankerUsSamples); const json searchTimingGraphRerankUsStats = summarizeSamples(searchTimingGraphRerankUsSamples); const json searchSubphraseFtsHitStats = summarizeSamples(searchSubphraseFtsHitSamples); const json searchAggressiveFtsHitStats = summarizeSamples(searchAggressiveFtsHitSamples); const json searchGraphExpansionFtsHitStats = summarizeSamples(searchGraphExpansionFtsHitSamples); const json searchGraphDocExpansionFtsHitStats = summarizeSamples(searchGraphDocExpansionFtsHitSamples); const json searchMultiVectorRawHitStats = summarizeSamples(searchMultiVectorRawHitSamples); const json searchGraphVectorRawHitStats = summarizeSamples(searchGraphVectorRawHitSamples); const json searchGraphQueryNeighborSeedDocStats = summarizeSamples(searchGraphQueryNeighborSeedDocSamples); const json listServiceExecStats = summarizeSamples(listServiceExecMsSamples); const json listDispatchServiceStats = summarizeSamples(listDispatchServiceMsSamples); const json listDispatchServiceUsStats = summarizeSamples(listDispatchServiceUsSamples); const json listDispatchMapStats = summarizeSamples(listDispatchMapMsSamples); const json listDispatchMapUsStats = summarizeSamples(listDispatchMapUsSamples); const json listDispatchTotalStats = summarizeSamples(listDispatchTotalMsSamples); const json listDispatchTotalUsStats = summarizeSamples(listDispatchTotalUsSamples); int totalOps = searchOps.load() + listOps.load() + grepOps.load() + statusOps.load() + getOps.load() + catOps.load(); int totalFails = searchFails.load() + listFails.load() + grepFails.load() + statusFails.load() + getFails.load() + catFails.load(); double opsPerSec = safeRate(static_cast(totalOps), globalElapsed); size_t invalidHashCount = 0; { std::lock_guard lk(invalidHashMutex); invalidHashCount = invalidHashes.size(); } // Memory watermarks from time series auto samples = sampler.getSamples(); double peakRss = 0.0, minRss = std::numeric_limits::max(); for (auto& s : samples) { peakRss = std::max(peakRss, s.snap.memoryUsageMb); minRss = std::min(minRss, s.snap.memoryUsageMb); } if (samples.empty()) minRss = 0.0; std::cout << "\n=== Large Corpus Results (instrumented) ===\n"; std::cout << " Corpus: " << finalSnap.documentsTotal << " docs\n"; std::cout << " Layout: " << kSearchClients << " search + " << kListClients << " list + " << kGrepClients << " grep + " << kStatusGetClients << " status/get/cat\n"; std::cout << " Wall time: " << std::fixed << std::setprecision(2) << globalElapsed << "s\n"; std::cout << " Total ops: " << totalOps << " (" << std::setprecision(1) << opsPerSec << " ops/s)\n"; std::cout << " Total failures: " << totalFails << "\n\n"; if (invalidHashCount > 0) { std::cout << " Invalid hash candidates quarantined: " << invalidHashCount << "\n\n"; } std::cout << " Op breakdown:\n"; std::cout << " Search: " << searchOps.load() << " ok, " << searchFails.load() << " fail\n"; std::cout << " List: " << listOps.load() << " ok, " << listFails.load() << " fail\n"; std::cout << " Grep: " << grepOps.load() << " ok, " << grepFails.load() << " fail\n"; std::cout << " Status: " << statusOps.load() << " ok, " << statusFails.load() << " fail\n"; std::cout << " Get: " << getOps.load() << " ok, " << getFails.load() << " fail\n\n"; std::cout << " Cat: " << catOps.load() << " ok, " << catFails.load() << " fail\n\n"; printPercentiles("Search latency", searchStats); printPercentiles("List latency ", listStats); printPercentiles("Grep latency ", grepStats); printPercentiles("Status latency", statusStats); printPercentiles("Get latency ", getStats); printPercentiles("Cat latency ", catStats); if (grepExecStats.value("count", 0) > 0 || grepWorkerScanStats.value("count", 0) > 0) { std::cout << "\n Grep server telemetry:\n"; std::cout << " execution_ms: mean=" << grepExecStats.value("mean", 0.0) << " p95=" << grepExecStats.value("p95", 0) << " max=" << grepExecStats.value("max", 0) << "\n"; std::cout << " metadata_enumeration_ms: mean=" << grepMetadataEnumerationStats.value("mean", 0.0) << " p95=" << grepMetadataEnumerationStats.value("p95", 0) << " max=" << grepMetadataEnumerationStats.value("max", 0) << "\n"; std::cout << " candidate_discovery_ms: mean=" << grepCandidateDiscoveryStats.value("mean", 0.0) << " p95=" << grepCandidateDiscoveryStats.value("p95", 0) << "\n"; std::cout << " path_filtering_ms: mean=" << grepPathFilteringStats.value("mean", 0.0) << " p95=" << grepPathFilteringStats.value("p95", 0) << "\n"; std::cout << " candidate_classification_ms: mean=" << grepCandidateClassificationStats.value("mean", 0.0) << " p95=" << grepCandidateClassificationStats.value("p95", 0) << "\n"; std::cout << " metadata_batch_fetch_ms: mean=" << grepMetadataBatchFetchStats.value("mean", 0.0) << " p95=" << grepMetadataBatchFetchStats.value("p95", 0) << "\n"; std::cout << " worker_scan_ms: mean=" << grepWorkerScanStats.value("mean", 0.0) << " p95=" << grepWorkerScanStats.value("p95", 0) << " max=" << grepWorkerScanStats.value("max", 0) << "\n"; std::cout << " content_retrieval_ms: mean=" << grepContentRetrievalStats.value("mean", 0.0) << " p95=" << grepContentRetrievalStats.value("p95", 0) << " max=" << grepContentRetrievalStats.value("max", 0) << "\n"; std::cout << " match_render_us: mean=" << grepMatchRenderStats.value("mean", 0.0) << " p95=" << grepMatchRenderStats.value("p95", 0) << " max=" << grepMatchRenderStats.value("max", 0) << "\n"; std::cout << " content_decode_ms_estimate: mean=" << grepContentDecodeEstimateStats.value("mean", 0.0) << " p95=" << grepContentDecodeEstimateStats.value("p95", 0) << " max=" << grepContentDecodeEstimateStats.value("max", 0) << "\n"; std::cout << " regex_scan_ms: mean=" << grepRegexScanStats.value("mean", 0.0) << " p95=" << grepRegexScanStats.value("p95", 0) << " max=" << grepRegexScanStats.value("max", 0) << "\n"; std::cout << " worker_critical_total_ms: mean=" << grepWorkerCriticalTotalStats.value("mean", 0.0) << " p95=" << grepWorkerCriticalTotalStats.value("p95", 0) << " max=" << grepWorkerCriticalTotalStats.value("max", 0) << "\n"; std::cout << " worker_critical_content_retrieval_ms: mean=" << grepWorkerCriticalRetrievalStats.value("mean", 0.0) << " p95=" << grepWorkerCriticalRetrievalStats.value("p95", 0) << " max=" << grepWorkerCriticalRetrievalStats.value("max", 0) << "\n"; std::cout << " worker_critical_regex_scan_ms: mean=" << grepWorkerCriticalRegexStats.value("mean", 0.0) << " p95=" << grepWorkerCriticalRegexStats.value("p95", 0) << " max=" << grepWorkerCriticalRegexStats.value("max", 0) << "\n"; std::cout << " worker_critical_other_ms: mean=" << grepWorkerCriticalOtherStats.value("mean", 0.0) << " p95=" << grepWorkerCriticalOtherStats.value("p95", 0) << " max=" << grepWorkerCriticalOtherStats.value("max", 0) << "\n"; } if (searchDispatchTotalStats.value("count", 0) > 0 || listDispatchTotalStats.value("count", 0) > 0) { std::cout << "\n Search/list server telemetry:\n"; std::cout << " search_total_ms: mean=" << searchDispatchTotalStats.value("mean", 0.0) << " p95=" << searchDispatchTotalStats.value("p95", 0) << " max=" << searchDispatchTotalStats.value("max", 0) << "\n"; std::cout << " list_total_ms: mean=" << listDispatchTotalStats.value("mean", 0.0) << " p95=" << listDispatchTotalStats.value("p95", 0) << " max=" << listDispatchTotalStats.value("max", 0) << "\n"; auto printSearchComponentUs = [&](const char* label, const json& stats) { if (stats.value("count", 0) <= 0) { return; } std::cout << " " << label << ": mean=" << (stats.value("mean", 0.0) / 1000.0) << "ms p95=" << (static_cast(stats.value("p95", 0)) / 1000.0) << "ms\n"; }; printSearchComponentUs("timing_embedding", searchTimingEmbeddingUsStats); printSearchComponentUs("timing_text", searchTimingTextUsStats); printSearchComponentUs("timing_path", searchTimingPathUsStats); printSearchComponentUs("timing_vector", searchTimingVectorUsStats); printSearchComponentUs("timing_entity_vector", searchTimingEntityVectorUsStats); printSearchComponentUs("timing_compressed_ann", searchTimingCompressedAnnUsStats); printSearchComponentUs("timing_kg", searchTimingKgUsStats); printSearchComponentUs("timing_metadata", searchTimingMetadataUsStats); printSearchComponentUs("timing_tag", searchTimingTagUsStats); printSearchComponentUs("timing_multi_vector", searchTimingMultiVectorUsStats); printSearchComponentUs("timing_concepts", searchTimingConceptsUsStats); printSearchComponentUs("timing_reranker", searchTimingRerankerUsStats); printSearchComponentUs("timing_graph_rerank", searchTimingGraphRerankUsStats); if (searchGraphDocExpansionFtsHitStats.value("count", 0) > 0) { std::cout << " graph_doc_expansion_fts_hit_count: mean=" << searchGraphDocExpansionFtsHitStats.value("mean", 0.0) << " p95=" << searchGraphDocExpansionFtsHitStats.value("p95", 0) << "\n"; } if (searchMultiVectorRawHitStats.value("count", 0) > 0) { std::cout << " multi_vector_raw_hit_count: mean=" << searchMultiVectorRawHitStats.value("mean", 0.0) << " p95=" << searchMultiVectorRawHitStats.value("p95", 0) << "\n"; } if (searchGraphVectorRawHitStats.value("count", 0) > 0) { std::cout << " graph_vector_raw_hit_count: mean=" << searchGraphVectorRawHitStats.value("mean", 0.0) << " p95=" << searchGraphVectorRawHitStats.value("p95", 0) << "\n"; } } std::cout << "\n Memory:\n"; std::cout << " Initial: " << std::setprecision(0) << initSnap.memoryUsageMb << " MB\n"; std::cout << " Peak: " << peakRss << " MB\n"; std::cout << " Min: " << minRss << " MB\n"; std::cout << " Final: " << finalSnap.memoryUsageMb << " MB\n"; std::cout << " Delta: " << (finalSnap.memoryUsageMb - initSnap.memoryUsageMb) << " MB\n\n"; if (!errorCategories.empty()) { std::cout << " Error categories:\n"; for (auto& [cat, count] : errorCategories) { std::cout << " " << cat << ": " << count << "\n"; } std::cout << "\n"; } // Hotspot analysis: find slowest ops std::sort(allTraces.begin(), allTraces.end(), [](const OpTrace& a, const OpTrace& b) { return a.latencyUs > b.latencyUs; }); std::cout << " Top 10 slowest operations:\n"; for (int i = 0; i < 10 && i < static_cast(allTraces.size()); ++i) { auto& tr = allTraces[i]; std::cout << " " << (i + 1) << ". " << tr.opType << " t" << tr.threadId << "#" << tr.opIndex << ": " << tr.latencyUs / 1000 << "ms" << (tr.success ? "" : " FAIL:" + tr.errorMsg) << " @" << tr.wallClockMs << "ms\n"; } std::cout << "\n"; // Thread-level hotspots std::map> perThreadSuccessLatUs; std::map perThreadFails; for (const auto& tr : allTraces) { if (tr.success) { pushMetricSample(perThreadSuccessLatUs[tr.threadId], tr.latencyUs); } else { perThreadFails[tr.threadId]++; } } struct ThreadHotspot { int threadId{0}; int64_t p95Us{0}; double meanUs{0.0}; int failures{0}; size_t okCount{0}; }; std::vector threadHotspots; threadHotspots.reserve(static_cast(kTotalClients)); for (int tid = 0; tid < kTotalClients; ++tid) { auto it = perThreadSuccessLatUs.find(tid); std::vector lats = (it != perThreadSuccessLatUs.end()) ? it->second : std::vector{}; auto stats = PercentileStats::compute(lats); threadHotspots.push_back( ThreadHotspot{tid, stats.p95, stats.mean, perThreadFails[tid], stats.count}); } std::sort(threadHotspots.begin(), threadHotspots.end(), [](const ThreadHotspot& a, const ThreadHotspot& b) { if (a.p95Us != b.p95Us) { return a.p95Us > b.p95Us; } return a.failures > b.failures; }); std::cout << " Slowest threads (p95):\n"; for (int i = 0; i < 5 && i < static_cast(threadHotspots.size()); ++i) { const auto& th = threadHotspots[i]; std::cout << " t" << th.threadId << ": p95=" << th.p95Us / 1000 << "ms mean=" << std::fixed << std::setprecision(1) << (th.meanUs / 1000.0) << "ms ok=" << th.okCount << " fail=" << th.failures << "\n"; } std::cout << "\n"; // Time-window hotspots struct WindowStats { int ops{0}; int fails{0}; int64_t totalLatUs{0}; }; std::map windows; for (const auto& tr : allTraces) { const auto boundedWallMs = clampMetricSample(tr.wallClockMs, kMaxBenchmarkWallMs); const int bucket = clampInt64ToInt(boundedWallMs / std::max(1, kHotspotWindowMs), 0, std::numeric_limits::max()); auto& ws = windows[bucket]; ws.ops++; ws.totalLatUs = checkedMetricAdd(ws.totalLatUs, tr.latencyUs); if (!tr.success) { ws.fails++; } } struct WindowHotspot { int bucket{0}; double avgMs{0.0}; int ops{0}; int fails{0}; }; std::vector windowHotspots; windowHotspots.reserve(windows.size()); for (const auto& [bucket, ws] : windows) { if (!isHotspotWindowEligible(ws.ops, kTotalClients)) { continue; } const double avgMs = (ws.ops > 0) ? (static_cast(ws.totalLatUs) / ws.ops / 1000.0) : 0.0; windowHotspots.push_back(WindowHotspot{bucket, avgMs, ws.ops, ws.fails}); } std::sort(windowHotspots.begin(), windowHotspots.end(), [](const WindowHotspot& a, const WindowHotspot& b) { if (a.avgMs != b.avgMs) { return a.avgMs > b.avgMs; } return a.fails > b.fails; }); std::cout << " Slowest time windows:\n"; for (int i = 0; i < 5 && i < static_cast(windowHotspots.size()); ++i) { const auto& hw = windowHotspots[i]; const int64_t startMs = checkedMetricMul(hw.bucket, kHotspotWindowMs, kMaxBenchmarkWallMs); const int64_t endMs = checkedMetricAdd(startMs, kHotspotWindowMs, kMaxBenchmarkWallMs); std::cout << " [" << startMs << ", " << endMs << ")ms" << " avg=" << std::fixed << std::setprecision(1) << hw.avgMs << "ms ops=" << hw.ops << " fail=" << hw.fails << "\n"; } std::cout << "\n"; // Latency distribution (histogram buckets) auto printHistogram = [](const std::string& label, std::vector& lats) { if (lats.empty()) return; std::sort(lats.begin(), lats.end()); // Buckets: <1ms, 1-10ms, 10-100ms, 100ms-1s, 1-10s, >10s int b0 = 0, b1 = 0, b2 = 0, b3 = 0, b4 = 0, b5 = 0; for (auto us : lats) { if (us < 1000) b0++; else if (us < 10000) b1++; else if (us < 100000) b2++; else if (us < 1000000) b3++; else if (us < 10000000) b4++; else b5++; } std::cout << " " << label << " histogram: <1ms=" << b0 << " 1-10ms=" << b1 << " 10-100ms=" << b2 << " 100ms-1s=" << b3 << " 1-10s=" << b4 << " >10s=" << b5 << "\n"; }; std::cout << " Latency distribution:\n"; printHistogram("Search", allSearchLat); printHistogram("List ", allListLat); printHistogram("Grep ", allGrepLat); printHistogram("Status", allStatusLat); printHistogram("Get ", allGetLat); printHistogram("Cat ", allCatLat); std::cout << "\n"; CHECK(totalOps > 0); // Soft-fail on errors — we want the data even if some ops time out if (totalFails > 0) { WARN("Had " << totalFails << " failures out of " << totalExpected << " total ops"); } // ───────────────────────────────────────────────────────────────────────── // JSONL output with full traces // ───────────────────────────────────────────────────────────────────────── json timeSeries = json::array(); for (auto& s : samples) { timeSeries.push_back(json{{"elapsed_ms", s.elapsedMs}, {"snapshot", s.snap.toJson()}}); } // Per-op latency trace array (sorted by wall clock) std::sort(allTraces.begin(), allTraces.end(), [](const OpTrace& a, const OpTrace& b) { return a.wallClockMs < b.wallClockMs; }); json traceArray = json::array(); for (auto& tr : allTraces) { json j{{"op", tr.opType}, {"tid", tr.threadId}, {"idx", tr.opIndex}, {"lat_us", tr.latencyUs}, {"ok", tr.success}, {"items", tr.resultCount}, {"wall_ms", tr.wallClockMs}}; if (!tr.errorMsg.empty()) j["error"] = tr.errorMsg; traceArray.push_back(std::move(j)); } json threadHotspotsJson = json::array(); for (const auto& th : threadHotspots) { threadHotspotsJson.push_back(json{{"tid", th.threadId}, {"p95_us", th.p95Us}, {"mean_us", th.meanUs}, {"ok", th.okCount}, {"fail", th.failures}}); } json windowHotspotsJson = json::array(); for (const auto& hw : windowHotspots) { const int64_t startMs = static_cast(hw.bucket) * kHotspotWindowMs; const int64_t endMs = startMs + kHotspotWindowMs; windowHotspotsJson.push_back(json{{"start_ms", startMs}, {"end_ms", endMs}, {"avg_ms", hw.avgMs}, {"ops", hw.ops}, {"fail", hw.fails}}); } const std::string recordTestName = externalAgentChurnProfile ? "large_corpus_reads_churn" : "large_corpus_reads"; const double mcpTransientRatio = (mcpQueryCalls.load() > 0) ? static_cast(mcpTransientSessionCalls.load()) / static_cast(mcpQueryCalls.load()) : 0.0; json record{ {"timestamp", isoTimestamp()}, {"test", recordTestName}, {"request_path", useMcpPath ? "mcp_query" : "daemon_ipc"}, {"backend", cfg.backend}, {"auth_mode", cfg.authMode}, {"data_dir", cfg.dataDir->string()}, {"num_clients", kTotalClients}, {"corpus_docs", finalSnap.documentsTotal}, {"layout", json{{"search_clients", kSearchClients}, {"list_clients", kListClients}, {"grep_clients", kGrepClients}, {"status_get_clients", kStatusGetClients}, {"ops_per_client", kOpsPerClient}, {"op_timeout_s", kOpTimeoutS}, {"status_every_n", kStatusEveryN}, {"cat_every_n", kCatEveryN}, {"list_inflight_cap", kListInflightCap}, {"mcp_session_churn_every_n", kMcpSessionChurnEveryN}, {"usage_profile", cfg.usageProfile}}}, {"total_ops", totalOps}, {"total_failures", totalFails}, {"elapsed_seconds", globalElapsed}, {"ops_per_sec", opsPerSec}, {"mcp", json{{"query_calls", mcpQueryCalls.load()}, {"transient_session_calls", mcpTransientSessionCalls.load()}, {"transient_session_ratio", mcpTransientRatio}}}, {"hash_pool", json{{"known_hashes", knownHashes.size()}, {"invalid_quarantined", invalidHashCount}}}, {"search", json{{"ops", searchOps.load()}, {"fails", searchFails.load()}, {"latency", searchStats.toJson()}}}, {"list", json{{"ops", listOps.load()}, {"fails", listFails.load()}, {"latency", listStats.toJson()}}}, {"grep", json{{"ops", grepOps.load()}, {"fails", grepFails.load()}, {"latency", grepStats.toJson()}}}, {"status", json{{"ops", statusOps.load()}, {"fails", statusFails.load()}, {"latency", statusStats.toJson()}}}, {"get", json{{"ops", getOps.load()}, {"fails", getFails.load()}, {"latency", getStats.toJson()}}}, {"cat", json{{"ops", catOps.load()}, {"fails", catFails.load()}, {"latency", catStats.toJson()}}}, {"grep_server", json{{"execution_time_ms", grepExecStats}, {"metadata_enumeration_ms", grepMetadataEnumerationStats}, {"phase_candidate_discovery_ms", grepCandidateDiscoveryStats}, {"phase_path_filtering_ms", grepPathFilteringStats}, {"phase_candidate_classification_ms", grepCandidateClassificationStats}, {"phase_metadata_batch_fetch_ms", grepMetadataBatchFetchStats}, {"docs_scanned", grepDocsScannedStats}, {"lines_scanned", grepLinesScannedStats}, {"bytes_scanned", grepBytesScannedStats}, {"content_retrieval_ms", grepContentRetrievalStats}, {"match_render_us", grepMatchRenderStats}, {"content_decode_ms_estimate", grepContentDecodeEstimateStats}, {"phase_worker_scan_ms", grepWorkerScanStats}, {"regex_scan_ms", grepRegexScanStats}, {"worker_critical_total_ms", grepWorkerCriticalTotalStats}, {"worker_critical_content_retrieval_ms", grepWorkerCriticalRetrievalStats}, {"worker_critical_regex_scan_ms", grepWorkerCriticalRegexStats}, {"worker_critical_other_ms", grepWorkerCriticalOtherStats}, {"regex_search_calls", grepRegexSearchCallsStats}, {"bmh_prefilter_skips", grepBmhPrefilterSkipsStats}}}, {"search_server", json{{"execution_time_ms", searchExecStats}, {"phase_dispatch_service_ms", searchDispatchServiceStats}, {"phase_dispatch_service_us", searchDispatchServiceUsStats}, {"phase_dispatch_map_sort_ms", searchDispatchMapSortStats}, {"phase_dispatch_map_sort_us", searchDispatchMapSortUsStats}, {"phase_dispatch_feedback_ms", searchDispatchFeedbackStats}, {"phase_dispatch_feedback_us", searchDispatchFeedbackUsStats}, {"phase_dispatch_response_ms", searchDispatchResponseStats}, {"phase_dispatch_response_us", searchDispatchResponseUsStats}, {"phase_dispatch_total_ms", searchDispatchTotalStats}, {"phase_dispatch_total_us", searchDispatchTotalUsStats}, {"timing_embedding_us", searchTimingEmbeddingUsStats}, {"timing_text_us", searchTimingTextUsStats}, {"timing_path_us", searchTimingPathUsStats}, {"timing_vector_us", searchTimingVectorUsStats}, {"timing_entity_vector_us", searchTimingEntityVectorUsStats}, {"timing_compressed_ann_us", searchTimingCompressedAnnUsStats}, {"timing_kg_us", searchTimingKgUsStats}, {"timing_metadata_us", searchTimingMetadataUsStats}, {"timing_tag_us", searchTimingTagUsStats}, {"timing_multi_vector_us", searchTimingMultiVectorUsStats}, {"timing_concepts_us", searchTimingConceptsUsStats}, {"timing_reranker_us", searchTimingRerankerUsStats}, {"timing_graph_rerank_us", searchTimingGraphRerankUsStats}, {"subphrase_fts_hit_count", searchSubphraseFtsHitStats}, {"aggressive_fts_hit_count", searchAggressiveFtsHitStats}, {"graph_expansion_fts_hit_count", searchGraphExpansionFtsHitStats}, {"graph_doc_expansion_fts_hit_count", searchGraphDocExpansionFtsHitStats}, {"multi_vector_raw_hit_count", searchMultiVectorRawHitStats}, {"graph_vector_raw_hit_count", searchGraphVectorRawHitStats}, {"graph_query_neighbor_seed_docs", searchGraphQueryNeighborSeedDocStats}}}, {"search_debug_samples", searchDebugSamples}, {"list_server", json{{"service_execution_time_ms", listServiceExecStats}, {"phase_dispatch_service_ms", listDispatchServiceStats}, {"phase_dispatch_service_us", listDispatchServiceUsStats}, {"phase_dispatch_map_ms", listDispatchMapStats}, {"phase_dispatch_map_us", listDispatchMapUsStats}, {"phase_dispatch_total_ms", listDispatchTotalStats}, {"phase_dispatch_total_us", listDispatchTotalUsStats}}}, {"memory", json{{"initial_mb", initSnap.memoryUsageMb}, {"peak_mb", peakRss}, {"min_mb", minRss}, {"final_mb", finalSnap.memoryUsageMb}, {"delta_mb", finalSnap.memoryUsageMb - initSnap.memoryUsageMb}}}, {"error_categories", errorCategories}, {"hotspots", json{{"window_ms", kHotspotWindowMs}, {"threads", threadHotspotsJson}, {"windows", windowHotspotsJson}}}, {"daemon_snapshot_final", finalSnap.toJson()}, {"time_series", timeSeries}, {"op_traces", traceArray}, {"tuning_profile", cfg.tuningProfile}, }; emitJsonl(cfg.outputPath, record); std::cout << " Results written to: " << cfg.outputPath.string() << "\n\n"; } // ============================================================================= // EMBEDDINGS-ENABLED BENCHMARK // ============================================================================= TEST_CASE("Multi-client ingestion: embeddings pipeline", "[!benchmark][multi-client][embeddings]") { // Full ingestion benchmark with real plugins loaded: ONNX embeddings, // symbol extraction, entity recognition. Measures the complete pipeline // including embedding generation which is the most expensive stage. // // Requires either: // YAMS_BENCH_ENABLE_PLUGINS=1 with plugins and models available, or // YAMS_BENCH_DATA_DIR pointing to a corpus (can combine both for // ingest-into-existing-corpus testing). auto cfg = BenchConfig::fromEnv(); if (!cfg.enablePlugins) { SKIP("Set YAMS_BENCH_ENABLE_PLUGINS=1 to run the embeddings benchmark"); } std::cout << "\n=== Embeddings Pipeline Benchmark ===\n"; std::cout << " Plugins: enabled\n"; std::cout << " Real model provider: " << (cfg.useRealModelProvider ? "yes" : "mock") << "\n"; if (cfg.pluginDir) std::cout << " Plugin dir: " << cfg.pluginDir->string() << "\n"; if (cfg.dataDir) std::cout << " Data dir: " << cfg.dataDir->string() << "\n"; constexpr int kWriterClients = 4; constexpr int kSearchClients = 4; constexpr int kTotalClients = kWriterClients + kSearchClients; auto opts = benchHarnessOptions(cfg); DaemonHarness harness(opts); // Plugin loading + model warmup needs extra time REQUIRE(harness.start(std::chrono::seconds(120))); // Allow async init to complete (plugin loading, model warmup, connection pools) std::this_thread::sleep_for(5s); // Monitor ClientConfig monCfg; monCfg.socketPath = harness.socketPath(); monCfg.autoStart = false; monCfg.requestTimeout = 30s; DaemonClient monitorClient(monCfg); auto initSnap = DaemonSnapshot::capture(monitorClient); std::cout << " Initial docs: " << initSnap.documentsTotal << "\n"; std::cout << " Docs per writer: " << cfg.docsPerClient << "\n"; TimeSeriesSampler sampler; sampler.start(monitorClient, 1s); // Latency collectors std::mutex addLatMutex, searchLatMutex; std::vector addLatencies, searchLatencies; std::atomic addOps{0}, searchOps{0}; std::atomic addFails{0}, searchFails{0}; // Collect hashes for search/get during ingest std::mutex hashMutex; std::vector ingestedHashes; std::atomic go{false}; std::atomic writersFinished{false}; auto globalStart = std::chrono::steady_clock::now(); std::vector threads; threads.reserve(kTotalClients); // Writer threads: ingest with embeddings pipeline active for (int w = 0; w < kWriterClients; ++w) { threads.emplace_back([&, w]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 60s; // embedding can be slow DaemonClient client(ccfg); while (!go.load()) std::this_thread::yield(); for (int i = 0; i < cfg.docsPerClient; ++i) { AddDocumentRequest req; req.name = "embed_w" + std::to_string(w) + "_" + std::to_string(i) + ".md"; req.content = generateDocument(w, i, cfg.docSizeBytes); req.tags = {"bench", "embeddings", "writer-" + std::to_string(w)}; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.streamingAddDocument(req), 60s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res && !res.value().hash.empty()) { addOps.fetch_add(1); { std::lock_guard lk(addLatMutex); addLatencies.push_back(latUs); } { std::lock_guard lk(hashMutex); ingestedHashes.push_back(res.value().hash); } } else { addFails.fetch_add(1); } } }); } // Search threads: query while ingest + embeddings are running for (int s = 0; s < kSearchClients; ++s) { threads.emplace_back([&]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); while (!go.load()) std::this_thread::yield(); static const char* queries[] = { "architecture design", "performance tuning parameters", "ingestion pipeline stages", "resource governance pressure", "IPC transport protocol", }; // Keep searching until writers finish int i = 0; while (!writersFinished.load()) { SearchRequest req; req.query = queries[i % 5]; req.limit = 10; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.search(req), 30s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { searchOps.fetch_add(1); std::lock_guard lk(searchLatMutex); searchLatencies.push_back(latUs); } else { searchFails.fetch_add(1); } ++i; } }); } go.store(true); // Wait for writers, then signal searchers to stop for (int w = 0; w < kWriterClients; ++w) threads[w].join(); writersFinished.store(true); // Join searchers for (int s = kWriterClients; s < kTotalClients; ++s) threads[s].join(); sampler.stop(); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); // Wait for embeddings pipeline to drain bool drained = waitForDrain(monitorClient, std::chrono::seconds(std::max(cfg.drainTimeoutSecs, 300))); auto finalSnap = DaemonSnapshot::capture(monitorClient); auto addStats = PercentileStats::compute(addLatencies); auto searchStats = PercentileStats::compute(searchLatencies); int totalOps = addOps.load() + searchOps.load(); int totalFails = addFails.load() + searchFails.load(); double addThroughput = safeRate(static_cast(addOps.load()), globalElapsed); double searchThroughput = safeRate(static_cast(searchOps.load()), globalElapsed); std::cout << "\n=== Embeddings Pipeline Results ===\n"; std::cout << " Wall time: " << std::fixed << std::setprecision(2) << globalElapsed << "s\n"; std::cout << " Total ops: " << totalOps << " Failures: " << totalFails << "\n\n"; std::cout << " Ingest:\n"; std::cout << " Added: " << addOps.load() << " (" << std::setprecision(1) << addThroughput << " docs/s) Fails: " << addFails.load() << "\n"; printPercentiles(" Add latency ", addStats); std::cout << "\n Search (concurrent with ingest):\n"; std::cout << " Queries: " << searchOps.load() << " (" << std::setprecision(1) << searchThroughput << " q/s) Fails: " << searchFails.load() << "\n"; printPercentiles(" Search latency", searchStats); std::cout << "\n Pipeline:\n"; std::cout << " Queue drained: " << (drained ? "yes" : "NO") << "\n"; std::cout << " Final docs_total: " << finalSnap.documentsTotal << "\n"; std::cout << " Final memory: " << finalSnap.memoryUsageMb << " MB\n\n"; CHECK(addOps.load() > 0); CHECK(drained); // Emit JSONL json timeSeries = json::array(); for (auto& s : sampler.getSamples()) { timeSeries.push_back(json{{"elapsed_ms", s.elapsedMs}, {"snapshot", s.snap.toJson()}}); } json record{ {"timestamp", isoTimestamp()}, {"test", "embeddings_pipeline"}, {"plugins_enabled", cfg.enablePlugins}, {"real_model_provider", cfg.useRealModelProvider}, {"data_dir", cfg.dataDir ? cfg.dataDir->string() : ""}, {"plugin_dir", cfg.pluginDir ? cfg.pluginDir->string() : ""}, {"writers", kWriterClients}, {"searchers", kSearchClients}, {"docs_per_writer", cfg.docsPerClient}, {"doc_size_bytes", cfg.docSizeBytes}, {"add", json{{"ops", addOps.load()}, {"fails", addFails.load()}, {"throughput_per_sec", addThroughput}, {"latency", addStats.toJson()}}}, {"search", json{{"ops", searchOps.load()}, {"fails", searchFails.load()}, {"throughput_per_sec", searchThroughput}, {"latency", searchStats.toJson()}}}, {"elapsed_seconds", globalElapsed}, {"drained", drained}, {"daemon_snapshot_final", finalSnap.toJson()}, {"time_series", timeSeries}, {"tuning_profile", cfg.tuningProfile}, }; emitJsonl(cfg.outputPath, record); } TEST_CASE("Multi-client ingestion: connection contention", "[!benchmark][multi-client][contention]") { // Stress test: many clients making rapid short-lived connections auto cfg = BenchConfig::fromEnv(); int burstClients = checkedMulToInt(cfg.numClients, 4, kMaxBenchClients * 4); // Higher contention int opsPerClient = 20; // Shorter burst per client DaemonHarness harness(benchHarnessOptions()); REQUIRE(harness.start(kStartTimeout)); // Monitor ClientConfig monCfg; monCfg.socketPath = harness.socketPath(); monCfg.autoStart = false; monCfg.requestTimeout = 10s; DaemonClient monitorClient(monCfg); (void)DaemonSnapshot::capture(monitorClient); // warmup connection std::atomic successCount{0}; std::atomic failCount{0}; std::atomic retryAfterCount{0}; std::mutex latMutex; std::vector allLatencies; auto globalStart = std::chrono::steady_clock::now(); std::vector threads; for (int c = 0; c < burstClients; ++c) { threads.emplace_back([&, c]() { std::vector localLats; for (int i = 0; i < opsPerClient; ++i) { // Alternate between status, list, and add to test different op types ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 10s; DaemonClient client(ccfg); auto t0 = std::chrono::steady_clock::now(); bool success = false; if (i % 3 == 0) { auto res = yams::cli::run_sync(client.status(), 10s); success = static_cast(res); if (res && res.value().retryAfterMs > 0) retryAfterCount.fetch_add(1); } else if (i % 3 == 1) { ListRequest req; req.limit = 5; auto res = yams::cli::run_sync(client.list(req), 10s); success = static_cast(res); } else { AddDocumentRequest req; req.name = "burst_c" + std::to_string(c) + "_" + std::to_string(i) + ".md"; req.content = "Burst test content from client " + std::to_string(c); auto res = yams::cli::run_sync(client.streamingAddDocument(req), 10s); success = static_cast(res); } auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); localLats.push_back(latUs); if (success) successCount.fetch_add(1); else failCount.fetch_add(1); } std::lock_guard lock(latMutex); allLatencies.insert(allLatencies.end(), localLats.begin(), localLats.end()); }); } for (auto& t : threads) t.join(); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); int totalOps = burstClients * opsPerClient; double throughput = static_cast(successCount.load()) / globalElapsed; auto postSnap = DaemonSnapshot::capture(monitorClient); auto latStats = PercentileStats::compute(allLatencies); std::cout << "\n=== Connection Contention (N=" << burstClients << " burst clients) ===\n"; std::cout << " Total ops: " << totalOps << " (success=" << successCount.load() << " fail=" << failCount.load() << ")\n"; std::cout << " Backpressure signals (retryAfter>0): " << retryAfterCount.load() << "\n"; std::cout << " Wall time: " << std::fixed << std::setprecision(2) << globalElapsed << "s\n"; std::cout << " Throughput: " << throughput << " ops/s\n"; printPercentiles("Op latency", latStats); std::cout << " Peak connections: " << postSnap.maxConnections << "\n"; std::cout << " Forced closes: " << postSnap.requestsProcessed << " total reqs processed\n\n"; // Failures acceptable under extreme contention, but should be minority double failRate = static_cast(failCount.load()) / static_cast(totalOps); CHECK(failRate < 0.20); // Less than 20% failure acceptable under burst json record{ {"timestamp", isoTimestamp()}, {"test", "connection_contention"}, {"burst_clients", burstClients}, {"ops_per_client", opsPerClient}, {"total_ops", totalOps}, {"success_count", successCount.load()}, {"fail_count", failCount.load()}, {"retry_after_count", retryAfterCount.load()}, {"elapsed_seconds", globalElapsed}, {"throughput_ops_per_sec", throughput}, {"latency", latStats.toJson()}, {"tuning_profile", cfg.tuningProfile}, }; emitJsonl(cfg.outputPath, record); } TEST_CASE("Multi-client ingestion: reader starvation under ingest load", "[!benchmark][multi-client][starvation]") { // Verify that readers are not starved when heavy ingestion is in progress auto cfg = BenchConfig::fromEnv(); DaemonHarness harness(benchHarnessOptions()); REQUIRE(harness.start(kStartTimeout)); // Pre-seed so readers have something to query { ClientConfig warmCfg; warmCfg.socketPath = harness.socketPath(); warmCfg.autoStart = false; warmCfg.requestTimeout = 30s; DaemonClient warmClient(warmCfg); for (int i = 0; i < 30; ++i) { AddDocumentRequest req; req.name = "preseed_" + std::to_string(i) + ".md"; req.content = generateDocument(0, i, cfg.docSizeBytes); yams::cli::run_sync(warmClient.streamingAddDocument(req), 30s); } (void)waitForDrain(warmClient, 30s); } // Writer clients (heavy ingestion) int numWriters = std::max(1, cfg.numClients - 1); // Reader clients (continuous reads) int numReaders = std::max(1, cfg.numClients / 2); std::atomic done{false}; std::atomic writerDocs{0}; std::atomic readerOps{0}; std::atomic readerFails{0}; std::mutex readerLatMutex; std::vector readerLatencies; auto globalStart = std::chrono::steady_clock::now(); // Launch writers std::vector writerThreads; for (int w = 0; w < numWriters; ++w) { writerThreads.emplace_back([&, w]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); for (int i = 0; i < cfg.docsPerClient; ++i) { AddDocumentRequest req; req.name = "writer" + std::to_string(w) + "_" + std::to_string(i) + ".md"; req.content = generateDocument(w, i, cfg.docSizeBytes); auto res = yams::cli::run_sync(client.streamingAddDocument(req), 30s); if (res) writerDocs.fetch_add(1); } }); } // Launch readers (run until writers finish) std::vector readerThreads; for (int r = 0; r < numReaders; ++r) { readerThreads.emplace_back([&]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 10s; DaemonClient client(ccfg); while (!done.load()) { auto t0 = std::chrono::steady_clock::now(); ListRequest req; req.limit = 10; auto res = yams::cli::run_sync(client.list(req), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { readerOps.fetch_add(1); std::lock_guard lock(readerLatMutex); readerLatencies.push_back(latUs); } else { readerFails.fetch_add(1); } // Small delay to avoid pure spin std::this_thread::sleep_for(10ms); } }); } // Wait for writers to finish for (auto& t : writerThreads) t.join(); done.store(true); // Wait for readers to stop for (auto& t : readerThreads) t.join(); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); auto readerStats = PercentileStats::compute(readerLatencies); std::cout << "\n=== Reader Starvation Test ===\n"; std::cout << " Writers: " << numWriters << " (ingested " << writerDocs.load() << " docs)\n"; std::cout << " Readers: " << numReaders << " (" << readerOps.load() << " ops, " << readerFails.load() << " fails)\n"; std::cout << " Wall time: " << std::fixed << std::setprecision(2) << globalElapsed << "s\n"; printPercentiles("Reader latency", readerStats); // Reader p99 should stay reasonable even under write load // This threshold can be tuned based on baseline measurements if (readerStats.count > 0) { std::cout << " Reader throughput: " << safeRate(static_cast(readerOps.load()), globalElapsed) << " ops/s\n\n"; } CHECK(readerOps.load() > 0); // Readers must make progress CHECK(readerFails.load() < readerOps.load()); // Most reads should succeed json record{ {"timestamp", isoTimestamp()}, {"test", "reader_starvation"}, {"num_writers", numWriters}, {"num_readers", numReaders}, {"docs_per_writer", cfg.docsPerClient}, {"total_writer_docs", writerDocs.load()}, {"total_reader_ops", readerOps.load()}, {"total_reader_fails", readerFails.load()}, {"elapsed_seconds", globalElapsed}, {"reader_latency", readerStats.toJson()}, {"tuning_profile", cfg.tuningProfile}, }; emitJsonl(cfg.outputPath, record); } TEST_CASE("Multi-client ingestion: file access under ingest load", "[!benchmark][multi-client][access]") { // Validate get/cat responsiveness while concurrent ingestion is running. auto cfg = BenchConfig::fromEnv(); DaemonHarness harness(benchHarnessOptions()); REQUIRE(harness.start(kStartTimeout)); // Seed a baseline corpus so readers always have accessible hashes. std::vector knownHashes; std::mutex knownHashesMutex; { ClientConfig warmCfg; warmCfg.socketPath = harness.socketPath(); warmCfg.autoStart = false; warmCfg.requestTimeout = 30s; DaemonClient warmClient(warmCfg); for (int i = 0; i < 64; ++i) { AddDocumentRequest req; req.name = "access_preseed_" + std::to_string(i) + ".md"; req.content = generateDocument(700, i, cfg.docSizeBytes); req.tags = {"bench", "access-preseed"}; auto res = yams::cli::run_sync(warmClient.streamingAddDocument(req), 30s); if (res && !res.value().hash.empty()) { std::lock_guard lk(knownHashesMutex); knownHashes.push_back(res.value().hash); } } (void)waitForDrain(warmClient, 30s); } const int numWriters = std::max(1, cfg.numClients - std::max(1, cfg.numClients / 3)); const int numReaders = std::max(1, cfg.numClients / 3); std::atomic done{false}; std::atomic writerDocs{0}; std::atomic readerOps{0}; std::atomic readerFails{0}; std::atomic readerAttempts{0}; std::atomic getOps{0}; std::atomic catOps{0}; std::mutex getLatMutex; std::mutex catLatMutex; std::vector getLatencies; std::vector catLatencies; auto pickHash = [&](std::mt19937& rng) -> std::string { std::lock_guard lk(knownHashesMutex); if (knownHashes.empty()) { return std::string{}; } std::uniform_int_distribution dist(0, knownHashes.size() - 1); return knownHashes[dist(rng)]; }; auto globalStart = std::chrono::steady_clock::now(); std::vector writerThreads; writerThreads.reserve(numWriters); for (int w = 0; w < numWriters; ++w) { writerThreads.emplace_back([&, w]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 30s; DaemonClient client(ccfg); for (int i = 0; i < cfg.docsPerClient; ++i) { AddDocumentRequest req; req.name = "access_writer" + std::to_string(w) + "_" + std::to_string(i) + ".md"; req.content = generateDocument(w, i, cfg.docSizeBytes); req.tags = {"bench", "access", "writer-" + std::to_string(w)}; auto res = yams::cli::run_sync(client.streamingAddDocument(req), 30s); if (res) { writerDocs.fetch_add(1); if (!res.value().hash.empty()) { std::lock_guard lk(knownHashesMutex); knownHashes.push_back(res.value().hash); } } } }); } std::vector readerThreads; readerThreads.reserve(numReaders); for (int r = 0; r < numReaders; ++r) { readerThreads.emplace_back([&, r]() { ClientConfig ccfg; ccfg.socketPath = harness.socketPath(); ccfg.autoStart = false; ccfg.requestTimeout = 10s; DaemonClient client(ccfg); std::mt19937 rng(static_cast(r * 1009 + 17)); bool doGet = true; while (!done.load()) { std::string hash = pickHash(rng); if (hash.empty()) { std::this_thread::sleep_for(5ms); continue; } readerAttempts.fetch_add(1); // Alternate get (metadata-focused) and cat (content-focused) if (doGet) { GetRequest req; req.hash = hash; req.metadataOnly = true; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.get(req), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { getOps.fetch_add(1); readerOps.fetch_add(1); std::lock_guard lk(getLatMutex); getLatencies.push_back(latUs); } else { readerFails.fetch_add(1); } } else { CatRequest req; req.hash = hash; auto t0 = std::chrono::steady_clock::now(); auto res = yams::cli::run_sync(client.cat(req), 10s); auto t1 = std::chrono::steady_clock::now(); auto latUs = std::chrono::duration_cast(t1 - t0).count(); if (res) { catOps.fetch_add(1); readerOps.fetch_add(1); std::lock_guard lk(catLatMutex); catLatencies.push_back(latUs); } else { readerFails.fetch_add(1); } } doGet = !doGet; std::this_thread::sleep_for(2ms); } }); } for (auto& t : writerThreads) t.join(); done.store(true); for (auto& t : readerThreads) t.join(); ClientConfig monCfg; monCfg.socketPath = harness.socketPath(); monCfg.autoStart = false; monCfg.requestTimeout = 10s; DaemonClient monitorClient(monCfg); bool drained = waitForDrain(monitorClient, std::chrono::seconds(cfg.drainTimeoutSecs)); auto globalEnd = std::chrono::steady_clock::now(); double globalElapsed = std::chrono::duration(globalEnd - globalStart).count(); auto getStats = PercentileStats::compute(getLatencies); auto catStats = PercentileStats::compute(catLatencies); std::cout << "\n=== File Access Under Ingest Load ===\n"; std::cout << " Writers: " << numWriters << " (ingested " << writerDocs.load() << " docs)\n"; std::cout << " Readers: " << numReaders << " (ops=" << readerOps.load() << ", attempts=" << readerAttempts.load() << ", get=" << getOps.load() << ", cat=" << catOps.load() << ", fails=" << readerFails.load() << ")\n"; std::cout << " Wall time: " << std::fixed << std::setprecision(2) << globalElapsed << "s\n"; printPercentiles("Get latency", getStats); printPercentiles("Cat latency", catStats); std::cout << " Queue drained: " << (drained ? "yes" : "NO") << "\n\n"; CHECK(writerDocs.load() > 0); CHECK(readerOps.load() > 0); CHECK(readerFails.load() < readerAttempts.load()); CHECK(drained); json record{ {"timestamp", isoTimestamp()}, {"test", "access_under_ingest_load"}, {"num_clients", cfg.numClients}, {"num_writers", numWriters}, {"num_readers", numReaders}, {"docs_per_writer", cfg.docsPerClient}, {"doc_size_bytes", cfg.docSizeBytes}, {"total_writer_docs", writerDocs.load()}, {"reader_attempts", readerAttempts.load()}, {"reader_ops", readerOps.load()}, {"get_ops", getOps.load()}, {"cat_ops", catOps.load()}, {"reader_fails", readerFails.load()}, {"elapsed_seconds", globalElapsed}, {"get_latency", getStats.toJson()}, {"cat_latency", catStats.toJson()}, {"drained", drained}, {"tuning_profile", cfg.tuningProfile}, }; emitJsonl(cfg.outputPath, record); }