html_translation_lib/pipeline/

concurrent_batch.rs

//! 并发批处理模块
//!
//! 提供高性能的并行文本翻译处理能力

use crate::error::{TranslationError, TranslationResult};
use crate::storage::memory_pool::GlobalMemoryManager;
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::{Semaphore, RwLock};
use tokio::time::timeout;
use futures::stream::{self, StreamExt};

/// 批处理配置
#[derive(Debug, Clone)]
pub struct BatchConfig {
    /// 批大小
    pub batch_size: usize,
    /// 最大并发数
    pub max_concurrency: usize,
    /// 处理超时时间
    pub timeout_duration: Duration,
    /// 重试次数
    pub max_retries: u32,
    /// 背压阈值
    pub backpressure_threshold: usize,
}

impl Default for BatchConfig {
    fn default() -> Self {
        Self {
            batch_size: 50,
            max_concurrency: 10,
            timeout_duration: Duration::from_secs(30),
            max_retries: 3,
            backpressure_threshold: 1000,
        }
    }
}

/// 批处理任务
#[derive(Debug)]
pub struct BatchTask {
    pub id: u64,
    pub texts: Vec<String>,
    pub priority: Priority,
    pub created_at: Instant,
    pub retry_count: u32,
}

/// 任务优先级
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Priority {
    Low = 0,
    Normal = 1,
    High = 2,
    Critical = 3,
}

/// 批处理结果
#[derive(Debug)]
pub struct BatchResult {
    pub task_id: u64,
    pub results: Vec<String>,
    pub processing_time: Duration,
    pub success: bool,
    pub error: Option<TranslationError>,
}

/// 并发批处理器
pub struct ConcurrentBatchProcessor {
    config: BatchConfig,
    memory_manager: Arc<GlobalMemoryManager>,
    /// 任务队列（按优先级排序）
    task_queue: Arc<Mutex<VecDeque<BatchTask>>>,
    /// 信号量控制并发数
    concurrency_limiter: Arc<Semaphore>,
    /// 统计信息
    stats: Arc<RwLock<BatchProcessingStats>>,
    /// 任务ID计数器
    next_task_id: Arc<Mutex<u64>>,
}

impl ConcurrentBatchProcessor {
    /// 创建新的并发批处理器
    pub fn new(config: BatchConfig, memory_manager: Arc<GlobalMemoryManager>) -> Self {
        let concurrency_limiter = Arc::new(Semaphore::new(config.max_concurrency));
        
        Self {
            config,
            memory_manager,
            task_queue: Arc::new(Mutex::new(VecDeque::new())),
            concurrency_limiter,
            stats: Arc::new(RwLock::new(BatchProcessingStats::default())),
            next_task_id: Arc::new(Mutex::new(0)),
        }
    }
    
    /// 提交批处理任务
    pub async fn submit_batch(
        &self,
        texts: Vec<String>,
        priority: Priority,
    ) -> TranslationResult<u64> {
        // 检查背压
        {
            let queue = self.task_queue.lock().unwrap();
            if queue.len() > self.config.backpressure_threshold {
                return Err(TranslationError::InternalError(
                    "任务队列已满，请稍后重试".to_string()
                ));
            }
        }
        
        let task_id = {
            let mut counter = self.next_task_id.lock().unwrap();
            *counter += 1;
            *counter
        };
        
        let task = BatchTask {
            id: task_id,
            texts,
            priority,
            created_at: Instant::now(),
            retry_count: 0,
        };
        
        // 插入到优先级队列中
        {
            let mut queue = self.task_queue.lock().unwrap();
            let insert_pos = queue
                .iter()
                .position(|t| t.priority < priority)
                .unwrap_or(queue.len());
            queue.insert(insert_pos, task);
        }
        
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.tasks_submitted += 1;
        }
        
