523 lines
21 KiB
Rust
523 lines
21 KiB
Rust
//! Playback orchestration: bridges the audio thread, the streaming cache,
|
||
//! and the webview.
|
||
|
||
use serde::Serialize;
|
||
use std::collections::{HashMap, VecDeque};
|
||
use std::path::PathBuf;
|
||
use std::sync::{Arc, Mutex};
|
||
|
||
static CACHE_PRUNE_LOCK: Mutex<()> = Mutex::new(());
|
||
use std::time::{Duration, Instant};
|
||
use tauri::{Emitter, Manager};
|
||
|
||
use player::{Player, PlayerEvent, StreamProgress};
|
||
|
||
// The audio thread keeps its responsive 60 ms command/end-detection loop, but
|
||
// the webview does not need 16-17 IPC/reactivity updates per second to move a
|
||
// progress indicator. Roughly four to five updates per second remain visually
|
||
// smooth while letting Chromium spend most playback time asleep.
|
||
const UI_POSITION_INTERVAL: Duration = Duration::from_millis(200);
|
||
const WAVEFORM_CACHE_MAX_ENTRIES: usize = 256;
|
||
const PROGRESSIVE_WAVEFORM_CACHE_MAX_ENTRIES: usize = 64;
|
||
const WAVEFORM_JOB_MAX_IDLE_ENTRIES: usize = 256;
|
||
|
||
/// Tiny insertion-order LRU used for decoded waveform data. Waveforms are
|
||
/// cheap individually, but a path-keyed session cache must not grow with every
|
||
/// track ever visited.
|
||
struct WaveformLru<T> {
|
||
entries: HashMap<String, T>,
|
||
order: VecDeque<String>,
|
||
capacity: usize,
|
||
}
|
||
|
||
impl<T: Clone> WaveformLru<T> {
|
||
fn new(capacity: usize) -> Self {
|
||
Self {
|
||
entries: HashMap::new(),
|
||
order: VecDeque::new(),
|
||
capacity,
|
||
}
|
||
}
|
||
|
||
fn get(&mut self, key: &str) -> Option<T> {
|
||
let value = self.entries.get(key)?.clone();
|
||
self.touch(key);
|
||
Some(value)
|
||
}
|
||
|
||
fn insert(&mut self, key: String, value: T) {
|
||
self.entries.insert(key.clone(), value);
|
||
self.touch(&key);
|
||
while self.entries.len() > self.capacity {
|
||
let Some(oldest) = self.order.pop_front() else {
|
||
break;
|
||
};
|
||
self.entries.remove(&oldest);
|
||
}
|
||
}
|
||
|
||
fn remove(&mut self, key: &str) {
|
||
self.entries.remove(key);
|
||
if let Some(index) = self.order.iter().position(|candidate| candidate == key) {
|
||
self.order.remove(index);
|
||
}
|
||
}
|
||
|
||
fn touch(&mut self, key: &str) {
|
||
if let Some(index) = self.order.iter().position(|candidate| candidate == key) {
|
||
self.order.remove(index);
|
||
}
|
||
self.order.push_back(key.to_string());
|
||
}
|
||
}
|
||
|
||
#[derive(Clone)]
|
||
pub(crate) struct ProgressiveWaveform {
|
||
decoded_bytes: u64,
|
||
next_decode_bytes: u64,
|
||
peaks: Vec<f32>,
|
||
}
|
||
|
||
impl ProgressiveWaveform {
|
||
pub(crate) fn new(decoded_bytes: u64, total_bytes: u64, peaks: Vec<f32>) -> Self {
|
||
Self {
|
||
decoded_bytes,
|
||
next_decode_bytes: next_progressive_waveform_threshold(decoded_bytes, total_bytes),
|
||
peaks,
|
||
}
|
||
}
|
||
|
||
/// Recompute only after the growing file has at least doubled. If a
|
||
/// transfer restarted and the file shrank, force a fresh decode instead of
|
||
/// returning peaks from bytes that no longer exist.
|
||
pub(crate) fn should_recompute(&self, available_bytes: u64) -> bool {
|
||
available_bytes < self.decoded_bytes
|
||
|| (available_bytes > self.decoded_bytes && available_bytes >= self.next_decode_bytes)
|
||
}
|
||
|
||
pub(crate) fn peaks(&self) -> Vec<f32> {
|
||
self.peaks.clone()
|
||
}
|
||
|
||
/// How much of the file these peaks were decoded from — the numerator of
|
||
/// the coverage the UI anchors its rise frontier to.
|
||
pub(crate) fn decoded_bytes(&self) -> u64 {
|
||
self.decoded_bytes
|
||
}
|
||
}
|
||
|
||
/// When the next partial re-decode is due. Early on the file doubles between
|
||
/// decodes — they're cheap and the shape fills in fast. But pure doubling made
|
||
/// the LAST gap span half the transfer (50% → 100%), so the drawn wave sat up
|
||
/// to 2× behind the buffer and the playhead visibly walked off its end
|
||
/// (decoded audio is always ahead of playback, the DRAWING wasn't). Past the
|
||
/// eighth-of-file mark the schedule switches to fixed total/8 steps: the wave
|
||
/// is never more than 12.5% behind the transfer, for at most ~7 extra decodes
|
||
/// spread across the download.
|
||
fn next_progressive_waveform_threshold(decoded_bytes: u64, total_bytes: u64) -> u64 {
|
||
let doubled = decoded_bytes
|
||
.saturating_mul(2)
|
||
.max(decoded_bytes.saturating_add(1));
|
||
let linear_step = decoded_bytes.saturating_add((total_bytes / 8).max(1));
|
||
doubled
|
||
.min(linear_step)
|
||
.min(total_bytes.max(decoded_bytes.saturating_add(1)))
|
||
}
|
||
|
||
pub struct PlayerState {
|
||
pub player: Player,
|
||
/// Active stream downloads: download id → progress shared with the player.
|
||
pub streams: Mutex<HashMap<u64, Arc<StreamProgress>>>,
|
||
/// Cache target for each active stream. This lets a library save adopt an
|
||
/// in-flight stream instead of requesting the same peer/path again.
|
||
pub stream_caches: Mutex<HashMap<u64, PathBuf>>,
|
||
/// Bounded completed-waveform LRU, keyed by file path + bucket count.
|
||
waveforms: Mutex<WaveformLru<Vec<f32>>>,
|
||
/// Last useful partial decode per active/recent stream. A growing file is
|
||
/// decoded again only after its byte length doubles.
|
||
progressive_waveforms: Mutex<WaveformLru<ProgressiveWaveform>>,
|
||
/// Per-key single-flight locks. The owned guard moves into the blocking
|
||
/// decoder so cancellation of the IPC future cannot expose overlapping
|
||
/// work while that decoder is still running.
|
||
waveform_jobs: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
|
||
/// Pre-buffered next-track downloads: download id → cache target.
|
||
/// Their events are swallowed (no UI, no library pipeline); play_stream
|
||
/// attaches to them instead of restarting the transfer.
|
||
pub preloads: Mutex<HashMap<u64, PathBuf>>,
|
||
/// Transfers whose completed cache file should also be organized into the
|
||
/// library. A save can subscribe to a stream/preload already in progress.
|
||
pub deferred_saves: Mutex<HashMap<u64, PathBuf>>,
|
||
/// The last transport state forwarded to the webview.
|
||
///
|
||
/// The audio thread is send-only — it reports by event and answers no
|
||
/// questions — so a webview that RELOADS has no way to learn that audio is
|
||
/// still running. It kept receiving position ticks and drew a moving
|
||
/// scrubber under the words "Nothing playing", because every other part of
|
||
/// the player store had been wiped with the page. This is the answer to
|
||
/// "what is the audio thread doing right now", mirrored here as the events
|
||
/// go past and read back by `player_state` at boot.
|
||
pub last_transport: Mutex<PlaybackSnapshot>,
|
||
}
|
||
|
||
/// What the audio thread is doing, as of its last event. Deliberately only
|
||
/// TRANSPORT: which song this is, and the queue around it, are the frontend's
|
||
/// own knowledge and are restored from its session snapshot.
|
||
#[derive(Serialize, Clone, Default)]
|
||
pub struct PlaybackSnapshot {
|
||
/// "idle" | "buffering" | "playing" | "paused" — the same vocabulary the
|
||
/// frontend's `player.status` uses, so a restore is a direct assignment.
|
||
pub status: String,
|
||
pub position: f64,
|
||
pub duration: Option<f64>,
|
||
}
|
||
|
||
#[derive(Serialize, Clone)]
|
||
#[serde(tag = "kind", rename_all = "kebab-case")]
|
||
enum PlayerUiEvent {
|
||
Buffering,
|
||
Playing,
|
||
Paused,
|
||
Stopped,
|
||
TrackEnded { gapless: bool },
|
||
Position { secs: f64, duration: Option<f64> },
|
||
Error { message: String },
|
||
}
|
||
|
||
/// Create the audio thread and forward its events to the webview.
|
||
pub fn init(app: &tauri::AppHandle) {
|
||
let output_device = crate::config::load(app).audio_output_device;
|
||
let (player, events) = Player::spawn_with_device(output_device);
|
||
app.manage(PlayerState {
|
||
player,
|
||
streams: Mutex::new(HashMap::new()),
|
||
stream_caches: Mutex::new(HashMap::new()),
|
||
waveforms: Mutex::new(WaveformLru::new(WAVEFORM_CACHE_MAX_ENTRIES)),
|
||
progressive_waveforms: Mutex::new(WaveformLru::new(PROGRESSIVE_WAVEFORM_CACHE_MAX_ENTRIES)),
|
||
waveform_jobs: Mutex::new(HashMap::new()),
|
||
preloads: Mutex::new(HashMap::new()),
|
||
deferred_saves: Mutex::new(HashMap::new()),
|
||
last_transport: Mutex::new(PlaybackSnapshot {
|
||
status: "idle".into(),
|
||
..Default::default()
|
||
}),
|
||
});
|
||
|
||
let handle = app.clone();
|
||
std::thread::Builder::new()
|
||
.name("player-events".into())
|
||
.spawn(move || {
|
||
let mut last_position_emit = Instant::now() - UI_POSITION_INTERVAL;
|
||
for ev in events {
|
||
let ui = match ev {
|
||
PlayerEvent::Buffering => {
|
||
last_position_emit = Instant::now() - UI_POSITION_INTERVAL;
|
||
PlayerUiEvent::Buffering
|
||
}
|
||
PlayerEvent::Playing => {
|
||
last_position_emit = Instant::now() - UI_POSITION_INTERVAL;
|
||
PlayerUiEvent::Playing
|
||
}
|
||
PlayerEvent::Paused => PlayerUiEvent::Paused,
|
||
PlayerEvent::Stopped => PlayerUiEvent::Stopped,
|
||
PlayerEvent::TrackEnded { gapless } => {
|
||
last_position_emit = Instant::now() - UI_POSITION_INTERVAL;
|
||
PlayerUiEvent::TrackEnded { gapless }
|
||
}
|
||
PlayerEvent::Position { secs, duration } => {
|
||
let now = Instant::now();
|
||
if now.duration_since(last_position_emit) < UI_POSITION_INTERVAL {
|
||
continue;
|
||
}
|
||
last_position_emit = now;
|
||
PlayerUiEvent::Position { secs, duration }
|
||
}
|
||
PlayerEvent::Error(message) => PlayerUiEvent::Error { message },
|
||
};
|
||
// Mirror the transport before emitting, so a webview that
|
||
// reloads mid-track can ask what it missed. TrackEnded is
|
||
// deliberately not recorded: the frontend owns the queue and
|
||
// decides what follows, and the next Buffering/Stopped says
|
||
// what actually happened.
|
||
// The OS media card rides this same mirror (see smtc.rs). The
|
||
// snapshot IS the transport the UI is about to be told about,
|
||
// so the card cannot drift from the window, and driving it
|
||
// from here costs no IPC at all. Copied out rather than used
|
||
// under the lock: the card's updates are cross-process calls
|
||
// and have no business holding the transport mutex.
|
||
let mut card = None;
|
||
if let Some(state) = handle.try_state::<PlayerState>() {
|
||
if let Ok(mut last) = state.last_transport.lock() {
|
||
match &ui {
|
||
PlayerUiEvent::Buffering => {
|
||
last.status = "buffering".into();
|
||
last.position = 0.0;
|
||
}
|
||
PlayerUiEvent::Playing => last.status = "playing".into(),
|
||
PlayerUiEvent::Paused => last.status = "paused".into(),
|
||
PlayerUiEvent::Stopped | PlayerUiEvent::Error { .. } => {
|
||
*last = PlaybackSnapshot {
|
||
status: "idle".into(),
|
||
..Default::default()
|
||
};
|
||
}
|
||
PlayerUiEvent::Position { secs, duration } => {
|
||
last.position = *secs;
|
||
if duration.is_some() {
|
||
last.duration = *duration;
|
||
}
|
||
}
|
||
PlayerUiEvent::TrackEnded { .. } => {}
|
||
}
|
||
card = Some((last.status.clone(), last.position, last.duration));
|
||
}
|
||
}
|
||
if let Some((status, position, duration)) = card {
|
||
crate::smtc::set_status(&handle, &status);
|
||
crate::smtc::set_timeline(&handle, position, duration);
|
||
// The taskbar preview's middle button wears play or pause
|
||
// off the same truth.
|
||
crate::thumbbar::set_playing(&handle, status == "playing");
|
||
}
|
||
let _ = handle.emit("soulseek:player", ui);
|
||
}
|
||
})
|
||
.expect("spawn player event forwarder");
|
||
}
|
||
|
||
impl PlayerState {
|
||
pub(crate) fn cached_waveform(&self, key: &str) -> Option<Vec<f32>> {
|
||
self.waveforms.lock().unwrap().get(key)
|
||
}
|
||
|
||
pub(crate) fn cache_waveform(&self, key: String, peaks: Vec<f32>) {
|
||
self.waveforms.lock().unwrap().insert(key, peaks);
|
||
}
|
||
|
||
pub(crate) fn remove_cached_waveform(&self, key: &str) {
|
||
self.waveforms.lock().unwrap().remove(key);
|
||
}
|
||
|
||
pub(crate) fn progressive_waveform(&self, key: &str) -> Option<ProgressiveWaveform> {
|
||
self.progressive_waveforms.lock().unwrap().get(key)
|
||
}
|
||
|
||
pub(crate) fn cache_progressive_waveform(&self, key: String, waveform: ProgressiveWaveform) {
|
||
self.progressive_waveforms
|
||
.lock()
|
||
.unwrap()
|
||
.insert(key, waveform);
|
||
}
|
||
|
||
pub(crate) fn remove_progressive_waveform(&self, key: &str) {
|
||
self.progressive_waveforms.lock().unwrap().remove(key);
|
||
}
|
||
|
||
pub(crate) fn waveform_job(&self, key: &str) -> Arc<tokio::sync::Mutex<()>> {
|
||
let mut jobs = self.waveform_jobs.lock().unwrap();
|
||
if jobs.len() >= WAVEFORM_JOB_MAX_IDLE_ENTRIES {
|
||
// An entry with only the map's Arc is idle and safe to forget.
|
||
// Active/waiting jobs may temporarily exceed the soft bound, but
|
||
// they must keep their shared lock until their decoder exits.
|
||
let idle: Vec<String> = jobs
|
||
.iter()
|
||
.filter(|(_, job)| Arc::strong_count(job) == 1)
|
||
.map(|(key, _)| key.clone())
|
||
.collect();
|
||
for idle_key in idle {
|
||
if jobs.len() < WAVEFORM_JOB_MAX_IDLE_ENTRIES {
|
||
break;
|
||
}
|
||
jobs.remove(&idle_key);
|
||
}
|
||
}
|
||
jobs.entry(key.to_string())
|
||
.or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
|
||
.clone()
|
||
}
|
||
|
||
pub(crate) fn release_waveform_job(&self, key: &str, job: &Arc<tokio::sync::Mutex<()>>) {
|
||
let mut jobs = self.waveform_jobs.lock().unwrap();
|
||
if jobs
|
||
.get(key)
|
||
.is_some_and(|current| Arc::ptr_eq(current, job) && Arc::strong_count(current) == 2)
|
||
{
|
||
jobs.remove(key);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Stable cache filename for a remote file: short hash + original name.
|
||
pub fn cache_path(app: &tauri::AppHandle, username: &str, remote_path: &str) -> Option<PathBuf> {
|
||
let dir = crate::paths::app_data(app).ok()?.join("stream-cache");
|
||
std::fs::create_dir_all(&dir).ok()?;
|
||
|
||
let mut hash: u64 = 0xcbf29ce484222325;
|
||
for b in username.bytes().chain(remote_path.bytes()) {
|
||
hash ^= b as u64;
|
||
hash = hash.wrapping_mul(0x100000001b3);
|
||
}
|
||
let name = remote_path.rsplit('\\').next().unwrap_or(remote_path);
|
||
let safe: String = name
|
||
.chars()
|
||
.map(|c| match c {
|
||
'<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*' => '_',
|
||
c if (c as u32) < 0x20 => '_',
|
||
c => c,
|
||
})
|
||
.collect();
|
||
Some(dir.join(format!("{hash:016x} {safe}")))
|
||
}
|
||
|
||
/// The in-flight download target for a cache file: `<name>.<ext>.partial`.
|
||
pub fn partial_path(cache: &std::path::Path) -> PathBuf {
|
||
cache.with_extension(format!(
|
||
"{}partial",
|
||
cache
|
||
.extension()
|
||
.map(|e| format!("{}.", e.to_string_lossy()))
|
||
.unwrap_or_default()
|
||
))
|
||
}
|
||
|
||
/// Enforce the configured streamed-track retention on the cache: drop files
|
||
/// not played within the retention window, then oldest-first beyond the size
|
||
/// cap. The playing track's file is held open by the OS, so its delete fails
|
||
/// and it is simply kept.
|
||
pub fn prune_cache(app: &tauri::AppHandle) {
|
||
// Completion/settings paths can request pruning concurrently. Serialize
|
||
// their directory snapshots so a stale running total cannot over-delete.
|
||
let _prune_guard = CACHE_PRUNE_LOCK
|
||
.lock()
|
||
.unwrap_or_else(|poisoned| poisoned.into_inner());
|
||
let cfg = crate::config::load(app);
|
||
let max_bytes = crate::config::stream_cache_max_mb(&cfg) * 1024 * 1024; // 0 = unlimited
|
||
let retention = std::time::Duration::from_secs(
|
||
crate::config::stream_retention_days(&cfg) as u64 * 24 * 60 * 60,
|
||
);
|
||
// In keep-forever mode streams land in the library, so the cache is only
|
||
// a transfer buffer — age expiry off, size cap still applies.
|
||
let age_prune = !crate::config::stream_keep_forever(&cfg);
|
||
|
||
let Ok(dir) = crate::paths::app_data(app).map(|d| d.join("stream-cache")) else {
|
||
return;
|
||
};
|
||
let Ok(entries) = std::fs::read_dir(&dir) else {
|
||
return;
|
||
};
|
||
|
||
let mut files: Vec<(std::time::SystemTime, u64, PathBuf)> = entries
|
||
.flatten()
|
||
.filter_map(|e| {
|
||
let path = e.path();
|
||
if path
|
||
.extension()
|
||
.is_some_and(|ext| ext.to_string_lossy().eq_ignore_ascii_case("partial"))
|
||
{
|
||
// Active downloads are transfer buffers, not evictable cache
|
||
// entries. Deleting one can make its final rename fail.
|
||
return None;
|
||
}
|
||
let md = e.metadata().ok()?;
|
||
if !md.is_file() {
|
||
return None;
|
||
}
|
||
Some((md.modified().ok()?, md.len(), path))
|
||
})
|
||
.collect();
|
||
files.sort_by_key(|(t, ..)| *t);
|
||
|
||
let now = std::time::SystemTime::now();
|
||
let mut total: u64 = files.iter().map(|(_, len, _)| len).sum();
|
||
let mut kept = Vec::new();
|
||
for (modified, len, path) in files {
|
||
let expired = age_prune
|
||
&& now
|
||
.duration_since(modified)
|
||
.map(|age| age > retention)
|
||
.unwrap_or(false);
|
||
if expired && std::fs::remove_file(&path).is_ok() {
|
||
total -= len;
|
||
continue;
|
||
}
|
||
kept.push((len, path));
|
||
}
|
||
|
||
if max_bytes == 0 {
|
||
return;
|
||
}
|
||
for (len, path) in kept {
|
||
if total <= max_bytes {
|
||
break;
|
||
}
|
||
if std::fs::remove_file(&path).is_ok() {
|
||
total -= len;
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Mark a cache file as freshly used so retention counts from the last play,
|
||
/// not the first stream.
|
||
pub fn touch(path: &std::path::Path) {
|
||
if let Ok(f) = std::fs::File::options().write(true).open(path) {
|
||
let _ = f.set_modified(std::time::SystemTime::now());
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn waveform_lru_evicts_oldest_and_refreshes_hits() {
|
||
let mut cache = WaveformLru::new(2);
|
||
cache.insert("a".into(), vec![1.0]);
|
||
cache.insert("b".into(), vec![2.0]);
|
||
assert_eq!(cache.get("a"), Some(vec![1.0]));
|
||
|
||
cache.insert("c".into(), vec![3.0]);
|
||
assert_eq!(cache.get("b"), None);
|
||
assert_eq!(cache.get("a"), Some(vec![1.0]));
|
||
assert_eq!(cache.get("c"), Some(vec![3.0]));
|
||
assert_eq!(cache.entries.len(), 2);
|
||
}
|
||
|
||
#[test]
|
||
fn progressive_waveform_waits_for_doubling_but_restarts_after_shrink() {
|
||
let waveform = ProgressiveWaveform::new(10, 100, vec![1.0]);
|
||
assert!(!waveform.should_recompute(10));
|
||
assert!(!waveform.should_recompute(19));
|
||
assert!(waveform.should_recompute(20));
|
||
assert!(waveform.should_recompute(9));
|
||
}
|
||
|
||
#[test]
|
||
fn progressive_thresholds_bound_lag_and_total_decode_cost() {
|
||
let total: u64 = 256 * 1024 * 1024;
|
||
let mut decoded: u64 = 1024 * 1024;
|
||
let mut cumulative = 0u64;
|
||
let mut steps = 0;
|
||
while decoded < total {
|
||
cumulative += decoded;
|
||
let next = next_progressive_waveform_threshold(decoded, total);
|
||
assert!(next > decoded, "thresholds must advance");
|
||
// The drawn wave may never fall more than an eighth of the file
|
||
// behind the transfer — the doubling-only schedule's half-file
|
||
// final gap is the bug this replaced (playhead walking off the
|
||
// stale wave's end mid-song).
|
||
assert!(
|
||
next - decoded <= total / 8,
|
||
"gap {} exceeds total/8",
|
||
next - decoded
|
||
);
|
||
decoded = next;
|
||
steps += 1;
|
||
}
|
||
assert_eq!(decoded, total);
|
||
// Early doubling + eighth-of-file steps: the whole progressive series
|
||
// costs a small constant number of full-file decodes, spread across
|
||
// the download.
|
||
assert!(steps <= 16, "too many recomputes: {steps}");
|
||
assert!(cumulative <= total * 4, "cumulative redecode too large");
|
||
}
|
||
}
|