aprender-serve 0.64.0

Pure Rust ML inference engine built from scratch - model serving for GGUF and safetensors

/// PARITY-053d: Document concurrent batch processing
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity053d_concurrent_processing() {
    println!("PARITY-053d: Concurrent Batch Processing");
    println!("========================================");
    println!();
    println!("  CONCURRENCY MODEL:");
    println!("    - Each request in batch spawns a tokio task");
    println!("    - Tasks run concurrently (not sequentially)");
    println!("    - process_batch() awaits all tasks before returning");
    println!();
    println!("  WHY CONCURRENT (not true batch inference):");
    println!("    - True batch inference requires model.forward_batch()");
    println!("    - Current model uses generate_with_cache() per request");
    println!("    - Concurrent processing still improves throughput:");
    println!("      * Overlaps CPU computation between requests");
    println!("      * Better utilization under high load");
    println!("      * Foundation for future true batch support");
    println!();
    println!("  THROUGHPUT IMPROVEMENT:");
    println!("    - Sequential: N * latency_per_request");
    println!("    - Concurrent: ~latency_per_request (with overhead)");
    println!("    - Actual speedup depends on model/hardware contention");
    println!();
    println!("  FUTURE: TRUE BATCH INFERENCE");
    println!("    - Requires model.forward_batch(token_ids_batch)");
    println!("    - Single GPU kernel launch for all requests");
    println!("    - 10x+ throughput improvement possible (PARITY-050)");

    // Test passes as documentation
    assert!(true, "PARITY-053d: Concurrency documented");
}

/// PARITY-053e: Document BatchProcessResult
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity053e_batch_process_result() {
    use crate::api::BatchProcessResult;

    println!("PARITY-053e: BatchProcessResult Structure");
    println!("=========================================");
    println!();

    // Create example result
    let result = BatchProcessResult {
        requests_processed: 16,
        was_batched: true,
        total_time_ms: 125.0,
        avg_latency_ms: 7.8,
    };

    println!("  FIELDS:");
    println!(
        "    requests_processed: {} (number of requests in batch)",
        result.requests_processed
    );
    println!(
        "    was_batched: {} (batch >= min_batch)",
        result.was_batched
    );
    println!(
        "    total_time_ms: {:.1} (wall clock time)",
        result.total_time_ms
    );
    println!(
        "    avg_latency_ms: {:.1} (per-request average)",
        result.avg_latency_ms
    );
    println!();
    println!("  METRICS CALCULATION:");
    println!("    throughput = requests_processed / (total_time_ms / 1000)");
    println!(
        "    throughput = {} / {:.3} = {:.1} req/s",
        result.requests_processed,
        result.total_time_ms / 1000.0,
        result.requests_processed as f64 / (result.total_time_ms / 1000.0)
    );

    // Verify structure
    assert_eq!(result.requests_processed, 16);
    assert!(result.was_batched);
    assert!((result.total_time_ms - 125.0).abs() < f64::EPSILON);
}

/// PARITY-053f: Summary and integration status
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity053f_summary() {
    println!("PARITY-053f: Batch Processor Implementation Summary");
    println!("===================================================");
    println!();
    println!("  IMPLEMENTATION STATUS: ✅ COMPLETE");
    println!();
    println!("  FUNCTIONS ADDED (api.rs):");
    println!("    ✅ spawn_batch_processor() - Creates channel and spawns task");
    println!("    ✅ batch_processor_task() - Main event loop with timeout batching");
    println!("    ✅ process_batch() - Concurrent request processing");
    println!();
    println!("  STRUCTS ADDED:");
    println!("    ✅ BatchProcessResult - Batch processing metrics");
    println!();
    println!("  BATCHING STRATEGY:");
    println!("    ✅ Size-triggered: batch >= optimal_batch (16)");
    println!("    ✅ Time-triggered: window_ms timeout (10ms)");
    println!("    ✅ Graceful shutdown: process remaining on channel close");
    println!();
    println!("  INTEGRATION STATUS:");
    println!("    ✅ spawn_batch_processor() ready for server startup");
    println!("    ✅ AppState.with_batch_config() wires channel to state");
    println!("    ⏳ Handler modification pending (use batch path)");
    println!();
    println!("  NEXT STEPS:");
    println!("    PARITY-054: Modify completions handler to use batch path");
    println!("    PARITY-055: Benchmark batch throughput");
    println!("    PARITY-056: Validate M4 parity achievement");

    // Test passes as documentation
    assert!(true, "PARITY-053f: Summary documented");
}