        Ok(task_id)
    }
    
    /// 并行处理批次队列
    pub async fn process_queue<F, Fut>(
        &self,
        translation_fn: F,
    ) -> TranslationResult<Vec<BatchResult>>
    where
        F: Fn(Vec<String>) -> Fut + Send + Sync + Clone + 'static,
        Fut: std::future::Future<Output = TranslationResult<Vec<String>>> + Send + 'static,
    {
        let mut results = Vec::new();
        let mut handles = Vec::new();
        
        // 启动工作线程
        for _ in 0..self.config.max_concurrency {
            let processor = self.clone_for_worker();
            let translation_fn = translation_fn.clone();
            
            let handle = tokio::spawn(async move {
                processor.worker_loop(translation_fn).await
            });
            
            handles.push(handle);
        }
        
        // 等待所有工作线程完成或超时
        let timeout_duration = self.config.timeout_duration * 2; // 给工作线程更多时间
        
        match timeout(timeout_duration, futures::future::join_all(handles)).await {
            Ok(worker_results) => {
                for worker_result in worker_results {
                    match worker_result {
                        Ok(batch_results) => results.extend(batch_results),
                        Err(e) => {
                            return Err(TranslationError::InternalError(
                                format!("工作线程错误: {e}")
                            ));
                        }
                    }
                }
            }
            Err(_) => {
                return Err(TranslationError::TimeoutError(
                    "批处理队列处理超时".to_string()
                ));
            }
        }
        
        Ok(results)
    }
    
    /// 工作线程循环
    async fn worker_loop<F, Fut>(&self, translation_fn: F) -> Vec<BatchResult>
    where
        F: Fn(Vec<String>) -> Fut + Send + Sync,
        Fut: std::future::Future<Output = TranslationResult<Vec<String>>> + Send,
    {
        let mut results = Vec::new();
        
        loop {
            // 获取下一个任务
            let task = {
                let mut queue = self.task_queue.lock().unwrap();
                queue.pop_front()
            };
            
            let task = match task {
                Some(task) => task,
                None => break, // 队列为空，退出
            };
            
            // 获取并发许可
            let _permit = match self.concurrency_limiter.try_acquire() {
                Ok(permit) => permit,
                Err(_) => {
                    // 无法获取许可，重新放回队列
                    let mut queue = self.task_queue.lock().unwrap();
                    queue.push_front(task);
                    break;
                }
            };
            
            // 处理任务
            let result = self.process_single_task(task, &translation_fn).await;
            results.push(result);
        }
        
        results
    }
    
    /// 处理单个任务
    async fn process_single_task<F, Fut>(
        &self,
        task: BatchTask,
        translation_fn: &F,
    ) -> BatchResult
    where
        F: Fn(Vec<String>) -> Fut + Send + Sync,
        Fut: std::future::Future<Output = TranslationResult<Vec<String>>> + Send,
    {
        let start_time = Instant::now();
        let task_id = task.id;
        
        // 分割任务为更小的批次
        let batches = self.split_into_batches(task.texts);
        let mut all_results = Vec::new();
        let mut has_error = None;
        
        // 并行处理所有批次
        let batch_futures = batches.into_iter().map(|batch| {
            let translation_fn = translation_fn;
            async move {
                let result = timeout(
                    self.config.timeout_duration,
                    translation_fn(batch)
                ).await;
                
                match result {
                    Ok(Ok(translations)) => Ok(translations),
                    Ok(Err(e)) => Err(e),
                    Err(_) => Err(TranslationError::TimeoutError(
                        "批次处理超时".to_string()
                    )),
                }
            }
        });
        
        let batch_results: Vec<_> = stream::iter(batch_futures)
            .buffer_unordered(self.config.max_concurrency.min(4))
            .collect()
            .await;
        
        // 收集结果
        for result in batch_results {
            match result {
                Ok(translations) => all_results.extend(translations),
                Err(e) => {
                    has_error = Some(e);
                    break;
                }
            }
        }
        
        let processing_time = start_time.elapsed();
        let success = has_error.is_none();
        
        // 更新统计
        {
            let mut stats = self.stats.write().await;
            stats.tasks_processed += 1;
            stats.total_processing_time += processing_time;
            
            if success {
                stats.tasks_succeeded += 1;
            } else {
                stats.tasks_failed += 1;
            }
            
            if processing_time > stats.max_processing_time {
                stats.max_processing_time = processing_time;
            }
            
            if processing_time < stats.min_processing_time {
                stats.min_processing_time = processing_time;
            }
        }
        
        BatchResult {
            task_id,
            results: all_results,
            processing_time,
            success,
            error: has_error,
        }
    }
    
    /// 将大任务分割为小批次
    fn split_into_batches(&self, texts: Vec<String>) -> Vec<Vec<String>> {
        texts
            .chunks(self.config.batch_size)
            .map(|chunk| chunk.to_vec())
            .collect()
    }
    
    /// 为工作线程创建克隆（共享状态）
    fn clone_for_worker(&self) -> Self {
        Self {
            config: self.config.clone(),
            memory_manager: Arc::clone(&self.memory_manager),
            task_queue: Arc::clone(&self.task_queue),
            concurrency_limiter: Arc::clone(&self.concurrency_limiter),
            stats: Arc::clone(&self.stats),
            next_task_id: Arc::clone(&self.next_task_id),
        }
    }
    
    /// 获取处理统计信息
    pub async fn get_stats(&self) -> BatchProcessingStats {
        self.stats.read().await.clone()
    }
    
    /// 获取队列状态
    pub fn get_queue_status(&self) -> QueueStatus {
        let queue = self.task_queue.lock().unwrap();
        let available_permits = self.concurrency_limiter.available_permits();
        
        let mut priority_counts = [0; 4];
        for task in queue.iter() {
            priority_counts[task.priority as usize] += 1;
        }
        
        QueueStatus {
            total_tasks: queue.len(),
            priority_distribution: priority_counts,
            available_workers: available_permits,
            oldest_task_age: queue
                .front()
                .map(|task| task.created_at.elapsed())
                .unwrap_or(Duration::ZERO),
        }
    }
    
    /// 清空队列
    pub fn clear_queue(&self) -> usize {
        let mut queue = self.task_queue.lock().unwrap();
        let count = queue.len();
        queue.clear();
        count
    }
}

