From e18eacf07ca141a39d0129c38960a8f1a93d9dad Mon Sep 17 00:00:00 2001 From: axpnet <45786925+axpnet@users.noreply.github.com> Date: Tue, 8 Sep 2026 07:52:42 +0200 Subject: [PATCH] perf(sftp): honour --parallel up to 16 connections, and say when a provider ceiling binds The SFTP pool ceiling was a literal 4, so --parallel 16 delivered 4 connections without a word. The comment above it asked for a live benchmark on the target server before raising it; the DAG engine review battery on the Hetzner lab (wired gigabit, 5000 x 4 KiB files, 2026-09-08) is that benchmark, on two binaries. From --parallel 4 to 16 rclone gained 47% on upload (687 s to 367 s) and 73% on download (422 s to 116 s); AeroFTP moved under 1% (1391 s to 1383 s, 1294 s to 1293 s) because the cap silently bound the flag. After the SFTP connection reuse of #735 made every leg four times faster the slope was unchanged: rclone still gained 47% and 75% from the same step, AeroFTP 4% and 3% (331 s to 316 s, 273 s to 264 s). The ceiling is now 16, the same range as --sftp-concurrency; the effective count stays the lower of the ceiling and --parallel, so the default of 4 connections is unchanged. On every provider, put -r and get -r now print a note when the provider ceiling is lower than the requested --parallel (text mode; quiet, JSON and machine stay silent): a flag that accepts 32 and delivers 4 without a word is worse than a lower documented limit. Tests: the SFTP ceiling equals the intra-file stream range; and the door itself, a pool-backed counting provider with the ceiling SftpProvider advertises, driven through the real provider executor and the batch DAG with a rendezvous that only opens when 8 downloads hold 8 sessions at once: --parallel 8 delivers 8, --parallel 32 delivers 16, and with the literal 4 put back the test fails with 4 against 8. Signed-off-by: axpnet <45786925+axpnet@users.noreply.github.com> --- CHANGELOG.md | 1 + docs/CLI-GUIDE.md | 2 +- src-tauri/src/bin/aeroftp_cli.rs | 22 +++ src-tauri/src/providers/sftp.rs | 41 ++++- src-tauri/src/transfer_dag_batch.rs | 242 ++++++++++++++++++++++++++++ 5 files changed, 303 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 15ce0a7e0..5ad2f3d58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 #### Changed - **The CLI remote scan of `sync`, `check` and `cryptcheck` runs on the provider's list pool, up to `--checkers` directories at once.** The CLI walked the remote tree with a serial walker of its own while the GUI scan already listed on clone workers; on the lab SFTP server a 5000-file `sync` spent 10.8 s in a no-op run and 14.7 s in a delta run against rclone's 4.2 s and 5.4 s, all of it in the scan. The three commands now use the shared walker (same filters, same completeness accounting, spinner fed by its progress hook), so a provider that lists on independent connections (SFTP, S3, WebDAV, B2, the HTTP clouds) lists in parallel and a single-session provider walks as before. `--checkers` therefore bounds something real now; its help text says so. Also fixed in the shared walker: a symlink to a directory is listed but no longer walked (the CLI walker already refused it, the GUI walker did not). - **The GUI sync scan lists SFTP directories on up to four connections, and scan workers stay warm across directories.** The parallel scanner cloned a fresh worker for every directory, which is a client copy for HTTP providers but would be a full SSH handshake on SFTP, so SFTP was left on the single locked session (the DAG engine review measured 4 sequential round trips per directory: 15 to 18 s for 5000 files in 51 folders, rclone 4 s). SFTP now exposes the clone-pool scan hooks once a connection exists, and the scanner keeps a worker that listed a directory cleanly for the next directory instead of re-dialling (HTTP clones do not opt in and keep their per-directory clone). The CLI `sync`, `check` and `cryptcheck` run on the same scanner since the entry above. +- **Multi-file SFTP transfers honour `--parallel` up to 16 connections, and the CLI says so when a provider ceiling binds a higher request.** The SFTP pool ceiling was a literal 4, so `--parallel 16` delivered 4 connections without a word. The DAG engine review battery on the Hetzner lab (wired gigabit, 5000 x 4 KiB files) measured what that cost: from 4 to 16 rclone gained 1.9x on upload and 3.6x on download, AeroFTP moved 1%. The ceiling is now 16, the same range as `--sftp-concurrency`; the effective count stays the lower of the ceiling and `--parallel`, so the default of 4 connections is unchanged. On every provider, `put -r` and `get -r` now print a note when the provider ceiling is lower than the requested `--parallel` (text mode; quiet and JSON stay silent). - **SFTP single-stream downloads read ahead by default (window 32), instead of one 256 KiB read per round trip.** The read-ahead tier existed behind `--sftp-readahead` and `AEROFTP_SFTP_READAHEAD`; the DAG engine review measured what the default cost on a 53 ms link with a 300 MiB file: 118 to 129 s serial, 38.8 s with a window of 16, 34.5 s with 32, 35.6 s with 64, rclone 30 s. The window now defaults to 32 when nothing asks for one; `AEROFTP_SFTP_READAHEAD=off` (or `0`) restores the serial read, an explicit `--sftp-readahead` or preset still wins, a bandwidth cap still takes the serial loop that owns the precise throttle, and the job-wide buffer budget still bounds the window per connection. - **`put -r --skip-restricted` uploads a tree whose names the destination partly forbids, setting aside and reporting the offending entries instead of refusing the whole batch.** The default is unchanged and deliberate: one restricted name (a tab in a file name bound for S3, for instance) still stops the batch before anything is created. The review battery of 2026-09-05 measured the two policies side by side on the same 20-file tree: AeroFTP uploaded 0 files, rclone 19 and reported the one it could not. With the flag a folder with a restricted name takes its children along, every skip is listed once, and the run ends `partial` with exit 4 because not everything requested landed. - **Single-file downloads are multi-threaded by default: files of 250 MiB or more are fetched with 4 concurrent range streams, the same stream count as rclone's default (4 streams above 256Mi; AeroFTP's cutoff stays 250 MiB, so files between the two thresholds split here and not there).** `--multi-thread-streams` used to default to `1`, so the DAG range scheduler, the strict 206 probe and the single-stream fallback all existed but only ran when a user found the flag; the May 2026 live matrix showed rclone ahead on every 1 GiB download because of that default alone, while the shared harness measured the two engines within noise once the streams were equal. The gate is unchanged: only backends that prove Range honesty (S3, Azure, SFTP as independent connections, WebDAV and Koofr after a live probe) split the file, everything else and every file below `--multi-thread-cutoff` stays single-stream, and `--multi-thread-streams 1` or `AEROFTP_MULTI_THREAD_STREAMS=1` restores the old behaviour. `pget` is unaffected. diff --git a/docs/CLI-GUIDE.md b/docs/CLI-GUIDE.md index 27f6ac187..6e25fa1b9 100644 --- a/docs/CLI-GUIDE.md +++ b/docs/CLI-GUIDE.md @@ -1915,7 +1915,7 @@ It also emits the transfer-scheduler surface: a `protocol_transfer_capabilities` | `--limit-rate ` | Speed limit (e.g., `1M`, `500K`) | | `--sftp-readahead ` | SFTP read-ahead window (2 to 1024). Default 32 when unset, which keeps up to 32 reads in flight on one connection; `AEROFTP_SFTP_READAHEAD=off` (or `0`) restores one read per round trip; a bandwidth cap always uses the serial loop | | `--bwlimit ` | Bandwidth schedule (e.g., `"08:00,512k 18:00,off"` or `"1M"`) | -| `--parallel ` | Number of parallel transfer workers for recursive/bulk operations | +| `--parallel ` | Number of parallel transfer workers for recursive/bulk operations (default 4, up to 32). Each provider has its own ceiling (SFTP 16 connections, FTP 5, S3 and B2 their clone pools, single-session providers 1); when the ceiling is lower than the request the CLI prints a note in text mode | | `--checkers ` | Directories listed at once by the remote scan of `sync`, `check` and `cryptcheck` (default 8, range 1-64), capped by the provider's list pool; single-session providers walk one directory at a time | | `--partial` | Resume interrupted transfers when the provider supports partial files or remote offsets | | `--include ` | Include only files matching glob pattern (repeatable) | diff --git a/src-tauri/src/bin/aeroftp_cli.rs b/src-tauri/src/bin/aeroftp_cli.rs index 4a2bfbfb1..b5f2444bc 100644 --- a/src-tauri/src/bin/aeroftp_cli.rs +++ b/src-tauri/src/bin/aeroftp_cli.rs @@ -7993,6 +7993,26 @@ fn create_spinner(msg: &str) -> ProgressBar { pb } +/// A flag that accepts 32 and delivers 4 without a word is worse than a lower +/// documented limit: when the provider's session ceiling binds the requested +/// `--parallel`, say so once on stderr (text mode only; quiet and JSON stay +/// silent). The DAG engine review battery found SFTP pinned at 4 for every +/// `--parallel` value while rclone scaled 3.6x from 4 to 16 on the same tree. +fn note_parallel_ceiling( + cli: &Cli, + settings: &ftp_client_gui_lib::transfer_settings::ResolvedTransferSettings, +) { + if cli.quiet || cli.json || cli.machine { + return; + } + if settings.max_concurrent < settings.requested_max_concurrent { + eprintln!( + "Note: this provider caps parallel workers at {} (requested --parallel {})", + settings.max_concurrent, settings.requested_max_concurrent + ); + } +} + fn effective_parallel_workers(cli: &Cli) -> usize { cli.parallel.clamp(1, 32) } @@ -9424,6 +9444,7 @@ async fn run_shared_provider_download_batch( | ProviderExecutorSessionModel::SftpConnectionPool { .. } | ProviderExecutorSessionModel::FtpConnectionPool { .. } ); + note_parallel_ceiling(cli, &runtime_settings); if !is_pool_backed { // Not pool-backed: return the still-connected provider so the // caller runs the legacy independent-connection batch. @@ -9646,6 +9667,7 @@ async fn run_shared_provider_upload_batch( | ProviderExecutorSessionModel::SftpConnectionPool { .. } | ProviderExecutorSessionModel::FtpConnectionPool { .. } ); + note_parallel_ceiling(cli, &runtime_settings); if !is_pool_backed { // Not pool-backed: return the still-connected provider so the // caller runs the legacy independent-connection batch. diff --git a/src-tauri/src/providers/sftp.rs b/src-tauri/src/providers/sftp.rs index 54e351adc..dd7316def 100644 --- a/src-tauri/src/providers/sftp.rs +++ b/src-tauri/src/providers/sftp.rs @@ -54,6 +54,9 @@ fn configure_aerorsync_metadata( /// connection from the pool, so the cap stays conservative; the live /// benchmark in master 9.6.2 says where it pays. const SFTP_MULTI_THREAD_MAX_STREAMS: usize = 16; +/// Independent SSH connections a multi-file job may hold (see +/// `transfer_executor_max_sessions`). +const SFTP_POOL_MAX_SESSIONS: u16 = 16; /// Job-wide guardrails for read-ahead. The byte budget includes one chunk per /// reader, one channel window, and the writer's current chunk. The handle cap @@ -2633,11 +2636,17 @@ impl StorageProvider for SftpProvider { } } - /// Conservative initial cap, mirroring the FTP pool clamp (1..8). - /// Each lease is a full independent SSH connection; raise only after a - /// live benchmark on the target server says it pays. + /// Ceiling of independent SSH connections one job may hold. Each lease is + /// a full connection. The former cap of 4 asked for "a live benchmark on + /// the target server" before being raised; the DAG engine review battery + /// on the Hetzner lab (wired gigabit, 5000 x 4 KiB files, 2026-09-08) is + /// that benchmark: at --parallel 16 rclone gained 1.9x on upload and 3.6x + /// on download over 4, while AeroFTP moved 1% because the cap silently + /// bound the flag. 16 matches rclone's range and the clamp already used by + /// --sftp-concurrency; the effective count stays min(ceiling, --parallel), + /// so the default of 4 connections is unchanged. fn transfer_executor_max_sessions(&self) -> u16 { - 4 + SFTP_POOL_MAX_SESSIONS } /// Produce an independent transfer worker. It is **not connected**: @@ -3970,6 +3979,30 @@ mod tests { assert!(worker.supports_transfer_worker_reuse()); } + #[test] + fn sftp_pool_ceiling_matches_the_intra_file_stream_range() { + // --parallel is documented up to 32 and clamped per provider; the SFTP + // ceiling follows the same 16 as --sftp-concurrency, so a request of + // 16 is honoured and a request of 4 still yields 4 connections. + let config = SftpConfig { + host: "example.com".to_string(), + port: 22, + username: "testuser".to_string(), + password: Some(secrecy::SecretString::from("testpass".to_string())), + private_key_path: None, + key_passphrase: None, + initial_path: None, + timeout_secs: 30, + trust_unknown_hosts: false, + }; + let provider = SftpProvider::new(config); + assert_eq!(provider.transfer_executor_max_sessions(), 16); + assert_eq!( + provider.transfer_executor_max_sessions() as usize, + SFTP_MULTI_THREAD_MAX_STREAMS + ); + } + #[test] fn sftp_workers_opt_into_warm_reuse_like_ftp() { // PD-FTP-2 pool semantics: a worker is recycled only when the provider diff --git a/src-tauri/src/transfer_dag_batch.rs b/src-tauri/src/transfer_dag_batch.rs index d34528705..bd0003841 100644 --- a/src-tauri/src/transfer_dag_batch.rs +++ b/src-tauri/src/transfer_dag_batch.rs @@ -3131,4 +3131,246 @@ mod tests { ); assert!(summary.metrics.slot_peak >= 1); } + + // ---- The provider ceiling is delivered, not only declared --------------- + + /// A pool-backed provider that counts the sessions actually in flight. + /// Every `clone_for_transfer` is a session; `download` holds its session + /// at a rendezvous until `parties` of them are open at once, so a batch + /// finishes only if the executor really runs that many in parallel. + struct SessionCountingProvider { + ceiling: u16, + in_flight: Arc, + peak: Arc, + rendezvous: Arc, + } + + impl SessionCountingProvider { + fn new(ceiling: u16, parties: usize) -> Self { + Self { + ceiling, + in_flight: Arc::new(AtomicUsize::new(0)), + peak: Arc::new(AtomicUsize::new(0)), + rendezvous: Arc::new(Barrier::new(parties)), + } + } + } + + #[async_trait] + impl crate::providers::StorageProvider for SessionCountingProvider { + fn as_any_mut(&mut self) -> &mut dyn std::any::Any { + self + } + fn provider_type(&self) -> crate::providers::ProviderType { + crate::providers::ProviderType::Sftp + } + fn display_name(&self) -> String { + "session-counting".to_string() + } + fn transfer_executor_kind(&self) -> crate::providers::ProviderTransferExecutorKind { + crate::providers::ProviderTransferExecutorKind::SftpConnectionPool + } + fn transfer_executor_max_sessions(&self) -> u16 { + self.ceiling + } + fn clone_for_transfer( + &self, + ) -> Result, crate::providers::ProviderError> + { + Ok(Box::new(Self { + ceiling: self.ceiling, + in_flight: Arc::clone(&self.in_flight), + peak: Arc::clone(&self.peak), + rendezvous: Arc::clone(&self.rendezvous), + })) + } + async fn connect(&mut self) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn disconnect(&mut self) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + fn is_connected(&self) -> bool { + true + } + async fn list( + &mut self, + _path: &str, + ) -> Result, crate::providers::ProviderError> { + Ok(Vec::new()) + } + async fn pwd(&mut self) -> Result { + Ok("/".to_string()) + } + async fn cd(&mut self, _path: &str) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn cd_up(&mut self) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn download( + &mut self, + _remote_path: &str, + local_path: &str, + _progress: Option>, + ) -> Result<(), crate::providers::ProviderError> { + let now = self.in_flight.fetch_add(1, AtomicOrdering::SeqCst) + 1; + self.peak.fetch_max(now, AtomicOrdering::SeqCst); + self.rendezvous.wait().await; + self.in_flight.fetch_sub(1, AtomicOrdering::SeqCst); + std::fs::write(local_path, b"ok").map_err(crate::providers::ProviderError::IoError) + } + async fn download_to_bytes( + &mut self, + _remote_path: &str, + ) -> Result, crate::providers::ProviderError> { + Ok(Vec::new()) + } + async fn upload( + &mut self, + _local_path: &str, + _remote_path: &str, + _progress: Option>, + ) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn mkdir(&mut self, _path: &str) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn delete(&mut self, _path: &str) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn rmdir(&mut self, _path: &str) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn rmdir_recursive( + &mut self, + _path: &str, + ) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn rename( + &mut self, + _from: &str, + _to: &str, + ) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn stat( + &mut self, + _path: &str, + ) -> Result { + Err(crate::providers::ProviderError::NotFound(_path.to_string())) + } + async fn size(&mut self, _path: &str) -> Result { + Ok(2) + } + async fn exists(&mut self, _path: &str) -> Result { + Ok(false) + } + async fn keep_alive(&mut self) -> Result<(), crate::providers::ProviderError> { + Ok(()) + } + async fn server_info(&mut self) -> Result { + Ok("mock".to_string()) + } + } + + /// Run a download batch of `files` entries at `requested` parallelism + /// through the real provider executor, resolved the way the CLI and the + /// GUI resolve it (capabilities from the provider, session model from the + /// capabilities), and return the peak number of sessions open at once. + async fn peak_sessions_delivered(ceiling: u16, requested: u32, files: usize) -> usize { + use crate::provider_transfer_executor::{ + resolve_provider_executor_runtime, ProviderDownloadExecutor, + }; + use crate::transfer_settings::{ + resolve_transfer_settings_for_capabilities, TransferSettingsInput, + }; + let dir = tempfile::tempdir().expect("tempdir"); + let counting = SessionCountingProvider::new(ceiling, files); + let peak = Arc::clone(&counting.peak); + let provider: Arc>>> = + Arc::new(Mutex::new(Some(Box::new(counting)))); + let (model, caps) = resolve_provider_executor_runtime(&provider, requested as usize).await; + let settings = resolve_transfer_settings_for_capabilities( + TransferSettingsInput { + max_concurrent: Some(requested), + ..TransferSettingsInput::default() + }, + &caps, + ); + let executor = Arc::new(ProviderDownloadExecutor::new( + Arc::new(CountingSink::default()), + provider, + settings, + tokio_util::sync::CancellationToken::new(), + model, + caps, + )); + let entries = (0..files) + .map(|i| { + entry_with_local( + &format!("f{i}"), + 2, + dir.path().join(format!("f{i}")).to_str().unwrap(), + ) + }) + .collect(); + let outcome = tokio::time::timeout( + Duration::from_secs(20), + execute_batch_dag_with_aimd( + Arc::new(CountingSink::default()) as Arc, + batch(entries, requested), + executor, + Arc::new(AtomicBool::new(false)), + None, + concurrency_proof_aimd(requested as usize), + ), + ) + .await; + match outcome { + Ok(result) => { + assert_eq!(result.completed, files as u32, "every file must land"); + peak.load(AtomicOrdering::SeqCst) + } + // The rendezvous never filled: fewer sessions than `files` were + // ever open at once. Report what was delivered instead of hanging. + Err(_) => peak.load(AtomicOrdering::SeqCst), + } + } + + /// The defect was not a ceiling that could not be raised: it was + /// `--parallel 8` reaching the pool as 4 with nothing said. This watches + /// the door: with the SFTP ceiling the provider advertises, 8 requested + /// files are open on 8 sessions at once, and a request above the ceiling + /// is delivered at the ceiling. + #[tokio::test] + async fn sftp_pool_delivers_the_requested_parallelism_up_to_the_provider_ceiling() { + use crate::providers::sftp::SftpProvider; + use crate::providers::{SftpConfig, StorageProvider}; + let sftp = SftpProvider::new(SftpConfig { + host: "example.com".to_string(), + port: 22, + username: "user".to_string(), + password: Some(secrecy::SecretString::from("pass".to_string())), + private_key_path: None, + key_passphrase: None, + initial_path: None, + timeout_secs: 30, + trust_unknown_hosts: false, + }); + let ceiling = sftp.transfer_executor_max_sessions(); + + assert_eq!( + peak_sessions_delivered(ceiling, 8, 8).await, + 8, + "--parallel 8 must open 8 sessions under the SFTP ceiling" + ); + assert_eq!( + peak_sessions_delivered(ceiling, 32, ceiling as usize).await, + ceiling as usize, + "a request above the ceiling is delivered at the ceiling" + ); + } }