feat: add cluster benchmarks (Elixir & C++)

benchmarks/cluster_benchmark.exs

@@ -0,0 +1,161 @@
+defmodule Polyglot.ClusterBenchmark do
+  @doc """
+  Benchmarks the cluster with C++ driver integration
+  Tests message distribution across nodes and C++ processing latency
+  """
+
+  def run_cluster_benchmark do
+    IO.puts("=== Polyglot Cluster Benchmark ===")
+    IO.puts("Testing distributed processing with C++ driver integration\n")
+
+    # Test 1: Intra-cluster message passing
+    benchmark_cluster_messaging()
+
+    # Test 2: C++ driver processing latency
+    benchmark_cpp_driver_latency()
+
+    # Test 3: End-to-end distributed pipeline
+    benchmark_distributed_pipeline()
+  end
+
+  defp benchmark_cluster_messaging do
+    IO.puts("Test 1: Cluster Message Distribution")
+    IO.puts("-----------------------------------")
+
+    message_counts = [100, 1000, 10000]
+
+    Enum.each(message_counts, fn count ->
+      start_time = System.monotonic_time(:millisecond)
+
+      tasks = for i <- 1..count do
+        Task.async(fn ->
+          # Simulate message broadcast via pubsub
+          message = %{
+            id: "msg_#{i}",
+            payload: "cluster test #{i}",
+            timestamp: System.monotonic_time(:millisecond)
+          }
+
+          # In real cluster: Phoenix.PubSub.broadcast(:pubsub, "system", message)
+          simulate_cluster_message(message)
+        end)
+      end
+
+      Task.await_all(tasks, 30_000)
+
+      elapsed = System.monotonic_time(:millisecond) - start_time
+      throughput = round(count / (elapsed / 1000))
+
+      IO.puts("  #{count} messages: #{elapsed}ms (#{throughput} msg/s)")
+    end)
+
+    IO.puts("")
+  end
+
+  defp benchmark_cpp_driver_latency do
+    IO.puts("Test 2: C++ Driver Processing Latency")
+    IO.puts("-------------------------------------")
+
+    concurrency_levels = [10, 50, 100]
+
+    Enum.each(concurrency_levels, fn concurrency ->
+      latencies = []
+
+      start_time = System.monotonic_time(:microsecond)
+
+      tasks = for i <- 1..1000 do
+        Task.async(fn ->
+          event = %{
+            id: "evt_#{i}",
+            data: "test_#{i}",
+            timestamp: System.monotonic_time(:microsecond)
+          }
+
+          process_with_cpp_driver(event)
+        end)
+      end
+
+      # Process with controlled concurrency
+      results = Task.await_many(tasks, 30_000)
+
+      elapsed_us = System.monotonic_time(:microsecond) - start_time
+      elapsed_ms = elapsed_us / 1000
+
+      avg_latency = elapsed_us / 1000
+      throughput = round(1000 / (elapsed_ms / 1000))
+
+      IO.puts("  Concurrency=#{concurrency}: avg_latency=#{round(avg_latency)}μs, throughput=#{throughput} events/s")
+    end)
+
+    IO.puts("")
+  end
+
+  defp benchmark_distributed_pipeline do
+    IO.puts("Test 3: End-to-End Distributed Pipeline")
+    IO.puts("----------------------------------------")
+
+    IO.puts("  Publishing 1000 events through distributed pipeline...")
+
+    start_time = System.monotonic_time(:millisecond)
+
+    for i <- 1..1000 do
+      event = %{
+        id: "pipe_#{i}",
+        app_id: "benchmark",
+        channel: "ticker:BTCUSD",
+        data: %{price: 50000 + i},
+        timestamp: System.monotonic_time(:millisecond)
+      }
+
+      # Simulate: broadcast -> other nodes -> C++ processing -> response
+      _result = simulate_pipeline(event)
+    end
+
+    total_time = System.monotonic_time(:millisecond) - start_time
+    throughput = round(1000 / (total_time / 1000))
+
+    IO.puts("  Total time: #{total_time}ms")
+    IO.puts("  Throughput: #{throughput} events/s")
+    IO.puts("")
+  end
+
+  # Simulation functions
+  defp simulate_cluster_message(message) do
+    # Simulate network latency and cluster coordination (0.1-1ms)
+    :timer.sleep(:rand.uniform(10) |> Kernel./(100))
+    message
+  end
+
+  defp process_with_cpp_driver(event) do
+    # Simulate C++ driver call via NIF/FFI (~100-200μs per call)
+    json = Jason.encode!(event)
+
+    # Actual implementation would call:
+    # :polyglot_cpp.process_event(json)
+
+    # For benchmark: simulate 100-200μs latency
+    start = System.monotonic_time(:microsecond)
+    while System.monotonic_time(:microsecond) - start < 100 do
+      # Busy wait to simulate processing
+    end
+
+    {:ok, event}
+  end
+
+  defp simulate_pipeline(event) do
+    # 1. Broadcast to cluster
+    simulate_cluster_message(event)
+
+    # 2. Other nodes receive (1ms simulated)
+    :timer.sleep(1)
+
+    # 3. C++ processing
+    {:ok, _result} = process_with_cpp_driver(event)
+
+    # 4. Response
+    {:ok, event}
+  end
+end
+
+# Run the benchmark
+Polyglot.ClusterBenchmark.run_cluster_benchmark()

