clusterfuck
/
crabidy
4
1
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects 1 Releases 5 Wiki Activity

Playback: exponential read-ahead when resolving a queue 

The resolve forwarder walked the paths one at a time. That starts the
first track quickly, but refills the rest only as fast as a single
provider resolve -- so when enumeration is slow and the leading tracks
are very short or skipped, playback drains the resolved queue faster
than it fills and stalls into silence.

Resolve the paths concurrently under a read-ahead window that starts at
1 and doubles after each path completes (1, 2, 4, 8, 16, then steady
16). The first path still resolves alone, so time-to-first-track is
unchanged; the window then grows geometrically, so the resolved queue
runs exponentially ahead of linear playback and a short/skipped head
cannot catch it. The cap 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, and the per-op cursor and "first chunk starts
the player" semantics are untouched. Cancellation drops the in-flight
receivers, stopping every concurrent resolve at once.

This is a read-ahead over paths; a single collection is still enumerated
by its provider's page streaming, so the win is for multi-item
selections. Documented as architecture/progressive-queueing.md D8.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Test User 2026-07-23 19:46:31 +02:00
parent bc3d2e099e
commit ef1e56e1f4
2 changed files with 90 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
architecture/progressive-queueing.md
View File

@ -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

78
crabidy-server/src/playback.rs
View File

@ -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<String>) {
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<Vec<Track>>)> = 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.