#include "../common/benchmark_tracker.h" #include "benchmark_base.h" #include #include #include #include #include #include #include #include #include namespace yams::benchmark { namespace { using yams::daemon::MessageFramer; using yams::test::BenchmarkTracker; std::string makeSnippet(size_t length, char seed) { std::string out(length, seed); for (size_t i = 0; i < length; ++i) { out[i] = static_cast(seed + static_cast(i % 23)); } return out; } daemon::SearchResult makeResult(size_t index, size_t snippetBytes) { daemon::SearchResult result; result.id = "doc-" + std::to_string(index); result.path = "/bench/doc-" + std::to_string(index) + ".txt"; result.title = "Benchmark Doc " + std::to_string(index); result.snippet = makeSnippet(snippetBytes, static_cast('a' + (index % 13))); result.score = 1.0 / static_cast(index + 1); result.metadata = {{"generator", "ipc_stream"}, {"bucket", std::to_string(index % 5)}}; return result; } daemon::Message makeMessage(uint64_t requestId, const daemon::Response& response) { daemon::Message msg; msg.version = daemon::PROTOCOL_VERSION; msg.requestId = requestId; msg.timestamp = std::chrono::steady_clock::now(); msg.expectsStreamingResponse = true; msg.payload = response; return msg; } } // namespace class StreamingFramerBenchmark : public BenchmarkBase { public: StreamingFramerBenchmark(std::string name, size_t resultsPerChunk, size_t chunkCount, size_t snippetBytes, const Config& config = Config()) : BenchmarkBase(std::move(name), config), resultsPerChunk_(resultsPerChunk), chunkCount_(chunkCount), snippetBytes_(snippetBytes) { buildPayloads(); } protected: size_t runIteration() override { const uint64_t requestId = requestCounter_++; std::size_t operations = 0; // Header frame frameBuffer_.clear(); frameBuffer_.reserve(reserveBytes_); auto headerMessage = makeMessage(requestId, headerResponse_); auto headerResult = framer_.frame_message_header_into(headerMessage, frameBuffer_); if (!headerResult) { throw std::runtime_error("Failed to frame streaming header: " + headerResult.error().message); } bytesAccumulated_ += frameBuffer_.size(); operations++; // Chunk frames for (size_t i = 0; i < chunkResponses_.size(); ++i) { frameBuffer_.clear(); auto chunkMessage = makeMessage(requestId, chunkResponses_[i]); bool last = (i + 1 == chunkResponses_.size()); auto chunkResult = framer_.frame_message_chunk_into(chunkMessage, frameBuffer_, last); if (!chunkResult) { throw std::runtime_error("Failed to frame streaming chunk: " + chunkResult.error().message); } bytesAccumulated_ += frameBuffer_.size(); operations++; } iterations_++; return operations; } void collectCustomMetrics(std::map& metrics) override { if (iterations_ == 0) return; metrics["bytes_per_iteration"] = static_cast(bytesAccumulated_) / static_cast(iterations_); metrics["chunks_per_iteration"] = static_cast(chunkCount_); metrics["results_per_chunk"] = static_cast(resultsPerChunk_); } private: void buildPayloads() { const size_t totalResults = resultsPerChunk_ * chunkCount_; daemon::SearchResponse headerPayload; headerPayload.totalCount = totalResults; headerPayload.elapsed = std::chrono::milliseconds(12); headerResponse_ = daemon::Response{std::in_place_type, headerPayload}; chunkResponses_.clear(); chunkResponses_.reserve(chunkCount_); size_t resultIndex = 0; for (size_t chunk = 0; chunk < chunkCount_; ++chunk) { daemon::SearchResponse chunkPayload; chunkPayload.totalCount = totalResults; chunkPayload.elapsed = std::chrono::milliseconds(20 + static_cast(chunk)); chunkPayload.results.reserve(resultsPerChunk_); for (size_t i = 0; i < resultsPerChunk_; ++i) { chunkPayload.results.push_back(makeResult(resultIndex++, snippetBytes_)); } daemon::Response chunkResp{std::in_place_type, chunkPayload}; chunkResponses_.push_back(std::move(chunkResp)); } reserveBytes_ = MessageFramer::HEADER_SIZE + resultsPerChunk_ * (snippetBytes_ + 256); frameBuffer_.reserve(reserveBytes_); } MessageFramer framer_; std::vector frameBuffer_; daemon::Response headerResponse_; std::vector chunkResponses_; size_t resultsPerChunk_; size_t chunkCount_; size_t snippetBytes_; size_t reserveBytes_ = MessageFramer::HEADER_SIZE * 2; size_t iterations_ = 0; uint64_t requestCounter_ = 1; size_t bytesAccumulated_ = 0; }; class UnaryFramerBenchmark : public BenchmarkBase { public: explicit UnaryFramerBenchmark(std::string name, size_t payloadBytes, const Config& config = Config()) : BenchmarkBase(std::move(name), config), payloadBytes_(payloadBytes) { buildPayload(); } protected: size_t runIteration() override { frameBuffer_.clear(); frameBuffer_.reserve(reserveBytes_); daemon::Message msg = makeMessage(requestId_++, response_); auto result = framer_.frame_message_into(msg, frameBuffer_); if (!result) { throw std::runtime_error("Failed to frame unary message: " + result.error().message); } bytesAccumulated_ += frameBuffer_.size(); iterations_++; return 1; } void collectCustomMetrics(std::map& metrics) override { if (iterations_ == 0) return; metrics["bytes_per_frame"] = static_cast(bytesAccumulated_) / static_cast(iterations_); } private: void buildPayload() { daemon::SuccessResponse success; success.message = std::string(payloadBytes_, 's'); response_ = daemon::Response{std::in_place_type, success}; reserveBytes_ = MessageFramer::HEADER_SIZE + payloadBytes_ + 64; frameBuffer_.reserve(reserveBytes_); } MessageFramer framer_; std::vector frameBuffer_; daemon::Response response_; size_t payloadBytes_; size_t reserveBytes_ = MessageFramer::HEADER_SIZE * 2; uint64_t requestId_ = 1000; size_t iterations_ = 0; size_t bytesAccumulated_ = 0; }; } // namespace yams::benchmark using yams::benchmark::BenchmarkBase; using yams::benchmark::StreamingFramerBenchmark; using yams::benchmark::UnaryFramerBenchmark; using yams::test::BenchmarkTracker; int main(int argc, char** argv) { BenchmarkBase::Config config; config.verbose = true; config.benchmark_iterations = 8; config.track_memory = false; // /proc/self/status not available on all platforms for (int i = 1; i < argc; ++i) { std::string arg = argv[i]; if (arg == "--quiet") { config.verbose = false; } else if (arg == "--iterations" && i + 1 < argc) { config.benchmark_iterations = std::stoi(argv[++i]); } else if (arg == "--output" && i + 1 < argc) { config.output_file = argv[++i]; } } std::cout << "YAMS IPC Streaming Benchmarks\n"; std::cout << "====================================\n\n"; std::filesystem::path outDir = "bench_results"; std::error_code ec; std::filesystem::create_directories(outDir, ec); if (ec) { std::cerr << "WARNING: unable to create bench_results directory: " << ec.message() << std::endl; } if (config.output_file.empty()) { config.output_file = (outDir / "ipc_stream_benchmarks.json").string(); } BenchmarkTracker tracker(outDir / "ipc_stream_benchmarks.json"); std::vector> benchmarks; benchmarks.push_back(std::make_unique("StreamingFramer_32x10_256B", 32, 10, 256, config)); benchmarks.push_back(std::make_unique("StreamingFramer_64x6_512B", 64, 6, 512, config)); benchmarks.push_back( std::make_unique("UnaryFramer_Success_8KB", 8 * 1024, config)); for (auto& bench : benchmarks) { auto result = bench->run(); BenchmarkTracker::BenchmarkResult trackerResult; trackerResult.name = result.name; trackerResult.value = result.duration_ms; trackerResult.unit = "ms"; trackerResult.timestamp = std::chrono::system_clock::now(); trackerResult.metrics = result.custom_metrics; tracker.recordResult(trackerResult); } tracker.generateReport(outDir / "ipc_stream_benchmark_report.json"); tracker.generateMarkdownReport(outDir / "ipc_stream_benchmark_report.md"); std::cout << "\n====================================\n"; std::cout << "Benchmark complete. Reports generated.\n"; return 0; }