/// 批处理统计信息
#[derive(Debug, Clone, Default)]
pub struct BatchProcessingStats {
    pub tasks_submitted: u64,
    pub tasks_processed: u64,
    pub tasks_succeeded: u64,
    pub tasks_failed: u64,
    pub total_processing_time: Duration,
    pub min_processing_time: Duration,
    pub max_processing_time: Duration,
}

impl BatchProcessingStats {
    /// 计算成功率
    pub fn success_rate(&self) -> f64 {
        if self.tasks_processed == 0 {
            return 0.0;
        }
        self.tasks_succeeded as f64 / self.tasks_processed as f64
    }
    
    /// 计算平均处理时间
    pub fn average_processing_time(&self) -> Duration {
        if self.tasks_processed == 0 {
            return Duration::ZERO;
        }
        self.total_processing_time / self.tasks_processed as u32
    }
    
    /// 计算吞吐量（任务/秒）
    pub fn throughput(&self) -> f64 {
        if self.total_processing_time.is_zero() {
            return 0.0;
        }
        self.tasks_processed as f64 / self.total_processing_time.as_secs_f64()
    }
}

/// 队列状态信息
#[derive(Debug, Clone)]
pub struct QueueStatus {
    pub total_tasks: usize,
    pub priority_distribution: [usize; 4], // [Low, Normal, High, Critical]
    pub available_workers: usize,
    pub oldest_task_age: Duration,
}

/// 自适应批处理器（根据系统负载调整参数）
pub struct AdaptiveBatchProcessor {
    inner: ConcurrentBatchProcessor,
    config: Arc<RwLock<BatchConfig>>,
    performance_monitor: PerformanceMonitor,
}

impl AdaptiveBatchProcessor {
    /// 创建自适应批处理器
    pub fn new(
        initial_config: BatchConfig,
        memory_manager: Arc<GlobalMemoryManager>,
    ) -> Self {
        let config = Arc::new(RwLock::new(initial_config.clone()));
        let inner = ConcurrentBatchProcessor::new(initial_config, memory_manager);
        let performance_monitor = PerformanceMonitor::new();
        
        Self {
            inner,
            config,
            performance_monitor,
        }
    }
    
