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..c9ff261 100644 --- a/README.md +++ b/README.md @@ -1,15 +1,15 @@
-stemd +Stemd -# stemd +# Stemd **Split any track into stems, on your own machine.** Drop a file on the window. Get `harmonics`, `vocals` and `drums` in a folder beside it. -The stemd window, with a drop zone reading Drop a track +The Stemd window: a drop zone reading Drop a track, and the Separated list below it [![License](https://img.shields.io/badge/license-MIT%20OR%20Apache--2.0-blue)](#license) [![Platforms](https://img.shields.io/badge/platforms-macOS%20%7C%20Windows%20%7C%20Linux-lightgrey)](#install) @@ -43,7 +43,7 @@ track bit for bit. **macOS.** Open the `macos-arm64` disk image from the [latest release](https://github.com/nsaintot/stemd/releases/latest) and drag -stemd to Applications. Signed and notarized, and the ticket is stapled to the +Stemd to Applications. Signed and notarized, and the ticket is stapled to the app as well as to the image, so it opens without argument and the first launch does not need a network. @@ -54,7 +54,7 @@ installer. On a machine with an NVIDIA card, tick the box at the end of the installer or run `install-cuda.cmd` once afterwards: about 1.2 GB of NVIDIA's own runtime, fetched -once and pinned by digest. No toolkit and no repository. Without it stemd runs on +once and pinned by digest. No toolkit and no repository. Without it Stemd runs on the CPU, correctly and far slower. **Linux.** Take the package for your distribution and architecture @@ -81,7 +81,7 @@ Drop a track on the window, or click to choose one. The stems appear in ### From another machine -stemd is also a server. It announces itself over mDNS, so a client does not need +Stemd is also a server. It announces itself over mDNS, so a client does not need to be told an address, and it speaks plain HTTP: ```bash @@ -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