// ==================== PARITY-054: Handler Batch Path Integration ====================
//
// OBJECTIVE: Modify completions handler to use batch path when enabled
//
// IMPLEMENTATION COMPLETE:
//   - Handler checks state.batch_enabled() before generation
//   - If enabled, sends ContinuousBatchRequest via batch_tx channel
//   - Awaits response via oneshot channel
//   - Falls back to single-request path on failure
//   - Backward compatible (batch disabled by default)
//
// RESPONSE CHANGES:
//   - model field: "batch-q4k-{batch_size}" instead of "cached-q4k"
//   - id prefix: "cmpl-batch-" instead of "cmpl-cached-"
// ================================================================================

/// PARITY-054a: Document handler batch path integration
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity054a_handler_batch_path() {
    println!("PARITY-054a: Handler Batch Path Integration");
    println!("===========================================");
    println!();
    println!("  LOCATION: src/api.rs::openai_completions_handler()");
    println!();
    println!("  BATCH PATH FLOW:");
    println!("    1. Check state.batch_enabled()");
    println!("    2. Get batch_tx from state.batch_request_tx()");
    println!("    3. Create oneshot channel for response");
    println!("    4. Build ContinuousBatchRequest");
    println!("    5. Send via batch_tx.send().await");
    println!("    6. Await response_rx.await");
    println!("    7. Extract generated_tokens(), decode, return response");
    println!();
    println!("  FALLBACK CONDITIONS:");
    println!("    - state.batch_enabled() returns false");
    println!("    - batch_tx.send() fails");
    println!("    - response_rx receives error (channel dropped)");
    println!("  -> Falls through to single-request path");

    // Test passes as documentation
    assert!(true, "PARITY-054a: Batch path documented");
}

/// PARITY-054b: Document response format changes
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity054b_response_format() {
    println!("PARITY-054b: Response Format Changes");
    println!("====================================");
    println!();
    println!("  SINGLE-REQUEST PATH:");
    println!("    id: \"cmpl-cached-{{timestamp}}\"");
    println!("    model: \"cached-q4k\"");
    println!();
    println!("  BATCH PATH:");
    println!("    id: \"cmpl-batch-{{timestamp}}\"");
    println!("    model: \"batch-q4k-{{batch_size}}\"");
    println!();
    println!("  EXAMPLE BATCH RESPONSE:");
    println!("    {{");
    println!("      \"id\": \"cmpl-batch-1734267890123\",");
    println!("      \"object\": \"text_completion\",");
    println!("      \"model\": \"batch-q4k-16\",");
    println!("      \"choices\": [{{ ... }}],");
    println!("      \"usage\": {{ ... }}");
    println!("    }}");
    println!();
    println!("  OBSERVABILITY:");
    println!("    - model field indicates batch size used");
    println!("    - Can track batch vs single requests in logs/metrics");

    // Test passes as documentation
    assert!(true, "PARITY-054b: Response format documented");
}

/// PARITY-054c: Document backward compatibility
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity054c_backward_compatibility() {
    println!("PARITY-054c: Backward Compatibility");
    println!("===================================");
    println!();
    println!("  DEFAULT BEHAVIOR:");
    println!("    - batch_enabled() returns false");
    println!("    - Handler uses single-request path");
    println!("    - No change to existing deployments");
    println!();
    println!("  OPT-IN BATCH MODE:");
    println!("    // During server startup");
    println!("    let batch_tx = spawn_batch_processor(model.clone(), config);");
    println!("    let state = AppState::with_cached_model(model)?");
    println!("        .with_batch_config(batch_tx, BatchConfig::default());");
    println!();
    println!("  GRACEFUL DEGRADATION:");
    println!("    - If batch channel fails, falls back to single-request");
    println!("    - If batch processor crashes, existing requests continue");
    println!("    - Metrics continue recording for both paths");

    // Test passes as documentation
    assert!(true, "PARITY-054c: Backward compatibility documented");
}