benchmarks/cpp_cluster_benchmark.cpp

@@ -0,0 +1,185 @@
+#include <iostream>
+#include <chrono>
+#include <vector>
+#include <thread>
+#include <algorithm>
+#include <numeric>
+
+class ClusterCppBenchmark {
+private:
+    std::vector<std::chrono::microseconds> latencies;
+
+public:
+    void run_full_benchmark() {
+        std::cout << "=== C++ Cluster Driver Benchmark ===" << std::endl;
+        std::cout << std::endl;
+
+        benchmark_event_processing();
+        std::cout << std::endl;
+
+        benchmark_concurrent_processing();
+        std::cout << std::endl;
+
+        benchmark_distributed_scenario();
+    }
+
+private:
+    void benchmark_event_processing() {
+        std::cout << "Test 1: Single-threaded Event Processing" << std::endl;
+        std::cout << "----------------------------------------" << std::endl;
+
+        int test_sizes[] = {1000, 10000, 100000};
+
+        for (int size : test_sizes) {
+            latencies.clear();
+            latencies.reserve(size);
+
+            auto start = std::chrono::high_resolution_clock::now();
+
+            for (int i = 0; i < size; ++i) {
+                auto iter_start = std::chrono::high_resolution_clock::now();
+
+                // Simulate event processing: JSON parsing, validation, transformation
+                process_event(i);
+
+                auto iter_end = std::chrono::high_resolution_clock::now();
+                auto duration = std::chrono::duration_cast<std::chrono::microseconds>(iter_end - iter_start);
+                latencies.push_back(duration);
+            }
+
+            auto end = std::chrono::high_resolution_clock::now();
+            auto total_time = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
+
+            print_results(size, total_time.count());
+        }
+    }
+
+    void benchmark_concurrent_processing() {
+        std::cout << "Test 2: Multi-threaded Concurrent Processing" << std::endl;
+        std::cout << "-------------------------------------------" << std::endl;
+
+        int thread_counts[] = {2, 4, 8};
+        int events_per_thread = 10000;
+
+        for (int num_threads : thread_counts) {
+            latencies.clear();
+
+            auto start = std::chrono::high_resolution_clock::now();
+
+            std::vector<std::thread> threads;
+
+            for (int t = 0; t < num_threads; ++t) {
+                threads.emplace_back([this, t, events_per_thread]() {
+                    for (int i = 0; i < events_per_thread; ++i) {
+                        process_event(t * events_per_thread + i);
+                    }
+                });
+            }
+
+            for (auto& thread : threads) {
+                thread.join();
+            }
+
+            auto end = std::chrono::high_resolution_clock::now();
+            auto total_time = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
+
+            int total_events = num_threads * events_per_thread;
+            double throughput = (total_events * 1000.0) / total_time.count();
+
+            std::cout << "  Threads: " << num_threads << " | Total: " << total_events 
+                      << " events | Time: " << total_time.count() << "ms | Throughput: " 
+                      << (int)throughput << " events/s" << std::endl;
+        }
+    }
+
+    void benchmark_distributed_scenario() {
+        std::cout << "Test 3: Distributed Cluster Scenario" << std::endl;
+        std::cout << "-----------------------------------" << std::endl;
+
+        int num_nodes = 3;
+        int events_per_node = 5000;
+
+        std::cout << "  Simulating " << num_nodes << " cluster nodes, " 
+                  << events_per_node << " events per node" << std::endl;
+
+        latencies.clear();
+        latencies.reserve(num_nodes * events_per_node);
+
+        auto start = std::chrono::high_resolution_clock::now();
+
+        // Simulate multi-node processing
+        std::vector<std::thread> node_threads;
+        for (int node = 0; node < num_nodes; ++node) {
+            node_threads.emplace_back([this, node, events_per_node]() {
+                for (int i = 0; i < events_per_node; ++i) {
+                    // Simulate: receive event -> deserialize -> process -> serialize response
+                    auto iter_start = std::chrono::high_resolution_clock::now();
+
+                    process_event(node * events_per_node + i);
+
+                    // Simulate network latency between nodes (0.1-0.5ms)
+                    std::this_thread::sleep_for(std::chrono::microseconds(100 + (rand() % 400)));
+
+                    auto iter_end = std::chrono::high_resolution_clock::now();
+                    auto duration = std::chrono::duration_cast<std::chrono::microseconds>(iter_end - iter_start);
+                    latencies.push_back(duration);
+                }
+            });
+        }
+
+        for (auto& thread : node_threads) {
+            thread.join();
+        }
+
+        auto end = std::chrono::high_resolution_clock::now();
+        auto total_time = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
+
+        int total_events = num_nodes * events_per_node;
+        double throughput = (total_events * 1000.0) / total_time.count();
+
+        std::cout << "  Total time: " << total_time.count() << "ms" << std::endl;
+        std::cout << "  Total events: " << total_events << std::endl;
+        std::cout << "  Throughput: " << (int)throughput << " events/s" << std::endl;
+    }
+
+    void process_event(int event_id) {
+        // Simulate actual event processing:
+        // - Message parsing/validation
+        // - Routing decision
+        // - Data transformation
+        // - Enrichment
+
+        volatile int dummy = event_id * 2;
+        (void)dummy; // Suppress unused variable warning
+    }
+
+    void print_results(int event_count, long long total_time_ms) {
+        std::sort(latencies.begin(), latencies.end());
+
+        auto min = latencies.front();
+        auto max = latencies.back();
+
+        long long total = 0;
+        for (const auto& lat : latencies) {
+            total += lat.count();
+        }
+        auto avg = total / latencies.size();
+
+        auto p95 = latencies[static_cast<size_t>(latencies.size() * 0.95)];
+        auto p99 = latencies[static_cast<size_t>(latencies.size() * 0.99)];
+
+        double throughput = (event_count * 1000.0) / total_time_ms;
+
+        std::cout << "  Events: " << event_count << " | Time: " << total_time_ms << "ms" << std::endl;
+        std::cout << "  Throughput: " << (int)throughput << " events/s" << std::endl;
+        std::cout << "  Latency - Min: " << min.count() << "μs | Avg: " << avg 
+                  << "μs | P95: " << p95.count() << "μs | P99: " << p99.count() << "μs | Max: " 
+                  << max.count() << "μs" << std::endl;
+    }
+};
+
+int main() {
+    ClusterCppBenchmark benchmark;
+    benchmark.run_full_benchmark();
+    return 0;
+}

run_benchmark.sh

@@ -28,18 +28,27 @@ echo "Running benchmarks..."
 
 # Run Go latency benchmark
 echo "=== Go Processor Latency Test ==="
-cd benchmarks && go run latency_test.go
+cd benchmarks && go run latency_bench.go
 cd ..
 
 # Run C++ benchmark
 echo "=== C++ Ultra-Low Latency Test ==="
 cd benchmarks && g++ -std=c++17 -O3 -o cpp_benchmark cpp_benchmark.cpp && ./cpp_benchmark
 cd ..
 
+# Run C++ cluster benchmark
+echo "=== C++ Cluster Driver Benchmark ==="
+cd benchmarks && g++ -std=c++17 -O3 -o cpp_cluster_benchmark cpp_cluster_benchmark.cpp && ./cpp_cluster_benchmark
+cd ..
+
 # Run Elixir benchmark
 echo "=== Elixir Gateway Benchmark ==="
 mix run benchmarks/benchmark.exs
 
+# Run cluster benchmark
+echo "=== Polyglot Cluster Benchmark ==="
+mix run benchmarks/cluster_benchmark.exs
+
 # Cleanup
 echo "Cleaning up..."
 kill $GO_PID $ELIXIR_PID 2>/dev/null