diff --git a/crates/pdfs-cli/src/main.rs b/crates/pdfs-cli/src/main.rs index 3270374..048da24 100644 --- a/crates/pdfs-cli/src/main.rs +++ b/crates/pdfs-cli/src/main.rs @@ -581,6 +581,15 @@ enum SyncCmd { #[arg(long)] folder: Option, }, + /// Cap upload and download speed, or show the caps with no options. + Limit { + /// Upload cap per second, e.g. `500K` or `2M`; `0` removes it. + #[arg(long, value_parser = parse_rate)] + up: Option, + /// Download cap per second, e.g. `500K` or `2M`; `0` removes it. + #[arg(long, value_parser = parse_rate)] + down: Option, + }, /// List uploads and changes not yet on Proton Drive. Queue, /// Retry a queued operation now (by id from `sync queue`), or every failed one. @@ -907,6 +916,7 @@ fn cmd_sync(action: SyncCmd) -> Result<()> { })?)? } SyncCmd::Queue => return cmd_sync_queue(), + SyncCmd::Limit { up, down } => return cmd_sync_limit(up, down), SyncCmd::Retry { id } => ok_or_bail(control_request(CtlRequest::RetryPendingOp { id })?)?, SyncCmd::Resume { folder: None } => { ok_or_bail(control_request(CtlRequest::SetSyncPaused { @@ -1845,6 +1855,8 @@ fn mount_once(mountpoint: Option) -> Result { pdfs_fuse::MountOptions { username, sweep_mode: config.resolved_conflict_sweep(), + upload_limit: config.upload_limit.unwrap_or(0), + download_limit: config.download_limit.unwrap_or(0), }, ); @@ -2565,6 +2577,47 @@ fn parse_pause_duration(text: &str) -> Result { } } +/// Read a bytes-per-second rate: a whole number with an optional `K`, `M` or +/// `G` suffix in the same binary units `human_bytes` prints. +fn parse_rate(text: &str) -> Result { + let text = text.trim(); + let (number, scale) = match text.char_indices().last() { + Some((at, 'K' | 'k')) => (&text[..at], 1024), + Some((at, 'M' | 'm')) => (&text[..at], 1_048_576), + Some((at, 'G' | 'g')) => (&text[..at], 1_073_741_824), + _ => (text, 1), + }; + number + .parse::() + .map(|n| n.saturating_mul(scale)) + .map_err(|_| "use a whole number with an optional K, M or G, e.g. 2M".to_string()) +} + +fn human_rate(rate: u64) -> String { + if rate == 0 { + "unlimited".to_string() + } else { + format!("{}/s", human_bytes(rate)) + } +} + +/// `pdfs sync limit`: set the bandwidth caps, keeping whichever one is not +/// given, or print both when neither is. +fn cmd_sync_limit(up: Option, down: Option) -> Result<()> { + let config = AppDirs::new()?.load_config(); + let upload = up.or(config.upload_limit).unwrap_or(0); + let download = down.or(config.download_limit).unwrap_or(0); + if up.is_none() && down.is_none() { + println!("Upload {}", human_rate(upload)); + println!("Download {}", human_rate(download)); + return Ok(()); + } + ok_or_bail(control_request(CtlRequest::SetBandwidthLimits { + upload, + download, + })?) +} + fn unix_now() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -3055,6 +3108,18 @@ mod pause_tests { assert!(parse_pause_duration("-1h").is_err()); assert!(parse_pause_duration("1ä").is_err()); } + + #[test] + fn rates_read_plain_bytes_and_binary_suffixes() { + assert_eq!(parse_rate("0"), Ok(0)); + assert_eq!(parse_rate("1500"), Ok(1500)); + assert_eq!(parse_rate("500K"), Ok(512_000)); + assert_eq!(parse_rate("2m"), Ok(2_097_152)); + assert!(parse_rate("").is_err()); + assert!(parse_rate("M").is_err()); + assert!(parse_rate("1.5M").is_err()); + assert!(parse_rate("2MB").is_err()); + } } #[cfg(test)] diff --git a/crates/pdfs-core/src/config.rs b/crates/pdfs-core/src/config.rs index 743bb3c..704790b 100644 --- a/crates/pdfs-core/src/config.rs +++ b/crates/pdfs-core/src/config.rs @@ -85,6 +85,15 @@ pub struct AppConfig { /// writes an explicit value here. Defaulted for configs predating the field. #[serde(default)] pub cache_budget: Option, + /// Upload cap in bytes per second shared by every upload. `None` or `0` + /// means no cap. Defaulted for configs predating the field. + #[serde(default)] + pub upload_limit: Option, + /// Download cap in bytes per second shared by every download, FUSE reads + /// included. `None` or `0` means no cap. Defaulted for configs predating + /// the field. + #[serde(default)] + pub download_limit: Option, /// Mountpoint the daemon mounts at. `None` means /// [`AppDirs::default_mountpoint`]; the Settings page writes an explicit /// path here. Defaulted for configs predating the field. @@ -189,6 +198,8 @@ impl Default for AppConfig { app_version: APP_VERSION.to_string(), user_agent: USER_AGENT.to_string(), cache_budget: None, + upload_limit: None, + download_limit: None, mountpoint: None, ignore_patterns: None, device_uid: None, @@ -535,6 +546,8 @@ mod tests { app_version: "external-drive-test-client@1.0.0".to_string(), user_agent: "test-agent/1.0".to_string(), cache_budget: Some(1234), + upload_limit: Some(500_000), + download_limit: None, mountpoint: Some("/tmp/x".to_string()), ignore_patterns: Some(vec!["build/".to_string()]), device_uid: Some("dev-uid".to_string()), @@ -554,6 +567,8 @@ mod tests { assert_eq!(decoded.app_version, "external-drive-test-client@1.0.0"); assert_eq!(decoded.user_agent, "test-agent/1.0"); assert_eq!(decoded.proton_accent, Some(true)); + assert_eq!(decoded.upload_limit, Some(500_000)); + assert_eq!(decoded.download_limit, None); assert_eq!(decoded.files_view, config.files_view); } diff --git a/crates/pdfs-core/src/control.rs b/crates/pdfs-core/src/control.rs index a6fc0da..99ee3c8 100644 --- a/crates/pdfs-core/src/control.rs +++ b/crates/pdfs-core/src/control.rs @@ -299,6 +299,10 @@ pub enum Request { /// persist it to config so the next mount keeps it. Replies with /// [`Response::Ok`]. SetCacheBudget { bytes: u64 }, + /// Cap uploads and downloads at these bytes per second (`0` = no cap), + /// live for running transfers, and persist both to config so the next + /// mount keeps them. Replies with [`Response::Ok`]. + SetBandwidthLimits { upload: u64, download: u64 }, /// Report on the health of the metadata database and content cache: sizes, /// row counts, and — when `deep` — SQLite's own integrity check. Replies /// with [`Response::CacheReport`]. @@ -2419,6 +2423,10 @@ mod tests { id: 3, paused: true, }, + Request::SetBandwidthLimits { + upload: 1_000_000, + download: 0, + }, Request::SyncNow { id: Some(3) }, Request::AdoptDevice { uid: Some("dev-1".into()), diff --git a/crates/pdfs-fuse/src/control.rs b/crates/pdfs-fuse/src/control.rs index 31a29f0..f7b8a23 100644 --- a/crates/pdfs-fuse/src/control.rs +++ b/crates/pdfs-fuse/src/control.rs @@ -899,6 +899,29 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U ))), } } + Ok(CtlRequest::SetBandwidthLimits { upload, download }) => { + core.transfers.set_limits(upload, download); + // Persisted like the cache budget: best-effort, the live caps are + // already applied. + match AppDirs::new().map(|dirs| { + let mut cfg = dirs.load_config(); + cfg.upload_limit = Some(upload); + cfg.download_limit = Some(download); + dirs.save_config(&cfg) + }) { + Ok(Ok(())) => CtlResponse::Ok { + message: format!( + "bandwidth limits set to {upload} B/s up, {download} B/s down" + ), + }, + Ok(Err(e)) => CtlResponse::error(CoreError::internal(format!( + "limits applied but config write failed: {e}" + ))), + Err(e) => CtlResponse::error(CoreError::internal(format!( + "limits applied but config unavailable: {e}" + ))), + } + } Ok(CtlRequest::ListDevices) => match core.list_devices() { Ok(items) => CtlResponse::Devices { items }, Err(e) => CtlResponse::error(e), diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs index 2a6652f..c5017df 100644 --- a/crates/pdfs-fuse/src/mount.rs +++ b/crates/pdfs-fuse/src/mount.rs @@ -241,6 +241,9 @@ pub struct MountOptions { pub username: String, /// What the background conflict sweep may do. See `docs/BUGS.md` B71. pub sweep_mode: SweepMode, + /// Upload and download caps in bytes per second (`0` = no cap). + pub upload_limit: u64, + pub download_limit: u64, } /// Spawn one FUSE session rooted at an arbitrary remote node. @@ -340,6 +343,8 @@ pub fn mount( let MountOptions { username, sweep_mode, + upload_limit, + download_limit, } = options; // Start the uptime clock here rather than at the first `pdfs diagnostics`, // so the age it reports is the daemon's own. @@ -419,6 +424,7 @@ pub fn mount( // Before anything can queue work against it: the drain thread below reaches // every mount's inode space through this registry, not through `core.state`. core.register_state(mountpoint); + core.transfers.set_limits(upload_limit, download_limit); // Writes queued by a previous run (or left behind by a crash) are still owed // an upload, and reads must be served from their staged blobs until they land. diff --git a/crates/pdfs-fuse/src/transfers.rs b/crates/pdfs-fuse/src/transfers.rs index ae8093e..2f483ad 100644 --- a/crates/pdfs-fuse/src/transfers.rs +++ b/crates/pdfs-fuse/src/transfers.rs @@ -13,6 +13,10 @@ //! always reflects exactly what is in flight — even if the work fails partway //! and unwinds. //! +//! The same per-block tick enforces the user's bandwidth limits: every byte a +//! transfer moves is charged to its direction's [`RateLimit`], and the thread +//! moving it sleeps off whatever the charge puts it over the cap. +//! //! [`Request::GetQueueStatus`]: pdfs_core::control::Request::GetQueueStatus use parking_lot::Mutex; @@ -20,7 +24,7 @@ use std::collections::HashMap; use std::io::{Read, Write}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::time::Instant; +use std::time::{Duration, Instant}; use pdfs_core::control::{JobItem, TransferDirection, TransferItem}; @@ -52,6 +56,59 @@ pub struct TransferRegistry { inner: Mutex>>, jobs: Mutex>>, next: AtomicU64, + upload_limit: RateLimit, + download_limit: RateLimit, +} + +/// How far ahead of its cap a direction may run before it is made to wait, so a +/// small file still goes out at full speed. +const RATE_BURST: Duration = Duration::from_secs(1); + +/// A bytes-per-second cap shared by every transfer in one direction; `0` is no +/// cap. +/// +/// A virtual clock rather than a refilled bucket: each charge pushes +/// `next_free` forward by the time its bytes take at the cap, so concurrent +/// transfers split the cap between them instead of each getting all of it. +#[derive(Default)] +struct RateLimit { + rate: AtomicU64, + next_free: Mutex>, +} + +impl RateLimit { + fn set(&self, rate: u64) { + self.rate.store(rate, Ordering::Relaxed); + // A new cap starts from now, not from the debt the old one ran up. + *self.next_free.lock() = None; + } + + /// Charge `n` bytes moved at `now`, returning how long the mover must wait + /// to stay under the cap. + fn charge(&self, n: u64, now: Instant) -> Duration { + let rate = self.rate.load(Ordering::Relaxed); + if rate == 0 || n == 0 { + return Duration::ZERO; + } + let mut next_free = self.next_free.lock(); + let start = next_free.filter(|t| *t > now).unwrap_or(now); + let done = start + Duration::from_secs_f64(n as f64 / rate as f64); + *next_free = Some(done); + done.saturating_duration_since(now + RATE_BURST) + } +} + +/// Sleep off a rate-limit wait. The bulk uploader reads on tokio workers, so +/// the sleep is announced to the runtime there rather than silently holding a +/// worker other tasks — FUSE reads among them — may need. +fn throttle(wait: Duration) { + use tokio::runtime::{Handle, RuntimeFlavor}; + match Handle::try_current() { + Ok(handle) if handle.runtime_flavor() == RuntimeFlavor::MultiThread => { + tokio::task::block_in_place(|| std::thread::sleep(wait)) + } + _ => std::thread::sleep(wait), + } } impl TransferRegistry { @@ -59,6 +116,20 @@ impl TransferRegistry { Arc::new(Self::default()) } + /// Cap uploads and downloads at these bytes per second; `0` lifts a cap. + /// Applies to transfers already running from their next block. + pub fn set_limits(&self, upload: u64, download: u64) { + self.upload_limit.set(upload); + self.download_limit.set(download); + } + + fn limit(&self, direction: TransferDirection) -> &RateLimit { + match direction { + TransferDirection::Upload => &self.upload_limit, + TransferDirection::Download => &self.download_limit, + } + } + /// Register a new transfer of `total` bytes (`0` = unknown), returning a /// guard that tracks progress and deregisters the transfer when dropped. pub fn begin( @@ -169,9 +240,17 @@ pub struct TransferGuard { } impl TransferGuard { - /// Record `n` more bytes moved. + /// Record `n` more bytes moved, then wait if they put the direction over + /// its cap. pub fn add(&self, n: u64) { self.entry.done.fetch_add(n, Ordering::Relaxed); + let wait = self + .reg + .limit(self.entry.direction) + .charge(n, Instant::now()); + if !wait.is_zero() { + throttle(wait); + } } } @@ -321,3 +400,37 @@ impl Read for CountingReader<'_, R> { Ok(n) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_uncapped_direction_never_waits() { + let limit = RateLimit::default(); + assert_eq!(limit.charge(u64::MAX / 2, Instant::now()), Duration::ZERO); + } + + #[test] + fn a_capped_direction_waits_once_past_its_burst() { + let limit = RateLimit::default(); + limit.set(1000); + let now = Instant::now(); + // One second of bytes is the burst: no wait yet. + assert_eq!(limit.charge(1000, now), Duration::ZERO); + // Two more seconds' worth at the same instant: wait out all but the burst. + assert_eq!(limit.charge(2000, now), Duration::from_secs(2)); + // Lifting the cap drops the debt with it. + limit.set(0); + assert_eq!(limit.charge(5000, now), Duration::ZERO); + } + + #[test] + fn concurrent_transfers_share_one_cap() { + let limit = RateLimit::default(); + limit.set(1000); + let now = Instant::now(); + assert_eq!(limit.charge(1500, now), Duration::from_millis(500)); + assert_eq!(limit.charge(1500, now), Duration::from_secs(2)); + } +} diff --git a/crates/pdfs-gui/src/app/main.rs b/crates/pdfs-gui/src/app/main.rs index 294c0a7..c2b7234 100644 --- a/crates/pdfs-gui/src/app/main.rs +++ b/crates/pdfs-gui/src/app/main.rs @@ -376,10 +376,13 @@ fn build_window(app: &adw::Application) { quota_checked_at: Cell::new(None), autostart_row: main_widgets.autostart_row.clone(), budget_row: main_widgets.budget_row.clone(), + upload_limit_row: main_widgets.upload_limit_row.clone(), + download_limit_row: main_widgets.download_limit_row.clone(), mountpoint_row: main_widgets.mountpoint_row.clone(), accent_row: main_widgets.accent_row.clone(), settings_suppress: Cell::new(false), budget_source: RefCell::new(None), + limit_source: RefCell::new(None), pins_expanded: Cell::new(false), pins_group: main_widgets.pins_group.clone(), pin_rows: RefCell::new(Vec::new()), diff --git a/crates/pdfs-gui/src/app/pages/status.rs b/crates/pdfs-gui/src/app/pages/status.rs index c1f85ff..5a13edd 100644 --- a/crates/pdfs-gui/src/app/pages/status.rs +++ b/crates/pdfs-gui/src/app/pages/status.rs @@ -37,6 +37,11 @@ pub(crate) struct StatusState { /// Cache-budget editor (GiB). Populated once from config; user edits drive a /// `SetCacheBudget` round-trip. Guarded by [`Self::settings_suppress`]. pub(crate) budget_row: adw::SpinRow, + /// Bandwidth caps (MiB/s, `0` = none). Populated once from config; an edit + /// to either sends both in one `SetBandwidthLimits`. Guarded by + /// [`Self::settings_suppress`]. + pub(crate) upload_limit_row: adw::SpinRow, + pub(crate) download_limit_row: adw::SpinRow, /// Shows where the primary mount lives; its Change button picks a new one. pub(crate) mountpoint_row: adw::ActionRow, /// "Proton purple accent" toggle. Guarded by [`Self::settings_suppress`]. @@ -49,6 +54,8 @@ pub(crate) struct StatusState { /// stack of toasts and a series of caps the user never asked to apply — so /// only the value they settle on is sent. pub(crate) budget_source: RefCell>, + /// Pending debounce for the bandwidth editors, as for the cache budget. + pub(crate) limit_source: RefCell>, pub(crate) pins_group: adw::PreferencesGroup, /// Whether the pin list is showing every pin or only the first /// [`PINS_COLLAPSED`]. A long pin list would otherwise push everything below @@ -119,6 +126,9 @@ pub(crate) struct MainWidgets { pub(crate) autostart_row: adw::SwitchRow, /// Cache soft-cap editor, in GiB; `0` = unlimited. pub(crate) budget_row: adw::SpinRow, + /// Upload and download caps, in MiB/s; `0` = unlimited. + pub(crate) upload_limit_row: adw::SpinRow, + pub(crate) download_limit_row: adw::SpinRow, /// Purges all unpinned cached content. pub(crate) purge_button: gtk4::Button, /// Shows the active mountpoint; its suffix button picks a new one. @@ -159,6 +169,24 @@ pub(crate) fn build_main_page() -> MainWidgets { mountpoint_row.add_suffix(&mountpoint_button); location_group.add(&mountpoint_row); + // Bandwidth caps in MiB/s. 0 = unlimited; the smallest cap is one 0.5 step, + // since anything slower would stall a single block for most of a minute. + let network_group = adw::PreferencesGroup::builder() + .title("Network") + .description("Limit how fast files move, shared by every transfer. 0 means no limit.") + .build(); + let limit_row = |title: &str| { + adw::SpinRow::builder() + .title(title) + .adjustment(>k4::Adjustment::new(0.0, 0.0, 1000.0, 0.5, 5.0, 0.0)) + .digits(1) + .build() + }; + let upload_limit_row = limit_row("Upload limit (MiB/s)"); + let download_limit_row = limit_row("Download limit (MiB/s)"); + network_group.add(&upload_limit_row); + network_group.add(&download_limit_row); + let appearance_group = adw::PreferencesGroup::builder().title("Appearance").build(); let accent_row = adw::SwitchRow::builder() .title("Proton purple accent") @@ -172,6 +200,7 @@ pub(crate) fn build_main_page() -> MainWidgets { .build(); general.add(&startup_group); general.add(&location_group); + general.add(&network_group); general.add(&appearance_group); // ---- Preferences: Storage @@ -327,6 +356,8 @@ pub(crate) fn build_main_page() -> MainWidgets { pins_group, autostart_row, budget_row, + upload_limit_row, + download_limit_row, purge_button, mountpoint_row, mountpoint_button, @@ -356,6 +387,9 @@ pub(crate) const BUDGET_DEBOUNCE: Duration = Duration::from_millis(600); /// Bytes per GiB, for the cache-budget editor's unit conversion. pub(crate) const GIB: f64 = 1024.0 * 1024.0 * 1024.0; +/// Bytes per MiB, for the bandwidth editors' unit conversion. +pub(crate) const MIB: f64 = 1024.0 * 1024.0; + /// Wire the Preferences controls: the cache-budget editor, the purge button, the /// start-on-login switch, the mountpoint chooser and the accent toggle. Initial widget state is /// read once from config / systemd here (the refresh loop owns only the live @@ -373,6 +407,12 @@ pub(crate) fn wire_settings( ui.status .budget_row .set_value(config.resolved_cache_budget() as f64 / GIB); + ui.status + .upload_limit_row + .set_value(config.upload_limit.unwrap_or(0) as f64 / MIB); + ui.status + .download_limit_row + .set_value(config.download_limit.unwrap_or(0) as f64 / MIB); ui.status .mountpoint_row .set_subtitle(&ui.dirs.resolved_mountpoint(&config).display().to_string()); @@ -407,6 +447,33 @@ pub(crate) fn wire_settings( *ui_budget.status.budget_source.borrow_mut() = Some(src); }); + // Bandwidth: either editor sends both caps, since the daemon sets them as a + // pair. Debounced like the budget. + for row in [&ui.status.upload_limit_row, &ui.status.download_limit_row] { + let ui_limit = ui.clone(); + row.connect_value_notify(move |_| { + if ui_limit.status.settings_suppress.get() { + return; + } + if let Some(src) = ui_limit.status.limit_source.borrow_mut().take() { + src.remove(); + } + let ui_fire = ui_limit.clone(); + let src = glib::timeout_add_local_once(BUDGET_DEBOUNCE, move || { + ui_fire.status.limit_source.borrow_mut().take(); + let upload = (ui_fire.status.upload_limit_row.value() * MIB).round() as u64; + let download = (ui_fire.status.download_limit_row.value() * MIB).round() as u64; + settings_request( + &ui_fire, + Request::SetBandwidthLimits { upload, download }, + "Bandwidth limits updated", + "Couldn't set bandwidth limits", + ); + }); + *ui_limit.status.limit_source.borrow_mut() = Some(src); + }); + } + // Purge: confirm, then drop all unpinned cached content via the daemon. let ui_purge = ui.clone(); purge_button.connect_clicked(move |_| { diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index c53260c..fceae59 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -31,6 +31,10 @@ Schema: **30** (unchanged). straight away. `pdfs sync pause --folder ` / `pdfs sync resume --folder ` do the same, and `pdfs sync list` marks paused folders. The pause is kept across restarts. On-demand folders upload through the shared queue, so only the global pause holds them back. +- **Bandwidth limits.** Preferences → General → Network caps upload and download speed in MiB/s, + shared by every transfer and applied to running ones from their next block; downloads include + reads through the mount. `pdfs sync limit --up 2M --down 0` does the same, and `pdfs sync limit` + on its own prints the current caps. The caps are kept in `config.json` across restarts. ### Changed - **New app shell.** Every page has its own header bar; the sidebar ends in a footer with sync