claude_utils/turbo/
executor.rs

1use futures::future::join_all;
2use serde_json::Value;
3use std::collections::VecDeque;
4use std::sync::Arc;
5use tokio::sync::{Mutex, Semaphore};
6use tokio::time::{sleep, Duration};
7use tracing::{error, info};
8
9pub struct ParallelExecutor {
10    semaphore: Arc<Semaphore>,
11    retry_queue: Arc<Mutex<VecDeque<Value>>>,
12}
13
14impl ParallelExecutor {
15    pub fn new(parallel_limit: usize) -> Self {
16        Self {
17            semaphore: Arc::new(Semaphore::new(parallel_limit)),
18            retry_queue: Arc::new(Mutex::new(VecDeque::new())),
19        }
20    }
21
22    /// Execute multiple operations in parallel
23    pub async fn execute_parallel(&self, operations: Vec<Value>) -> Vec<Result<Value, String>> {
24        info!("🚀 Turbo: Executing {} operations in parallel", operations.len());
25        
26        let tasks: Vec<_> = operations
27            .into_iter()
28            .map(|op| {
29                let sem = self.semaphore.clone();
30                async move {
31                    let _permit = match sem.acquire().await {
32                        Ok(permit) => permit,
33                        Err(e) => {
34                            error!("Failed to acquire semaphore: {}", e);
35                            return Err(format!("Semaphore error: {e}"));
36                        }
37                    };
38                    self.execute_single(op).await
39                }
40            })
41            .collect();
42
43        join_all(tasks).await
44    }
45
46    /// Execute a single operation with retry logic
47    async fn execute_single(&self, operation: Value) -> Result<Value, String> {
48        let mut retry_count = 0;
49        const MAX_RETRIES: u32 = 3;
50
51        loop {
52            match self.perform_operation(&operation).await {
53                Ok(result) => return Ok(result),
54                Err(e) if retry_count < MAX_RETRIES => {
55                    retry_count += 1;
56                    let delay = Duration::from_millis(100 * (2_u64.pow(retry_count)));
57                    info!("Retry {}/{} after {:?}: {}", retry_count, MAX_RETRIES, delay, e);
58                    sleep(delay).await;
59                }
60                Err(e) => {
61                    error!("Operation failed after {} retries: {}", MAX_RETRIES, e);
62                    return Err(e);
63                }
64            }
65        }
66    }
67
68    /// Perform the actual operation (placeholder - will integrate with MCP)
69    async fn perform_operation(&self, operation: &Value) -> Result<Value, String> {
70        // This will be integrated with the actual MCP server
71        // For now, simulate processing
72        if let Some(method) = operation.get("method").and_then(|m| m.as_str()) {
73            match method {
74                "file.edit" => {
75                    // Simulate file edit
76                    Ok(serde_json::json!({
77                        "success": true,
78                        "file": operation.get("params").and_then(|p| p.get("file")),
79                        "turbo": true
80                    }))
81                }
82                "command.execute" => {
83                    // Simulate command execution
84                    Ok(serde_json::json!({
85                        "success": true,
86                        "output": "Command executed",
87                        "turbo": true
88                    }))
89                }
90                _ => Err(format!("Unknown method: {method}")),
91            }
92        } else {
93            Err("No method specified".to_string())
94        }
95    }
96
97    /// Queue operation for retry
98    pub async fn queue_retry(&self, operation: Value) {
99        self.retry_queue.lock().await.push_back(operation);
100    }
101
102    /// Process retry queue
103    pub async fn process_retries(&self) -> Vec<Result<Value, String>> {
104        let mut queue = self.retry_queue.lock().await;
105        let operations: Vec<Value> = queue.drain(..).collect();
106        drop(queue);
107
108        if !operations.is_empty() {
109            info!("Processing {} queued retries", operations.len());
110            self.execute_parallel(operations).await
111        } else {
112            Vec::new()
113        }
114    }
115}
claude_utils/turbo/executor.rs

claude_utils/turbo/
executor.rs
