diff --git a/architecture/progressive-queueing.md b/architecture/progressive-queueing.md index a16e79f..4408408 100644 --- a/architecture/progressive-queueing.md +++ b/architecture/progressive-queueing.md @@ -130,7 +130,8 @@ Options considered: Other provider commands keep flowing while (possibly several) resolves run. - **Playback side**: each queue op registers a *pending op* (id from an `AtomicU64`, kind, insertion cursor) and spawns a forwarder task that - drives `ProviderCommand::ResolveTracks` per path (sequentially, preserving + drives `ProviderCommand::ResolveTracks` per path (concurrently, under the + exponential read-ahead window of D8, but forwarding chunks in strict multi-path order) and forwards each chunk to the playback channel as `PlaybackCommand::ApplyResolvedChunk { op_id, tracks }`, followed by `ResolveFinished { op_id }`. Queue state is only ever mutated inside the @@ -193,6 +194,41 @@ appended at render time only and never enters `self.list`, so selection, removal, and `get_size` cannot reach it — no new input states, nothing to misclick. +### D8 — Exponential read-ahead across paths + +The forwarder of D4 first resolved paths one at a time. That starts the first +track quickly (good) but fills the rest only as fast as one provider resolve +at a time. When enumeration is slow (per-path Tidal/YouTube/fyyd round trips) +and the leading tracks are very short or `is_skipped`, playback drains the +resolved queue faster than a sequential resolver refills it — and stalls into +silence, the exact thing progressive queueing exists to avoid. + +Options considered: + +1. **Sequential per path** (original): simplest, but a short/skipped head can + outrun a slow resolver. +2. **Resolve every path at once**: fills fastest, but a large marked + selection fires an unbounded burst of concurrent provider calls (rate + limits, memory) the instant playback starts — wasteful when the user skips + away after two tracks. +3. **Exponential read-ahead window**: a concurrency window that starts at 1 + and doubles (1, 2, 4, 8, 16, then steady 16) after each path completes. + +**Decision: (3).** The first path resolves alone, so time-to-first-track is +unchanged from (1); the window then grows geometrically, so the resolved +queue runs exponentially ahead of linear playback and a short/skipped head +cannot catch it after the first couple of tracks. The cap (16) bounds +concurrent provider load. Chunks are still forwarded in strict path order — +the forwarder fully drains the oldest in-flight resolve before the next, so +concurrency never reorders the queue (D5's per-op cursor and "first chunk +starts the player" are untouched). Cancellation (D6) drops the in-flight +receivers, stopping every concurrent resolve at once. + +This is a read-ahead over *paths*. A single collection path (one album or +playlist) is still enumerated by its provider's `resolve_tracks_into` — the +per-page streaming of D3 is that path's read-ahead — so the window is the win +for multi-item selections; single-collection latency stays a provider concern. + ## Flows ```d2 diff --git a/crabidy-server/src/playback.rs b/crabidy-server/src/playback.rs index 5f67186..2844c82 100644 --- a/crabidy-server/src/playback.rs +++ b/crabidy-server/src/playback.rs @@ -9,7 +9,7 @@ use crabidy_core::proto::crabidy::{ Queue as ProtoQueue, QueueTrack, Track, TrackPosition, }; use crabidy_core::ProviderError; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use tracing::{debug, debug_span, error, info, instrument, trace, warn, Instrument}; @@ -441,17 +441,25 @@ impl Playback { /// Registers a pending resolve operation and spawns its forwarder task. /// - /// The forwarder resolves `paths` one after the other (preserving the - /// request's path order): for each path it sends - /// `ProviderCommand::ResolveTracks` with a fresh bounded chunk channel - /// and forwards every chunk to the playback loop as - /// `PlaybackCommand::ApplyResolvedChunk`; after the last path it sends - /// `ResolveFinished`. When the op's cancellation flag is set, the - /// forwarder drops the chunk receiver instead — the provider's next - /// send fails and the fetch stops. Queue state is never touched here: - /// mutations happen only when the loop processes the forwarded - /// commands. An immediate `Queue` broadcast (unchanged tracks, - /// `resolving = true`) gives clients instant feedback. + /// The forwarder resolves `paths` with an *exponential read-ahead*: a + /// concurrency window that starts at 1 and doubles (1, 2, 4, 8, 16, then + /// steady 16) after each path completes. It keeps up to `window` paths + /// resolving at once (each via `ProviderCommand::ResolveTracks` on its own + /// bounded chunk channel) but forwards chunks to the playback loop as + /// `PlaybackCommand::ApplyResolvedChunk` in strict path order, so the queue + /// keeps the requested order and the first chunk still lands — and starts + /// playback — as fast as a single resolve. The first track therefore plays + /// as soon as one resolve yields it, while the window fills the queue far + /// ahead of playback: a very short or skipped leading track cannot outrun + /// resolution and cause a silent stall (architecture/progressive-queueing.md D8). + /// + /// After the last path it sends `ResolveFinished`. When the op's + /// cancellation flag is set, the forwarder stops launching resolves and + /// drops its chunk receivers — each provider's next send fails and the + /// fetch stops. Queue state is never touched here: mutations happen only + /// when the loop processes the forwarded commands. An immediate `Queue` + /// broadcast (unchanged tracks, `resolving = true`) gives clients instant + /// feedback. fn start_resolve(&self, kind: ResolveKind, paths: Vec) { let op = PendingResolve::new(kind); let cancelled = op.cancel_flag(); @@ -477,26 +485,45 @@ impl Playback { let playback_tx = self.playback_tx.clone(); tokio::spawn( async move { - for path in &paths { + // Exponential read-ahead window: 1, 2, 4, 8, 16, then steady. + const MAX_WINDOW: usize = 16; + let mut window = 1usize; + let mut next = 0usize; + let mut inflight: VecDeque<(String, flume::Receiver>)> = VecDeque::new(); + 'resolve: loop { if cancelled.load(Ordering::Relaxed) { break; } - let (chunk_tx, chunk_rx) = flume::bounded(4); - let message = ProviderMessage::new(ProviderCommand::ResolveTracks { - path: path.clone(), - chunk_tx, - }); - if provider_tx.send_async(message).await.is_err() { - error!("provider channel closed"); - break; + // Top the window up with fresh concurrent resolves. Each + // resolves in the background into its own bounded channel + // while we drain the oldest one below. + while inflight.len() < window && next < paths.len() { + let path = paths[next].clone(); + next += 1; + let (chunk_tx, chunk_rx) = flume::bounded(4); + let message = ProviderMessage::new(ProviderCommand::ResolveTracks { + path: path.clone(), + chunk_tx, + }); + if provider_tx.send_async(message).await.is_err() { + error!("provider channel closed"); + break 'resolve; + } + inflight.push_back((path, chunk_rx)); } + // Drain the oldest (lowest path index) resolve to + // completion before the next, so chunks reach the loop in + // request order even though resolution ran concurrently. + let Some((path, chunk_rx)) = inflight.pop_front() else { + break; + }; let mut forwarded = 0usize; while let Ok(tracks) = chunk_rx.recv_async().await { - // On cancellation this loop exits and drops - // chunk_rx; the provider's next send fails and the - // fetch stops. + // On cancellation this drops chunk_rx (and, on the next + // iteration, the rest of `inflight`); each provider's + // next send then fails and its fetch stops. if cancelled.load(Ordering::Relaxed) { - break; + break 'resolve; } forwarded += tracks.len(); let apply = PlaybackCommand::ApplyResolvedChunk { op_id, tracks }; @@ -512,6 +539,7 @@ impl Playback { if forwarded == 0 && !cancelled.load(Ordering::Relaxed) { warn!(path, "path resolved to no playable tracks"); } + window = (window * 2).min(MAX_WINDOW); } // Always reported — also for cancelled or empty ops — so // the pending map can never leak a stuck resolving flag.