Method process_batch

core/src/embeddings.rs:811–916

Process a batch of embedding requests with lock-free concurrency control

(
        &self,
        batch: EmbeddingBatchRequest,
    )

    /// Process a batch of embedding requests with lock-free concurrency control
    pub async fn process_batch(
        &self,
        batch: EmbeddingBatchRequest,
    ) -> GraphBitResult<EmbeddingBatchResponse> {
        let start_time = std::time::Instant::now();
        let max_concurrency = batch.max_concurrency.unwrap_or(self.max_concurrency);

        let mut tasks = Vec::with_capacity(batch.requests.len());
        let current_requests = Arc::clone(&self.current_requests);

        for request in batch.requests {
            let config = self.config.clone();
            let current_requests = Arc::clone(&current_requests);

            let task = tokio::spawn(async move {
                // Wait for available slot using atomic operations (no semaphore bottleneck)
                loop {
                    let current = current_requests.load(std::sync::atomic::Ordering::Acquire);
                    if current < max_concurrency {
                        match current_requests.compare_exchange(
                            current,
                            current + 1,
                            std::sync::atomic::Ordering::AcqRel,
                            std::sync::atomic::Ordering::Acquire,
                        ) {
                            Ok(_) => break,     // Successfully acquired slot
                            Err(_) => continue, // Retry
                        }
                    }
                    tokio::task::yield_now().await;
                }

                // Execute the request
                let result = async {
                    let provider = EmbeddingProviderFactory::create_provider(config)?;
                    provider.generate_embeddings(request).await
                }
                .await;

                // Release slot
                current_requests.fetch_sub(1, std::sync::atomic::Ordering::AcqRel);

                result
            });

            tasks.push(task);
        }

        // Wait for all tasks with optional timeout
        let responses = if let Some(timeout_ms) = batch.timeout_ms {
            let timeout_duration = tokio::time::Duration::from_millis(timeout_ms);
            match tokio::time::timeout(timeout_duration, futures::future::join_all(tasks)).await {
                Ok(results) => results,
                Err(_) => return Err(GraphBitError::llm("Batch request timed out".to_string())),
            }
        } else {
            futures::future::join_all(tasks).await
        };