From 71f5c4a1c5a456cc307047625842d9b3a54b3423 Mon Sep 17 00:00:00 2001 From: nsaintot Date: Thu, 12 Nov 2026 00:07:47 +0100 Subject: [PATCH 1/2] Redesign the window - New look: a drop zone with an animated globe, glass panels, Instrument Sans, a status bar with the server switch, and an empty-state placeholder for the Separated list. - Own title bar on macOS and Windows, with the name and version centred; Windows draws its own window buttons and resize grips. Linux unchanged. - The window opens centred and remembers its position and size. - Animations run at the same speed at any refresh rate. - Stemd is capitalised in the app bundle, installer, shortcuts and docs. - The server switch turns the API off (503) and cancels client jobs; drops carry on. A running or queued drop can be stopped from its row. - A second launch on the same cache folder stops at once, whatever its port. Needs Rust 1.89 for the file lock. - Windows: the MLX GPU thread no longer exits, which froze the window for about 3 s while Windows reported the crash. - Drops: a full queue fails the job instead of leaving it queued forever; cached stems survive a failed copy; cache hits skip the queue order. - The API checks disk space after the cache lookup, so cache hits are served on a low disk. - mDNS registration runs off the UI thread. --- .github/workflows/cd.yml | 6 +- Cargo.lock | 63 ++ Cargo.toml | 5 +- README.md | 2 +- crates/stemd-cli/src/discovery.rs | 2 +- crates/stemd-core/src/stemfmt.rs | 6 +- crates/stemd-mlx/src/device.rs | 20 +- crates/stemd-server/Cargo.toml | 10 +- crates/stemd-server/src/api.rs | 40 +- crates/stemd-server/src/cache.rs | 143 ++- crates/stemd-server/src/disk.rs | 180 ++++ crates/stemd-server/src/drops.rs | 316 +++++- crates/stemd-server/src/main.rs | 12 +- crates/stemd-server/src/models/download.rs | 13 +- crates/stemd-server/src/models/preset.rs | 8 +- crates/stemd-server/src/pieces.rs | 27 +- crates/stemd-server/src/queue.rs | 141 ++- crates/stemd-server/src/serve.rs | 14 +- crates/stemd-server/src/serving.rs | 122 +++ crates/stemd-server/src/settings.rs | 18 +- crates/stemd-server/src/shutdown.rs | 4 +- crates/stemd-server/src/startup.rs | 111 ++- crates/stemd-server/src/switch.rs | 1 + crates/stemd-server/src/testkit.rs | 4 +- crates/stemd-server/src/ui/blur.rs | 424 ++++++++ crates/stemd-server/src/ui/dropzone.rs | 739 ++++++++++---- crates/stemd-server/src/ui/globe.rs | 771 ++++++++++++++ crates/stemd-server/src/ui/header.rs | 826 +++++++-------- crates/stemd-server/src/ui/icons.rs | 131 +++ crates/stemd-server/src/ui/logs.rs | 209 ++-- crates/stemd-server/src/ui/mod.rs | 676 ++++++++++--- crates/stemd-server/src/ui/probe.rs | 504 +++++----- crates/stemd-server/src/ui/quit.rs | 61 +- crates/stemd-server/src/ui/recents.rs | 426 ++++---- crates/stemd-server/src/ui/section.rs | 126 +-- crates/stemd-server/src/ui/starting.rs | 472 +++++++-- crates/stemd-server/src/ui/status.rs | 323 ++++-- crates/stemd-server/src/ui/theme.rs | 1051 ++++++++++++++++---- crates/stemd-server/src/ui/titlebar.rs | 253 +++++ crates/stemd-server/src/ui/widgets.rs | 267 +++++ docs/building.md | 14 +- packaging/linux/THIRD-PARTY | 4 + packaging/linux/stemd.desktop | 2 +- resources/fonts/InstrumentSans-OFL.txt | 93 ++ resources/fonts/InstrumentSans.ttf | Bin 0 -> 194336 bytes scripts/bundle-app.sh | 10 +- scripts/bundle.sh | 8 +- scripts/installer-windows.nsi | 32 +- scripts/package-windows.ps1 | 2 +- 49 files changed, 6768 insertions(+), 1924 deletions(-) create mode 100644 crates/stemd-server/src/disk.rs create mode 100644 crates/stemd-server/src/serving.rs create mode 100644 crates/stemd-server/src/ui/blur.rs create mode 100644 crates/stemd-server/src/ui/globe.rs create mode 100644 crates/stemd-server/src/ui/icons.rs create mode 100644 crates/stemd-server/src/ui/titlebar.rs create mode 100644 crates/stemd-server/src/ui/widgets.rs create mode 100644 resources/fonts/InstrumentSans-OFL.txt create mode 100644 resources/fonts/InstrumentSans.ttf diff --git a/.github/workflows/cd.yml b/.github/workflows/cd.yml index 2bfee3a..d3fdde9 100644 --- a/.github/workflows/cd.yml +++ b/.github/workflows/cd.yml @@ -215,7 +215,7 @@ jobs: if: env.SIGNED != 'true' run: scripts/bundle-app.sh - name: Start check - run: dist/stemd.app/Contents/MacOS/stemd --version + run: dist/Stemd.app/Contents/MacOS/stemd --version - name: Remove credentials if: always() run: | @@ -271,14 +271,14 @@ jobs: Test-Version dist\stemd-windows\stemd-server.exe $setup = Get-Item dist\stemd-*-setup.exe Start-Process $setup.FullName -ArgumentList '/S' -Wait - $installed = Join-Path $env:LOCALAPPDATA 'Programs\stemd' + $installed = Join-Path $env:LOCALAPPDATA 'Programs\Stemd' Test-Version (Join-Path $installed 'stemd-server.exe') Test-Version (Join-Path $installed 'stemd-cli.exe') # The runner has no NVIDIA driver. - name: CPU separation timeout-minutes: 30 run: | - $installed = Join-Path $env:LOCALAPPDATA 'Programs\stemd' + $installed = Join-Path $env:LOCALAPPDATA 'Programs\Stemd' .github/scripts/cpu-separation.ps1 -Server (Join-Path $installed 'stemd-server.exe') -Cli (Join-Path $installed 'stemd-cli.exe') - uses: actions/upload-artifact@v4 with: diff --git a/Cargo.lock b/Cargo.lock index cad6290..dcd5983 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,6 +8,8 @@ version = "0.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3b7f7f85a7e5f68090000ed7622545829afd484d210358702ae4cb97dd0c320" dependencies = [ + "enumn", + "serde", "uuid", ] @@ -26,6 +28,7 @@ dependencies = [ "cfg-if", "getrandom 0.3.4", "once_cell", + "serde", "version_check", "zerocopy", ] @@ -312,6 +315,9 @@ name = "bitflags" version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +dependencies = [ + "serde_core", +] [[package]] name = "block-buffer" @@ -859,6 +865,7 @@ checksum = "6758be723a3f298bbfda4db75748bc2ba0abafe096b6383c7c32da264764fbc3" dependencies = [ "bytemuck", "emath", + "serde", ] [[package]] @@ -877,6 +884,7 @@ dependencies = [ "glow", "glutin", "glutin-winit", + "home", "image", "js-sys", "log", @@ -887,6 +895,8 @@ dependencies = [ "percent-encoding", "profiling", "raw-window-handle", + "ron", + "serde", "static_assertions", "wasm-bindgen", "wasm-bindgen-futures", @@ -911,6 +921,8 @@ dependencies = [ "log", "nohash-hasher", "profiling", + "ron", + "serde", "smallvec", "unicode-segmentation", ] @@ -950,6 +962,7 @@ dependencies = [ "objc2-ui-kit 0.3.2", "profiling", "raw-window-handle", + "serde", "smithay-clipboard", "web-time", "webbrowser", @@ -984,6 +997,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd4ec073c9898516584d8c6cfdcee95b530b3d941cd5031ef4050aa36812308b" dependencies = [ "bytemuck", + "serde", ] [[package]] @@ -995,6 +1009,17 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "enumn" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f9ed6b3789237c8a0c1c505af1c7eb2c560df6186f01b098c3a1064ea532f38" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "epaint" version = "0.35.0" @@ -1013,6 +1038,7 @@ dependencies = [ "parking_lot", "profiling", "self_cell", + "serde", "skrifa", "smallvec", "unicode-general-category", @@ -1153,6 +1179,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b38ad915f6dadd993ced50848a8291a543bd41ca62bc10740d5e64e2ab4cfd7" dependencies = [ "bytemuck", + "serde", ] [[package]] @@ -1483,6 +1510,15 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dfa686283ad6dd069f105e5ab091b04c62850d3e4cf5d67debad1933f55023df" +[[package]] +name = "home" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "hound" version = "3.5.1" @@ -2904,6 +2940,19 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "ron" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97b3fe97c18ff24299ea56f802c5693eb34266fb0bc8317f4e101debc8a3ceaf" +dependencies = [ + "bitflags 2.13.1", + "serde", + "serde_derive", + "typeid", + "unicode-ident", +] + [[package]] name = "rubato" version = "5.0.0" @@ -3246,6 +3295,9 @@ name = "smallvec" version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +dependencies = [ + "serde", +] [[package]] name = "smithay-client-toolkit" @@ -3422,6 +3474,7 @@ dependencies = [ "dirs", "eframe", "if-addrs", + "libc", "parking_lot", "serde", "serde_json", @@ -3435,6 +3488,7 @@ dependencies = [ "tracing", "tracing-subscriber", "ureq", + "windows-sys 0.61.2", "winit", "winresource", "zip", @@ -3981,6 +4035,12 @@ dependencies = [ "rustc-hash 2.1.3", ] +[[package]] +name = "typeid" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c" + [[package]] name = "typenum" version = "1.20.1" @@ -4081,6 +4141,9 @@ name = "uuid" version = "1.24.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" +dependencies = [ + "serde_core", +] [[package]] name = "valuable" diff --git a/Cargo.toml b/Cargo.toml index 219b2f2..199b0a8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,9 +15,8 @@ members = [ [workspace.package] version = "0.1.4" edition = "2024" -# Stated so cargo says "requires rustc 1.85" rather than failing to parse the -# edition. It is the edition's floor, not a tested one. -rust-version = "1.85" +# 1.89 for `File::try_lock`, which keeps a second server off a cache in use. +rust-version = "1.89" license = "MIT OR Apache-2.0" repository = "https://github.com/nsaintot/stemd" readme = "README.md" diff --git a/README.md b/README.md index 1141dd8..427b32b 100644 --- a/README.md +++ b/README.md @@ -126,7 +126,7 @@ by the audio and by everything else that changes the output. ## Build from source -Needs `cmake`, Rust 1.85 or later, and a recursive clone. Per-platform +Needs `cmake`, Rust 1.89 or later, and a recursive clone. Per-platform toolchains, packaging and every flag are in [docs/building.md](docs/building.md). diff --git a/crates/stemd-cli/src/discovery.rs b/crates/stemd-cli/src/discovery.rs index 80500ce..4115df3 100644 --- a/crates/stemd-cli/src/discovery.rs +++ b/crates/stemd-cli/src/discovery.rs @@ -30,7 +30,7 @@ pub fn discover(timeout: Duration) -> Result { let _ = daemon.shutdown(); bail!( "no stemd server found on the network after {}s — is it running? \ - start it with `open dist/stemd.app`, or pass --host host:port", + start it with `open dist/Stemd.app`, or pass --host host:port", timeout.as_secs() ) } diff --git a/crates/stemd-core/src/stemfmt.rs b/crates/stemd-core/src/stemfmt.rs index 4c2df75..d47e73b 100644 --- a/crates/stemd-core/src/stemfmt.rs +++ b/crates/stemd-core/src/stemfmt.rs @@ -131,9 +131,9 @@ impl StemFormat { match self { Self::Flac => "FLAC", Self::Wav => "WAV", - Self::Mp3 => "MP3 320", - Self::Pcm16 => "PCM 16", - Self::Pcm32 => "PCM 32", + Self::Mp3 => "MP3", + Self::Pcm16 => "PCM 16-bit", + Self::Pcm32 => "PCM 32-bit float", } } } diff --git a/crates/stemd-mlx/src/device.rs b/crates/stemd-mlx/src/device.rs index 844d23f..4819d50 100644 --- a/crates/stemd-mlx/src/device.rs +++ b/crates/stemd-mlx/src/device.rs @@ -74,7 +74,7 @@ impl Accelerator { if !gpu { return (Self::Cpu, false); } - if !gpu_computes() { + if !gpu_computes_on_kept_thread() { return (Self::demote(), true); } let which = if cfg!(target_vendor = "apple") { @@ -118,6 +118,24 @@ impl std::fmt::Display for Accelerator { } } +/// [`gpu_computes`] on a thread that never exits. +/// +/// MLX keeps CUDA state per thread, and on Windows its destructor throws when the +/// thread exits. Windows reports that as a crash and freezes the whole process for +/// about 3 s while it writes the report. +fn gpu_computes_on_kept_thread() -> bool { + let (tx, rx) = std::sync::mpsc::channel(); + let spawned = std::thread::Builder::new() + .name("stemd-gpu-probe".into()) + .spawn(move || { + let _ = tx.send(gpu_computes()); + loop { + std::thread::park(); + } + }); + spawned.is_ok() && rx.recv().unwrap_or(false) +} + /// Whether the GPU MLX offered returns right answers. /// /// One operation of each kind the models use, on inputs small enough that the diff --git a/crates/stemd-server/Cargo.toml b/crates/stemd-server/Cargo.toml index f4da714..dccca20 100644 --- a/crates/stemd-server/Cargo.toml +++ b/crates/stemd-server/Cargo.toml @@ -22,7 +22,8 @@ clap = { version = "4.6.6", features = ["derive"] } tokio-util = { version = "0.7.19", features = ["io"] } parking_lot = "0.12.5" thiserror.workspace = true -eframe = { version = "0.35.0", default-features = false, features = ["glow", "default_fonts"] } +# `persistence` saves the window's position and size between launches. +eframe = { version = "0.35.0", default-features = false, features = ["glow", "default_fonts", "persistence"] } if-addrs = "0.15.0" # First-run model fetch: HTTP plus a pinned digest. ureq = "3.4.0" @@ -51,6 +52,12 @@ winit = { version = "0.30.13", default-features = false, features = ["x11"] } # download only there. See `cuda`. [target.'cfg(windows)'.dependencies] zip = { version = "2", default-features = false, features = ["deflate"] } +# Free disk space, see `disk.rs`. +windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } + +[target.'cfg(unix)'.dependencies] +# statvfs and sysctl, see `disk.rs`. +libc = "0.2" # `default-features = false` above is what keeps wgpu and accesskit out, but it # also drops `x11` and `wayland`, and those are not extras — winit has no Linux @@ -62,6 +69,7 @@ zip = { version = "2", default-features = false, features = ["deflate"] } eframe = { version = "0.35.0", default-features = false, features = [ "glow", "default_fonts", + "persistence", "x11", "wayland", ] } diff --git a/crates/stemd-server/src/api.rs b/crates/stemd-server/src/api.rs index 3abe2d3..2eaa5f0 100644 --- a/crates/stemd-server/src/api.rs +++ b/crates/stemd-server/src/api.rs @@ -31,7 +31,6 @@ use stemd_core::{Audio, DspMode, OutputRate, Parts, PcmFormat, Stage, resample}; use tokio_util::io::ReaderStream; use crate::cache::{Cache, Ingredients, Output}; -use crate::discovery::Advertiser; use crate::jobs::JobStore; use crate::logbuf::LogBuffer; use crate::models::Preset; @@ -53,9 +52,9 @@ pub struct AppState { /// `/v1/health` so a client can refuse locally instead of uploading a track /// it will be told to take back. pub max_track_secs: f64, - /// Present unless --no-mdns. Shared so every exit path can withdraw the - /// advertisement rather than relying on Drop, which macOS skips on Cmd-Q. - pub advertiser: Option>, + /// Whether the API answers, and its mDNS advertisement. Shared so every + /// exit path can withdraw the advertisement: macOS skips Drop on Cmd-Q. + pub serving: crate::serving::Serving, } pub fn router(state: Arc) -> Router { @@ -67,9 +66,30 @@ pub fn router(state: Arc) -> Router { .route("/v1/jobs/{id}/stems/{name}", get(get_stem)) .route("/v1/logs", get(get_logs)) .merge(crate::session_api::routes()) + .route_layer(axum::middleware::from_fn_with_state( + Arc::clone(&state), + refuse_when_off, + )) .with_state(state) } +/// 503 for every request while the server is turned off in the window. +async fn refuse_when_off( + State(state): State>, + request: axum::extract::Request, + next: axum::middleware::Next, +) -> Response { + if state.serving.is_on() { + next.run(request).await + } else { + ApiError::new( + StatusCode::SERVICE_UNAVAILABLE, + "the server is turned off in the Stemd window", + ) + .into_response() + } +} + #[derive(Debug, Serialize)] struct Health { version: &'static str, @@ -518,6 +538,16 @@ async fn create_job( return Ok((StatusCode::OK, Json(job.view())).into_response()); } + // Stems are written to the cache before they are served, and a full disk + // also leaves the system no room to swap. + if let Err(full) = crate::disk::ensure_room(state.cache.root()) { + job.fail(full.to_string()); + return Err(ApiError::new( + StatusCode::INSUFFICIENT_STORAGE, + full.to_string(), + )); + } + let mix = decode_upload(body, submission, &job).await?; enqueue(&state, &job, mix, submission)?; Ok((StatusCode::ACCEPTED, Json(job.view())).into_response()) @@ -538,6 +568,7 @@ fn enqueue( mix, output: submission.output, model_id: state.switcher.published_identity(), + from_api: true, }) .map_err(|full| { job.fail(format!("{full}")); @@ -626,6 +657,7 @@ async fn get_stem( // Only once the bytes are provably readable. Marking on request would let a // failed fetch count as consumption and keep an entry nobody can use. + entry.hand_out(); entry.mark_fetched(&name); Ok(Response::builder() diff --git a/crates/stemd-server/src/cache.rs b/crates/stemd-server/src/cache.rs index 9b65981..871449b 100644 --- a/crates/stemd-server/src/cache.rs +++ b/crates/stemd-server/src/cache.rs @@ -168,6 +168,9 @@ pub struct Entry { /// Set when the cache deletes this entry's directory, so a job holding the /// last reference can tell that its stems are gone. evicted: AtomicBool, + /// Set once [`Cache::get`] or a stem download gives the entry to a caller, + /// so a drop that made it does not delete it from under them. + handed_out: AtomicBool, state: Mutex, } @@ -192,6 +195,11 @@ impl Entry { self.evicted.load(Ordering::SeqCst) } + /// Record that a client outside the window has the entry. See `handed_out`. + pub fn hand_out(&self) { + self.handed_out.store(true, Ordering::SeqCst); + } + /// Record that `name` was pulled, and restart the idle clock. /// /// Touching here is what stops rule 1 from collecting an entry between a @@ -258,6 +266,8 @@ pub struct Cache { /// Makes staging directory names unique. The worker is serialised today, so /// two publishes cannot overlap, this keeps that from being load-bearing. seq: AtomicU64, + /// Whether the cap shrinks to keep [`crate::disk::FLOOR`] free. + disk_aware: AtomicBool, } /// Where entries live when no directory is configured. @@ -284,6 +294,40 @@ pub fn default_dir() -> Result { }) } +/// The cache root `args` names, or [`default_dir`]. +pub fn root(args: &crate::cli::Args) -> Result { + match args.cache_dir.clone() { + Some(dir) => Ok(dir), + None => default_dir(), + } +} + +/// The lock file in the cache root, which the startup clear keeps. +const LOCK_FILE: &str = "stemd.lock"; + +/// Lock `root` for this process, or fail if another stemd holds it, whatever +/// port that one serves on. The system releases the lock when the process ends. +pub fn lock(root: &Path) -> Result { + std::fs::create_dir_all(root).with_context(|| format!("creating {}", root.display()))?; + let path = root.join(LOCK_FILE); + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&path) + .with_context(|| format!("opening {}", path.display()))?; + match file.try_lock() { + Ok(()) => Ok(file), + Err(std::fs::TryLockError::WouldBlock) => anyhow::bail!( + "another stemd is already running with its cache in {}", + root.display() + ), + Err(std::fs::TryLockError::Error(err)) => { + Err(err).with_context(|| format!("locking {}", path.display())) + } + } +} + impl Cache { /// Prepare `root`, discarding whatever a previous run left behind. /// @@ -300,6 +344,7 @@ impl Cache { max_bytes, unfetched_ttl, seq: AtomicU64::new(0), + disk_aware: AtomicBool::new(true), })) } @@ -316,6 +361,7 @@ impl Cache { let entries = self.entries.lock(); let entry = entries.get(key).cloned()?; entry.touch(); + entry.hand_out(); entry }; @@ -354,6 +400,21 @@ impl Cache { (tracks, freed) } + /// Delete `entry` and its stems now, unless it was replaced under its key or + /// [`Cache::get`] has handed it to anyone. Returns whether it went. + pub fn discard_unshared(&self, entry: &Arc) -> bool { + let mut entries = self.entries.lock(); + let ours = entries + .get(&entry.key) + .is_some_and(|held| Arc::ptr_eq(held, entry)); + if !ours || entry.handed_out.load(Ordering::SeqCst) { + return false; + } + entries.remove(&entry.key); + self.delete(entry); + true + } + /// One consistent snapshot, rather than three locks that can disagree. pub fn stats(&self) -> Stats { let entries = self.entries.lock(); @@ -387,8 +448,16 @@ impl Cache { let dir = self.root.join(key); let staging = self.staging_dir(key)?; - let files = write_stems(&staging, &dir, &shipped, output.format)?; - install(&staging, &dir)?; + let written = write_stems(&staging, &dir, &shipped, output.format) + .and_then(|files| install(&staging, &dir).map(|()| files)); + let files = match written { + Ok(files) => files, + Err(err) => { + // A half-written entry would hold its bytes until the next start. + let _ = std::fs::remove_dir_all(&staging); + return Err(err); + } + }; let entry = Arc::new(Entry { key: key.to_owned(), @@ -405,6 +474,7 @@ impl Cache { derived: output.parts.derived(), separation_secs, evicted: AtomicBool::new(false), + handed_out: AtomicBool::new(false), state: Mutex::new(State { unfetched: shipped.iter().map(|(name, _)| (*name).to_owned()).collect(), last_used: Instant::now(), @@ -481,13 +551,36 @@ impl Cache { /// the store under it, so waiting for the cap would make rule 1 nearly /// unreachable. fn under_pressure(&self, entries: &HashMap>) -> bool { - total_bytes(entries) > self.max_bytes / 2 + let total = total_bytes(entries); + total > self.budget(total) / 2 + } + + /// The cap, or less when the disk is short of [`crate::disk::FLOOR`]. + fn budget(&self, total: u64) -> u64 { + if self.disk_aware.load(Ordering::Relaxed) { + crate::disk::budget(self.max_bytes, total, &self.root) + } else { + self.max_bytes + } + } + + /// Hold to the cap alone, whatever the disk has free, so tests do not + /// depend on the machine running them. + #[cfg(test)] + pub fn ignore_disk(&self) { + self.disk_aware.store(false, Ordering::Relaxed); + } + + /// The directory the stems live in. + pub fn root(&self) -> &Path { + &self.root } /// Drop least-recently-used entries until the store fits. fn enforce_cap(&self, entries: &mut HashMap>) -> (usize, u64) { let mut total = total_bytes(entries); - if total <= self.max_bytes { + let budget = self.budget(total); + if total <= budget { return (0, 0); } @@ -504,7 +597,7 @@ impl Cache { let mut count = 0; let mut freed = 0; for (_, key) in order { - if total <= self.max_bytes { + if total <= budget { break; } if let Some(entry) = entries.remove(&key) { @@ -534,6 +627,9 @@ impl Cache { fn clear_contents(dir: &Path) -> std::io::Result<()> { for entry in std::fs::read_dir(dir)? { let path = entry?.path(); + if path.file_name().is_some_and(|name| name == LOCK_FILE) { + continue; + } if path.is_dir() { std::fs::remove_dir_all(&path)?; } else { @@ -724,6 +820,19 @@ mod tests { use crate::testkit::{cache, publish}; + #[test] + fn a_second_lock_on_the_same_root_fails_and_survives_the_startup_clear() { + let root = std::env::temp_dir().join(format!("stemd-lock-test-{}", std::process::id())); + let held = lock(&root).unwrap(); + let err = lock(&root).unwrap_err(); + assert!(err.to_string().contains("already running"), "{err:#}"); + clear_contents(&root).unwrap(); + assert!(root.join(LOCK_FILE).exists()); + drop(held); + lock(&root).unwrap(); + std::fs::remove_dir_all(&root).unwrap(); + } + /// A 200 must be followed by stems the client can actually download. /// /// The TTL is zero so every published entry is collectable at once; a nonzero one @@ -818,6 +927,30 @@ mod tests { assert_eq!(cache.stats().tracks, 0); } + /// What a drop does once its stems are copied out: that entry goes, others stay. + #[test] + fn discarding_takes_one_entry_and_its_stems() { + let cache = cache(u64::MAX, Duration::from_secs(60)); + let dropped = publish(&cache, "dropped", 64, 1.0); + let kept = publish(&cache, "kept", 32, 1.0); + + assert!(cache.discard_unshared(&dropped)); + assert!(!dropped.dir.exists() && dropped.evicted()); + assert!(cache.get(&dropped.key).is_none()); + assert!(kept.dir.exists() && cache.get(&kept.key).is_some()); + assert_eq!(cache.stats().tracks, 1); + } + + /// An entry someone was handed by a hit stays, whoever made it. + #[test] + fn an_entry_handed_out_is_not_discarded() { + let cache = cache(u64::MAX, Duration::from_secs(60)); + let made = publish(&cache, "shared", 64, 1.0); + let _hit = cache.get(&made.key).expect("a hit"); + assert!(!cache.discard_unshared(&made)); + assert!(made.dir.exists()); + } + #[test] fn clearing_takes_the_index_and_the_stems_together() { let cache = cache(u64::MAX, Duration::from_secs(60)); diff --git a/crates/stemd-server/src/disk.rs b/crates/stemd-server/src/disk.rs new file mode 100644 index 0000000..13ce365 --- /dev/null +++ b/crates/stemd-server/src/disk.rs @@ -0,0 +1,180 @@ +//! Free disk space and memory pressure, read so work is refused or trimmed +//! before the machine runs out. +//! +//! With the disk full, macOS cannot grow swap, and a separation that needs more +//! memory stalls the whole process inside an allocation. + +use std::path::Path; + +/// Free space below which new work is refused and caches stop growing. +pub const FLOOR: u64 = 2_000_000_000; + +/// Bytes free to this user on the volume holding `path`, or `None` if it cannot +/// be read. A path that does not exist yet is measured at its nearest existing +/// ancestor. +pub fn free_bytes(path: &Path) -> Option { + let existing = path.ancestors().find(|p| p.exists())?; + free_at(existing) +} + +/// Whether the volume holding `path` is under [`FLOOR`]. +pub fn low(path: &Path) -> bool { + free_bytes(path).is_some_and(|free| free < FLOOR) +} + +/// Work refused for want of disk space. +#[derive(Debug, thiserror::Error)] +#[error( + "not enough disk space: {:.1} GB free, {:.1} GB needed; free some space and try again", + *free as f64 / 1e9, + *needed as f64 / 1e9 +)] +pub struct DiskFull { + pub free: u64, + pub needed: u64, +} + +/// Fail unless the volume holding `path` has [`FLOOR`] free. Passes when the +/// free space cannot be read. +pub fn ensure_room(path: &Path) -> Result<(), DiskFull> { + ensure_room_for(path, 0) +} + +/// Fail unless the volume holding `path` can take `bytes` and keep [`FLOOR`] +/// free. +pub fn ensure_room_for(path: &Path, bytes: u64) -> Result<(), DiskFull> { + let needed = bytes.saturating_add(FLOOR); + match free_bytes(path) { + Some(free) if free < needed => Err(DiskFull { free, needed }), + _ => Ok(()), + } +} + +/// Room a cache may hold: its cap, reduced so the volume keeps [`FLOOR`] free. +/// `used` is what the cache already holds on that volume. +pub fn budget(cap: u64, used: u64, path: &Path) -> u64 { + free_bytes(path).map_or(cap, |free| fit(cap, used, free)) +} + +/// `cap`, cut so that a cache holding `used` with `free` left keeps [`FLOOR`] free. +const fn fit(cap: u64, used: u64, free: u64) -> u64 { + let room = (used + free).saturating_sub(FLOOR); + if cap < room { cap } else { room } +} + +/// Whether the system is short of memory: macOS's warning level or worse. +/// Always false elsewhere. +pub fn memory_pressure() -> bool { + #[cfg(target_os = "macos")] + { + let mut level: libc::c_int = 0; + let mut size = std::mem::size_of::(); + // SAFETY: the name is NUL-terminated and `level` is a live int of `size` bytes. + let read = unsafe { + libc::sysctlbyname( + c"kern.memorystatus_vm_pressure_level".as_ptr(), + (&raw mut level).cast(), + &raw mut size, + std::ptr::null_mut(), + 0, + ) + }; + // 1 normal, 2 warning, 4 critical. + read == 0 && level >= 2 + } + #[cfg(not(target_os = "macos"))] + { + false + } +} + +/// Bytes of physical memory, or `None` where it is not read. +pub fn physical_memory() -> Option { + #[cfg(unix)] + { + // SAFETY: sysconf only reads system constants. + let (pages, size) = unsafe { + ( + libc::sysconf(libc::_SC_PHYS_PAGES), + libc::sysconf(libc::_SC_PAGESIZE), + ) + }; + (pages > 0 && size > 0).then(|| pages as u64 * size as u64) + } + #[cfg(not(unix))] + { + None + } +} + +#[cfg(unix)] +fn free_at(path: &Path) -> Option { + use std::os::unix::ffi::OsStrExt; + let name = std::ffi::CString::new(path.as_os_str().as_bytes()).ok()?; + // SAFETY: an all-zero statvfs is a valid value for the call to overwrite. + let mut stat: libc::statvfs = unsafe { std::mem::zeroed() }; + // SAFETY: `name` is NUL-terminated and `stat` is a live statvfs. + if unsafe { libc::statvfs(name.as_ptr(), &raw mut stat) } != 0 { + return None; + } + #[allow(clippy::useless_conversion)] + Some(u64::from(stat.f_bavail) * u64::from(stat.f_frsize)) +} + +#[cfg(windows)] +fn free_at(path: &Path) -> Option { + use std::os::windows::ffi::OsStrExt; + let wide: Vec = path.as_os_str().encode_wide().chain(Some(0)).collect(); + let mut free = 0u64; + // SAFETY: `wide` is NUL-terminated and `free` is a live u64; the other two + // outputs are optional. + let ok = unsafe { + windows_sys::Win32::Storage::FileSystem::GetDiskFreeSpaceExW( + wide.as_ptr(), + &raw mut free, + std::ptr::null_mut(), + std::ptr::null_mut(), + ) + }; + (ok != 0).then_some(free) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_temp_volume_has_a_free_size() { + let free = free_bytes(&std::env::temp_dir()).expect("statvfs on the temp dir"); + assert!(free > 0); + } + + #[test] + fn a_missing_path_is_measured_where_it_would_go() { + let missing = std::env::temp_dir().join("stemd-no-such-dir/below/it"); + assert!(free_bytes(&missing).is_some()); + } + + #[test] + fn a_refusal_says_what_is_free_and_what_is_needed() { + let full = DiskFull { + free: 1_200_000_000, + needed: 2_300_000_000, + }; + assert_eq!( + full.to_string(), + "not enough disk space: 1.2 GB free, 2.3 GB needed; free some space and try again" + ); + } + + #[test] + fn the_budget_leaves_the_floor_free() { + let gb = 1_000_000_000; + // Plenty free: the cap stands. + assert_eq!(fit(4 * gb, gb, 50 * gb), 4 * gb); + // 3 GB free with 1 GB held: 2 GB may be held in all. + assert_eq!(fit(4 * gb, gb, 3 * gb), 2 * gb); + // Under the floor already: nothing. + assert_eq!(fit(4 * gb, 0, gb), 0); + } +} diff --git a/crates/stemd-server/src/drops.rs b/crates/stemd-server/src/drops.rs index 5fdbb6d..8714b66 100644 --- a/crates/stemd-server/src/drops.rs +++ b/crates/stemd-server/src/drops.rs @@ -12,13 +12,15 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use anyhow::{Context, Result}; -use parking_lot::Mutex; +use parking_lot::{Condvar, Mutex}; use stemd_core::{Audio, DspMode, PART_NAMES, PcmFormat, Progress, Stage}; use crate::api::AppState; use crate::cache::{Entry, Ingredients, Output}; +use crate::jobs::Job; use crate::queue::QueuedWork; use crate::settings::DropParts; @@ -79,6 +81,10 @@ pub struct Dropped { /// File name without the extension, which is what the window shows. pub title: String, state: Mutex, + /// The job this drop waits on, once claimed. Whoever takes it releases it. + job: Mutex>>, + /// Set by [`Drops::stop`]. + stopped: AtomicBool, } impl Dropped { @@ -100,6 +106,20 @@ impl Dropped { secs: 1.0, cached: false, })), + job: Mutex::new(None), + stopped: AtomicBool::new(false), + }) + } + + /// A drop part way through, for the window's tests. + #[cfg(test)] + pub fn working(title: &str, progress: Progress) -> Arc { + Arc::new(Self { + source: PathBuf::from(format!("/Music/{title}.wav")), + title: title.to_owned(), + state: Mutex::new(State::Working(progress)), + job: Mutex::new(None), + stopped: AtomicBool::new(false), }) } @@ -110,12 +130,105 @@ impl Dropped { pub fn is_running(&self) -> bool { matches!(self.state(), State::Reading | State::Working(_)) } + + fn stopped(&self) -> bool { + self.stopped.load(Ordering::SeqCst) + } + + /// Let go of the job, and cancel it if nobody else waits on it. Returns + /// whether this drop was the last waiter. + fn release(&self, state: &AppState) -> bool { + let Some(job) = self.job.lock().take() else { + return false; + }; + if job.release() > 0 { + return false; + } + if !job.view().progress.stage.is_terminal() { + state.queue.cancel(&job.id); + state.store.remove(&job.id); + } + true + } } /// The window's list of drops, newest first. #[derive(Default)] pub struct Drops { recent: Mutex>>, + turns: Turnstile, +} + +/// Queue positions in drop order: a short track decodes faster than a long one +/// dropped before it, and must still be queued after it. +#[derive(Default)] +struct Turnstile { + next: Mutex, + moved: Condvar, +} + +#[derive(Default)] +struct Line { + issued: u64, + serving: u64, +} + +impl Turnstile { + fn ticket(&self) -> u64 { + let mut line = self.next.lock(); + line.issued += 1; + line.issued - 1 + } +} + +/// One drop's place in line. [`Turn::wait`] takes the turn and [`Turn::pass`] +/// hands it on; dropping a turn that was never passed does both, so a drop +/// that fails or panics cannot hold up the ones behind it. +struct Turn<'a> { + turns: &'a Turnstile, + ticket: u64, + waited: bool, + passed: bool, +} + +impl<'a> Turn<'a> { + const fn new(turns: &'a Turnstile, ticket: u64) -> Self { + Self { + turns, + ticket, + waited: false, + passed: false, + } + } + + /// Block until every earlier drop has been queued or has failed. + fn wait(&mut self) { + if self.waited { + return; + } + let mut line = self.turns.next.lock(); + while line.serving != self.ticket { + self.turns.moved.wait(&mut line); + } + self.waited = true; + } + + /// Let the next drop go. + fn pass(&mut self) { + if self.passed { + return; + } + self.wait(); + self.turns.next.lock().serving += 1; + self.turns.moved.notify_all(); + self.passed = true; + } +} + +impl Drop for Turn<'_> { + fn drop(&mut self) { + self.pass(); + } } impl Drops { @@ -133,25 +246,61 @@ impl Drops { /// track takes long enough to drop frames, and the window has to keep /// painting the progress of the job this creates. pub fn accept(self: &Arc, state: &Arc, path: PathBuf) -> Arc { + // A second drop of a track still in progress would write the same folder. + if let Some(running) = self + .recent + .lock() + .iter() + .find(|d| d.source == path && d.is_running()) + { + return Arc::clone(running); + } let dropped = Arc::new(Dropped { title: title_of(&path), source: path, state: Mutex::new(State::Reading), + job: Mutex::new(None), + stopped: AtomicBool::new(false), }); self.remember(Arc::clone(&dropped)); let state = Arc::clone(state); + let drops = Arc::clone(self); + let ticket = self.turns.ticket(); let started = Arc::clone(&dropped); - std::thread::Builder::new() + let spawned = std::thread::Builder::new() .name("stemd-drop".into()) .spawn(move || { - if let Err(err) = run(&state, &started) { - tracing::error!("{}: {err:#}", started.title); - started.set(State::Failed(format!("{err:#}"))); + let mut turn = Turn::new(&drops.turns, ticket); + let ran = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run(&state, &started, &mut turn) + })); + let failure = match ran { + Ok(Ok(())) => None, + Ok(Err(err)) => Some(format!("{err:#}")), + Err(payload) => Some(format!( + "the drop panicked: {}", + crate::panics::message(&*payload) + )), + }; + // Marked failed before the turn is handed on, which may wait. + if let Some(why) = failure { + if started.stopped() { + tracing::info!("{}: stopped", started.title); + } else { + tracing::error!("{}: {why}", started.title); + } + started.set(State::Failed(why)); } - }) - .expect("spawning the drop worker"); + turn.pass(); + }); + if let Err(err) = spawned { + // Hand the turn on, or every later drop waits for it. + std::mem::drop(Turn::new(&self.turns, ticket)); + tracing::error!("{}: could not start: {err}", dropped.title); + dropped.set(State::Failed(format!("could not start: {err}"))); + } dropped } @@ -168,12 +317,18 @@ impl Drops { self.recent.lock().retain(|other| !Arc::ptr_eq(other, item)); } + /// Stop a drop that is still running and take it out of the list. Its job + /// is cancelled unless an API client also waits on it. + pub fn stop(&self, state: &AppState, item: &Arc) { + item.stopped.store(true, Ordering::SeqCst); + item.release(state); + self.forget(item); + } + /// Forget a drop and delete the stems it wrote. /// - /// Deleted outright rather than moved to the Trash, which would need Finder - /// automation and a permission prompt inside an `.app`. The separation is still in - /// the cache, so dropping the same file again writes the folder back without - /// running the model. + /// The folder is deleted outright. Moving it to the Trash needs Finder + /// automation and a permission prompt inside an `.app`. pub fn discard(&self, item: &Arc) { if let State::Done(finished) = item.state() { remove_stems(&finished.dir); @@ -204,8 +359,11 @@ fn remove_stems(dir: &Path) { } /// Decode, separate, and write the stems out. -fn run(state: &Arc, item: &Arc) -> Result<()> { +fn run(state: &Arc, item: &Arc, turn: &mut Turn) -> Result<()> { let mix = read(state, &item.source)?; + if item.stopped() { + anyhow::bail!("stopped"); + } let settings = state.settings.get(); let output = Output { format: settings.format, @@ -216,8 +374,20 @@ fn run(state: &Arc, item: &Arc) -> Result<()> { parts: settings.drop_parts.request(), }; - let (entry, cached) = separate(state, item, &mix, output)?; - let finished = write_out(&item.source, &entry, cached, settings.drop_parts)?; + let (entry, origin) = separate(state, item, mix, output, turn)?; + let written = write_out( + &item.source, + &entry, + matches!(origin, Origin::Cached), + settings.drop_parts, + ); + // The folder is the drop's only copy. The last drop waiting on the job + // deletes the entry, unless an API client holds the job or fetched the + // entry. Kept when writing failed, so a second drop needs no separation. + if matches!(origin, Origin::Queued) && item.release(state) && written.is_ok() { + state.cache.discard_unshared(&entry); + } + let finished = written?; tracing::info!( "{}: {} parts written to {}", item.title, @@ -266,13 +436,23 @@ fn read(state: &AppState, path: &Path) -> Result