ferrum_engine/
tts_engine.rs

1//! TTS service — concurrent slot-based serving of TtsModelExecutor.
2//!
3//! Multiple TTS requests can be processed in parallel, each on its own
4//! executor slot. Slots share nothing (each has its own model weights +
5//! KV cache). Future: Phase 2 will share weights across slots to reduce
6//! memory.
7//!
8//! The struct is named `TtsService` to free the `TtsEngine` identifier
9//! for the trait of the same name in `ferrum-interfaces`.
10
11use async_trait::async_trait;
12use ferrum_interfaces::engine::{InferenceEngine, TtsEngine};
13use ferrum_models::executor::tts_executor::TtsModelExecutor;
14use ferrum_types::{EngineConfig, EngineMetrics, EngineStatus, FerrumError, ModelId, Result};
15use parking_lot::Mutex;
16use std::sync::atomic::{AtomicUsize, Ordering};
17use std::sync::Arc;
18
19/// Concurrent TTS service. Implements [`TtsEngine`] (the modality trait)
20/// and [`InferenceEngine`] (lifecycle).
21pub struct TtsService {
22    slots: Vec<Arc<Mutex<TtsModelExecutor>>>,
23    /// Semaphore to limit concurrent slot usage.
24    semaphore: tokio::sync::Semaphore,
25    config: EngineConfig,
26    sample_rate: u32,
27    active_requests: AtomicUsize,
28}
29
30impl TtsService {
31    /// Create with a single executor (backward compatible).
32    pub fn new(executor: TtsModelExecutor, model_id: ModelId) -> Self {
33        let sr = executor.sample_rate() as u32;
34        let mut config = ferrum_types::EngineConfig::default();
35        config.model.model_id = model_id;
36        config.backend.device = ferrum_types::Device::CPU;
37        Self {
38            slots: vec![Arc::new(Mutex::new(executor))],
39            semaphore: tokio::sync::Semaphore::new(1),
40            config,
41            sample_rate: sr,
42            active_requests: AtomicUsize::new(0),
43        }
44    }
45
46    /// Create with multiple executor slots for concurrent serving.
47    pub fn new_multi(executors: Vec<TtsModelExecutor>, model_id: ModelId) -> Self {
48        let n = executors.len().max(1);
49        let sr = executors
50            .first()
51            .map(|e| e.sample_rate() as u32)
52            .unwrap_or(24000);
53        let mut config = ferrum_types::EngineConfig::default();
54        config.model.model_id = model_id;
55        config.backend.device = ferrum_types::Device::CPU;
56        let slots: Vec<_> = executors
57            .into_iter()
58            .map(|e| Arc::new(Mutex::new(e)))
59            .collect();
60        Self {
61            slots,
62            semaphore: tokio::sync::Semaphore::new(n),
63            config,
64            sample_rate: sr,
65            active_requests: AtomicUsize::new(0),
66        }
67    }
68
69    /// Number of available slots.
70    pub fn num_slots(&self) -> usize {
71        self.slots.len()
72    }
73}
74
75#[async_trait]
76impl InferenceEngine for TtsService {
77    async fn status(&self) -> EngineStatus {
78        EngineStatus {
79            is_ready: true,
80            loaded_models: vec![],
81            active_requests: self.active_requests.load(Ordering::Relaxed),
82            queued_requests: self.slots.len() - self.semaphore.available_permits(),
83            memory_usage: ferrum_types::MemoryUsage {
84                total_bytes: 0,
85                used_bytes: 0,
86                free_bytes: 0,
87                gpu_memory_bytes: None,
88                cpu_memory_bytes: None,
89                cache_memory_bytes: 0,
90                utilization_percent: 0.0,
91            },
92            uptime_seconds: 0,
93            last_heartbeat: chrono::Utc::now(),
94            version: env!("CARGO_PKG_VERSION").to_string(),
95        }
96    }
97
98    async fn shutdown(&self) -> Result<()> {
99        Ok(())
100    }
101
102    fn config(&self) -> &EngineConfig {
103        &self.config
104    }
105
106    fn metrics(&self) -> EngineMetrics {
107        crate::modality_stubs::inert_metrics()
108    }
109
110    async fn health_check(&self) -> ferrum_types::HealthStatus {
111        crate::modality_stubs::inert_health()
112    }
113}
114
115#[async_trait]
116impl TtsEngine for TtsService {
117    async fn synthesize_speech(
118        &self,
119        text: &str,
120        language: Option<&str>,
121        chunk_frames: usize,
122    ) -> Result<Vec<Vec<f32>>> {
123        // Acquire a slot (waits if all slots busy).
124        let _permit = self
125            .semaphore
126            .acquire()
127            .await
128            .map_err(|_| FerrumError::model("TTS semaphore closed"))?;
129
130        self.active_requests.fetch_add(1, Ordering::Relaxed);
131
132        // Find an unlocked slot (semaphore guarantees at least one is available).
133        let slot = self
134            .slots
135            .iter()
136            .find(|s| s.try_lock().is_some())
137            .unwrap_or(&self.slots[0])
138            .clone();
139
140        let text = text.to_string();
141        let lang = language.unwrap_or("auto").to_string();
142        let _active = &self.active_requests;
143
144        // Run TTS on blocking thread (model forward is CPU/GPU bound).
145        let result = tokio::task::spawn_blocking(move || {
146            let mut executor = slot.lock();
147            executor.synthesize_streaming(&text, &lang, chunk_frames, |_, _| {})
148        })
149        .await
150        .map_err(|e| FerrumError::model(format!("TTS task panic: {e}")))?;
151
152        self.active_requests.fetch_sub(1, Ordering::Relaxed);
153        result
154    }
155
156    fn tts_sample_rate(&self) -> u32 {
157        self.sample_rate
158    }
159}
ferrum_engine/tts_engine.rs

ferrum_engine/
tts_engine.rs
