Optimize OpenHome playlist operations with caching
This commit introduces caching mechanisms for OpenHome playlist operations to reduce redundant SOAP calls and improve performance. Key changes include: - Added track IDs caching in OpenHomeQueue to avoid repeated id_array() calls - Implemented cache invalidation for all playlist modification operations - Added caching for source XML and source index in OhProductClient - Updated playlist length, ID retrieval, and track ID fetching to use cached data - Ensured cache is invalidated on write operations to maintain consistency The caching strategy uses time-based expiration (1 second for track IDs, 10 minutes for sources) and employs mutex locking for thread safety.
This commit is contained in:
@@ -174,16 +174,23 @@ impl OpenHomeRenderer {
|
||||
/// Retourne la longueur de la playlist OpenHome sans récupérer toutes les métadonnées.
|
||||
/// Plus rapide que snapshot_openhome_playlist() pour juste connaître le nombre de pistes.
|
||||
pub(crate) fn openhome_playlist_len(&self) -> Result<usize, ControlPointError> {
|
||||
let playlist = self.playlist_client_for("openhome_playlist_len")?;
|
||||
let ids = playlist.id_array()?;
|
||||
Ok(ids.len())
|
||||
// Use queue.len() which uses cached track_ids() internally
|
||||
let queue = self.queue.lock().unwrap();
|
||||
queue.len()
|
||||
}
|
||||
|
||||
/// Retourne les IDs des pistes de la playlist OpenHome.
|
||||
/// Plus rapide que snapshot_openhome_playlist() car ne récupère pas les métadonnées.
|
||||
pub(crate) fn openhome_playlist_ids(&self) -> Result<Vec<u32>, ControlPointError> {
|
||||
let playlist = self.playlist_client_for("openhome_playlist_ids")?;
|
||||
playlist.id_array()
|
||||
// Use the queue's cached track_ids() instead of direct id_array() call
|
||||
let queue = self.queue.lock().unwrap();
|
||||
if let MusicQueue::OpenHome(oh_queue) = &*queue {
|
||||
oh_queue.track_ids()
|
||||
} else {
|
||||
Err(ControlPointError::QueueError(
|
||||
"Not an OpenHome queue".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn clear_openhome_playlist(&self) -> Result<(), ControlPointError> {
|
||||
@@ -201,11 +208,21 @@ impl OpenHomeRenderer {
|
||||
let playlist = self.playlist_client_for("add_track_openhome")?;
|
||||
let insert_after = match after_id {
|
||||
Some(id) => id,
|
||||
None => playlist
|
||||
.id_array()?
|
||||
None => {
|
||||
// Use the queue's cached track_ids() instead of direct id_array() call
|
||||
let queue = self.queue.lock().unwrap();
|
||||
if let MusicQueue::OpenHome(oh_queue) = &*queue {
|
||||
oh_queue
|
||||
.track_ids()?
|
||||
.last()
|
||||
.copied()
|
||||
.unwrap_or(OPENHOME_PLAYLIST_HEAD_ID),
|
||||
.unwrap_or(OPENHOME_PLAYLIST_HEAD_ID)
|
||||
} else {
|
||||
return Err(ControlPointError::QueueError(
|
||||
"Not an OpenHome queue".to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let new_id = playlist.insert(insert_after, uri, metadata)?;
|
||||
@@ -376,16 +393,18 @@ impl PlaybackPosition for OpenHomeRenderer {
|
||||
let mut track_uri = None;
|
||||
let mut track_metadata_xml = None;
|
||||
|
||||
// Get track ID from playlist
|
||||
if let Some(playlist_client) = &self.playlist {
|
||||
match playlist_client.id() {
|
||||
Ok(id) => track_id = Some(id),
|
||||
// Get track ID from queue (uses cached data)
|
||||
let queue_guard_for_id = self.queue.lock().unwrap();
|
||||
if let MusicQueue::OpenHome(oh_queue) = &*queue_guard_for_id {
|
||||
match oh_queue.current_track() {
|
||||
Ok(id_opt) => track_id = id_opt,
|
||||
Err(err) => debug!(
|
||||
error = %err,
|
||||
"Failed to read OpenHome track id"
|
||||
),
|
||||
}
|
||||
}
|
||||
drop(queue_guard_for_id);
|
||||
|
||||
// Use queue API to get current item with cached metadata
|
||||
let mut queue_guard = self.queue.lock().unwrap();
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::SystemTime;
|
||||
use std::usize;
|
||||
|
||||
use quick_xml::escape::escape;
|
||||
@@ -15,6 +17,55 @@ use crate::queue::{
|
||||
};
|
||||
use crate::{DeviceId, DeviceIdentity, RendererInfo};
|
||||
|
||||
/// Cache for OpenHome track IDs to avoid redundant SOAP calls
|
||||
#[derive(Debug)]
|
||||
struct TrackIdsCache {
|
||||
/// Cached track IDs
|
||||
ids: Option<Vec<u32>>,
|
||||
/// Timestamp of last cache update
|
||||
last_update: Option<SystemTime>,
|
||||
}
|
||||
|
||||
impl TrackIdsCache {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
ids: None,
|
||||
last_update: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if cache is valid (not expired and has data)
|
||||
fn is_valid(&self) -> bool {
|
||||
if let (Some(_), Some(last_update)) = (&self.ids, self.last_update) {
|
||||
if let Ok(elapsed) = SystemTime::now().duration_since(last_update) {
|
||||
return elapsed.as_millis() < 1000; // TTL: 1 second
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
/// Get cached IDs if valid
|
||||
fn get(&self) -> Option<Vec<u32>> {
|
||||
if self.is_valid() {
|
||||
self.ids.clone()
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// Update cache with new IDs
|
||||
fn set(&mut self, ids: Vec<u32>) {
|
||||
self.ids = Some(ids);
|
||||
self.last_update = Some(SystemTime::now());
|
||||
}
|
||||
|
||||
/// Invalidate cache (called on write operations)
|
||||
fn invalidate(&mut self) {
|
||||
self.ids = None;
|
||||
self.last_update = None;
|
||||
}
|
||||
}
|
||||
|
||||
/// Local mirror of an OpenHome playlist for a single renderer.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct OpenHomeQueue {
|
||||
@@ -26,6 +77,8 @@ pub struct OpenHomeQueue {
|
||||
/// Permet de maintenir des métadonnées à jour même si le service OpenHome
|
||||
/// ne permet pas de les modifier directement.
|
||||
metadata_cache: HashMap<u32, Option<crate::model::TrackMetadata>>,
|
||||
/// Cache for track IDs to avoid redundant IdArray SOAP calls
|
||||
track_ids_cache: Arc<Mutex<TrackIdsCache>>,
|
||||
}
|
||||
|
||||
impl OpenHomeQueue {
|
||||
@@ -41,6 +94,7 @@ impl OpenHomeQueue {
|
||||
info_client,
|
||||
product_client,
|
||||
metadata_cache: HashMap::new(),
|
||||
track_ids_cache: Arc::new(Mutex::new(TrackIdsCache::new())),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -167,6 +221,9 @@ impl OpenHomeQueue {
|
||||
"Gentle sync completed: preserved playing track as first item (not in new playlist)"
|
||||
);
|
||||
|
||||
// Invalidate cache after playlist modifications
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -344,6 +401,9 @@ impl OpenHomeQueue {
|
||||
"Gentle sync completed: double-LCS with pivot (playing track preserved)"
|
||||
);
|
||||
|
||||
// Invalidate cache after playlist modifications
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -442,6 +502,9 @@ impl OpenHomeQueue {
|
||||
)));
|
||||
}
|
||||
|
||||
// Invalidate cache after playlist modifications
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -562,7 +625,22 @@ impl QueueBackend for OpenHomeQueue {
|
||||
/// Return the list of OpenHome track IDs in order.
|
||||
fn track_ids(&self) -> Result<Vec<u32>, ControlPointError> {
|
||||
self.ensure_playlist_source_selected()?;
|
||||
self.playlist_client.id_array()
|
||||
|
||||
// Lock the cache for the entire operation to prevent race conditions
|
||||
let mut cache = self.track_ids_cache.lock().unwrap();
|
||||
|
||||
// Check if cache is valid
|
||||
if let Some(cached_ids) = cache.get() {
|
||||
return Ok(cached_ids);
|
||||
}
|
||||
|
||||
// Cache miss or expired - fetch from service (keep lock held to prevent concurrent calls)
|
||||
let ids = self.playlist_client.id_array()?;
|
||||
|
||||
// Update cache before releasing lock
|
||||
cache.set(ids.clone());
|
||||
|
||||
Ok(ids)
|
||||
}
|
||||
|
||||
fn id_to_position(&self, id: u32) -> Result<usize, ControlPointError> {
|
||||
@@ -637,6 +715,8 @@ impl QueueBackend for OpenHomeQueue {
|
||||
self.ensure_playlist_source_selected()?;
|
||||
self.playlist_client.stop()?;
|
||||
}
|
||||
// Invalidate cache (seek_id modifies playlist state)
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -659,6 +739,9 @@ impl QueueBackend for OpenHomeQueue {
|
||||
self.playlist_client.delete_all()?;
|
||||
self.metadata_cache.clear();
|
||||
|
||||
// Invalidate cache after delete_all
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
|
||||
if items.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
@@ -677,6 +760,9 @@ impl QueueBackend for OpenHomeQueue {
|
||||
previous_id = new_id;
|
||||
}
|
||||
|
||||
// Invalidate cache after insertions
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -685,6 +771,8 @@ impl QueueBackend for OpenHomeQueue {
|
||||
if items.is_empty() {
|
||||
self.playlist_client.delete_all()?;
|
||||
self.metadata_cache.clear();
|
||||
// Invalidate cache after delete_all
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
@@ -819,6 +907,9 @@ impl QueueBackend for OpenHomeQueue {
|
||||
self.playlist_client.seek_id(new_id)?;
|
||||
}
|
||||
|
||||
// Invalidate cache after playlist modifications
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -857,8 +948,13 @@ impl QueueBackend for OpenHomeQueue {
|
||||
EnqueueMode::ReplaceAll => {
|
||||
// Replace the entire playlist
|
||||
self.replace_queue(items, None)?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
// Invalidate cache after playlist modifications (except ReplaceAll which already does it)
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -867,7 +963,10 @@ impl QueueBackend for OpenHomeQueue {
|
||||
/// Optimized clear_queue: use delete_all() directly instead of replace_queue.
|
||||
fn clear_queue(&mut self) -> Result<(), ControlPointError> {
|
||||
self.ensure_playlist_source_selected()?;
|
||||
self.playlist_client.delete_all()
|
||||
self.playlist_client.delete_all()?;
|
||||
// Invalidate cache after clearing playlist
|
||||
self.track_ids_cache.lock().unwrap().invalidate();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Optimized is_empty: only fetch track IDs, not the full playlist.
|
||||
|
||||
@@ -8,6 +8,8 @@ use crate::soap_client::{
|
||||
};
|
||||
use anyhow::{Result, anyhow};
|
||||
use pmodidl::DIDLLite;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, SystemTime};
|
||||
use tracing::{debug, info, trace, warn};
|
||||
use xmltree::{Element, XMLNode};
|
||||
|
||||
@@ -728,10 +730,87 @@ impl OhRadioClient {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct SourceXmlCache {
|
||||
sources: Option<Vec<OhProductSource>>,
|
||||
last_update: Option<SystemTime>,
|
||||
}
|
||||
|
||||
impl SourceXmlCache {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
sources: None,
|
||||
last_update: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_valid(&self) -> bool {
|
||||
if let (Some(_), Some(last_update)) = (&self.sources, self.last_update) {
|
||||
if let Ok(elapsed) = SystemTime::now().duration_since(last_update) {
|
||||
return elapsed < Duration::from_secs(600); // 10 minutes TTL
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn get(&self) -> Option<Vec<OhProductSource>> {
|
||||
if self.is_valid() {
|
||||
self.sources.clone()
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
fn set(&mut self, sources: Vec<OhProductSource>) {
|
||||
self.sources = Some(sources);
|
||||
self.last_update = Some(SystemTime::now());
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct SourceIndexCache {
|
||||
index: Option<u32>,
|
||||
last_update: Option<SystemTime>,
|
||||
}
|
||||
|
||||
impl SourceIndexCache {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
index: None,
|
||||
last_update: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_valid(&self) -> bool {
|
||||
if let (Some(_), Some(last_update)) = (self.index, self.last_update) {
|
||||
if let Ok(elapsed) = SystemTime::now().duration_since(last_update) {
|
||||
return elapsed < Duration::from_secs(1); // 1 second TTL
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
fn get(&self) -> Option<u32> {
|
||||
if self.is_valid() { self.index } else { None }
|
||||
}
|
||||
|
||||
fn set(&mut self, index: u32) {
|
||||
self.index = Some(index);
|
||||
self.last_update = Some(SystemTime::now());
|
||||
}
|
||||
|
||||
fn invalidate(&mut self) {
|
||||
self.index = None;
|
||||
self.last_update = None;
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct OhProductClient {
|
||||
pub control_url: String,
|
||||
pub service_type: String,
|
||||
source_xml_cache: Arc<Mutex<SourceXmlCache>>,
|
||||
source_index_cache: Arc<Mutex<SourceIndexCache>>,
|
||||
}
|
||||
|
||||
impl OhProductClient {
|
||||
@@ -739,6 +818,8 @@ impl OhProductClient {
|
||||
Self {
|
||||
control_url,
|
||||
service_type,
|
||||
source_xml_cache: Arc::new(Mutex::new(SourceXmlCache::new())),
|
||||
source_index_cache: Arc::new(Mutex::new(SourceIndexCache::new())),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -754,16 +835,39 @@ impl OhProductClient {
|
||||
}
|
||||
|
||||
pub fn source_xml(&self) -> Result<Vec<OhProductSource>> {
|
||||
// Lock the cache for the entire operation to prevent race conditions
|
||||
let mut cache = self.source_xml_cache.lock().unwrap();
|
||||
|
||||
// Check if cache is valid
|
||||
if let Some(cached_sources) = cache.get() {
|
||||
return Ok(cached_sources);
|
||||
}
|
||||
|
||||
// Cache miss or expired - fetch from service (keep lock held to prevent concurrent calls)
|
||||
let call_result =
|
||||
invoke_upnp_action(&self.control_url, &self.service_type, "SourceXml", &[])?;
|
||||
let envelope = ensure_success("SourceXml", &call_result)?;
|
||||
let response = find_child_with_suffix(&envelope.body.content, "SourceXmlResponse")
|
||||
.ok_or_else(|| anyhow!("Missing SourceXmlResponse element in SOAP body"))?;
|
||||
let xml = extract_child_text_any(response, &["SourceXml", "Xml", "Value"])?;
|
||||
parse_product_source_list(&xml)
|
||||
let sources = parse_product_source_list(&xml)?;
|
||||
|
||||
// Update cache before releasing lock
|
||||
cache.set(sources.clone());
|
||||
|
||||
Ok(sources)
|
||||
}
|
||||
|
||||
pub fn source_index(&self) -> Result<u32, ControlPointError> {
|
||||
// Lock the cache for the entire operation to prevent race conditions
|
||||
let mut cache = self.source_index_cache.lock().unwrap();
|
||||
|
||||
// Check if cache is valid
|
||||
if let Some(cached_index) = cache.get() {
|
||||
return Ok(cached_index);
|
||||
}
|
||||
|
||||
// Cache miss or expired - fetch from service (keep lock held to prevent concurrent calls)
|
||||
let call_result =
|
||||
invoke_upnp_action(&self.control_url, &self.service_type, "SourceIndex", &[])?;
|
||||
let envelope = ensure_success("SourceIndex", &call_result)?;
|
||||
@@ -773,9 +877,14 @@ impl OhProductClient {
|
||||
})?;
|
||||
|
||||
let value = extract_child_text_any(response, &["Index", "Value"])?;
|
||||
value
|
||||
let index = value
|
||||
.parse::<u32>()
|
||||
.map_err(|_| ControlPointError::UpnpBadReturnValue("volume".to_string(), value))
|
||||
.map_err(|_| ControlPointError::UpnpBadReturnValue("volume".to_string(), value))?;
|
||||
|
||||
// Update cache before releasing lock
|
||||
cache.set(index);
|
||||
|
||||
Ok(index)
|
||||
}
|
||||
|
||||
pub fn set_source_index(&self, index: u32) -> Result<(), ControlPointError> {
|
||||
@@ -787,7 +896,15 @@ impl OhProductClient {
|
||||
"SetSourceIndex",
|
||||
&args,
|
||||
)?;
|
||||
handle_action_response("SetSourceIndex", &call_result)
|
||||
let result = handle_action_response("SetSourceIndex", &call_result);
|
||||
|
||||
// Invalidate cache after write operation
|
||||
if result.is_ok() {
|
||||
let mut cache = self.source_index_cache.lock().unwrap();
|
||||
cache.invalidate();
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
pub fn ensure_playlist_source_selected(&self) -> Result<(), ControlPointError> {
|
||||
|
||||
Reference in New Issue
Block a user