eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

🚀 Optimise OpenHome playlist sync performance (-75% SOAP calls)

Browse Source 
- Augmente le batch ReadList de 64 à 256 (-75% appels SOAP)
- Élimine les doubles appel à queue_snapshot() dans sync_queue()–50% appels SOAP en moins
- Introduit lcs_flags_optimized() pour élaguer préfixe/suffixes communs (LCS O(n) vs quadratique)
- Met en place un polling adaptatif : 500ms actif / 5s veille
- Factorise parse_duration() et invalidation de caches via méthodes utilitaires
This commit is contained in:
Eric Coissac
2026-04-06 16:59:23 +02:00
parent cba1c01b1b
commit 22b3a67417
7 changed files with 648 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
pmocontrol/src/music_renderer/musicrenderer.rs
View File

@@ -9,7 +9,7 @@
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::SystemTime;
use std::time::{Duration, SystemTime};
use pmodidl::{DIDLLite, MediaMetadataParser};
use tracing::{debug, error};
@@ -312,17 +312,37 @@ impl MusicRenderer {
/// Main loop for the watcher thread.
fn watcher_loop(&self, strategy: WatchStrategy, stop_flag: Arc<AtomicBool>) {
let Some(interval) = strategy.polling_interval() else {
let Some(base_interval) = strategy.polling_interval() else {
// Pure push strategy - no polling needed (future implementation)
return;
};
let short_interval = base_interval;
let long_interval = Duration::from_millis(5_000);
let inactivity_threshold = Duration::from_secs(10);
let mut tick: u32 = 0;
let mut next_poll_time = SystemTime::now();
let mut last_activity_time = SystemTime::now();
while !stop_flag.load(Ordering::SeqCst) {
let is_active = {
let watched = self
.watched_state
.lock()
.expect("WatchedState mutex poisoned");
watched.is_active
};
let interval = if is_active {
short_interval
} else {
long_interval
};
if self.is_online() {
self.poll_and_emit_changes(tick);
last_activity_time = SystemTime::now();
}
tick = tick.wrapping_add(1);
@@ -341,6 +361,17 @@ impl MusicRenderer {
);
next_poll_time = SystemTime::now();
}
// Reset is_active flag if no activity for threshold
if let Ok(elapsed) = SystemTime::now().duration_since(last_activity_time) {
if elapsed >= inactivity_threshold {
let mut watched = self
.watched_state
.lock()
.expect("WatchedState mutex poisoned");
watched.is_active = false;
}
}
}
debug!(
@@ -1012,6 +1043,15 @@ impl MusicRenderer {
/// Démarre ou reprend la lecture. Si une queue non vide existe,
/// joue le track courant de la queue automatiquement (comportement unifié pour tous les backends).
pub fn play(&self) -> Result<(), ControlPointError> {
// Mark renderer as active for adaptive polling
{
let mut watched = self
.watched_state
.lock()
.expect("WatchedState mutex poisoned");
watched.is_active = true;
}
// Vérifier si on a une queue non vide
let backend = self.lock_backend_for("play");
let queue_not_empty = backend.len().unwrap_or(0) > 0;
@@ -1034,6 +1074,15 @@ impl MusicRenderer {
/// Transport control: stop
#[track_caller]
pub fn stop(&self) -> Result<(), ControlPointError> {
// Mark renderer as active for adaptive polling
{
let mut watched = self
.watched_state
.lock()
.expect("WatchedState mutex poisoned");
watched.is_active = true;
}
// Reset the has_played flag when stopping playback.
// This ensures that if we start a new track, the flag will be false
// until PLAYING state is observed, preventing auto-advance on
@@ -1067,6 +1116,15 @@ impl MusicRenderer {
/// Transport control: seek to relative time
pub fn seek_rel_time(&self, hhmmss: &str) -> Result<(), ControlPointError> {
// Mark renderer as active for adaptive polling
{
let mut watched = self
.watched_state
.lock()
.expect("WatchedState mutex poisoned");
watched.is_active = true;
}
self.lock_backend_for("seek_rel_time").seek_rel_time(hhmmss)
}
@@ -1314,6 +1372,15 @@ impl MusicRenderer {
/// - If the current track is NOT in the new items, it's preserved as the first item
/// - If there's no current track, the queue is simply replaced
pub fn sync_queue(&self, items: Vec<PlaybackItem>) -> Result<(), ControlPointError> {
// Mark renderer as active for adaptive polling
{
let mut watched = self
.watched_state
.lock()
.expect("WatchedState mutex poisoned");
watched.is_active = true;
}
let mut backend = self.lock_backend_for("sync_queue");
backend.sync_queue(items)?;
drop(backend);

2
pmocontrol/src/music_renderer/watcher.rs
View File

@@ -91,6 +91,8 @@ pub struct WatchedState {
pub metadata: Option<TrackMetadata>,
/// Last known stream state (continuous stream vs bounded media)
pub is_stream: Option<bool>,
/// Flag indicating if renderer was active recently (used for adaptive polling)
pub is_active: bool,
}
// ============================================================================

66
pmocontrol/src/queue/interne.rs
View File

@@ -65,25 +65,7 @@ impl InternalQueue {
/// Vérifie si une durée a diminué (format HH:MM:SS).
/// Retourne true si new_duration < old_duration.
fn duration_decreased(old_duration: &str, new_duration: &str) -> bool {
let parse_duration = |dur: &str| -> Option<u32> {
let parts: Vec<&str> = dur.split(':').collect();
if parts.len() == 3 {
let h: u32 = parts[0].parse().ok()?;
let m: u32 = parts[1].parse().ok()?;
let s: u32 = parts[2].parse().ok()?;
Some(h * 3600 + m * 60 + s)
} else {
None
}
};
if let (Some(old_secs), Some(new_secs)) =
(parse_duration(old_duration), parse_duration(new_duration))
{
new_secs < old_secs
} else {
false // Impossible de parser: considérer que ça n'a pas diminué
}
super::stream_duration_decreased(old_duration, new_duration)
}
/// Protège les durées des streams contre la diminution.
@@ -164,45 +146,23 @@ impl InternalQueue {
// Même chanson sur un stream: vérifier que la durée n'a pas diminué
let should_update = match (&old_meta.duration, &new_meta.duration) {
(Some(old_dur), Some(new_dur)) => {
// Parser les durées (format HH:MM:SS)
let parse_duration = |dur: &str| -> Option<u32> {
let parts: Vec<&str> = dur.split(':').collect();
if parts.len() == 3 {
let h: u32 = parts[0].parse().ok()?;
let m: u32 = parts[1].parse().ok()?;
let s: u32 = parts[2].parse().ok()?;
Some(h * 3600 + m * 60 + s)
} else {
None
}
};
if let (Some(old_secs), Some(new_secs)) =
(parse_duration(old_dur), parse_duration(new_dur))
{
if new_secs < old_secs {
// Durée a diminué: garder l'ancienne
tracing::trace!(
"InternalQueue merge_metadata: uri={}, REJECTING update (same stream track, duration decreased): {} -> {}",
if super::stream_duration_decreased(old_dur, new_dur) {
tracing::trace!(
"InternalQueue merge_metadata: uri={}, REJECTING update (same stream track, duration decreased): {} -> {}",
uri,
old_dur,
new_dur
);
false
} else {
if super::stream_duration_increased(old_dur, new_dur) {
tracing::debug!(
"InternalQueue merge_metadata: uri={}, same stream track, duration increased: {} -> {}",
uri,
old_dur,
new_dur
);
false
} else {
// Durée a augmenté ou est égale: accepter
if new_secs > old_secs {
tracing::debug!(
"InternalQueue merge_metadata: uri={}, same stream track, duration increased: {} -> {}",
uri,
old_dur,
new_dur
);
}
true
}
} else {
// Impossible de parser: accepter par défaut
true
}
}

18
pmocontrol/src/queue/mod.rs
View File

@@ -15,6 +15,24 @@ pub(crate) use interne::InternalQueue;
pub(crate) use openhome::OpenHomeQueue;
use crate::{RendererInfo, errors::ControlPointError};
use crate::music_renderer::time_utils::parse_time_flexible;
/// Returns true if `new_dur` < `old_dur` (both parseable as HH:MM:SS/MM:SS/SS).
/// Used to protect stream durations from decreasing for the same track.
pub(super) fn stream_duration_decreased(old_dur: &str, new_dur: &str) -> bool {
match (parse_time_flexible(old_dur).ok(), parse_time_flexible(new_dur).ok()) {
(Some(old_secs), Some(new_secs)) => new_secs < old_secs,
_ => false,
}
}
/// Returns true if `new_dur` > `old_dur` (both parseable as HH:MM:SS/MM:SS/SS).
pub(super) fn stream_duration_increased(old_dur: &str, new_dur: &str) -> bool {
match (parse_time_flexible(old_dur).ok(), parse_time_flexible(new_dur).ok()) {
(Some(old_secs), Some(new_secs)) => new_secs > old_secs,
_ => false,
}
}
pub trait QueueFromRendererInfo {
fn from_renderer_info(renderer: &RendererInfo) -> Result<Self, ControlPointError>

174
pmocontrol/src/queue/openhome.rs
View File

@@ -218,6 +218,19 @@ impl OpenHomeQueue {
}
}
/// Invalide les caches track_ids et read_list (après insert/delete sans impact sur la piste courante).
fn invalidate_track_caches(&self) {
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
}
/// Invalide tous les caches (après delete_all, seek, stop — opérations qui changent la piste courante).
fn invalidate_all_caches(&self) {
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.current_track_id_cache.lock().unwrap().invalidate();
}
/// Met à jour les métadonnées d'un item de la queue à l'index spécifié.
///
/// Contrairement au service OpenHome qui ne permet pas de modifier les métadonnées,
@@ -274,45 +287,23 @@ impl OpenHomeQueue {
new_metadata.as_ref().and_then(|m| m.duration.as_ref()),
) {
(Some(cached_dur), Some(new_dur)) => {
// Parser les durées (format HH:MM:SS)
let parse_duration = |dur: &str| -> Option<u32> {
let parts: Vec<&str> = dur.split(':').collect();
if parts.len() == 3 {
let h: u32 = parts[0].parse().ok()?;
let m: u32 = parts[1].parse().ok()?;
let s: u32 = parts[2].parse().ok()?;
Some(h * 3600 + m * 60 + s)
} else {
None
}
};
if let (Some(cached_secs), Some(new_secs)) =
(parse_duration(cached_dur), parse_duration(new_dur))
{
if new_secs < cached_secs {
// Durée a diminué pour la même chanson: refuser
tracing::trace!(
"OpenHome cache_metadata: track_id={}, REJECTING update (same track, duration decreased): {} -> {}",
if super::stream_duration_decreased(cached_dur, new_dur) {
tracing::trace!(
"OpenHome cache_metadata: track_id={}, REJECTING update (same track, duration decreased): {} -> {}",
track_id,
cached_dur,
new_dur
);
false
} else {
if super::stream_duration_increased(cached_dur, new_dur) {
tracing::debug!(
"OpenHome cache_metadata: track_id={}, same track, duration increased: {} -> {}",
track_id,
cached_dur,
new_dur
);
false
} else {
// Durée a augmenté ou est égale: accepter
if new_secs > cached_secs {
tracing::debug!(
"OpenHome cache_metadata: track_id={}, same track, duration increased: {} -> {}",
track_id,
cached_dur,
new_dur
);
}
true
}
} else {
// Impossible de parser: accepter par défaut
true
}
}
@@ -452,8 +443,7 @@ impl OpenHomeQueue {
);
// Invalidate cache after playlist modifications
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.invalidate_track_caches();
Ok(())
}
@@ -550,11 +540,9 @@ impl OpenHomeQueue {
new_items: Vec<PlaybackItem>,
pivot_idx_new: usize,
pivot_id: usize,
snapshot: &QueueSnapshot,
current_track_ids: &[u32],
) -> Result<(), ControlPointError> {
// Get current state from OpenHome
let snapshot = self.queue_snapshot()?;
let current_track_ids = self.track_ids()?;
// Find the pivot index in our current state
let pivot_idx = current_track_ids
.iter()
@@ -576,10 +564,10 @@ impl OpenHomeQueue {
let new_after = &new_items[pivot_idx_new + 1..];
// LCS on the AFTER part (using fresh data from OpenHome)
let (keep_old_after, keep_new_after) = lcs_flags(&old_after, new_after);
let (keep_old_after, keep_new_after) = lcs_flags_optimized(&old_after, new_after);
// LCS on the BEFORE part (using fresh data from OpenHome)
let (keep_old_before, keep_new_before) = lcs_flags(&old_before, new_before);
let (keep_old_before, keep_new_before) = lcs_flags_optimized(&old_before, new_before);
// Delete items marked for deletion in AFTER part (reverse order)
self.delete_marked_items(&old_ids_after, &keep_old_after, "AFTER pivot")?;
@@ -631,8 +619,7 @@ impl OpenHomeQueue {
);
// Invalidate cache after playlist modifications
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.invalidate_track_caches();
Ok(())
}
@@ -641,12 +628,9 @@ impl OpenHomeQueue {
fn replace_queue_standard_lcs(
&mut self,
items: Vec<PlaybackItem>,
_current_index: Option<usize>,
snapshot: &QueueSnapshot,
current_track_ids: &[u32],
) -> Result<(), ControlPointError> {
// Get current state from OpenHome
let snapshot = self.queue_snapshot()?;
let current_track_ids = self.track_ids()?;
debug!(
renderer = self.renderer_id.0.as_str(),
current_count = snapshot.items.len(),
@@ -658,7 +642,7 @@ impl OpenHomeQueue {
"LCS input: current vs desired items"
);
let (keep_current, keep_desired) = lcs_flags(&snapshot.items, &items);
let (keep_current, keep_desired) = lcs_flags_optimized(&snapshot.items, &items);
let items_to_keep = keep_current.iter().filter(|&&k| k).count();
let items_to_delete = keep_current.iter().filter(|&k| !k).count();
@@ -768,8 +752,7 @@ impl OpenHomeQueue {
}
// Invalidate cache after playlist modifications
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.invalidate_track_caches();
Ok(())
}
@@ -883,6 +866,52 @@ fn lcs_flags(current: &[PlaybackItem], desired: &[PlaybackItem]) -> (Vec<bool>,
(keep_current, keep_desired)
}
fn lcs_flags_optimized(
current: &[PlaybackItem],
desired: &[PlaybackItem],
) -> (Vec<bool>, Vec<bool>) {
if current.is_empty() {
return (vec![], vec![true; desired.len()]);
}
if desired.is_empty() {
return (vec![true; current.len()], vec![]);
}
let leading = current
.iter()
.zip(desired.iter())
.take_while(|(c, d)| items_match(c, d))
.count();
let c_tail = &current[leading..];
let d_tail = &desired[leading..];
let trailing = c_tail
.iter()
.rev()
.zip(d_tail.iter().rev())
.take_while(|(c, d)| items_match(c, d))
.count();
let c_mid = &c_tail[..c_tail.len().saturating_sub(trailing)];
let d_mid = &d_tail[..d_tail.len().saturating_sub(trailing)];
if c_mid.is_empty() && d_mid.is_empty() {
return (vec![true; current.len()], vec![true; desired.len()]);
}
let (keep_c_mid, keep_d_mid) = lcs_flags(c_mid, d_mid);
let mut keep_current = vec![true; leading];
keep_current.extend(keep_c_mid);
keep_current.extend(vec![true; trailing]);
let mut keep_desired = vec![true; leading];
keep_desired.extend(keep_d_mid);
keep_desired.extend(vec![true; trailing]);
(keep_current, keep_desired)
}
impl QueueBackend for OpenHomeQueue {
fn len(&self) -> Result<usize, ControlPointError> {
Ok(self.track_ids()?.len())
@@ -983,7 +1012,7 @@ impl QueueBackend for OpenHomeQueue {
// Read metadata for all tracks (batched), with 500ms cache to avoid
// redundant SOAP calls during sync_queue (which calls queue_snapshot twice).
const MAX_BATCH: usize = 64;
const MAX_BATCH: usize = 256;
let mut entries = Vec::with_capacity(ids.len());
for chunk in ids.chunks(MAX_BATCH) {
if let Some(cached) = self.read_list_cache.lock().unwrap().get(chunk) {
@@ -1056,9 +1085,7 @@ impl QueueBackend for OpenHomeQueue {
self.playlist_client.stop()?;
}
// Invalidate caches (seek_id/stop modifies playlist state and current track)
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.current_track_id_cache.lock().unwrap().invalidate();
self.invalidate_all_caches();
Ok(())
}
@@ -1082,9 +1109,7 @@ impl QueueBackend for OpenHomeQueue {
self.metadata_cache.lock().unwrap().clear();
// Invalidate caches after delete_all (clears queue and current track)
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.current_track_id_cache.lock().unwrap().invalidate();
self.invalidate_all_caches();
if items.is_empty() {
return Ok(());
@@ -1105,8 +1130,7 @@ impl QueueBackend for OpenHomeQueue {
}
// Invalidate cache after insertions
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.invalidate_track_caches();
Ok(())
}
@@ -1131,9 +1155,7 @@ impl QueueBackend for OpenHomeQueue {
self.playlist_client.delete_all()?;
self.metadata_cache.lock().unwrap().clear();
// Invalidate caches after delete_all (clears queue and current track)
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.current_track_id_cache.lock().unwrap().invalidate();
self.invalidate_all_caches();
// DIAGNOSTIC: Log state after delete_all
let post_current_track = self.playlist_client.id().ok();
@@ -1218,7 +1240,15 @@ impl QueueBackend for OpenHomeQueue {
pivot_idx
);
self.replace_queue_with_pivot(items, pivot_idx, playing_id)?;
let current_ids_for_pivot: Vec<u32> =
snapshot.items.iter().map(|i| i.backend_id as u32).collect();
self.replace_queue_with_pivot(
items,
pivot_idx,
playing_id,
&snapshot,
&current_ids_for_pivot,
)?;
} else {
// CASE 1: Currently playing item NOT in the new playlist
// Keep it as first item and append the new playlist after it
@@ -1250,7 +1280,9 @@ impl QueueBackend for OpenHomeQueue {
renderer = self.renderer_id.0.as_str(),
"No currently playing item, using standard LCS sync"
);
self.replace_queue_standard_lcs(items, Some(0))?;
let current_ids_for_lcs: Vec<u32> =
snapshot.items.iter().map(|i| i.backend_id as u32).collect();
self.replace_queue_standard_lcs(items, &snapshot, &current_ids_for_lcs)?;
}
// DIAGNOSTIC: Log state after sync completes
@@ -1319,8 +1351,7 @@ impl QueueBackend for OpenHomeQueue {
}
// Invalidate cache after playlist modifications
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.invalidate_track_caches();
Ok(())
}
@@ -1365,8 +1396,7 @@ impl QueueBackend for OpenHomeQueue {
}
// Invalidate cache after playlist modifications (except ReplaceAll which already does it)
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.invalidate_track_caches();
Ok(())
}
@@ -1378,9 +1408,7 @@ impl QueueBackend for OpenHomeQueue {
self.ensure_playlist_source_selected()?;
self.playlist_client.delete_all()?;
// Invalidate caches after clearing playlist (clears queue and current track)
self.track_ids_cache.lock().unwrap().invalidate();
self.read_list_cache.lock().unwrap().invalidate();
self.current_track_id_cache.lock().unwrap().invalidate();
self.invalidate_all_caches();
Ok(())
}