    /// 自适应调整配置
    pub async fn adapt_configuration(&self) {
        let stats = self.inner.get_stats().await;
        let queue_status = self.inner.get_queue_status();
        
        let mut config = self.config.write().await;
        
        // 根据成功率调整重试次数
        if stats.success_rate() < 0.8 && config.max_retries < 5 {
            config.max_retries += 1;
        } else if stats.success_rate() > 0.95 && config.max_retries > 1 {
            config.max_retries = (config.max_retries - 1).max(1);
        }
        
        // 根据队列长度调整并发数
        if queue_status.total_tasks > config.backpressure_threshold / 2 {
            config.max_concurrency = (config.max_concurrency + 2).min(20);
        } else if queue_status.total_tasks < 10 && config.max_concurrency > 5 {
            config.max_concurrency = (config.max_concurrency - 1).max(5);
        }
        
        // 根据平均处理时间调整超时
        let avg_time = stats.average_processing_time();
        if avg_time > config.timeout_duration {
            config.timeout_duration = avg_time + Duration::from_secs(5);
        }
    }
}

/// 性能监控器
#[derive(Debug)]
pub struct PerformanceMonitor {
    start_time: Instant,
}

impl PerformanceMonitor {
    fn new() -> Self {
        Self {
            start_time: Instant::now(),
        }
    }
    
    /// 获取运行时间
    pub fn uptime(&self) -> Duration {
        self.start_time.elapsed()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    
    async fn mock_translation_fn(texts: Vec<String>) -> TranslationResult<Vec<String>> {
        // 模拟翻译延迟
        tokio::time::sleep(Duration::from_millis(10)).await;
        
        // 简单的模拟翻译
        let translations = texts.iter()
            .map(|text| format!("translated_{text}"))
            .collect();
        
        Ok(translations)
    }
    
    #[tokio::test]
    async fn test_concurrent_batch_processor() {
        let config = BatchConfig::default();
        let memory_manager = Arc::new(GlobalMemoryManager::new());
        let processor = ConcurrentBatchProcessor::new(config, memory_manager);
        
        // 提交一些任务
        let task1 = vec!["Hello".to_string(), "World".to_string()];
        let task2 = vec!["Foo".to_string(), "Bar".to_string()];
        
        let id1 = processor.submit_batch(task1, Priority::Normal).await.unwrap();
        let id2 = processor.submit_batch(task2, Priority::High).await.unwrap();
        
        assert!(id1 != id2);
        
        // 处理队列
        let results = processor.process_queue(mock_translation_fn).await.unwrap();
        
        assert_eq!(results.len(), 2);
        
        // 检查统计信息
        let stats = processor.get_stats().await;
        assert_eq!(stats.tasks_processed, 2);
    }
    
    #[tokio::test]
    async fn test_priority_ordering() {
        let config = BatchConfig::default();
        let memory_manager = Arc::new(GlobalMemoryManager::new());
        let processor = ConcurrentBatchProcessor::new(config, memory_manager);
        
        // 按不同优先级提交任务
        processor.submit_batch(vec!["low".to_string()], Priority::Low).await.unwrap();
        processor.submit_batch(vec!["high".to_string()], Priority::High).await.unwrap();
        processor.submit_batch(vec!["normal".to_string()], Priority::Normal).await.unwrap();
        
        let queue_status = processor.get_queue_status();
        assert_eq!(queue_status.total_tasks, 3);
        
        // 高优先级任务应该排在前面
        let queue = processor.task_queue.lock().unwrap();
        assert_eq!(queue[0].priority, Priority::High);
        assert_eq!(queue[1].priority, Priority::Normal);
        assert_eq!(queue[2].priority, Priority::Low);
    }
    
    #[test]
    fn test_batch_processing_stats() {
        let mut stats = BatchProcessingStats::default();
        stats.tasks_processed = 100;
        stats.tasks_succeeded = 95;
        stats.total_processing_time = Duration::from_secs(50);
        
        assert_eq!(stats.success_rate(), 0.95);
        assert_eq!(stats.average_processing_time(), Duration::from_millis(500));
        assert_eq!(stats.throughput(), 2.0);
    }
}