diff --git a/CHANGELOG.md b/CHANGELOG.md index 4979c15c..89a2f366 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Performance +- **Roughly double bssh-server single-connection SFTP write throughput by fixing channel fragmentation and the serial write path** (#187). Three independent bottlenecks identified in the issue's code audit are addressed. First, `build_russh_config` now advertises a 65535-byte `maximum_packet_size` (the russh cap; library default 32768) and an 8 MiB `window_size` (library default 2 MiB), both configurable via `server.maximum_packet_size` / `server.window_size` (env: `BSSH_MAX_PACKET_SIZE` / `BSSH_WINDOW_SIZE`), so a 256 KiB SFTP write is no longer chopped into 8 CHANNEL_DATA packets that each pay the full cipher/copy/scheduling cost. Second, the SFTP server loop in `bssh-russh-sftp` is restructured from a strict read-process-write-flush cycle into a reader task with a bounded read-ahead queue plus an in-order processor: responses are flushed once per burst instead of once per request, and consecutive queued `SSH_FXP_WRITE` requests to the same handle at sequential offsets are coalesced into a single handler call (bounded by `max_write_coalesce_len`, default 256 KiB) while every merged request id still receives its own status reply, preserving SFTP response ordering and error semantics. The bssh write/read handlers additionally track each open file's cursor and elide the per-chunk `seek` for sequential transfers (append handles are exempt since O_APPEND ignores the cursor). Third, the workspace gains a tuned `[profile.release]` (`lto = "fat"`, `codegen-units = 1`) and documented `RUSTFLAGS` target-CPU guidance for self-builds. Local before/after benchmark (loopback, OpenSSH sftp client, 1 GiB random file, 3-run averages, 20-core Linux box): upload 71 MiB/s to 437 MiB/s (6.2x), download 73 MiB/s to 282 MiB/s (3.9x); an ablation with the code changes but library-default channel sizing lands at 333 / 291 MiB/s, attributing most of the gain to the pipelined write path and the rest of the upload gain to the larger packets. Numbers on the reporter's NIC-limited Xeon will differ; the methodology is documented in the PR. Remaining plan items (per-packet copies inside russh, AES-CTR internals) live in upstream russh since #212 removed the fork and are out of scope here. + ### Added - **Make server-side SSH compression configurable instead of hard-disabled** (#220). The #215 workaround forced the server to advertise only `none` compression for every deployment, with no way for operators to opt back in. A new `server.compression` option (YAML `server.compression: true`, env `BSSH_COMPRESSION`, builder `ServerConfigBuilder::compression`) now controls whether `build_russh_config` advertises russh's default compression list (`none`, `zlib`, `zlib@openssh.com`) or only `none`. The default stays off, so behavior is unchanged unless explicitly enabled, and enabling it logs a warning: russh's delayed-zlib (`zlib@openssh.com`) transport still desyncs a few packets after compression activates post-auth (reproduced on russh 0.61.1 and 0.62.1), dropping clients that negotiate it mid-session. Both settings are covered by unit tests, and the option plus the desync caveat are documented in the man page and config docs. diff --git a/Cargo.toml b/Cargo.toml index 79e75e73..cf962f3e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -104,3 +104,14 @@ path = "src/bin/bssh_server.rs" name = "bssh-keygen" path = "src/bin/bssh_keygen.rs" + +# Tuned release profile (issue #187): fat LTO plus a single codegen unit let +# the compiler inline across the russh framing / cipher / SFTP hot path, +# which per-packet overhead makes worthwhile for bulk SFTP/SCP transfers. +# For an additional boost on known deployment hardware, build with a pinned +# target CPU, e.g.: +# RUSTFLAGS="-C target-cpu=x86-64-v3" cargo build --release # AVX2-class servers +# RUSTFLAGS="-C target-cpu=native" cargo build --release # this machine only +[profile.release] +lto = "fat" +codegen-units = 1 diff --git a/README.md b/README.md index 3c1327c5..327a32d1 100644 --- a/README.md +++ b/README.md @@ -1446,6 +1446,16 @@ Read [ARCHITECTURE](ARCHITECTURE.md) documentation for more information. cargo build ``` +Release builds use a tuned profile (`lto = "fat"`, `codegen-units = 1`) so the SSH framing, cipher, and SFTP hot paths inline across crates. For maximum throughput on known hardware, additionally pin the target CPU: + +```bash +# Portable AVX2-class server build (Haswell/Zen and newer) +RUSTFLAGS="-C target-cpu=x86-64-v3" cargo build --release + +# Optimized only for the build machine itself +RUSTFLAGS="-C target-cpu=native" cargo build --release +``` + ### Testing ```bash cargo test diff --git a/crates/bssh-russh-sftp/src/server/mod.rs b/crates/bssh-russh-sftp/src/server/mod.rs index a3e5a9ba..f7557040 100644 --- a/crates/bssh-russh-sftp/src/server/mod.rs +++ b/crates/bssh-russh-sftp/src/server/mod.rs @@ -3,6 +3,7 @@ mod reply; use bytes::Bytes; use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use tokio::sync::mpsc; pub use self::handler::Handler; pub use self::reply::StatusReply; @@ -35,12 +36,30 @@ macro_rules! into_wrap { pub struct Config { /// Maximum allowed size of SFTP packets sent by clients. Default: 256 KiB. pub max_client_packet_len: u32, + + /// Maximum number of client requests read ahead of the one currently + /// being processed. Read-ahead lets the transport keep delivering (and + /// decrypting) requests while the handler is blocked on file I/O, and it + /// feeds the sequential-write coalescer. Bounded so a client cannot make + /// the server buffer unlimited data; the worst-case buffered bytes are + /// `max_read_ahead * max_client_packet_len` per session. Default: 16. + pub max_read_ahead: usize, + + /// Maximum number of bytes merged into a single coalesced `SSH_FXP_WRITE` + /// handler call. Consecutive queued WRITE requests targeting the same + /// handle at strictly sequential offsets are merged into one handler + /// invocation (one seek + one write instead of one per request), and every + /// merged request id still receives its own status reply. Set to 0 to + /// disable coalescing. Default: 256 KiB. + pub max_write_coalesce_len: usize, } impl Default for Config { fn default() -> Self { Self { max_client_packet_len: 262144, + max_read_ahead: 16, + max_write_coalesce_len: 262144, } } } @@ -76,23 +95,191 @@ where } } -async fn process_handler(stream: &mut S, handler: &mut H, cfg: &Config) -> Result<(), Error> +/// A client packet after framing and decoding, as seen by the processor loop. +enum Queued { + /// A well-formed request. + Request(Packet), + /// A frame that could not be decoded; answered with `SSH_FXP_STATUS` + /// `BadMessage` (id 0), matching the previous serial-loop behavior. + Malformed, + /// The reader failed to obtain a frame (I/O error or EOF). + ReadError(Error), +} + +fn decode(item: Result) -> Queued { + match item { + Ok(mut bytes) => match Packet::try_from(&mut bytes) { + Ok(packet) => Queued::Request(packet), + Err(_) => Queued::Malformed, + }, + Err(err) => Queued::ReadError(err), + } +} + +/// Encode and send one response packet without flushing. Flushing is deferred +/// to the moment the request queue runs empty so a burst of pipelined +/// requests is answered with one flush instead of one per request. +async fn send_response(writer: &mut W, response: Packet) -> Result<(), Error> where + W: AsyncWrite + Unpin, +{ + let bytes = Bytes::try_from(response)?; + writer.write_all(&bytes).await?; + Ok(()) +} + +/// Drive one SFTP session over `stream` until EOF. +/// +/// Architecture: a reader task frames client packets and feeds a bounded +/// queue (`Config::max_read_ahead`), so the transport keeps delivering +/// requests while the handler is busy with file I/O. The processor loop +/// consumes the queue strictly in order: requests are handled one at a time +/// against `&mut handler` and responses are written in request order, so +/// response ordering and error semantics are identical to the previous +/// serial read -> process -> write -> flush loop. Two optimizations apply on +/// top: +/// +/// - **Deferred flush**: responses are flushed only when the queue is +/// momentarily empty (or the session ends) instead of after every request. +/// - **Sequential write coalescing**: consecutive queued `SSH_FXP_WRITE` +/// requests for the same handle at strictly contiguous offsets are merged +/// into a single handler call (bounded by +/// `Config::max_write_coalesce_len`). Every merged request id receives its +/// own status reply carrying the outcome of the merged write; on failure +/// all merged ids receive the same error, which is the conservative +/// superset of what a partially-failed serial sequence would report. +async fn process_stream(stream: S, handler: &mut H, cfg: &Config) +where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, H: Handler + Send, - S: AsyncRead + AsyncWrite + Unpin, { - let mut bytes = read_packet(stream, cfg.max_client_packet_len).await?; + let (mut read_half, mut write_half) = tokio::io::split(stream); - let response = match Packet::try_from(&mut bytes) { - Ok(request) => process_request(request, handler).await, - Err(_) => Packet::error(0, StatusCode::BadMessage), - }; + let (tx, mut rx) = mpsc::channel::>(cfg.max_read_ahead.max(1)); + let max_packet_len = cfg.max_client_packet_len; + let reader = tokio::spawn(async move { + loop { + let item = read_packet(&mut read_half, max_packet_len).await; + // Stop on EOF; keep reading after other errors to preserve the + // previous loop's behavior (it warned and retried). + let stop = matches!(item, Err(Error::UnexpectedEof)); + if tx.send(item).await.is_err() || stop { + break; + } + } + }); - let packet = Bytes::try_from(response)?; - stream.write_all(&packet).await?; - stream.flush().await?; + // Holds a packet dequeued by the coalescer that did not merge into the + // current write; it must be processed next to preserve ordering. + let mut pending: Option = None; - Ok(()) + 'session: loop { + let queued = match pending.take() { + Some(queued) => queued, + None => match rx.try_recv() { + Ok(item) => decode(item), + Err(mpsc::error::TryRecvError::Empty) => { + // No request ready: flush buffered responses before + // blocking so the client is never left waiting on + // replies we already produced. + if let Err(err) = write_half.flush().await { + warn!("sftp: flush failed: {err}"); + break 'session; + } + match rx.recv().await { + Some(item) => decode(item), + None => break 'session, + } + } + Err(mpsc::error::TryRecvError::Disconnected) => break 'session, + }, + }; + + match queued { + Queued::ReadError(Error::UnexpectedEof) => break 'session, + Queued::ReadError(err) => { + warn!("{}", err); + } + Queued::Malformed => { + if let Err(err) = + send_response(&mut write_half, Packet::error(0, StatusCode::BadMessage)).await + { + warn!("{}", err); + } + } + Queued::Request(Packet::Write(mut write)) => { + // Coalesce strictly sequential queued writes to the same + // handle into one handler call. + let mut ids = vec![write.id]; + while write.data.len() < cfg.max_write_coalesce_len { + let Ok(item) = rx.try_recv() else { + // Empty or disconnected: nothing more to merge now. + // A disconnect is surfaced by the next dequeue. + break; + }; + match decode(item) { + Queued::Request(Packet::Write(next)) + if next.handle == write.handle + && write.offset.checked_add(write.data.len() as u64) + == Some(next.offset) + && write.data.len() + next.data.len() + <= cfg.max_write_coalesce_len => + { + ids.push(next.id); + write.data.extend_from_slice(&next.data); + } + other => { + pending = Some(other); + break; + } + } + } + + let reply: StatusReply = match handler + .write(write.id, write.handle, write.offset, write.data) + .await + { + Ok(status) => StatusReply { + status_code: status.status_code, + error_message: Some(status.error_message), + language_tag: Some(status.language_tag), + }, + Err(err) => err.into(), + }; + + for id in ids { + let response = Packet::Status(Status { + id, + status_code: reply.status_code, + error_message: reply + .error_message + .clone() + .unwrap_or_else(|| reply.status_code.to_string()), + language_tag: reply + .language_tag + .clone() + .unwrap_or_else(|| "en-US".to_string()), + }); + if let Err(err) = send_response(&mut write_half, response).await { + warn!("{}", err); + } + } + } + Queued::Request(request) => { + let response = process_request(request, handler).await; + if let Err(err) = send_response(&mut write_half, response).await { + warn!("{}", err); + } + } + } + } + + if let Err(err) = write_half.flush().await { + debug!("sftp: final flush failed: {err}"); + } + reader.abort(); + + debug!("sftp stream ended"); } /// Run processing stream as SFTP @@ -105,20 +292,406 @@ where } /// Run processing stream as SFTP with custom configuration -pub async fn run_with_config(mut stream: S, mut handler: H, cfg: Config) +pub async fn run_with_config(stream: S, mut handler: H, cfg: Config) where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, H: Handler + Send + 'static, { tokio::spawn(async move { - loop { - match process_handler(&mut stream, &mut handler, &cfg).await { - Err(Error::UnexpectedEof) => break, - Err(err) => warn!("{}", err), - Ok(_) => (), + process_stream(stream, &mut handler, &cfg).await; + }); +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use bytes::{Buf, BytesMut}; + use tokio::io::AsyncReadExt; + + use super::*; + use crate::protocol::{Handle, OpenFlags, Write}; + + /// Call log entry for the mock handler's write method. + #[derive(Debug, Clone, PartialEq)] + struct WriteCall { + handle: String, + offset: u64, + len: usize, + } + + /// State shared between the test body and the mock handler. + #[derive(Debug, Default)] + struct Shared { + /// Sparse file image keyed by handle. + files: HashMap>, + /// Every write call the handler received, in order. + write_calls: Vec, + /// Offsets whose writes must fail with `StatusCode::Failure`. + fail_offsets: Vec, + } + + /// Mock handler. `open` sleeps 50 ms before replying, which gives the + /// reader task ample time to enqueue every already-sent request. Tests + /// exploit this: sending OPEN followed by a burst of WRITEs makes the + /// queue state during write processing deterministic, so coalescing + /// expectations can be exact instead of timing-tolerant. + #[derive(Debug, Clone, Default)] + struct MockHandler { + shared: Arc>, + } + + impl Handler for MockHandler { + type Error = StatusCode; + + fn unimplemented(&self) -> Self::Error { + StatusCode::OpUnsupported + } + + fn open( + &mut self, + id: u32, + filename: String, + _pflags: OpenFlags, + _attrs: crate::protocol::FileAttributes, + ) -> impl std::future::Future> + Send { + let shared = Arc::clone(&self.shared); + async move { + // Let the reader task queue all pipelined requests sent + // after this OPEN before the processor resumes. + tokio::time::sleep(Duration::from_millis(50)).await; + shared + .lock() + .unwrap() + .files + .insert(filename.clone(), Vec::new()); + Ok(Handle { + id, + handle: filename, + }) } } - debug!("sftp stream ended"); - }); + fn write( + &mut self, + id: u32, + handle: String, + offset: u64, + data: Vec, + ) -> impl std::future::Future> + Send { + let shared = Arc::clone(&self.shared); + async move { + let mut shared = shared.lock().unwrap(); + if shared.fail_offsets.contains(&offset) { + return Err(StatusCode::Failure); + } + shared.write_calls.push(WriteCall { + handle: handle.clone(), + offset, + len: data.len(), + }); + let file = shared.files.entry(handle).or_default(); + let end = offset as usize + data.len(); + if file.len() < end { + file.resize(end, 0); + } + file[offset as usize..end].copy_from_slice(&data); + Ok(Status { + id, + status_code: StatusCode::Ok, + error_message: String::new(), + language_tag: "en".to_string(), + }) + } + } + + async fn close(&mut self, id: u32, _handle: String) -> Result { + Ok(Status { + id, + status_code: StatusCode::Ok, + error_message: String::new(), + language_tag: "en".to_string(), + }) + } + } + + fn encode(packet: Packet) -> Bytes { + Bytes::try_from(packet).expect("packet must encode") + } + + /// OPEN request used as a queue-priming barrier (see [`MockHandler`]). + fn open_packet(id: u32, filename: &str) -> Bytes { + encode(Packet::Open(crate::protocol::Open { + id, + filename: filename.to_string(), + pflags: OpenFlags::WRITE, + attrs: crate::protocol::FileAttributes::default(), + })) + } + + fn write_packet(id: u32, handle: &str, offset: u64, data: Vec) -> Bytes { + encode(Packet::Write(Write { + id, + handle: handle.to_string(), + offset, + data, + })) + } + + /// Feed `requests` into a session and collect one decoded response per + /// request. + async fn run_session( + requests: Vec, + expected_responses: usize, + shared: Arc>, + cfg: Config, + ) -> Vec { + let (client, server) = tokio::io::duplex(1 << 20); + + let session = tokio::spawn(async move { + let mut handler = MockHandler { shared }; + process_stream(server, &mut handler, &cfg).await; + }); + + let (mut client_rd, mut client_wr) = tokio::io::split(client); + for request in &requests { + client_wr.write_all(request).await.unwrap(); + } + client_wr.shutdown().await.unwrap(); + + let mut responses = Vec::new(); + let mut buf = BytesMut::new(); + while responses.len() < expected_responses { + let mut chunk = [0u8; 4096]; + let n = tokio::time::timeout(Duration::from_secs(5), client_rd.read(&mut chunk)) + .await + .expect("timed out waiting for responses") + .expect("read must succeed"); + assert!(n > 0, "stream closed before all responses arrived"); + buf.extend_from_slice(&chunk[..n]); + + loop { + if buf.len() < 4 { + break; + } + let length = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize; + if buf.len() < 4 + length { + break; + } + buf.advance(4); + let mut frame = buf.split_to(length).freeze(); + responses.push(Packet::try_from(&mut frame).expect("response must decode")); + } + } + + session.await.unwrap(); + responses + } + + fn status_of(packet: &Packet) -> (u32, StatusCode) { + match packet { + Packet::Status(status) => (status.id, status.status_code), + other => panic!("expected status packet, got {other:?}"), + } + } + + fn response_id(packet: &Packet) -> u32 { + match packet { + Packet::Status(status) => status.id, + Packet::Handle(handle) => handle.id, + other => panic!("unexpected response packet: {other:?}"), + } + } + + #[tokio::test] + async fn sequential_writes_coalesce_into_one_handler_call() { + let shared = Arc::new(Mutex::new(Shared::default())); + let payload: Vec = (0..128u32).flat_map(|i| i.to_be_bytes()).collect(); + let chunk = payload.len() / 4; + + // OPEN primes the queue: while its handler sleeps, the reader + // enqueues all four WRITEs, so they coalesce into one handler call. + let mut requests = vec![open_packet(10, "h")]; + for (index, part) in payload.chunks(chunk).enumerate() { + requests.push(write_packet( + index as u32 + 1, + "h", + (index * chunk) as u64, + part.to_vec(), + )); + } + requests.push(encode(Packet::Close(crate::protocol::Close { + id: 99, + handle: "h".to_string(), + }))); + + let responses = run_session(requests, 6, Arc::clone(&shared), Config::default()).await; + + // Every request id gets a reply, in request order. + let ids: Vec = responses.iter().map(response_id).collect(); + assert_eq!(ids, vec![10, 1, 2, 3, 4, 99]); + for response in &responses[1..] { + assert_eq!(status_of(response).1, StatusCode::Ok); + } + + let shared = shared.lock().unwrap(); + // Coalescing must not change the file image. + assert_eq!(shared.files.get("h"), Some(&payload)); + // All four queued sequential writes merge into a single call. + assert_eq!( + shared.write_calls, + vec![WriteCall { + handle: "h".into(), + offset: 0, + len: payload.len() + }] + ); + } + + #[tokio::test] + async fn non_contiguous_writes_are_not_merged() { + let shared = Arc::new(Mutex::new(Shared::default())); + // Two queued writes with a hole between them must stay two calls. + let requests = vec![ + open_packet(10, "h"), + write_packet(1, "h", 0, vec![0xAA; 16]), + write_packet(2, "h", 64, vec![0xBB; 16]), + ]; + + let responses = run_session(requests, 3, Arc::clone(&shared), Config::default()).await; + assert_eq!( + responses[1..].iter().map(status_of).collect::>(), + vec![(1, StatusCode::Ok), (2, StatusCode::Ok)] + ); + + let shared = shared.lock().unwrap(); + assert_eq!( + shared.write_calls, + vec![ + WriteCall { + handle: "h".into(), + offset: 0, + len: 16 + }, + WriteCall { + handle: "h".into(), + offset: 64, + len: 16 + }, + ] + ); + } + + #[tokio::test] + async fn different_handles_are_not_merged() { + let shared = Arc::new(Mutex::new(Shared::default())); + let requests = vec![ + open_packet(10, "a"), + write_packet(1, "a", 0, vec![0xAA; 16]), + write_packet(2, "b", 16, vec![0xBB; 16]), + ]; + + let responses = run_session(requests, 3, Arc::clone(&shared), Config::default()).await; + assert_eq!( + responses[1..].iter().map(status_of).collect::>(), + vec![(1, StatusCode::Ok), (2, StatusCode::Ok)] + ); + + let shared = shared.lock().unwrap(); + assert_eq!(shared.write_calls.len(), 2); + assert_eq!(shared.write_calls[0].handle, "a"); + assert_eq!(shared.write_calls[1].handle, "b"); + } + + #[tokio::test] + async fn coalesce_respects_byte_budget() { + let shared = Arc::new(Mutex::new(Shared::default())); + let cfg = Config { + max_write_coalesce_len: 32, + ..Config::default() + }; + // Three queued sequential 16-byte writes with a 32-byte budget merge + // as [32, 16]. + let requests = vec![ + open_packet(10, "h"), + write_packet(1, "h", 0, vec![1; 16]), + write_packet(2, "h", 16, vec![2; 16]), + write_packet(3, "h", 32, vec![3; 16]), + ]; + + let responses = run_session(requests, 4, Arc::clone(&shared), cfg).await; + for (index, response) in responses[1..].iter().enumerate() { + assert_eq!(status_of(response), (index as u32 + 1, StatusCode::Ok)); + } + + let shared = shared.lock().unwrap(); + assert_eq!( + shared.write_calls, + vec![ + WriteCall { + handle: "h".into(), + offset: 0, + len: 32 + }, + WriteCall { + handle: "h".into(), + offset: 32, + len: 16 + }, + ] + ); + let mut expected = vec![1u8; 16]; + expected.extend_from_slice(&[2; 16]); + expected.extend_from_slice(&[3; 16]); + assert_eq!(shared.files.get("h"), Some(&expected)); + } + + #[tokio::test] + async fn merged_write_failure_reports_error_to_every_merged_id() { + let shared = Arc::new(Mutex::new(Shared { + fail_offsets: vec![0], + ..Shared::default() + })); + let requests = vec![ + open_packet(10, "h"), + write_packet(1, "h", 0, vec![1; 16]), + write_packet(2, "h", 16, vec![2; 16]), + ]; + + let responses = run_session(requests, 3, Arc::clone(&shared), Config::default()).await; + for (index, response) in responses[1..].iter().enumerate() { + let (id, code) = status_of(response); + assert_eq!(id, index as u32 + 1); + assert_eq!( + code, + StatusCode::Failure, + "every merged id must observe the write failure" + ); + } + + let shared = shared.lock().unwrap(); + assert!(shared.write_calls.is_empty()); + } + + #[tokio::test] + async fn coalescing_disabled_with_zero_budget() { + let shared = Arc::new(Mutex::new(Shared::default())); + let cfg = Config { + max_write_coalesce_len: 0, + ..Config::default() + }; + let requests = vec![ + open_packet(10, "h"), + write_packet(1, "h", 0, vec![1; 16]), + write_packet(2, "h", 16, vec![2; 16]), + ]; + + let responses = run_session(requests, 3, Arc::clone(&shared), cfg).await; + assert_eq!(responses.len(), 3); + + let shared = shared.lock().unwrap(); + assert_eq!(shared.write_calls.len(), 2, "no merging with zero budget"); + } } diff --git a/docs/man/bssh-server.8 b/docs/man/bssh-server.8 index be428727..9fe2c804 100644 --- a/docs/man/bssh-server.8 +++ b/docs/man/bssh-server.8 @@ -150,6 +150,13 @@ Keepalive interval in seconds (default: 60) Advertise SSH transport compression, "true" or "false" (default: false). See the \fBserver\fR configuration section for the caveat about enabling it. .TP +.B BSSH_MAX_PACKET_SIZE +Maximum SSH channel packet size in bytes (default: 65535; clamped to +4096-65535) +.TP +.B BSSH_WINDOW_SIZE +SSH channel flow-control window size in bytes (default: 8388608) +.TP .B BSSH_AUTH_METHODS Comma-separated authentication methods (publickey,password) .TP @@ -169,7 +176,14 @@ Command timeout in seconds (default: 3600) .TP .B server Network and connection settings (bind_address, port, host_keys, -max_connections, timeout, keepalive_interval, compression). +max_connections, timeout, keepalive_interval, compression, +maximum_packet_size, window_size). +\fBmaximum_packet_size\fR (default 65535) and \fBwindow_size\fR (default +8388608) size the SSH channel: larger packets amortize the per-packet +cipher and copy overhead of the transport, so bulk SFTP/SCP transfers are +markedly faster than with the library defaults (32768 / 2097152). +\fBmaximum_packet_size\fR is clamped to 4096-65535 and \fBwindow_size\fR is +raised to at least one packet. \fBcompression\fR controls whether the server advertises SSH transport compression (\fBzlib\fR, \fBzlib@openssh.com\fR). It defaults to \fBfalse\fR, advertising only \fBnone\fR so clients use the uncompressed transport, diff --git a/src/server/config/loader.rs b/src/server/config/loader.rs index 4a3f99f2..458979b5 100644 --- a/src/server/config/loader.rs +++ b/src/server/config/loader.rs @@ -56,6 +56,8 @@ use std::path::{Path, PathBuf}; /// - `BSSH_MAX_CONNECTIONS` - Maximum concurrent connections /// - `BSSH_KEEPALIVE_INTERVAL` - Keepalive interval in seconds /// - `BSSH_COMPRESSION` - Advertise SSH transport compression ("true"/"false"; default false, see issue #215) +/// - `BSSH_MAX_PACKET_SIZE` - Maximum SSH channel packet size in bytes (default 65535, clamped to at most 65535) +/// - `BSSH_WINDOW_SIZE` - SSH channel flow-control window size in bytes (default 8388608) /// - `BSSH_AUTH_METHODS` - Comma-separated auth methods (e.g., "publickey,password") /// - `BSSH_AUTHORIZED_KEYS_DIR` - Directory for authorized_keys files /// - `BSSH_AUTHORIZED_KEYS_PATTERN` - Pattern for authorized_keys paths @@ -263,6 +265,28 @@ fn apply_env_overrides(mut config: ServerFileConfig) -> Result ); } + // BSSH_MAX_PACKET_SIZE + if let Ok(size_str) = std::env::var("BSSH_MAX_PACKET_SIZE") { + config.server.maximum_packet_size = size_str + .parse() + .context(format!("Invalid BSSH_MAX_PACKET_SIZE value: {size_str}"))?; + tracing::debug!( + maximum_packet_size = config.server.maximum_packet_size, + "Applied BSSH_MAX_PACKET_SIZE override" + ); + } + + // BSSH_WINDOW_SIZE + if let Ok(size_str) = std::env::var("BSSH_WINDOW_SIZE") { + config.server.window_size = size_str + .parse() + .context(format!("Invalid BSSH_WINDOW_SIZE value: {size_str}"))?; + tracing::debug!( + window_size = config.server.window_size, + "Applied BSSH_WINDOW_SIZE override" + ); + } + // BSSH_AUTH_METHODS (comma-separated: "publickey,password") if let Ok(methods_str) = std::env::var("BSSH_AUTH_METHODS") { use super::types::AuthMethod; @@ -501,6 +525,26 @@ auth: assert!(result.is_err()); } + #[test] + #[serial_test::serial] + fn test_env_override_channel_sizing() { + let _port = EnvGuard::remove("BSSH_PORT"); + let _packet = EnvGuard::set("BSSH_MAX_PACKET_SIZE", "32768"); + let _window = EnvGuard::set("BSSH_WINDOW_SIZE", "4194304"); + let config = apply_env_overrides(ServerFileConfig::default()).unwrap(); + assert_eq!(config.server.maximum_packet_size, 32768); + assert_eq!(config.server.window_size, 4194304); + } + + #[test] + #[serial_test::serial] + fn test_env_override_channel_sizing_invalid() { + let _port = EnvGuard::remove("BSSH_PORT"); + let _packet = EnvGuard::set("BSSH_MAX_PACKET_SIZE", "not-a-number"); + let result = apply_env_overrides(ServerFileConfig::default()); + assert!(result.is_err()); + } + #[test] #[serial_test::serial] fn test_env_override_auth_methods() { diff --git a/src/server/config/mod.rs b/src/server/config/mod.rs index df6eed04..6f021cb5 100644 --- a/src/server/config/mod.rs +++ b/src/server/config/mod.rs @@ -230,6 +230,24 @@ pub struct ServerConfig { /// compression. #[serde(default)] pub compression: bool, + + /// Maximum SSH channel packet size in bytes advertised to clients. + /// + /// Larger packets amortize per-packet cipher and copy overhead; russh's + /// library default of 32768 fragments a 256 KiB SFTP write into 8 + /// CHANNEL_DATA packets (see + /// ). Clamped to at most + /// 65535, which russh requires. Default: 65535. + #[serde(default = "default_maximum_packet_size")] + pub maximum_packet_size: u32, + + /// SSH channel flow-control window size in bytes advertised to clients. + /// + /// Bounds in-flight client data per channel (and thus per-channel + /// buffering). Default: 8 MiB, four times the russh library default, so + /// bulk uploads do not stall on window-adjust round trips. + #[serde(default = "default_window_size")] + pub window_size: u32, } fn default_max_sessions_per_user() -> usize { @@ -310,6 +328,17 @@ fn default_true() -> bool { true } +fn default_maximum_packet_size() -> u32 { + // russh rejects channel packets larger than a TCP frame (65535); + // advertise the cap to minimize SFTP write fragmentation (issue #187). + 65535 +} + +fn default_window_size() -> u32 { + // 8 MiB: four times the russh default; see ServerSettings::window_size. + 8 * 1024 * 1024 +} + impl Default for ServerConfig { fn default() -> Self { Self { @@ -337,6 +366,8 @@ impl Default for ServerConfig { max_sessions_per_user: default_max_sessions_per_user(), session_timeout_secs: 0, compression: false, + maximum_packet_size: default_maximum_packet_size(), + window_size: default_window_size(), } } } @@ -643,6 +674,20 @@ impl ServerConfigBuilder { self } + /// Set the maximum SSH channel packet size in bytes. + /// + /// Values above 65535 are clamped when the russh config is built. + pub fn maximum_packet_size(mut self, size: u32) -> Self { + self.config.maximum_packet_size = size; + self + } + + /// Set the SSH channel flow-control window size in bytes. + pub fn window_size(mut self, size: u32) -> Self { + self.config.window_size = size; + self + } + /// Build the ServerConfig. pub fn build(self) -> ServerConfig { self.config @@ -714,6 +759,8 @@ impl ServerFileConfig { max_sessions_per_user: self.security.max_sessions_per_user, session_timeout_secs: self.security.session_timeout, compression: self.server.compression, + maximum_packet_size: self.server.maximum_packet_size, + window_size: self.server.window_size, } } } @@ -833,6 +880,22 @@ mod tests { assert!(server_config.compression); } + #[test] + fn channel_sizing_defaults_and_threads_into_server_config() { + // #187: tuned channel-sizing defaults, overridable via the file + // config, must propagate into ServerConfig. + let mut file_config = ServerFileConfig::default(); + file_config.server.host_keys = vec![PathBuf::from("/test/key")]; + assert_eq!(file_config.server.maximum_packet_size, 65535); + assert_eq!(file_config.server.window_size, 8 * 1024 * 1024); + + file_config.server.maximum_packet_size = 32768; + file_config.server.window_size = 4 * 1024 * 1024; + let server_config = file_config.into_server_config(); + assert_eq!(server_config.maximum_packet_size, 32768); + assert_eq!(server_config.window_size, 4 * 1024 * 1024); + } + #[test] fn test_config_new() { let config = ServerConfig::new(); diff --git a/src/server/config/types.rs b/src/server/config/types.rs index a66827d1..db993a22 100644 --- a/src/server/config/types.rs +++ b/src/server/config/types.rs @@ -140,6 +140,30 @@ pub struct ServerSettings { /// Default: false #[serde(default)] pub compression: bool, + + /// Maximum SSH channel packet size in bytes advertised to clients. + /// + /// Larger packets amortize the per-packet cipher, copy, and scheduling + /// overhead of the SSH transport: with the russh library default of + /// 32768, a 256 KiB SFTP write is fragmented into 8 CHANNEL_DATA packets, + /// roughly halving single-connection SFTP throughput on slower CPUs (see + /// ). Values above 65535 are + /// clamped because russh rejects packets larger than a TCP frame. + /// + /// Default: 65535 + #[serde(default = "default_maximum_packet_size")] + pub maximum_packet_size: u32, + + /// SSH channel flow-control window size in bytes advertised to clients. + /// + /// Bounds how much data a client may send on a channel before waiting + /// for a window adjustment, and therefore bounds the per-channel receive + /// buffering. The russh library default is 2 MiB; bssh-server defaults + /// higher to keep bulk SFTP/SCP uploads from stalling on window updates. + /// + /// Default: 8388608 (8 MiB) + #[serde(default = "default_window_size")] + pub window_size: u32, } /// Authentication configuration. @@ -605,6 +629,20 @@ fn default_keepalive() -> u64 { 60 } +fn default_maximum_packet_size() -> u32 { + // The russh transport rejects channel packets larger than a TCP frame + // (65535); advertise the cap so SFTP writes fragment as little as + // possible (issue #187). + 65535 +} + +fn default_window_size() -> u32 { + // 8 MiB: four times the russh default, sized so bulk uploads are not + // throttled by window-adjust round trips while keeping worst-case + // per-channel buffering bounded. + 8 * 1024 * 1024 +} + fn default_auth_methods() -> Vec { vec![AuthMethod::PublicKey] } @@ -653,6 +691,8 @@ impl Default for ServerSettings { timeout: default_timeout(), keepalive_interval: default_keepalive(), compression: false, + maximum_packet_size: default_maximum_packet_size(), + window_size: default_window_size(), } } } diff --git a/src/server/mod.rs b/src/server/mod.rs index a4b5197c..48d80dd2 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -218,6 +218,37 @@ impl BsshServer { } }; + // Channel sizing (issue #187): russh's library defaults + // (maximum_packet_size 32768, window_size 2 MiB) fragment a 256 KiB + // SFTP write into 8 CHANNEL_DATA packets, multiplying per-packet + // cipher and copy overhead. Advertise larger, configurable values. + // russh rejects channel packets above a TCP frame, so clamp there. + const RUSSH_MAX_PACKET_SIZE: u32 = 65535; + // Floor keeps a misconfiguration from advertising a packet size that + // cannot even carry an SFTP header round trip. + const MIN_PACKET_SIZE: u32 = 4096; + let maximum_packet_size = self + .config + .maximum_packet_size + .clamp(MIN_PACKET_SIZE, RUSSH_MAX_PACKET_SIZE); + if maximum_packet_size != self.config.maximum_packet_size { + tracing::warn!( + configured = self.config.maximum_packet_size, + effective = maximum_packet_size, + "maximum_packet_size out of range [{MIN_PACKET_SIZE}, {RUSSH_MAX_PACKET_SIZE}], clamped" + ); + } + // A window smaller than one packet would deadlock the channel before + // the first packet completes; keep it at least one packet wide. + let window_size = self.config.window_size.max(maximum_packet_size); + if window_size != self.config.window_size { + tracing::warn!( + configured = self.config.window_size, + effective = window_size, + "window_size smaller than maximum_packet_size, raised" + ); + } + Ok(russh::server::Config { keys, preferred, @@ -225,6 +256,8 @@ impl BsshServer { auth_rejection_time_initial: Some(Duration::from_secs(0)), max_auth_attempts: self.config.max_auth_attempts as usize, inactivity_timeout: self.config.idle_timeout(), + maximum_packet_size, + window_size, ..Default::default() }) } @@ -505,6 +538,81 @@ mod tests { ); } + #[test] + fn test_build_russh_config_channel_sizing_defaults() { + // #187: by default the server advertises the russh packet-size cap + // and an 8 MiB window instead of the library defaults (32768 / 2 MiB) + // so SFTP writes are not fragmented into 8 packets each. + let key = concat!( + env!("CARGO_MANIFEST_DIR"), + "/test_keys/ssh_host_ed25519_key" + ); + let config = ServerConfig::builder().host_key(key).build(); + let server = BsshServer::new(config); + + let russh_config = server + .build_russh_config() + .expect("config should build with a valid host key"); + assert_eq!(russh_config.maximum_packet_size, 65535); + assert_eq!(russh_config.window_size, 8 * 1024 * 1024); + } + + #[test] + fn test_build_russh_config_channel_sizing_clamped() { + // Out-of-range values are clamped: packet size to [4096, 65535] and + // the window to at least one packet. + let key = concat!( + env!("CARGO_MANIFEST_DIR"), + "/test_keys/ssh_host_ed25519_key" + ); + let config = ServerConfig::builder() + .host_key(key) + .maximum_packet_size(1_000_000) + .window_size(1024) + .build(); + let server = BsshServer::new(config); + + let russh_config = server + .build_russh_config() + .expect("config should build with a valid host key"); + assert_eq!(russh_config.maximum_packet_size, 65535); + assert_eq!( + russh_config.window_size, 65535, + "window must be raised to at least one packet" + ); + + let config = ServerConfig::builder() + .host_key(key) + .maximum_packet_size(16) + .build(); + let server = BsshServer::new(config); + let russh_config = server + .build_russh_config() + .expect("config should build with a valid host key"); + assert_eq!(russh_config.maximum_packet_size, 4096); + } + + #[test] + fn test_build_russh_config_channel_sizing_custom() { + // In-range custom values pass through unchanged. + let key = concat!( + env!("CARGO_MANIFEST_DIR"), + "/test_keys/ssh_host_ed25519_key" + ); + let config = ServerConfig::builder() + .host_key(key) + .maximum_packet_size(32768) + .window_size(2 * 1024 * 1024) + .build(); + let server = BsshServer::new(config); + + let russh_config = server + .build_russh_config() + .expect("config should build with a valid host key"); + assert_eq!(russh_config.maximum_packet_size, 32768); + assert_eq!(russh_config.window_size, 2 * 1024 * 1024); + } + #[tokio::test] async fn test_session_count() { let config = ServerConfig::builder().host_key("/nonexistent/key").build(); diff --git a/src/server/sftp.rs b/src/server/sftp.rs index 20cc3105..40f28086 100644 --- a/src/server/sftp.rs +++ b/src/server/sftp.rs @@ -153,8 +153,14 @@ enum OpenHandle { File { file: File, path: PathBuf, - #[allow(dead_code)] flags: OpenFlags, + /// Tracked logical file position, used to elide redundant seeks for + /// strictly sequential reads and writes (SFTP clients address every + /// transfer chunk by absolute offset, so bulk transfers would + /// otherwise pay one seek per chunk). `None` means the position is + /// unknown (append mode, or an interrupted operation) and the next + /// positioned operation must seek. + pos: Option, }, /// An open directory listing. Dir { @@ -668,13 +674,21 @@ impl russh_sftp::server::Handler for SftpHandler { let file = opts.open(&path).await?; - // Store the handle + // Store the handle. Append-mode files carry no tracked position: + // the kernel writes at EOF regardless of the cursor, so seek + // elision must not apply. + let pos = if pflags.contains(OpenFlags::APPEND) { + None + } else { + Some(0) + }; handles.lock().await.insert( handle_id.clone(), OpenHandle::File { file, path, flags: pflags, + pos, }, ); @@ -710,13 +724,20 @@ impl russh_sftp::server::Handler for SftpHandler { let mut handles_guard = handles.lock().await; let handle_entry = handles_guard.get_mut(&handle); - let file = match handle_entry { - Some(OpenHandle::File { file, .. }) => file, + let (file, pos) = match handle_entry { + Some(OpenHandle::File { file, pos, .. }) => (file, pos), _ => return Err(SftpError::invalid_handle()), }; - // Seek to offset - file.seek(SeekFrom::Start(offset)).await?; + // Seek only when the file cursor is not already at the requested + // offset (sequential bulk reads hit the fast path). Mark the + // position unknown while operations are in flight so an error + // path can never leave a stale tracked position behind. + let need_seek = *pos != Some(offset); + *pos = None; + if need_seek { + file.seek(SeekFrom::Start(offset)).await?; + } // Read data let mut buffer = vec![0u8; capped_len as usize]; @@ -726,6 +747,8 @@ impl russh_sftp::server::Handler for SftpHandler { return Err(SftpError::eof()); } + *pos = Some(offset + bytes_read as u64); + buffer.truncate(bytes_read); tracing::trace!( @@ -755,16 +778,31 @@ impl russh_sftp::server::Handler for SftpHandler { let mut handles_guard = handles.lock().await; let handle_entry = handles_guard.get_mut(&handle); - let file = match handle_entry { - Some(OpenHandle::File { file, .. }) => file, + let (file, flags, pos) = match handle_entry { + Some(OpenHandle::File { + file, flags, pos, .. + }) => (file, *flags, pos), _ => return Err(SftpError::invalid_handle()), }; - // Seek to offset - file.seek(SeekFrom::Start(offset)).await?; - - // Write data - file.write_all(&data).await?; + if flags.contains(OpenFlags::APPEND) { + // O_APPEND writes always land at EOF regardless of the file + // cursor, so seeking is pointless and the position stays + // untracked (`None`, set at open time). + file.write_all(&data).await?; + } else { + // Seek only when the cursor is not already at the requested + // offset; sequential uploads take the fast path. The tracked + // position is cleared first so a failed seek or partial + // write leaves it unknown rather than stale. + let need_seek = *pos != Some(offset); + *pos = None; + if need_seek { + file.seek(SeekFrom::Start(offset)).await?; + } + file.write_all(&data).await?; + *pos = Some(offset + data.len() as u64); + } tracing::trace!( handle = %handle,