/// PARITY-054d: Document batch request structure
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity054d_batch_request_structure() {
    println!("PARITY-054d: Batch Request Structure");
    println!("====================================");
    println!();
    println!("  ContinuousBatchRequest FIELDS:");
    println!("    prompt_tokens: Vec<u32>     // Tokenized input");
    println!("    max_tokens: usize           // Generation limit");
    println!("    temperature: f32            // Sampling temperature");
    println!("    top_k: usize               // Top-k sampling");
    println!("    response_tx: oneshot::Sender // Response channel");
    println!("    submitted_at: Instant      // For latency tracking");
    println!();
    println!("  CONSTRUCTED FROM:");
    println!("    - prompt_tokens: tokenizer.encode(&request.prompt)");
    println!("    - max_tokens: request.max_tokens.unwrap_or(256)");
    println!("    - temperature: request.temperature.unwrap_or(0.7) as f32");
    println!("    - top_k: 1 if temperature == 0.0 else 40");
    println!();
    println!("  CHANNEL LIFECYCLE:");
    println!("    1. Handler creates oneshot channel");
    println!("    2. response_tx moved into ContinuousBatchRequest");
    println!("    3. Handler awaits response_rx");
    println!("    4. Batch processor sends via response_tx");
    println!("    5. Handler receives ContinuousBatchResponse");

    // Test passes as documentation
    assert!(true, "PARITY-054d: Request structure documented");
}

/// PARITY-054e: Document error handling
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity054e_error_handling() {
    println!("PARITY-054e: Error Handling");
    println!("===========================");
    println!();
    println!("  ERROR SCENARIOS:");
    println!();
    println!("  1. Batch send fails (batch_tx.send().is_err()):");
    println!("     -> Fall through to single-request path");
    println!("     -> No error returned to client");
    println!();
    println!("  2. Response channel dropped (response_rx error):");
    println!("     -> Fall through to single-request path");
    println!("     -> No error returned to client");
    println!();
    println!("  3. Token decode fails:");
    println!("     -> Return 500 Internal Server Error");
    println!("     -> Record failure metric");
    println!();
    println!("  RESILIENCE:");
    println!("    - Batch path failures are non-fatal");
    println!("    - Single-request path always available");
    println!("    - Client receives response either way");

    // Test passes as documentation
    assert!(true, "PARITY-054e: Error handling documented");
}

/// PARITY-054f: Summary and integration complete
#[test]
#[cfg(feature = "gpu")]
#[serial_test::serial]
fn test_parity054f_summary() {
    println!("PARITY-054f: Handler Batch Integration Summary");
    println!("==============================================");
    println!();
    println!("  IMPLEMENTATION STATUS: ✅ COMPLETE");
    println!();
    println!("  HANDLER MODIFICATION:");
    println!("    ✅ Check state.batch_enabled() before generation");
    println!("    ✅ Create oneshot channel for response");
    println!("    ✅ Send ContinuousBatchRequest via batch_tx");
    println!("    ✅ Await response via oneshot receiver");
    println!("    ✅ Fall back to single-request on failure");
    println!();
    println!("  RESPONSE CHANGES:");
    println!("    ✅ id: \"cmpl-batch-{{timestamp}}\"");
    println!("    ✅ model: \"batch-q4k-{{batch_size}}\"");
    println!();
    println!("  BACKWARD COMPATIBLE:");
    println!("    ✅ batch_enabled() false by default");
    println!("    ✅ No change to existing deployments");
    println!("    ✅ Graceful fallback on batch failure");
    println!();
    println!("  BATCH INFERENCE PATH COMPLETE:");
    println!("    ✅ PARITY-052: Batch request queuing (structs)");
    println!("    ✅ PARITY-053: Batch processor task");
    println!("    ✅ PARITY-054: Handler batch integration");
    println!();
    println!("  NEXT STEPS:");
    println!("    PARITY-055: Benchmark batch throughput");
    println!("    PARITY-056: Validate M4 parity achievement");

    // Test passes as documentation
    assert!(true, "PARITY-054f: Summary documented");
}