fix: resolve deadlock in audio cache with single mutex

This commit is contained in:
vicrodh committed 2026-01-09 10:05:29 -06:00
1 parent 28bd316c89
commit 913b80e09a
1 file changed
+75 -161
+75 -161
View File
@@ -5,9 +5,8 @@
//! - Pre-fetching of next track in queue //! - Pre-fetching of next track in queue
//! - Background download tasks //! - Background download tasks
use std::collections::HashMap; use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex}; use std::sync::Mutex;
use tokio::sync::mpsc;
/// Cached audio data for a track /// Cached audio data for a track
#[derive(Clone)] #[derive(Clone)]
@@ -17,18 +16,23 @@ pub struct CachedTrack {
pub size_bytes: usize, pub size_bytes: usize,
} }
/// Internal cache state - all in one struct to avoid deadlocks
struct CacheState {
/// Cached tracks keyed by track ID
tracks: HashMap<u64, CachedTrack>,
/// Order of access for LRU eviction (most recent at back)
access_order: Vec<u64>,
/// Current cache size in bytes
current_size: usize,
/// Track IDs currently being fetched
fetching: HashSet<u64>,
}
/// Audio cache manager with LRU eviction /// Audio cache manager with LRU eviction
pub struct AudioCache { pub struct AudioCache {
/// Cached tracks keyed by track ID state: Mutex<CacheState>,
cache: Mutex<HashMap<u64, CachedTrack>>,
/// Order of access for LRU eviction (most recent at back)
access_order: Mutex<Vec<u64>>,
/// Maximum cache size in bytes (default: 500MB) /// Maximum cache size in bytes (default: 500MB)
max_size_bytes: usize, max_size_bytes: usize,
/// Current cache size in bytes
current_size: Mutex<usize>,
/// Track IDs currently being fetched
fetching: Mutex<std::collections::HashSet<u64>>,
} }
impl Default for AudioCache { impl Default for AudioCache {
@@ -41,49 +45,52 @@ impl AudioCache {
/// Create a new cache with specified max size in bytes /// Create a new cache with specified max size in bytes
pub fn new(max_size_bytes: usize) -> Self { pub fn new(max_size_bytes: usize) -> Self {
Self { Self {
cache: Mutex::new(HashMap::new()), state: Mutex::new(CacheState {
access_order: Mutex::new(Vec::new()), tracks: HashMap::new(),
access_order: Vec::new(),
current_size: 0,
fetching: HashSet::new(),
}),
max_size_bytes, max_size_bytes,
current_size: Mutex::new(0),
fetching: Mutex::new(std::collections::HashSet::new()),
} }
} }
/// Get a track from cache if available /// Get a track from cache if available
pub fn get(&self, track_id: u64) -> Option<CachedTrack> { pub fn get(&self, track_id: u64) -> Option<CachedTrack> {
let cache = self.cache.lock().unwrap(); let mut state = self.state.lock().unwrap();
if let Some(track) = cache.get(&track_id) {
// Update access order (move to back = most recently used)
let mut order = self.access_order.lock().unwrap();
order.retain(|&id| id != track_id);
order.push(track_id);
let track = state.tracks.get(&track_id).cloned();
if track.is_some() {
// Update access order (move to back = most recently used)
state.access_order.retain(|&id| id != track_id);
state.access_order.push(track_id);
log::debug!("Cache hit for track {}", track_id); log::debug!("Cache hit for track {}", track_id);
Some(track.clone())
} else { } else {
log::debug!("Cache miss for track {}", track_id); log::debug!("Cache miss for track {}", track_id);
None
} }
track
} }
/// Check if a track is in cache without updating access order /// Check if a track is in cache without updating access order
pub fn contains(&self, track_id: u64) -> bool { pub fn contains(&self, track_id: u64) -> bool {
self.cache.lock().unwrap().contains_key(&track_id) self.state.lock().unwrap().tracks.contains_key(&track_id)
} }
/// Check if a track is currently being fetched /// Check if a track is currently being fetched
pub fn is_fetching(&self, track_id: u64) -> bool { pub fn is_fetching(&self, track_id: u64) -> bool {
self.fetching.lock().unwrap().contains(&track_id) self.state.lock().unwrap().fetching.contains(&track_id)
} }
/// Mark a track as being fetched /// Mark a track as being fetched
pub fn mark_fetching(&self, track_id: u64) { pub fn mark_fetching(&self, track_id: u64) {
self.fetching.lock().unwrap().insert(track_id); self.state.lock().unwrap().fetching.insert(track_id);
} }
/// Unmark a track as being fetched /// Unmark a track as being fetched
pub fn unmark_fetching(&self, track_id: u64) { pub fn unmark_fetching(&self, track_id: u64) {
self.fetching.lock().unwrap().remove(&track_id); self.state.lock().unwrap().fetching.remove(&track_id);
} }
/// Insert a track into cache, evicting old entries if needed /// Insert a track into cache, evicting old entries if needed
@@ -101,51 +108,13 @@ impl AudioCache {
return; return;
} }
let mut state = self.state.lock().unwrap();
// Evict old entries to make room // Evict old entries to make room
self.evict_to_fit(size); while state.current_size + size > self.max_size_bytes && !state.access_order.is_empty() {
let oldest_id = state.access_order.remove(0);
let cached = CachedTrack { if let Some(track) = state.tracks.remove(&oldest_id) {
track_id, state.current_size = state.current_size.saturating_sub(track.size_bytes);
data,
size_bytes: size,
};
let mut cache = self.cache.lock().unwrap();
// If track already exists, update size tracking
if let Some(existing) = cache.get(&track_id) {
let mut current = self.current_size.lock().unwrap();
*current = current.saturating_sub(existing.size_bytes);
}
cache.insert(track_id, cached);
// Update size and access order
*self.current_size.lock().unwrap() += size;
let mut order = self.access_order.lock().unwrap();
order.retain(|&id| id != track_id);
order.push(track_id);
log::info!(
"Cached track {} ({} bytes). Cache size: {}/{} bytes",
track_id,
size,
*self.current_size.lock().unwrap(),
self.max_size_bytes
);
}
/// Evict oldest entries until we have room for new_size bytes
fn evict_to_fit(&self, new_size: usize) {
let mut current = self.current_size.lock().unwrap();
let mut cache = self.cache.lock().unwrap();
let mut order = self.access_order.lock().unwrap();
while *current + new_size > self.max_size_bytes && !order.is_empty() {
// Remove oldest (front of order)
let oldest_id = order.remove(0);
if let Some(track) = cache.remove(&oldest_id) {
*current = current.saturating_sub(track.size_bytes);
log::debug!( log::debug!(
"Evicted track {} ({} bytes) from cache", "Evicted track {} ({} bytes) from cache",
oldest_id, oldest_id,
@@ -153,24 +122,52 @@ impl AudioCache {
); );
} }
} }
// If track already exists, update size tracking
if let Some(existing) = state.tracks.get(&track_id) {
state.current_size = state.current_size.saturating_sub(existing.size_bytes);
}
let cached = CachedTrack {
track_id,
data,
size_bytes: size,
};
state.tracks.insert(track_id, cached);
state.current_size += size;
// Update access order
state.access_order.retain(|&id| id != track_id);
state.access_order.push(track_id);
log::info!(
"Cached track {} ({} bytes). Cache size: {}/{} bytes",
track_id,
size,
state.current_size,
self.max_size_bytes
);
} }
/// Clear all cached data /// Clear all cached data
pub fn clear(&self) { pub fn clear(&self) {
self.cache.lock().unwrap().clear(); let mut state = self.state.lock().unwrap();
self.access_order.lock().unwrap().clear(); state.tracks.clear();
*self.current_size.lock().unwrap() = 0; state.access_order.clear();
self.fetching.lock().unwrap().clear(); state.current_size = 0;
state.fetching.clear();
log::info!("Cache cleared"); log::info!("Cache cleared");
} }
/// Get cache statistics /// Get cache statistics
pub fn stats(&self) -> CacheStats { pub fn stats(&self) -> CacheStats {
let state = self.state.lock().unwrap();
CacheStats { CacheStats {
cached_tracks: self.cache.lock().unwrap().len(), cached_tracks: state.tracks.len(),
current_size_bytes: *self.current_size.lock().unwrap(), current_size_bytes: state.current_size,
max_size_bytes: self.max_size_bytes, max_size_bytes: self.max_size_bytes,
fetching_count: self.fetching.lock().unwrap().len(), fetching_count: state.fetching.len(),
} }
} }
} }
@@ -183,86 +180,3 @@ pub struct CacheStats {
pub max_size_bytes: usize, pub max_size_bytes: usize,
pub fetching_count: usize, pub fetching_count: usize,
} }
/// Pre-fetcher for downloading next tracks in background
pub struct Prefetcher {
/// Sender for prefetch requests
tx: mpsc::Sender<PrefetchRequest>,
}
/// Request to prefetch a track
pub struct PrefetchRequest {
pub track_id: u64,
pub url: String,
}
impl Prefetcher {
/// Create a new prefetcher with the given cache
pub fn new(cache: Arc<AudioCache>) -> Self {
let (tx, mut rx) = mpsc::channel::<PrefetchRequest>(10);
// Spawn background task for prefetching
tokio::spawn(async move {
while let Some(request) = rx.recv().await {
// Skip if already cached or being fetched
if cache.contains(request.track_id) || cache.is_fetching(request.track_id) {
continue;
}
cache.mark_fetching(request.track_id);
log::info!("Prefetching track {}...", request.track_id);
match download_audio(&request.url).await {
Ok(data) => {
cache.insert(request.track_id, data);
log::info!("Prefetch complete for track {}", request.track_id);
}
Err(e) => {
log::warn!("Prefetch failed for track {}: {}", request.track_id, e);
}
}
cache.unmark_fetching(request.track_id);
}
});
Self { tx }
}
/// Request prefetch of a track (non-blocking)
pub fn prefetch(&self, track_id: u64, url: String) {
let tx = self.tx.clone();
tokio::spawn(async move {
let _ = tx.send(PrefetchRequest { track_id, url }).await;
});
}
}
/// Download audio from URL
async fn download_audio(url: &str) -> Result<Vec<u8>, String> {
use std::time::Duration;
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(120))
.connect_timeout(Duration::from_secs(10))
.build()
.map_err(|e| format!("Failed to create HTTP client: {}", e))?;
let response = client
.get(url)
.header("User-Agent", "Mozilla/5.0")
.send()
.await
.map_err(|e| format!("Failed to fetch audio: {}", e))?;
if !response.status().is_success() {
return Err(format!("HTTP error: {}", response.status()));
}
let bytes = response
.bytes()
.await
.map_err(|e| format!("Failed to read audio bytes: {}", e))?;
Ok(bytes.to_vec())
}