From 54b71e065c7e5b88f07017bdb6d664b7b43f64f0 Mon Sep 17 00:00:00 2001 From: usabarashi <19676305+usabarashi@users.noreply.github.com> Date: Tue, 21 Apr 2026 22:59:17 +0900 Subject: [PATCH 1/2] feat: implement memory management and enhance daemon synthesis handling --- src/infrastructure/core.rs | 8 +-- src/infrastructure/daemon/server.rs | 6 ++- src/infrastructure/daemon/state.rs | 9 ++-- src/infrastructure/daemon/state/executor.rs | 56 ++++++++++++++++----- src/infrastructure/daemon/state/policy.rs | 2 +- src/infrastructure/memory.rs | 23 +++++++++ src/infrastructure/mod.rs | 1 + 7 files changed, 83 insertions(+), 22 deletions(-) create mode 100644 src/infrastructure/memory.rs diff --git a/src/infrastructure/core.rs b/src/infrastructure/core.rs index d0880f4..c9778f2 100644 --- a/src/infrastructure/core.rs +++ b/src/infrastructure/core.rs @@ -142,11 +142,13 @@ impl VoicevoxCore { /// /// Returns an error if the model file cannot be opened or the core fails to unload it. pub fn unload_voice_model_by_path(&self, model_path: &Path) -> Result<()> { - let model = open_voice_model_file(model_path)?; - let voice_model_id = model.id(); + let voice_model_id = open_voice_model_file(model_path)?.id(); self.synthesizer .unload_voice_model(voice_model_id) - .map_err(|e| anyhow!("Failed to unload model: {e}")) + .map_err(|e| anyhow!("Failed to unload model: {e}"))?; + + crate::infrastructure::memory::release_unused_allocator_memory(); + Ok(()) } } diff --git a/src/infrastructure/daemon/server.rs b/src/infrastructure/daemon/server.rs index cd0e747..659876e 100644 --- a/src/infrastructure/daemon/server.rs +++ b/src/infrastructure/daemon/server.rs @@ -11,7 +11,9 @@ use tokio::time::timeout; use tokio_util::codec::{Framed, LengthDelimitedCodec}; use crate::infrastructure::daemon::state::DaemonState; -use crate::infrastructure::ipc::{DaemonRequest, MAX_DAEMON_REQUEST_FRAME_BYTES, OwnedResponse}; +use crate::infrastructure::ipc::{ + DaemonRequest, MAX_DAEMON_REQUEST_FRAME_BYTES, MAX_DAEMON_RESPONSE_FRAME_BYTES, OwnedResponse, +}; const SOCKET_DIR_MODE: u32 = 0o700; const SOCKET_FILE_MODE: u32 = 0o600; @@ -116,7 +118,7 @@ async fn handle_client_with_limit( permits: Arc, ) -> Result<()> { let codec = LengthDelimitedCodec::builder() - .max_frame_length(MAX_DAEMON_REQUEST_FRAME_BYTES) + .max_frame_length(MAX_DAEMON_REQUEST_FRAME_BYTES.max(MAX_DAEMON_RESPONSE_FRAME_BYTES)) .new_codec(); let mut framed = Framed::new(stream, codec); diff --git a/src/infrastructure/daemon/state.rs b/src/infrastructure/daemon/state.rs index e2b46f9..418d33a 100644 --- a/src/infrastructure/daemon/state.rs +++ b/src/infrastructure/daemon/state.rs @@ -52,9 +52,12 @@ impl DaemonState { /// Returns an error if VOICEVOX core initialization fails, model discovery fails, /// or the style-to-model mapping cannot be constructed. pub fn new() -> Result { - let core = crate::infrastructure::core::VoicevoxCore::new()?; - let catalog = ModelCatalog::new(&core)?; - let synthesis_executor = DaemonSynthesisExecutor::new(core); + let catalog_core = crate::infrastructure::core::VoicevoxCore::new()?; + let catalog = ModelCatalog::new(&catalog_core)?; + drop(catalog_core); + crate::infrastructure::memory::release_unused_allocator_memory(); + + let synthesis_executor = DaemonSynthesisExecutor::new(); let synthesis_policy = SerializedSynthesisPolicy::new(synthesis_executor); Ok(Self { diff --git a/src/infrastructure/daemon/state/executor.rs b/src/infrastructure/daemon/state/executor.rs index c808836..366674d 100644 --- a/src/infrastructure/daemon/state/executor.rs +++ b/src/infrastructure/daemon/state/executor.rs @@ -6,7 +6,7 @@ use super::catalog::{ModelCatalog, TargetResolution}; use super::result::{DaemonServiceError, DaemonServiceErrorKind, DaemonServiceResult}; pub(super) struct DaemonSynthesisExecutor { - core: VoicevoxCore, + core: Option, } /// RAII guard that unloads a voice model on drop. @@ -40,12 +40,30 @@ impl Drop for ModelUnloadGuard<'_> { } impl DaemonSynthesisExecutor { - pub(super) fn new(core: VoicevoxCore) -> Self { - Self { core } + pub(super) fn new() -> Self { + Self { core: None } + } + + fn ensure_core(&mut self) -> Result<(), DaemonServiceError> { + if self.core.is_none() { + self.core = Some(VoicevoxCore::new().map_err(|error| { + DaemonServiceError::new( + DaemonServiceErrorKind::ModelLoadFailed, + format!("Failed to initialize VOICEVOX core for synthesis: {error}"), + ) + })?); + } + + Ok(()) + } + + fn release_core(&mut self) { + self.core = None; + crate::infrastructure::memory::release_unused_allocator_memory(); } pub(super) fn synthesize( - &self, + &mut self, catalog: &ModelCatalog, text: String, requested_id: u32, @@ -62,26 +80,38 @@ impl DaemonSynthesisExecutor { }; let model_path = catalog.get_model_path(model_id); - if let Err(error) = self.core.load_specific_model(model_id) { + self.ensure_core()?; + + let core = self + .core + .as_ref() + .expect("core must be initialized by ensure_core"); + + if let Err(error) = core.load_specific_model(model_id) { crate::infrastructure::logging::error(&format!( "Failed to load model {model_id}: {error}" )); + self.release_core(); return Err(DaemonServiceError::new( DaemonServiceErrorKind::ModelLoadFailed, format!("Failed to load model {model_id} for synthesis: {error}"), )); } - // RAII guard ensures the model is always unloaded, even on panic or - // task cancellation. Matches DaemonRequestHandling.tla ClientDisconnect: - // mutex_holder = c => model_loaded' = FALSE - let _model_guard = ModelUnloadGuard { - core: &self.core, - model_id, - model_path, + let synthesis_result = { + // RAII guard ensures the model is always unloaded, even on panic or + // task cancellation. Matches DaemonRequestHandling.tla ClientDisconnect: + // mutex_holder = c => model_loaded' = FALSE + let _model_guard = ModelUnloadGuard { + core, + model_id, + model_path, + }; + + core.synthesize_with_rate(&text, style_id, rate) }; - let synthesis_result = self.core.synthesize_with_rate(&text, style_id, rate); + self.release_core(); match synthesis_result { Ok(wav_data) => Ok(DaemonServiceResult::SynthesizeResult { wav_data }), diff --git a/src/infrastructure/daemon/state/policy.rs b/src/infrastructure/daemon/state/policy.rs index 415521d..92ab095 100644 --- a/src/infrastructure/daemon/state/policy.rs +++ b/src/infrastructure/daemon/state/policy.rs @@ -26,7 +26,7 @@ impl SerializedSynthesisPolicy { requested_id: u32, rate: f32, ) -> Result { - let executor = self.executor.lock().await; + let mut executor = self.executor.lock().await; executor.synthesize(catalog, text, requested_id, rate) } } diff --git a/src/infrastructure/memory.rs b/src/infrastructure/memory.rs new file mode 100644 index 0000000..a010196 --- /dev/null +++ b/src/infrastructure/memory.rs @@ -0,0 +1,23 @@ +/// Asks the platform allocator to return unused process memory to the OS. +/// +/// This is a best-effort RSS reduction hook for large native allocations released +/// by VOICEVOX Core. The return value is the allocator-reported released byte +/// count when the platform exposes one. +#[cfg(target_os = "macos")] +pub fn release_unused_allocator_memory() -> usize { + use libc::size_t; + use std::ffi::c_void; + + unsafe extern "C" { + fn malloc_zone_pressure_relief(zone: *mut c_void, goal: size_t) -> size_t; + } + + // A null zone asks libmalloc to examine all zones. A zero goal requests as + // much relief as the allocator can currently provide. + unsafe { malloc_zone_pressure_relief(std::ptr::null_mut(), 0) as usize } +} + +#[cfg(not(target_os = "macos"))] +pub fn release_unused_allocator_memory() -> usize { + 0 +} diff --git a/src/infrastructure/mod.rs b/src/infrastructure/mod.rs index b4dfc4c..54906f4 100644 --- a/src/infrastructure/mod.rs +++ b/src/infrastructure/mod.rs @@ -4,6 +4,7 @@ pub mod download; pub mod ipc; pub mod logging; pub mod mcp_instructions; +pub mod memory; pub mod onnxruntime; pub mod openjtalk; pub mod paths; From cf1aa75b07509c52d398900f5a59052d1eabfae9 Mon Sep 17 00:00:00 2001 From: usabarashi <19676305+usabarashi@users.noreply.github.com> Date: Tue, 21 Apr 2026 23:41:53 +0900 Subject: [PATCH 2/2] fix: keep daemon IPC frame limits asymmetric --- src/infrastructure/daemon/server.rs | 17 ++++--- src/infrastructure/daemon/state/executor.rs | 50 ++++++++------------- 2 files changed, 29 insertions(+), 38 deletions(-) diff --git a/src/infrastructure/daemon/server.rs b/src/infrastructure/daemon/server.rs index 659876e..b3b3d26 100644 --- a/src/infrastructure/daemon/server.rs +++ b/src/infrastructure/daemon/server.rs @@ -8,7 +8,7 @@ use tokio::net::{UnixListener, UnixStream}; use tokio::signal; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tokio::time::timeout; -use tokio_util::codec::{Framed, LengthDelimitedCodec}; +use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec}; use crate::infrastructure::daemon::state::DaemonState; use crate::infrastructure::ipc::{ @@ -117,12 +117,17 @@ async fn handle_client_with_limit( state: Arc, permits: Arc, ) -> Result<()> { - let codec = LengthDelimitedCodec::builder() - .max_frame_length(MAX_DAEMON_REQUEST_FRAME_BYTES.max(MAX_DAEMON_RESPONSE_FRAME_BYTES)) + let request_codec = LengthDelimitedCodec::builder() + .max_frame_length(MAX_DAEMON_REQUEST_FRAME_BYTES) .new_codec(); - let mut framed = Framed::new(stream, codec); + let response_codec = LengthDelimitedCodec::builder() + .max_frame_length(MAX_DAEMON_RESPONSE_FRAME_BYTES) + .new_codec(); + let (reader, writer) = stream.into_split(); + let mut framed_read = FramedRead::new(reader, request_codec); + let mut framed_write = FramedWrite::new(writer, response_codec); - while let Some(frame) = timeout(CLIENT_IDLE_TIMEOUT, framed.next()) + while let Some(frame) = timeout(CLIENT_IDLE_TIMEOUT, framed_read.next()) .await .map_err(|_| anyhow!("Client idle timeout"))? { @@ -150,7 +155,7 @@ async fn handle_client_with_limit( break; }; - if let Err(error) = framed.send(response_data.into()).await { + if let Err(error) = framed_write.send(response_data.into()).await { log_client_error("Client stream write error", &error); break; } diff --git a/src/infrastructure/daemon/state/executor.rs b/src/infrastructure/daemon/state/executor.rs index 366674d..8166eaf 100644 --- a/src/infrastructure/daemon/state/executor.rs +++ b/src/infrastructure/daemon/state/executor.rs @@ -5,9 +5,7 @@ use crate::infrastructure::core::VoicevoxCore; use super::catalog::{ModelCatalog, TargetResolution}; use super::result::{DaemonServiceError, DaemonServiceErrorKind, DaemonServiceResult}; -pub(super) struct DaemonSynthesisExecutor { - core: Option, -} +pub(super) struct DaemonSynthesisExecutor; /// RAII guard that unloads a voice model on drop. /// @@ -20,6 +18,14 @@ struct ModelUnloadGuard<'a> { model_path: Option<&'a Path>, } +struct AllocatorReliefGuard; + +impl Drop for AllocatorReliefGuard { + fn drop(&mut self) { + crate::infrastructure::memory::release_unused_allocator_memory(); + } +} + impl Drop for ModelUnloadGuard<'_> { fn drop(&mut self) { let Some(model_path) = self.model_path else { @@ -41,25 +47,7 @@ impl Drop for ModelUnloadGuard<'_> { impl DaemonSynthesisExecutor { pub(super) fn new() -> Self { - Self { core: None } - } - - fn ensure_core(&mut self) -> Result<(), DaemonServiceError> { - if self.core.is_none() { - self.core = Some(VoicevoxCore::new().map_err(|error| { - DaemonServiceError::new( - DaemonServiceErrorKind::ModelLoadFailed, - format!("Failed to initialize VOICEVOX core for synthesis: {error}"), - ) - })?); - } - - Ok(()) - } - - fn release_core(&mut self) { - self.core = None; - crate::infrastructure::memory::release_unused_allocator_memory(); + Self } pub(super) fn synthesize( @@ -80,18 +68,18 @@ impl DaemonSynthesisExecutor { }; let model_path = catalog.get_model_path(model_id); - self.ensure_core()?; - - let core = self - .core - .as_ref() - .expect("core must be initialized by ensure_core"); + let _allocator_relief = AllocatorReliefGuard; + let core = VoicevoxCore::new().map_err(|error| { + DaemonServiceError::new( + DaemonServiceErrorKind::ModelLoadFailed, + format!("Failed to initialize VOICEVOX core for synthesis: {error}"), + ) + })?; if let Err(error) = core.load_specific_model(model_id) { crate::infrastructure::logging::error(&format!( "Failed to load model {model_id}: {error}" )); - self.release_core(); return Err(DaemonServiceError::new( DaemonServiceErrorKind::ModelLoadFailed, format!("Failed to load model {model_id} for synthesis: {error}"), @@ -103,7 +91,7 @@ impl DaemonSynthesisExecutor { // task cancellation. Matches DaemonRequestHandling.tla ClientDisconnect: // mutex_holder = c => model_loaded' = FALSE let _model_guard = ModelUnloadGuard { - core, + core: &core, model_id, model_path, }; @@ -111,8 +99,6 @@ impl DaemonSynthesisExecutor { core.synthesize_with_rate(&text, style_id, rate) }; - self.release_core(); - match synthesis_result { Ok(wav_data) => Ok(DaemonServiceResult::SynthesizeResult { wav_data }), Err(error) => Err(DaemonServiceError::new(