Skip to content

perf: Optimize TTS pool using deadpool (Issue #32) #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NTLx wants to merge 4 commits into
base: main
Choose a base branch
from
Open

perf: Optimize TTS pool using deadpool (Issue #32) #43

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
aff57bb
perf: Optimize TTS pool using deadpool (Issue #32)
NTLx Jan 31, 2026
a696b56
feat: add timeout configuration to TTS session pool
NTLx Feb 1, 2026
1ab5d7e
perf(tts): add idle and max-lifetime eviction to TTS pool
NTLx Jul 28, 2026
d936e78
fix(tts): honor idle worker floor and propagate pool errors
NTLx Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 2 additions & 0 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ serde_json = "1.0"
toml = "0.8"
rmp-serde = "1"

deadpool = { version = "0.10", features = ["rt_tokio_1"] }
async-trait = "0.1"
cached = { version = "0.56.0", features = ["async"] }

anyhow = "1.0"
Expand Down
29 changes: 17 additions & 12 deletions src/services/ws/stable/claude.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -223,17 +223,17 @@ impl RunSessionState {

let mut rx_list = LinkedList::new();
for chunk in finished_output {
let (tts_response_tx, tts_response_rx) = tokio::sync::mpsc::unbounded_channel();
if let Err(e) = tts_req_tx.send((chunk.to_string(), tts_response_tx)).await {
log::error!(
"{}:{:x} error sending tts request: {}",
self.session.id,
self.session.request_id,
e
);
} else {
rx_list.push_back((chunk, tts_response_rx));
}
let tts_response_rx = super::tts::submit_request(tts_req_tx, chunk.to_string())
.await
.map_err(|e| {
anyhow::anyhow!(
"{}:{:x} error sending tts request: {}",
self.session.id,
self.session.request_id,
e
)
})?;
rx_list.push_back((chunk, tts_response_rx));
}

for (text_chunk, mut tts_response_rx) in rx_list {
Expand Down Expand Up @@ -746,7 +746,12 @@ pub async fn run_session_manager(
mut session_rx: tokio::sync::mpsc::UnboundedReceiver<Session>,
notifications: Arc<RwLock<ClaudeNotifications>>,
) -> anyhow::Result<()> {
let mut tts_session_pool = super::tts::TTSSessionPool::new(tts.clone(), 4);
let mut tts_session_pool = super::tts::TTSSessionPool::new(
tts.clone(),
super::tts::DEFAULT_TTS_IDLE_WORKERS,
super::tts::DEFAULT_TTS_MAX_WORKERS,
super::tts::DEFAULT_TTS_IDLE_TIMEOUT,
);
let (tts_req_tx, tts_req_rx) = tokio::sync::mpsc::channel(128);

let mut sessions: HashMap<
Expand Down
10 changes: 7 additions & 3 deletions src/services/ws/stable/gemini.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,12 @@ pub async fn run_session_manager(
> = HashMap::new();

let tts_req_tx = if let Some(tts) = tts {
let mut tts_session_pool = super::tts::TTSSessionPool::new(tts.clone(), 4);
let mut tts_session_pool = super::tts::TTSSessionPool::new(
tts.clone(),
super::tts::DEFAULT_TTS_IDLE_WORKERS,
super::tts::DEFAULT_TTS_MAX_WORKERS,
super::tts::DEFAULT_TTS_IDLE_TIMEOUT,
);
let (tts_req_tx, tts_req_rx) = tokio::sync::mpsc::channel(128);

tokio::spawn(async move {
Expand Down Expand Up @@ -471,8 +476,7 @@ async fn run_session_with_tts(
gemini::types::ServerContent::Interrupted(_) => {}
gemini::types::ServerContent::TurnComplete(_) => {
let (chunks_tx, chunks_rx) = tokio::sync::mpsc::unbounded_channel();
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
tts_req_tx.send((llm_text.clone(), tx)).await?;
let rx = super::tts::submit_request(tts_req_tx, llm_text.clone()).await?;
chunks_tx.send((llm_text.clone(), rx))?;
asr_text.clear();
llm_text = String::with_capacity(1024);
Expand Down
73 changes: 17 additions & 56 deletions src/services/ws/stable/llm.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,19 @@ pub type ChunksRx = tokio::sync::mpsc::UnboundedReceiver<(String, super::tts::TT

use tokio::time::Duration;

async fn queue_tts_request(
tts_tx: &mut super::tts::TTSRequestTx,
chunks_tx: &ChunksTx,
text: String,
) -> anyhow::Result<()> {
let tts_resp_rx = super::tts::submit_request(tts_tx, text.clone()).await?;

chunks_tx
.send((text, tts_resp_rx))
.map_err(|e| anyhow::anyhow!("error sending tts chunks receiver: {e}"))?;
Ok(())
}

#[cached::proc_macro::cached(time = 60, size = 100, result = true)]
async fn load_url_content(url: String) -> anyhow::Result<String> {
let client = reqwest::Client::new();
Expand Down Expand Up @@ -214,18 +227,7 @@ pub async fn chat(
continue;
}

let (tts_resp_tx, tts_resp_rx) = tokio::sync::mpsc::unbounded_channel();

tts_tx
.send((chunk_.to_string(), tts_resp_tx))
.await
.map_err(|e| anyhow::anyhow!("error sending tts request for llm chunk: {e}"))?;

chunks_tx
.send((chunk_.to_string(), tts_resp_rx))
.map_err(|e| {
anyhow::anyhow!("error sending tts chunks receiver for llm chunk: {e}")
})?;
queue_tts_request(tts_tx, &chunks_tx, chunk_.to_string()).await?;
}
Ok(StableLLMResponseChunk::Functions(functions)) => {
log::info!("llm functions: {:#?}", functions);
Expand All @@ -234,20 +236,7 @@ pub async fn chat(
if let Some(message) = chat_session.get_tool_call_message(&function) {
log::info!("tool {} call message: {}", &function.function.name, message);
if !message.is_empty() {
let (tts_resp_tx, tts_resp_rx) = tokio::sync::mpsc::unbounded_channel();

tts_tx
.send((message.to_string(), tts_resp_tx))
.await
.map_err(|e| {
anyhow::anyhow!("error sending tts request for llm chunk: {e}")
})?;

chunks_tx.send((message, tts_resp_rx)).map_err(|e| {
anyhow::anyhow!(
"error sending tts chunks receiver for llm chunk: {e}"
)
})?;
queue_tts_request(tts_tx, &chunks_tx, message).await?;
}
}
chat_session.execute_tool(&function).await?
Expand Down Expand Up @@ -381,22 +370,7 @@ pub async fn responses(
continue;
}

let (tts_resp_tx, tts_resp_rx) = tokio::sync::mpsc::unbounded_channel();

tts_tx
.send((chunk_.to_string(), tts_resp_tx))
.await
.map_err(|e| {
anyhow::anyhow!("error sending tts request for llm responses chunk: {e}")
})?;

chunks_tx
.send((chunk_.to_string(), tts_resp_rx))
.map_err(|e| {
anyhow::anyhow!(
"error sending tts chunks receiver for llm responses chunk: {e}"
)
})?;
queue_tts_request(tts_tx, &chunks_tx, chunk_.to_string()).await?;
}
LLMResponsesChunk::Functions(functions) => {
log::info!("llm responses functions: {:#?}", functions);
Expand All @@ -405,20 +379,7 @@ pub async fn responses(
if let Some(message) = responses_session.get_tool_call_message(&function) {
log::info!("tool {} call message: {}", &function.function.name, message);
if !message.is_empty() {
let (tts_resp_tx, tts_resp_rx) = tokio::sync::mpsc::unbounded_channel();

tts_tx
.send((message.to_string(), tts_resp_tx))
.await
.map_err(|e| {
anyhow::anyhow!("error sending tts request for llm chunk: {e}")
})?;

chunks_tx.send((message, tts_resp_rx)).map_err(|e| {
anyhow::anyhow!(
"error sending tts chunks receiver for llm chunk: {e}"
)
})?;
queue_tts_request(tts_tx, &chunks_tx, message).await?;
}
}
let result = responses_session.execute_tool(&function).await;
Expand Down
7 changes: 6 additions & 1 deletion src/services/ws/stable/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -380,7 +380,12 @@ pub async fn run_session_manager(
tokio::sync::mpsc::UnboundedSender<(Session, Option<llm::MixPrompts>)>,
> = HashMap::new();

let mut tts_session_pool = tts::TTSSessionPool::new(tts.clone(), 4);
let mut tts_session_pool = tts::TTSSessionPool::new(
tts.clone(),
tts::DEFAULT_TTS_IDLE_WORKERS,
tts::DEFAULT_TTS_MAX_WORKERS,
tts::DEFAULT_TTS_IDLE_TIMEOUT,
);
let (tts_req_tx, tts_req_rx) = tokio::sync::mpsc::channel(128);

tokio::spawn(async move {
Expand Down
Loading