Files
foxhunt/tests/rdtsc_performance_validation.rs
jgrusewski 6093eac7bf 🔧 Tonic 0.14 Upgrade: Auto-generated and build system changes
Wave 64-65 cleanup: Proto regeneration and build system updates from Tonic 0.12→0.14 upgrade

Files updated:
- Cargo.lock: Dependency resolution for Tonic 0.14.2
- All build.rs: Updated for tonic-prost-build
- Proto files: Regenerated with tonic-prost 0.14
- Examples/tests: Updated for new gRPC API

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-10-03 07:34:26 +02:00

717 lines
23 KiB
Rust

//! RDTSC Performance Validation Tests
//!
//! This module implements REAL hardware-level performance testing using RDTSC (Read Time-Stamp Counter)
//! to validate the Foxhunt HFT system meets its <14ns latency requirements.
//!
//! ## Test Coverage:
//! 1. **RDTSC Precision Testing**: Validate timing accuracy to nanosecond level
//! 2. **Lock-free Operations**: Test atomic operations and lock-free data structures
//! 3. **SIMD Performance**: Validate vectorized operations meet performance targets
//! 4. **CPU Cache Performance**: Test memory access patterns and cache efficiency
//! 5. **System Call Overhead**: Measure OS interaction costs
//! 6. **Thread Context Switching**: Validate multi-threading performance
//! 7. **Memory Allocation**: Test allocation patterns and performance
//! 8. **Network I/O Latency**: Measure network stack performance
//!
//! ## Performance Targets:
//! - RDTSC timing resolution: <14ns
//! - Lock-free queue operations: <100ns
//! - SIMD operations: <1μs
//! - Cache miss penalty: <50ns
//! - System call overhead: <1μs
//! - Context switch time: <10μs
//! - Memory allocation: <100ns
//! - Network round-trip: <50μs
#![warn(missing_docs)]
#![warn(clippy::all)]
#![allow(clippy::too_many_arguments)]
#![allow(unused_crate_dependencies)]
use criterion::{black_box, BenchmarkId, Criterion};
use std::arch::x86_64::*;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tracing::{error, info, warn};
// REMOVED: rustc_private feature usage is unstable and not needed
// #[cfg(target_os = "linux")]
// extern crate libc;
// use trading_engine::prelude::*; // REMOVED - prelude does not exist
/// RDTSC performance validation test suite
pub struct RdtscPerformanceValidator {
/// Number of iterations for statistical significance
iterations: usize,
/// TSC frequency for time conversion
tsc_frequency: u64,
/// Test results storage
results: Vec<PerformanceTestResult>,
}
/// Individual performance test result
#[derive(Debug, Clone)]
pub struct PerformanceTestResult {
/// Test name
pub name: String,
/// Measured latency in nanoseconds
pub latency_ns: u64,
/// Whether test passed the target
pub passed: bool,
/// Target latency in nanoseconds
pub target_ns: u64,
/// Number of samples taken
pub samples: usize,
/// Standard deviation of measurements
pub std_dev_ns: f64,
/// Percentile measurements
pub percentiles: PerformancePercentiles,
}
/// Performance percentile measurements
#[derive(Debug, Clone)]
pub struct PerformancePercentiles {
/// 50th percentile (median)
pub p50_ns: u64,
/// 95th percentile
pub p95_ns: u64,
/// 99th percentile
pub p99_ns: u64,
/// 99.9th percentile
pub p999_ns: u64,
/// Maximum measurement
pub max_ns: u64,
/// Minimum measurement
pub min_ns: u64,
}
impl RdtscPerformanceValidator {
/// Create a new RDTSC performance validator
pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
let iterations = 100_000; // Large sample size for statistical significance
let tsc_frequency = Self::calibrate_tsc_frequency()?;
info!("Initialized RDTSC Performance Validator");
info!("TSC Frequency: {} Hz", tsc_frequency);
info!("Test iterations: {}", iterations);
Ok(Self {
iterations,
tsc_frequency,
results: Vec::new(),
})
}
/// Run comprehensive RDTSC performance validation
pub async fn run_comprehensive_validation(&mut self) -> Result<(), Box<dyn std::error::Error>> {
info!("🚀 Starting comprehensive RDTSC performance validation");
// Warm up the CPU and stabilize frequency
self.cpu_warmup().await?;
// Test 1: RDTSC timing precision
info!("Test 1/8: RDTSC timing precision");
self.test_rdtsc_precision().await?;
// Test 2: Lock-free atomic operations
info!("Test 2/8: Lock-free atomic operations");
self.test_lockfree_operations().await?;
// Test 3: SIMD vectorized operations
info!("Test 3/8: SIMD vectorized operations");
self.test_simd_operations().await?;
// Test 4: CPU cache performance
info!("Test 4/8: CPU cache performance");
self.test_cache_performance().await?;
// Test 5: System call overhead
info!("Test 5/8: System call overhead");
self.test_syscall_overhead().await?;
// Test 6: Thread context switching
info!("Test 6/8: Thread context switching");
self.test_context_switching().await?;
// Test 7: Memory allocation performance
info!("Test 7/8: Memory allocation performance");
self.test_memory_allocation().await?;
// Test 8: Network I/O latency
info!("Test 8/8: Network I/O latency");
self.test_network_io().await?;
// Generate comprehensive report
self.generate_performance_report().await?;
Ok(())
}
/// Test RDTSC timing precision
async fn test_rdtsc_precision(&mut self) -> Result<(), Box<dyn std::error::Error>> {
let mut measurements = Vec::with_capacity(self.iterations);
for _ in 0..self.iterations {
let start_tsc = unsafe { _rdtsc() };
// Minimal operation to measure timing resolution
black_box(1 + 1);
let end_tsc = unsafe { _rdtsc() };
let cycles = end_tsc - start_tsc;
let nanoseconds = self.cycles_to_nanoseconds(cycles);
measurements.push(nanoseconds);
}
let result = self.analyze_measurements("RDTSC Precision", &measurements, 14)?;
self.results.push(result.clone());
if result.passed {
info!(
"✅ RDTSC precision: {}ns (target: <14ns)",
result.latency_ns
);
} else {
warn!(
"⚠️ RDTSC precision: {}ns (target: <14ns)",
result.latency_ns
);
}
Ok(())
}
/// Test lock-free atomic operations
async fn test_lockfree_operations(&mut self) -> Result<(), Box<dyn std::error::Error>> {
let counter = Arc::new(AtomicU64::new(0));
let mut measurements = Vec::with_capacity(self.iterations);
for _ in 0..self.iterations {
let start_tsc = unsafe { _rdtsc() };
// Atomic increment operation
counter.fetch_add(1, Ordering::Relaxed);
let end_tsc = unsafe { _rdtsc() };
let cycles = end_tsc - start_tsc;
let nanoseconds = self.cycles_to_nanoseconds(cycles);
measurements.push(nanoseconds);
}
let result =
self.analyze_measurements("Lock-free Atomic Operations", &measurements, 100)?;
self.results.push(result.clone());
if result.passed {
info!(
"✅ Lock-free operations: {}ns (target: <100ns)",
result.latency_ns
);
} else {
warn!(
"⚠️ Lock-free operations: {}ns (target: <100ns)",
result.latency_ns
);
}
Ok(())
}
/// Test SIMD vectorized operations
async fn test_simd_operations(&mut self) -> Result<(), Box<dyn std::error::Error>> {
// Check for AVX2 support
if !is_x86_feature_detected!("avx2") {
warn!("AVX2 not supported, skipping SIMD tests");
return Ok(());
}
let data = vec![1.0f32; 256]; // 256 floats for AVX2 processing
let mut measurements = Vec::with_capacity(self.iterations);
for _ in 0..self.iterations {
let start_tsc = unsafe { _rdtsc() };
// SIMD vector addition
unsafe {
self.simd_vector_add(&data);
}
let end_tsc = unsafe { _rdtsc() };
let cycles = end_tsc - start_tsc;
let nanoseconds = self.cycles_to_nanoseconds(cycles);
measurements.push(nanoseconds);
}
let result = self.analyze_measurements("SIMD Operations", &measurements, 1000)?;
self.results.push(result.clone());
if result.passed {
info!(
"✅ SIMD operations: {}ns (target: <1000ns)",
result.latency_ns
);
} else {
warn!(
"⚠️ SIMD operations: {}ns (target: <1000ns)",
result.latency_ns
);
}
Ok(())
}
/// Test CPU cache performance
async fn test_cache_performance(&mut self) -> Result<(), Box<dyn std::error::Error>> {
// Test L1 cache performance with sequential access
let cache_line_size = 64;
let l1_cache_size = 32 * 1024; // 32KB typical L1 cache
let data_size = l1_cache_size / std::mem::size_of::<u64>();
let data = vec![0u64; data_size];
let mut measurements = Vec::with_capacity(self.iterations);
for _ in 0..self.iterations {
let start_tsc = unsafe { _rdtsc() };
// Sequential memory access pattern (cache-friendly)
let mut sum = 0u64;
for i in (0..data.len()).step_by(cache_line_size / std::mem::size_of::<u64>()) {
sum = sum.wrapping_add(unsafe { *data.get_unchecked(i) });
}
black_box(sum);
let end_tsc = unsafe { _rdtsc() };
let cycles = end_tsc - start_tsc;
let nanoseconds = self.cycles_to_nanoseconds(cycles);
measurements.push(nanoseconds);
}
let result = self.analyze_measurements("Cache Performance", &measurements, 50)?;
self.results.push(result.clone());
if result.passed {
info!(
"✅ Cache performance: {}ns (target: <50ns)",
result.latency_ns
);
} else {
warn!(
"⚠️ Cache performance: {}ns (target: <50ns)",
result.latency_ns
);
}
Ok(())
}
/// Test system call overhead
async fn test_syscall_overhead(&mut self) -> Result<(), Box<dyn std::error::Error>> {
let mut measurements = Vec::with_capacity(self.iterations);
for _ in 0..self.iterations {
let start_tsc = unsafe { _rdtsc() };
// Minimal system call - get process ID
unsafe {
libc::getpid();
}
let end_tsc = unsafe { _rdtsc() };
let cycles = end_tsc - start_tsc;
let nanoseconds = self.cycles_to_nanoseconds(cycles);
measurements.push(nanoseconds);
}
let result = self.analyze_measurements("System Call Overhead", &measurements, 1000)?;
self.results.push(result.clone());
if result.passed {
info!(
"✅ System call overhead: {}ns (target: <1000ns)",
result.latency_ns
);
} else {
warn!(
"⚠️ System call overhead: {}ns (target: <1000ns)",
result.latency_ns
);
}
Ok(())
}
/// Test thread context switching
async fn test_context_switching(&mut self) -> Result<(), Box<dyn std::error::Error>> {
use std::sync::mpsc;
use std::thread;
let mut measurements = Vec::with_capacity(100); // Fewer iterations due to overhead
for _ in 0..100 {
let (tx, rx) = mpsc::channel();
let (tx_back, rx_back) = mpsc::channel();
let start_tsc = unsafe { _rdtsc() };
// Create thread and measure round-trip time
let handle = thread::spawn(move || {
let _msg = rx.recv().unwrap();
tx_back.send(()).unwrap();
});
tx.send(()).unwrap();
rx_back.recv().unwrap();
let end_tsc = unsafe { _rdtsc() };
let cycles = end_tsc - start_tsc;
let nanoseconds = self.cycles_to_nanoseconds(cycles);
measurements.push(nanoseconds);
handle.join().unwrap();
}
let result = self.analyze_measurements("Context Switching", &measurements, 10000)?;
self.results.push(result.clone());
if result.passed {
info!(
"✅ Context switching: {}ns (target: <10000ns)",
result.latency_ns
);
} else {
warn!(
"⚠️ Context switching: {}ns (target: <10000ns)",
result.latency_ns
);
}
Ok(())
}
/// Test memory allocation performance
async fn test_memory_allocation(&mut self) -> Result<(), Box<dyn std::error::Error>> {
let mut measurements = Vec::with_capacity(self.iterations);
for _ in 0..self.iterations {
let start_tsc = unsafe { _rdtsc() };
// Small allocation typical for HFT operations
let _data: Vec<u64> = Vec::with_capacity(64);
let end_tsc = unsafe { _rdtsc() };
let cycles = end_tsc - start_tsc;
let nanoseconds = self.cycles_to_nanoseconds(cycles);
measurements.push(nanoseconds);
}
let result = self.analyze_measurements("Memory Allocation", &measurements, 100)?;
self.results.push(result.clone());
if result.passed {
info!(
"✅ Memory allocation: {}ns (target: <100ns)",
result.latency_ns
);
} else {
warn!(
"⚠️ Memory allocation: {}ns (target: <100ns)",
result.latency_ns
);
}
Ok(())
}
/// Test network I/O latency (localhost loopback)
async fn test_network_io(&mut self) -> Result<(), Box<dyn std::error::Error>> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
// Start a simple echo server
let listener = TcpListener::bind("127.0.0.1:0").await?;
let addr = listener.local_addr()?;
let server_handle = tokio::spawn(async move {
while let Ok((mut socket, _)) = listener.accept().await {
tokio::spawn(async move {
let mut buf = [0; 8];
if socket.read_exact(&mut buf).await.is_ok() {
let _ = socket.write_all(&buf).await;
}
});
}
});
// Give server time to start
tokio::time::sleep(Duration::from_millis(10)).await;
let mut measurements = Vec::with_capacity(1000); // Fewer iterations for network I/O
for _ in 0..1000 {
let start_tsc = unsafe { _rdtsc() };
// Connect and send/receive data
if let Ok(mut stream) = TcpStream::connect(addr).await {
let test_data = [0x42u8; 8];
if stream.write_all(&test_data).await.is_ok() {
let mut response = [0u8; 8];
let _ = stream.read_exact(&mut response).await;
}
}
let end_tsc = unsafe { _rdtsc() };
let cycles = end_tsc - start_tsc;
let nanoseconds = self.cycles_to_nanoseconds(cycles);
measurements.push(nanoseconds);
}
server_handle.abort();
let result = self.analyze_measurements("Network I/O", &measurements, 50000)?;
self.results.push(result.clone());
if result.passed {
info!("✅ Network I/O: {}ns (target: <50000ns)", result.latency_ns);
} else {
warn!(
"⚠️ Network I/O: {}ns (target: <50000ns)",
result.latency_ns
);
}
Ok(())
}
/// Analyze measurement data and calculate statistics
fn analyze_measurements(
&self,
name: &str,
measurements: &[u64],
target_ns: u64,
) -> Result<PerformanceTestResult, Box<dyn std::error::Error>> {
if measurements.is_empty() {
return Err("No measurements available".into());
}
let mut sorted = measurements.to_vec();
sorted.sort_unstable();
let sum: u64 = measurements.iter().sum();
let mean = sum / measurements.len() as u64;
let variance = measurements
.iter()
.map(|&x| {
let diff = x as f64 - mean as f64;
diff * diff
})
.sum::<f64>()
/ measurements.len() as f64;
let std_dev = variance.sqrt();
let percentiles = PerformancePercentiles {
p50_ns: sorted[sorted.len() * 50 / 100],
p95_ns: sorted[sorted.len() * 95 / 100],
p99_ns: sorted[sorted.len() * 99 / 100],
p999_ns: sorted[sorted.len() * 999 / 1000],
max_ns: sorted[sorted.len() - 1],
min_ns: sorted[0],
};
let result = PerformanceTestResult {
name: name.to_string(),
latency_ns: percentiles.p95_ns, // Use P95 for pass/fail determination
passed: percentiles.p95_ns <= target_ns,
target_ns,
samples: measurements.len(),
std_dev_ns: std_dev,
percentiles,
};
Ok(result)
}
/// Generate comprehensive performance report
async fn generate_performance_report(&self) -> Result<(), Box<dyn std::error::Error>> {
info!("📊 RDTSC Performance Validation Report");
info!("=====================================");
let mut total_tests = 0;
let mut passed_tests = 0;
for result in &self.results {
total_tests += 1;
if result.passed {
passed_tests += 1;
}
let status = if result.passed {
"✅ PASS"
} else {
"❌ FAIL"
};
info!("{} - {}", status, result.name);
info!(
" P50: {}ns, P95: {}ns, P99: {}ns, P99.9: {}ns",
result.percentiles.p50_ns,
result.percentiles.p95_ns,
result.percentiles.p99_ns,
result.percentiles.p999_ns
);
info!(
" Target: {}ns, Samples: {}, StdDev: {:.2}ns",
result.target_ns, result.samples, result.std_dev_ns
);
info!("");
}
let success_rate = (passed_tests as f64 / total_tests as f64) * 100.0;
info!(
"Overall Results: {}/{} tests passed ({:.1}%)",
passed_tests, total_tests, success_rate
);
if passed_tests == total_tests {
info!("🎉 All performance tests PASSED - System meets HFT requirements!");
} else {
warn!("⚠️ Some performance tests FAILED - Review system configuration");
}
Ok(())
}
/// CPU warmup to stabilize frequency and cache state
async fn cpu_warmup(&self) -> Result<(), Box<dyn std::error::Error>> {
info!("Warming up CPU and stabilizing frequency...");
let warmup_start = Instant::now();
let warmup_duration = Duration::from_secs(2);
while warmup_start.elapsed() < warmup_duration {
// CPU-intensive work to boost frequency
let mut sum = 0u64;
for i in 0..10000 {
sum = sum.wrapping_add(i);
}
black_box(sum);
}
info!("CPU warmup completed");
Ok(())
}
/// Calibrate TSC frequency
fn calibrate_tsc_frequency() -> Result<u64, Box<dyn std::error::Error>> {
let start_time = Instant::now();
let start_tsc = unsafe { _rdtsc() };
std::thread::sleep(Duration::from_millis(100));
let end_time = Instant::now();
let end_tsc = unsafe { _rdtsc() };
let elapsed_ns = (end_time - start_time).as_nanos() as u64;
let elapsed_cycles = end_tsc - start_tsc;
let frequency = (elapsed_cycles * 1_000_000_000) / elapsed_ns;
Ok(frequency)
}
/// Convert TSC cycles to nanoseconds
fn cycles_to_nanoseconds(&self, cycles: u64) -> u64 {
(cycles * 1_000_000_000) / self.tsc_frequency
}
/// SIMD vector addition using AVX2
#[target_feature(enable = "avx2")]
unsafe fn simd_vector_add(&self, data: &[f32]) {
let chunks = data.chunks_exact(8);
for chunk in chunks {
let a = _mm256_loadu_ps(chunk.as_ptr());
let b = _mm256_set1_ps(1.0);
let result = _mm256_add_ps(a, b);
black_box(result);
}
}
}
/// Run RDTSC performance validation tests
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_rdtsc_performance_validation() {
let mut validator =
RdtscPerformanceValidator::new().expect("Failed to create RDTSC validator");
validator
.run_comprehensive_validation()
.await
.expect("RDTSC performance validation failed");
// Check that critical tests pass
let critical_tests = ["RDTSC Precision", "Lock-free Atomic Operations"];
for test_name in &critical_tests {
let result = validator
.results
.iter()
.find(|r| r.name == *test_name)
.expect(&format!("Test '{}' not found", test_name));
assert!(
result.passed,
"Critical test '{}' failed: {}ns > {}ns target",
test_name, result.latency_ns, result.target_ns
);
}
}
#[tokio::test]
async fn test_individual_rdtsc_precision() {
let mut validator =
RdtscPerformanceValidator::new().expect("Failed to create RDTSC validator");
validator
.test_rdtsc_precision()
.await
.expect("RDTSC precision test failed");
let result = &validator.results[0];
assert_eq!(result.name, "RDTSC Precision");
// RDTSC should provide sub-nanosecond precision on modern CPUs
// but we allow up to 14ns due to system variability
info!("RDTSC precision: {}ns (target: <14ns)", result.latency_ns);
}
#[tokio::test]
async fn test_simd_operations() {
if !is_x86_feature_detected!("avx2") {
println!("Skipping SIMD test - AVX2 not supported");
return;
}
let mut validator =
RdtscPerformanceValidator::new().expect("Failed to create RDTSC validator");
validator
.test_simd_operations()
.await
.expect("SIMD operations test failed");
let result = &validator.results[0];
assert_eq!(result.name, "SIMD Operations");
info!("SIMD operations: {}ns (target: <1000ns)", result.latency_ns);
}
}