Implement comprehensive performance and load testing suite

+644

1 changed file

expand all

tests

+644

tests/performance_tests.rs

··· 1 + use std::sync::Arc; 2 + use std::time::{Duration, Instant}; 3 + use tokio::time::sleep; 4 + use serde_json::{json, Value}; 5 + use reqwest::Client; 6 + use gigabrain::Graph; 7 + use gigabrain::server::{ServerConfig, rest::RestServer}; 8 + use futures::future::join_all; 9 + use std::sync::atomic::{AtomicU64, Ordering}; 10 + 11 + /// Performance and load testing suite 12 + /// This module tests the system's performance characteristics under various load conditions, 13 + /// measuring throughput, latency, and resource utilization patterns. 14 + 15 + #[derive(Clone)] 16 + pub struct PerformanceTestClient { 17 + client: Client, 18 + base_url: String, 19 + } 20 + 21 + impl PerformanceTestClient { 22 + pub fn new(base_url: &str) -> Self { 23 + Self { 24 + client: Client::builder() 25 + .timeout(Duration::from_secs(30)) 26 + .build() 27 + .expect("Failed to create client"), 28 + base_url: base_url.to_string(), 29 + } 30 + } 31 + 32 + pub async fn post_json(&self, path: &str, body: Value) -> Result<reqwest::Response, reqwest::Error> { 33 + self.client 34 + .post(&format!("{}{}", self.base_url, path)) 35 + .header("Content-Type", "application/json") 36 + .json(&body) 37 + .send() 38 + .await 39 + } 40 + 41 + pub async fn get(&self, path: &str) -> Result<reqwest::Response, reqwest::Error> { 42 + self.client 43 + .get(&format!("{}{}", self.base_url, path)) 44 + .send() 45 + .await 46 + } 47 + 48 + pub async fn put_json(&self, path: &str, body: Value) -> Result<reqwest::Response, reqwest::Error> { 49 + self.client 50 + .put(&format!("{}{}", self.base_url, path)) 51 + .header("Content-Type", "application/json") 52 + .json(&body) 53 + .send() 54 + .await 55 + } 56 + 57 + pub async fn delete(&self, path: &str) -> Result<reqwest::Response, reqwest::Error> { 58 + self.client 59 + .delete(&format!("{}{}", self.base_url, path)) 60 + .send() 61 + .await 62 + } 63 + } 64 + 65 + pub struct PerformanceTestServer { 66 + _handle: tokio::task::JoinHandle<()>, 67 + base_url: String, 68 + graph: Arc<Graph>, 69 + } 70 + 71 + impl PerformanceTestServer { 72 + pub async fn start() -> Result<Self, Box<dyn std::error::Error>> { 73 + use std::net::TcpListener; 74 + 75 + let listener = TcpListener::bind("127.0.0.1:0")?; 76 + let port = listener.local_addr()?.port(); 77 + drop(listener); 78 + 79 + let graph = Arc::new(Graph::new()); 80 + let graph_clone = graph.clone(); 81 + let config = ServerConfig::default(); 82 + let server = RestServer::new(graph_clone, config); 83 + 84 + let handle = tokio::spawn(async move { 85 + if let Err(e) = server.serve(port).await { 86 + eprintln!("Performance test server error: {}", e); 87 + } 88 + }); 89 + 90 + sleep(Duration::from_millis(200)).await; 91 + 92 + let base_url = format!("http://localhost:{}", port); 93 + 94 + Ok(Self { 95 + _handle: handle, 96 + base_url, 97 + graph, 98 + }) 99 + } 100 + 101 + pub fn client(&self) -> PerformanceTestClient { 102 + PerformanceTestClient::new(&self.base_url) 103 + } 104 + 105 + pub fn graph(&self) -> Arc<Graph> { 106 + self.graph.clone() 107 + } 108 + } 109 + 110 + #[derive(Debug)] 111 + pub struct PerformanceMetrics { 112 + pub total_operations: u64, 113 + pub successful_operations: u64, 114 + pub failed_operations: u64, 115 + pub total_duration: Duration, 116 + pub average_latency: Duration, 117 + pub min_latency: Duration, 118 + pub max_latency: Duration, 119 + pub operations_per_second: f64, 120 + } 121 + 122 + impl PerformanceMetrics { 123 + pub fn new(operations: &[Duration], successful: u64, failed: u64, total_duration: Duration) -> Self { 124 + let total_operations = successful + failed; 125 + let (min_latency, max_latency, average_latency) = if operations.is_empty() { 126 + (Duration::ZERO, Duration::ZERO, Duration::ZERO) 127 + } else { 128 + let min = *operations.iter().min().unwrap_or(&Duration::ZERO); 129 + let max = *operations.iter().max().unwrap_or(&Duration::ZERO); 130 + let avg = Duration::from_nanos( 131 + operations.iter().map(|d| d.as_nanos() as u64).sum::<u64>() / operations.len() as u64 132 + ); 133 + (min, max, avg) 134 + }; 135 + 136 + let operations_per_second = if total_duration.as_secs_f64() > 0.0 { 137 + successful as f64 / total_duration.as_secs_f64() 138 + } else { 139 + 0.0 140 + }; 141 + 142 + Self { 143 + total_operations, 144 + successful_operations: successful, 145 + failed_operations: failed, 146 + total_duration, 147 + average_latency, 148 + min_latency, 149 + max_latency, 150 + operations_per_second, 151 + } 152 + } 153 + } 154 + 155 + #[cfg(test)] 156 + mod tests { 157 + use super::*; 158 + 159 + async fn setup() -> PerformanceTestServer { 160 + PerformanceTestServer::start().await.expect("Failed to start test server") 161 + } 162 + 163 + #[tokio::test] 164 + async fn test_node_creation_throughput() { 165 + let server = setup().await; 166 + let client = server.client(); 167 + 168 + const NUM_OPERATIONS: usize = 1000; 169 + let start_time = Instant::now(); 170 + let mut latencies = Vec::with_capacity(NUM_OPERATIONS); 171 + let successful = AtomicU64::new(0); 172 + let failed = AtomicU64::new(0); 173 + 174 + // Sequential node creation test 175 + for i in 0..NUM_OPERATIONS { 176 + let operation_start = Instant::now(); 177 + 178 + let node = json!({ 179 + "labels": ["PerfTest"], 180 + "properties": { 181 + "id": i, 182 + "name": format!("Node_{}", i), 183 + "value": i * 2 184 + } 185 + }); 186 + 187 + match client.post_json("/api/v1/nodes", node).await { 188 + Ok(response) if response.status() == 200 => { 189 + successful.fetch_add(1, Ordering::Relaxed); 190 + } 191 + _ => { 192 + failed.fetch_add(1, Ordering::Relaxed); 193 + } 194 + } 195 + 196 + latencies.push(operation_start.elapsed()); 197 + } 198 + 199 + let total_duration = start_time.elapsed(); 200 + let metrics = PerformanceMetrics::new( 201 + &latencies, 202 + successful.load(Ordering::Relaxed), 203 + failed.load(Ordering::Relaxed), 204 + total_duration, 205 + ); 206 + 207 + println!("Node Creation Performance:"); 208 + println!(" Total Operations: {}", metrics.total_operations); 209 + println!(" Successful: {}", metrics.successful_operations); 210 + println!(" Failed: {}", metrics.failed_operations); 211 + println!(" Total Duration: {:?}", metrics.total_duration); 212 + println!(" Average Latency: {:?}", metrics.average_latency); 213 + println!(" Min Latency: {:?}", metrics.min_latency); 214 + println!(" Max Latency: {:?}", metrics.max_latency); 215 + println!(" Operations/sec: {:.2}", metrics.operations_per_second); 216 + 217 + // Performance assertions 218 + assert!(metrics.operations_per_second > 100.0, "Should handle at least 100 ops/sec"); 219 + assert!(metrics.average_latency < Duration::from_millis(50), "Average latency should be under 50ms"); 220 + assert!(metrics.successful_operations as f64 / metrics.total_operations as f64 > 0.95, "Success rate should be >95%"); 221 + } 222 + 223 + #[tokio::test] 224 + async fn test_concurrent_node_creation_load() { 225 + let server = setup().await; 226 + let client = server.client(); 227 + 228 + const NUM_CONCURRENT: usize = 50; 229 + const OPS_PER_WORKER: usize = 20; 230 + 231 + let start_time = Instant::now(); 232 + let successful = Arc::new(AtomicU64::new(0)); 233 + let failed = Arc::new(AtomicU64::new(0)); 234 + 235 + // Spawn concurrent workers 236 + let mut handles = Vec::with_capacity(NUM_CONCURRENT); 237 + 238 + for worker_id in 0..NUM_CONCURRENT { 239 + let client = client.clone(); 240 + let successful = successful.clone(); 241 + let failed = failed.clone(); 242 + 243 + let handle = tokio::spawn(async move { 244 + let mut worker_latencies = Vec::with_capacity(OPS_PER_WORKER); 245 + 246 + for op_id in 0..OPS_PER_WORKER { 247 + let operation_start = Instant::now(); 248 + 249 + let node = json!({ 250 + "labels": ["ConcurrentPerfTest"], 251 + "properties": { 252 + "worker_id": worker_id, 253 + "op_id": op_id, 254 + "timestamp": start_time.elapsed().as_millis() 255 + } 256 + }); 257 + 258 + match client.post_json("/api/v1/nodes", node).await { 259 + Ok(response) if response.status() == 200 => { 260 + successful.fetch_add(1, Ordering::Relaxed); 261 + } 262 + _ => { 263 + failed.fetch_add(1, Ordering::Relaxed); 264 + } 265 + } 266 + 267 + worker_latencies.push(operation_start.elapsed()); 268 + } 269 + 270 + worker_latencies 271 + }); 272 + 273 + handles.push(handle); 274 + } 275 + 276 + // Wait for all workers to complete 277 + let results = join_all(handles).await; 278 + let total_duration = start_time.elapsed(); 279 + 280 + // Collect all latencies 281 + let mut all_latencies = Vec::new(); 282 + for result in results { 283 + if let Ok(latencies) = result { 284 + all_latencies.extend(latencies); 285 + } 286 + } 287 + 288 + let metrics = PerformanceMetrics::new( 289 + &all_latencies, 290 + successful.load(Ordering::Relaxed), 291 + failed.load(Ordering::Relaxed), 292 + total_duration, 293 + ); 294 + 295 + println!("Concurrent Node Creation Performance:"); 296 + println!(" Workers: {}", NUM_CONCURRENT); 297 + println!(" Ops per Worker: {}", OPS_PER_WORKER); 298 + println!(" Total Operations: {}", metrics.total_operations); 299 + println!(" Successful: {}", metrics.successful_operations); 300 + println!(" Failed: {}", metrics.failed_operations); 301 + println!(" Total Duration: {:?}", metrics.total_duration); 302 + println!(" Average Latency: {:?}", metrics.average_latency); 303 + println!(" Min Latency: {:?}", metrics.min_latency); 304 + println!(" Max Latency: {:?}", metrics.max_latency); 305 + println!(" Operations/sec: {:.2}", metrics.operations_per_second); 306 + 307 + // Performance assertions for concurrent load 308 + assert!(metrics.operations_per_second > 200.0, "Should handle at least 200 concurrent ops/sec"); 309 + assert!(metrics.average_latency < Duration::from_millis(100), "Average latency should be under 100ms under load"); 310 + assert!(metrics.successful_operations as f64 / metrics.total_operations as f64 > 0.90, "Success rate should be >90% under load"); 311 + } 312 + 313 + #[tokio::test] 314 + async fn test_read_performance_with_scaling_data() { 315 + let server = setup().await; 316 + let client = server.client(); 317 + 318 + // First, create a significant amount of data 319 + const NUM_NODES: usize = 500; 320 + let mut node_ids = Vec::with_capacity(NUM_NODES); 321 + 322 + println!("Setting up {} nodes for read performance test...", NUM_NODES); 323 + 324 + for i in 0..NUM_NODES { 325 + let node = json!({ 326 + "labels": ["ReadPerfTest"], 327 + "properties": { 328 + "id": i, 329 + "data": format!("data_value_{}", i), 330 + "category": i % 10, 331 + "active": i % 2 == 0 332 + } 333 + }); 334 + 335 + if let Ok(response) = client.post_json("/api/v1/nodes", node).await { 336 + if response.status() == 200 { 337 + if let Ok(body) = response.json::<Value>().await { 338 + if let Some(node_id) = body["node_id"].as_u64() { 339 + node_ids.push(node_id); 340 + } 341 + } 342 + } 343 + } 344 + } 345 + 346 + println!("Created {} nodes, starting read performance test...", node_ids.len()); 347 + 348 + // Now test read performance 349 + const NUM_READ_OPS: usize = 500; 350 + let start_time = Instant::now(); 351 + let mut latencies = Vec::with_capacity(NUM_READ_OPS); 352 + let successful = AtomicU64::new(0); 353 + let failed = AtomicU64::new(0); 354 + 355 + for i in 0..NUM_READ_OPS { 356 + let operation_start = Instant::now(); 357 + 358 + // Read random nodes 359 + if let Some(&node_id) = node_ids.get(i % node_ids.len()) { 360 + match client.get(&format!("/api/v1/nodes/{}", node_id)).await { 361 + Ok(response) if response.status() == 200 => { 362 + successful.fetch_add(1, Ordering::Relaxed); 363 + } 364 + _ => { 365 + failed.fetch_add(1, Ordering::Relaxed); 366 + } 367 + } 368 + } 369 + 370 + latencies.push(operation_start.elapsed()); 371 + } 372 + 373 + let total_duration = start_time.elapsed(); 374 + let metrics = PerformanceMetrics::new( 375 + &latencies, 376 + successful.load(Ordering::Relaxed), 377 + failed.load(Ordering::Relaxed), 378 + total_duration, 379 + ); 380 + 381 + println!("Read Performance with {} nodes in database:", node_ids.len()); 382 + println!(" Read Operations: {}", metrics.total_operations); 383 + println!(" Successful: {}", metrics.successful_operations); 384 + println!(" Failed: {}", metrics.failed_operations); 385 + println!(" Total Duration: {:?}", metrics.total_duration); 386 + println!(" Average Latency: {:?}", metrics.average_latency); 387 + println!(" Min Latency: {:?}", metrics.min_latency); 388 + println!(" Max Latency: {:?}", metrics.max_latency); 389 + println!(" Reads/sec: {:.2}", metrics.operations_per_second); 390 + 391 + // Read performance should be even better than writes 392 + assert!(metrics.operations_per_second > 500.0, "Should handle at least 500 reads/sec"); 393 + assert!(metrics.average_latency < Duration::from_millis(10), "Read latency should be under 10ms"); 394 + assert!(metrics.successful_operations as f64 / metrics.total_operations as f64 > 0.98, "Read success rate should be >98%"); 395 + } 396 + 397 + #[tokio::test] 398 + async fn test_mixed_workload_performance() { 399 + let server = setup().await; 400 + let client = server.client(); 401 + 402 + const DURATION_SECONDS: u64 = 10; 403 + const NUM_WORKERS: usize = 20; 404 + 405 + let start_time = Instant::now(); 406 + let end_time = start_time + Duration::from_secs(DURATION_SECONDS); 407 + 408 + let create_count = Arc::new(AtomicU64::new(0)); 409 + let read_count = Arc::new(AtomicU64::new(0)); 410 + let update_count = Arc::new(AtomicU64::new(0)); 411 + let delete_count = Arc::new(AtomicU64::new(0)); 412 + let error_count = Arc::new(AtomicU64::new(0)); 413 + 414 + // Create some initial nodes for read/update/delete operations 415 + let mut initial_nodes = Vec::new(); 416 + for i in 0..100 { 417 + let node = json!({ 418 + "labels": ["MixedWorkload"], 419 + "properties": {"id": i} 420 + }); 421 + if let Ok(response) = client.post_json("/api/v1/nodes", node).await { 422 + if response.status() == 200 { 423 + if let Ok(body) = response.json::<Value>().await { 424 + if let Some(node_id) = body["node_id"].as_u64() { 425 + initial_nodes.push(node_id); 426 + } 427 + } 428 + } 429 + } 430 + } 431 + 432 + let initial_nodes = Arc::new(initial_nodes); 433 + 434 + // Spawn workers with mixed operations 435 + let mut handles = Vec::with_capacity(NUM_WORKERS); 436 + 437 + for worker_id in 0..NUM_WORKERS { 438 + let client = client.clone(); 439 + let create_count = create_count.clone(); 440 + let read_count = read_count.clone(); 441 + let update_count = update_count.clone(); 442 + let delete_count = delete_count.clone(); 443 + let error_count = error_count.clone(); 444 + let initial_nodes = initial_nodes.clone(); 445 + 446 + let handle = tokio::spawn(async move { 447 + let mut rng_state = worker_id; 448 + 449 + while Instant::now() < end_time { 450 + // Simple pseudo-random operation selection 451 + rng_state = rng_state.wrapping_mul(1103515245).wrapping_add(12345); 452 + let operation = rng_state % 100; 453 + 454 + let result = if operation < 40 { 455 + // 40% reads 456 + if !initial_nodes.is_empty() { 457 + let node_id = initial_nodes[rng_state as usize % initial_nodes.len()]; 458 + let response = client.get(&format!("/api/v1/nodes/{}", node_id)).await; 459 + if matches!(response, Ok(r) if r.status() == 200) { 460 + read_count.fetch_add(1, Ordering::Relaxed); 461 + true 462 + } else { 463 + false 464 + } 465 + } else { 466 + false 467 + } 468 + } else if operation < 70 { 469 + // 30% creates 470 + let node = json!({ 471 + "labels": ["MixedWorkload"], 472 + "properties": { 473 + "worker_id": worker_id, 474 + "timestamp": Instant::now().elapsed().as_millis() 475 + } 476 + }); 477 + let response = client.post_json("/api/v1/nodes", node).await; 478 + if matches!(response, Ok(r) if r.status() == 200) { 479 + create_count.fetch_add(1, Ordering::Relaxed); 480 + true 481 + } else { 482 + false 483 + } 484 + } else if operation < 85 { 485 + // 15% updates 486 + if !initial_nodes.is_empty() { 487 + let node_id = initial_nodes[rng_state as usize % initial_nodes.len()]; 488 + let update = json!({ 489 + "labels": ["MixedWorkload"], 490 + "properties": { 491 + "updated_by": worker_id, 492 + "update_time": Instant::now().elapsed().as_millis() 493 + } 494 + }); 495 + let response = client.put_json(&format!("/api/v1/nodes/{}", node_id), update).await; 496 + if matches!(response, Ok(r) if r.status() == 200) { 497 + update_count.fetch_add(1, Ordering::Relaxed); 498 + true 499 + } else { 500 + false 501 + } 502 + } else { 503 + false 504 + } 505 + } else { 506 + // 15% stats queries (lightweight reads) 507 + let response = client.get("/api/v1/stats").await; 508 + if matches!(response, Ok(r) if r.status() == 200) { 509 + read_count.fetch_add(1, Ordering::Relaxed); 510 + true 511 + } else { 512 + false 513 + } 514 + }; 515 + 516 + if !result { 517 + error_count.fetch_add(1, Ordering::Relaxed); 518 + } 519 + 520 + // Small delay to prevent overwhelming 521 + sleep(Duration::from_millis(1)).await; 522 + } 523 + }); 524 + 525 + handles.push(handle); 526 + } 527 + 528 + // Wait for all workers to complete 529 + join_all(handles).await; 530 + 531 + let total_duration = start_time.elapsed(); 532 + let creates = create_count.load(Ordering::Relaxed); 533 + let reads = read_count.load(Ordering::Relaxed); 534 + let updates = update_count.load(Ordering::Relaxed); 535 + let deletes = delete_count.load(Ordering::Relaxed); 536 + let errors = error_count.load(Ordering::Relaxed); 537 + let total_ops = creates + reads + updates + deletes; 538 + 539 + println!("Mixed Workload Performance ({} seconds, {} workers):", DURATION_SECONDS, NUM_WORKERS); 540 + println!(" Creates: {}", creates); 541 + println!(" Reads: {}", reads); 542 + println!(" Updates: {}", updates); 543 + println!(" Deletes: {}", deletes); 544 + println!(" Errors: {}", errors); 545 + println!(" Total Operations: {}", total_ops); 546 + println!(" Total Duration: {:?}", total_duration); 547 + println!(" Mixed Operations/sec: {:.2}", total_ops as f64 / total_duration.as_secs_f64()); 548 + println!(" Error Rate: {:.2}%", (errors as f64 / (total_ops + errors) as f64) * 100.0); 549 + 550 + // Mixed workload assertions 551 + assert!(total_ops > 0, "Should complete some operations"); 552 + assert!(total_ops as f64 / total_duration.as_secs_f64() > 100.0, "Should handle at least 100 mixed ops/sec"); 553 + assert!((errors as f64 / (total_ops + errors) as f64) < 0.1, "Error rate should be <10%"); 554 + } 555 + 556 + #[tokio::test] 557 + async fn test_memory_usage_under_load() { 558 + let server = setup().await; 559 + let client = server.client(); 560 + 561 + // Get initial stats 562 + let initial_response = client.get("/api/v1/stats").await.expect("Initial stats failed"); 563 + let initial_stats: Value = initial_response.json().await.expect("Parse failed"); 564 + let initial_nodes = initial_stats["node_count"].as_u64().unwrap_or(0); 565 + 566 + // Create a large number of nodes 567 + const NUM_LARGE_NODES: usize = 1000; 568 + let mut created_nodes = Vec::new(); 569 + 570 + let start_time = Instant::now(); 571 + 572 + for i in 0..NUM_LARGE_NODES { 573 + let large_node = json!({ 574 + "labels": ["MemoryTest", "LargeNode"], 575 + "properties": { 576 + "id": i, 577 + "large_text": "x".repeat(1000), // 1KB of text 578 + "metadata": { 579 + "created": start_time.elapsed().as_millis(), 580 + "worker": "memory_test", 581 + "size": "large" 582 + }, 583 + "tags": (0..20).map(|j| format!("tag_{}", j)).collect::<Vec<_>>() 584 + } 585 + }); 586 + 587 + match client.post_json("/api/v1/nodes", large_node).await { 588 + Ok(response) if response.status() == 200 => { 589 + if let Ok(body) = response.json::<Value>().await { 590 + if let Some(node_id) = body["node_id"].as_u64() { 591 + created_nodes.push(node_id); 592 + } 593 + } 594 + } 595 + _ => { 596 + // Continue even if some fail 597 + } 598 + } 599 + 600 + // Check progress periodically 601 + if i % 100 == 0 { 602 + println!("Created {} large nodes...", i); 603 + } 604 + } 605 + 606 + let creation_duration = start_time.elapsed(); 607 + 608 + // Get final stats 609 + let final_response = client.get("/api/v1/stats").await.expect("Final stats failed"); 610 + let final_stats: Value = final_response.json().await.expect("Parse failed"); 611 + let final_nodes = final_stats["node_count"].as_u64().unwrap_or(0); 612 + 613 + // Test some read operations on the large dataset 614 + let read_start = Instant::now(); 615 + let mut successful_reads = 0; 616 + 617 + for i in 0..100 { 618 + if let Some(&node_id) = created_nodes.get(i * (created_nodes.len() / 100)) { 619 + if let Ok(response) = client.get(&format!("/api/v1/nodes/{}", node_id)).await { 620 + if response.status() == 200 { 621 + successful_reads += 1; 622 + } 623 + } 624 + } 625 + } 626 + 627 + let read_duration = read_start.elapsed(); 628 + 629 + println!("Memory Usage Under Load Test:"); 630 + println!(" Large Nodes Created: {}", created_nodes.len()); 631 + println!(" Initial Node Count: {}", initial_nodes); 632 + println!(" Final Node Count: {}", final_nodes); 633 + println!(" Creation Duration: {:?}", creation_duration); 634 + println!(" Successful Sample Reads: {}/100", successful_reads); 635 + println!(" Read Duration: {:?}", read_duration); 636 + println!(" Nodes/sec during creation: {:.2}", created_nodes.len() as f64 / creation_duration.as_secs_f64()); 637 + 638 + // Memory usage assertions 639 + assert!(created_nodes.len() > 500, "Should create most of the large nodes"); 640 + assert!(successful_reads > 90, "Should be able to read most nodes even with large dataset"); 641 + assert!(creation_duration.as_secs() < 60, "Should create nodes in reasonable time"); 642 + assert!(read_duration < Duration::from_secs(5), "Should read sample nodes quickly"); 643 + } 644 + }

Configure Feed

Configure Feed