From a4fe4afa20eebd7fc5e2ec3ac7d4150ef0a98b98 Mon Sep 17 00:00:00 2001 From: ExceptionallyHandsome Date: Sun, 3 Jan 2021 06:15:56 +0300 Subject: [PATCH] Fallback to buffered IO disk operations when not on Linux Going cross-platform. --- cratetorrent/Cargo.toml | 5 +- cratetorrent/src/disk/error.rs | 12 ++ cratetorrent/src/disk/io/file.rs | 211 ++++++++++++++++++++++++++---- cratetorrent/src/disk/io/piece.rs | 140 +++++++++++++++----- cratetorrent/src/lib.rs | 10 +- 5 files changed, 310 insertions(+), 68 deletions(-) diff --git a/cratetorrent/Cargo.toml b/cratetorrent/Cargo.toml index b373e3b..103d7d1 100644 --- a/cratetorrent/Cargo.toml +++ b/cratetorrent/Cargo.toml @@ -20,7 +20,6 @@ hex = "0.4" lazy_static = "1.4" log = "0.4" lru = "0.6" -nix = "0.19" percent-encoding = "2.1" reqwest = "0.10" serde = "1.0" @@ -32,6 +31,10 @@ sha-1 = "0.9" tokio = { version = "0.2", features = ["blocking", "macros", "rt-threaded", "stream", "sync", "tcp", "time"] } tokio-util = { version = "0.3", features = ["codec"] } url = "2.2" +cfg-if = "1.0" + +[target.'cfg(target_os = "linux")'.dependencies] +nix = "0.19" [dev-dependencies] mockito = "0.28" diff --git a/cratetorrent/src/disk/error.rs b/cratetorrent/src/disk/error.rs index e9384eb..043af13 100644 --- a/cratetorrent/src/disk/error.rs +++ b/cratetorrent/src/disk/error.rs @@ -46,6 +46,12 @@ pub(crate) enum WriteError { Io(std::io::Error), } +impl From for WriteError { + fn from(e: std::io::Error) -> Self { + Self::Io(e) + } +} + impl fmt::Display for WriteError { fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { match self { @@ -72,6 +78,12 @@ pub(crate) enum ReadError { Io(std::io::Error), } +impl From for ReadError { + fn from(e: std::io::Error) -> Self { + Self::Io(e) + } +} + impl fmt::Display for ReadError { fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { match self { diff --git a/cratetorrent/src/disk/io/file.rs b/cratetorrent/src/disk/io/file.rs index 755e385..3945565 100644 --- a/cratetorrent/src/disk/io/file.rs +++ b/cratetorrent/src/disk/io/file.rs @@ -1,19 +1,62 @@ use std::{ fs::{File, OpenOptions}, - os::unix::io::AsRawFd, path::Path, + io::ErrorKind }; -use nix::sys::uio::{preadv, pwritev}; - use crate::{ disk::error::*, - iovecs, - iovecs::{IoVec, IoVecs}, storage_info::FileSlice, FileInfo, }; +use cfg_if::cfg_if; + +#[cfg(target_os = "linux")] +use crate::iovecs::{self, IoVec, IoVecs}; + + +// a convenience macro for handling ErrorKind::Interrupted, +// zero-byte reads and the like +macro_rules! unwrap_read_res { + ($res:expr) => { + match $res { + Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Err(ReadError::MissingData), + Err(e) if e.kind() == ErrorKind::Interrupted => continue, + Err(e) => return Err(e.into()), + + // If there was nothing to read from file, it means we tried to + // read a piece from a portion of a file not yet downloaded or + // otherwise missing + Ok(0) => return Err(ReadError::MissingData), + + Ok(written) => written, + } + } +} + +// a convenience macro for handling ErrorKind::Interrupted, +// zero-byte writes and the like +macro_rules! unwrap_write_res { + ($res:expr) => { + match $res { + Err(e) if e.kind() == ErrorKind::Interrupted => continue, + Err(e) => return Err(e.into()), + + // Write syscalls never return 0 these days (unless buf.len == 0). + // If that happened, something went terribly wrong; + // io::ErrorKind even has a dedicated variant for it! + // Anyway, we don't expect them to happen. + Ok(0) => { + return Err(std::io::Error::new(ErrorKind::WriteZero, "failed to write whole buffer").into()); + } + + Ok(written) => written, + } + } +} + + pub(crate) struct TorrentFile { pub info: FileInfo, pub handle: File, @@ -46,7 +89,7 @@ impl TorrentFile { } /// Writes to file at most the slice length number of bytes of blocks at the - /// file slice's offset, using pwritev, called repeteadly until all blocks are + /// file slice's offset, using pwritev, called repeatedly until all blocks are /// written to disk. /// /// It returns the slice of blocks that weren't written to disk. That is, it @@ -58,11 +101,15 @@ impl TorrentFile { /// /// Since the syscall may be invoked repeatedly to perform disk IO, this /// means that this operation is not guaranteed to be atomic. - pub fn write<'a>( + #[cfg(target_os = "linux")] + pub fn write_vectored<'a>( &self, file_slice: FileSlice, blocks: &'a mut [IoVec<&'a [u8]>], ) -> Result<&'a mut [IoVec<&'a [u8]>], WriteError> { + use std::os::unix::io::AsRawFd; + use nix::sys::uio::{pwritev}; + let mut iovecs = IoVecs::bounded(blocks, file_slice.len as usize); // the write buffer cannot be larger than the file slice we want to // write to @@ -80,16 +127,18 @@ impl TorrentFile { // transferred to disk (or an error occurs) let mut total_write_count = 0; while !iovecs.as_slice().is_empty() { - let write_count = pwritev( + let write_res = pwritev( self.handle.as_raw_fd(), iovecs.as_slice(), file_slice.offset as i64, ) - .map_err(|e| { - log::warn!("File {:?} write error: {}", self.info.path, e); - // FIXME: convert actual error here - WriteError::Io(std::io::Error::last_os_error()) - })?; + .map_err(|sys_err| { + let err = convert_nix_error(sys_err); + log::warn!("File {:?} write error: {}", self.info.path, err); + err + }); + + let write_count = unwrap_write_res!(write_res); // tally up the total write count total_write_count += write_count; @@ -109,8 +158,21 @@ impl TorrentFile { Ok(iovecs.into_tail()) } + pub fn write<'a>(&self, file_slice: FileSlice, blocks: &'a [u8]) -> Result<&'a [u8], WriteError> { + let len = file_slice.len as usize; + let content = &blocks[..len]; + + write_all_at(&self.handle, content, file_slice.offset) + .map_err(|err| { + log::warn!("File {:?} write error: {}", self.info.path, err); + err + })?; + + Ok(&blocks[len..]) + } + /// Reads from file at most the slice length number of bytes of blocks at - /// the file slice's offset, using preadv, called repeteadly until all + /// the file slice's offset, using preadv, called repeatedly until all /// blocks are read from disk. /// /// It returns the slice of block buffers that weren't filled by the @@ -122,11 +184,15 @@ impl TorrentFile { /// /// Since the syscall may be invoked repeatedly to perform disk IO, this /// means that this operation is not guaranteed to be atomic. - pub fn read<'a>( + #[cfg(target_os = "linux")] + pub fn read_vectored<'a>( &self, file_slice: FileSlice, mut iovecs: &'a mut [IoVec<&'a mut [u8]>], ) -> Result<&'a mut [IoVec<&'a mut [u8]>], ReadError> { + use std::os::unix::io::AsRawFd; + use nix::sys::uio::preadv; + // This is simpler than the write implementation as the preadv method // stops reading in from the file if reaching EOF. We do need to advance // the iovecs read buffer cursor after a read as we may want to read @@ -138,23 +204,18 @@ impl TorrentFile { // transferred to disk (or an error occurs) let mut total_read_count = 0; while !iovecs.is_empty() && (total_read_count as u64) < file_slice.len { - let read_count = preadv( + let read_res = preadv( self.handle.as_raw_fd(), iovecs, file_slice.offset as i64, ) - .map_err(|e| { - log::warn!("File {:?} read error: {}", self.info.path, e); - // FIXME: convert actual error here - ReadError::Io(std::io::Error::last_os_error()) - })?; + .map_err(|sys_err| { + let err = convert_nix_error(sys_err); + log::warn!("File {:?} read error: {}", self.info.path, err); + err + }); - // if there was nothing to read from file it means we tried to - // read a piece from a portion of a file not yet downloaded or - // otherwise missing - if read_count == 0 { - return Err(ReadError::MissingData); - } + let read_count = unwrap_read_res!(read_res); // tally up the total read count total_read_count += read_count; @@ -166,4 +227,100 @@ impl TorrentFile { Ok(iovecs) } + + pub fn read<'a>(&self, file_slice: FileSlice, blocks: &'a mut [u8]) -> Result<&'a mut [u8], ReadError> { + let len = file_slice.len as usize; + + read_exact_at(&self.handle, &mut blocks[..len], file_slice.offset) + .map_err(|err| { + log::warn!("File {:?} read error: {}", self.info.path, err); + err + })?; + + Ok(&mut blocks[len..]) + } +} + +#[cfg(target_os = "linux")] +fn convert_nix_error(err: nix::Error) -> std::io::Error { + match err { + nix::Error::Sys(errno) => std::io::Error::from_raw_os_error(errno as i32), + + // preadv/pwrite never return them, but better safe than sorry + nix::Error::InvalidPath | + nix::Error::InvalidUtf8 | + nix::Error::UnsupportedOperation => { + log::warn!("Unexpected nix::Error kind ({}), falling back to last_os_error", err); + std::io::Error::last_os_error() + }, + } +} + +// Here's where the most effective implementation of `write_at/read_at` is chosen. +// The trick is, linux-specific APIs do not modify the "seek cursor", while Windows' and others' do. +// This discrepancy doesn't matter because nothing in this crate relies on this cursor +// (the read/write offset is always provided explicitly). +// +// FIXME: On the other hand, if `TorrentInfo.handle` will ever become exposed in our public API, +// this "jumping cursor" can potentially be observed by the outside world. It shouldn't be +// a problem either because, well, I fail to see why someone would want to interact with +// the file in a way where the cursor's positioning matters. But, at least, this should be +// clearly documented and the user must be warned to perform his own seeks if need for +// unaccounted reads/writes presents itself... +cfg_if! { + if #[cfg(any(unix, target_os = "redox", target_os = "vxworks", target_os = "hermit"))] { + use std::os::unix::fs::FileExt; + + #[inline] + fn write_all_at(file: &File, buf: &[u8], offset: u64) -> Result<(), WriteError> { + file.write_all_at(buf, offset).map_err(Into::into) + } + + #[inline] + fn read_exact_at(file: &File, buf: &mut [u8], offset: u64) -> Result<(), ReadError> { + file.read_exact_at(buf, offset) + .map_err(|e| match e.kind() { + ErrorKind::UnexpectedEof => ReadError::MissingData, + _ => e.into() + }) + + } + } else if #[cfg(windows)] { + use std::os::windows::fs::FileExt; + + fn write_all_at(file: &File, mut buf: &[u8], mut offset: u64) -> Result<(), WriteError> { + while !buf.is_empty() { + let written = unwrap_write_res!(file.seek_write(buf, offset)); + offset += written; + buf = &buf[written..]; + } + + Ok(()) + } + + fn read_exact_at(file: &File, mut buf: &mut [u8], mut offset: u64) -> Result<(), ReadError> { + while !buf.is_empty() { + let read = unwrap_read_res!(file.seek_read(buf, offset)); + offset += read; + buf = &mut buf[read..]; + } + + Ok(()) + } + } else { + fn write_all_at(file: &File, buf: &[u8], offset: u64) -> Result<(), WriteError> { + file.seek(SeekFrom(offset))?; + file.write_all(buf).map_err(Into::into) + } + + fn read_exact_at(file: &File, buf: &mut [u8], offset: u64) -> Result<(), ReadError> { + file.seek(SeekFrom(offset))?; + file.read_exact(buf).map_err(|e| match e.kind() { + ErrorKind::UnexpectedEof => ReadError::MissingData, + _ => e.into() + }) + } + } } + + diff --git a/cratetorrent/src/disk/io/piece.rs b/cratetorrent/src/disk/io/piece.rs index 74f9dc4..0ad9842 100644 --- a/cratetorrent/src/disk/io/piece.rs +++ b/cratetorrent/src/disk/io/piece.rs @@ -5,14 +5,17 @@ use std::{ }; use sha1::{Digest, Sha1}; +use cfg_if::cfg_if; use crate::{ block_count, block_len, disk::{error::*, io::file::TorrentFile}, - iovecs::IoVec, CachedBlock, FileIndex, Sha1Hash, }; +#[cfg(target_os = "linux")] +use crate::iovecs::IoVec; + /// An in-progress piece download that keeps in memory the so far downloaded /// blocks and the expected hash of the piece. pub(crate) struct Piece { @@ -88,15 +91,49 @@ impl Piece { torrent_piece_offset: u64, files: &[sync::RwLock], ) -> Result<(), WriteError> { - // convert the blocks to IO slices that the underlying - // systemcall can deal with - let mut blocks: Vec<_> = self - .blocks - .values() - .map(|b| IoVec::from_slice(&b)) - .collect(); - // the actual slice of blocks being worked on - let mut bufs = blocks.as_mut_slice(); + cfg_if! { + if #[cfg(target_os = "linux")] { + // convert the blocks to IO slices that the underlying + // pwritev syscall can deal with + let mut blocks: Vec<_> = self + .blocks + .values() + .map(|b| IoVec::from_slice(&b)) + .collect(); + // the actual slice of blocks being worked on + let mut bufs = blocks.as_mut_slice(); + } else { + // (the same situation is in the `read` method below) + // + // We don't have pwritev The Convenient. Instead, we have two options: + // + // 1. Coalesce the slices to a continuous preallocated memory buffer and + // write it at once. + // 2. Iterate through the slices and issue a separate write call for each. + // + // The first option implies preallocating a short-living buffer + // and destroying it shortly after. The second option + // implies issuing a big number of writes, and possibly disk I/O operations. + // + // We've chosen the first option. Piece size is commonly less than 1MB + // which is less than the page size, so the (re)allocation should be fast enough. + // Careful benchmarking, while necessary when the project gets to the optimization phase, + // is something I don't feel like doing just yet, and thus left as TODO + // and an exercise to the reader. + // + // (another potential optimization is to preserve the buffer between `Piece::write` calls, + // storing it in a `thread_local!` storage, but then the function will cease to be reentrant) + let len = self.blocks.values().map(|s|s.len()).sum(); + let mut buffer = self + .blocks + .values() + .fold(Vec::with_capacity(len), |mut buf, slice| { + buf.extend_from_slice(slice); + buf + }); + let mut buffer = &buffer; + } + } // loop through all files piece overlaps with and write that part of // piece to file @@ -119,18 +156,24 @@ impl Piece { // an empty file slice shouldn't occur as it would mean that piece // was thought to span fewer files than it actually does debug_assert!(file_slice.len > 0); - // the write buffer should still contain bytes to write - debug_assert!(!bufs.is_empty()); - debug_assert!(!bufs[0].as_slice().is_empty()); - // write to file - let tail = file.write(file_slice, bufs)?; + cfg_if! { + if #[cfg(target_os = "linux")] { + // the write buffer should still contain bytes to write + debug_assert!(!bufs.is_empty()); + debug_assert!(!bufs[0].as_slice().is_empty()); - // `write_vectored_at` only writes at most `slice.len` bytes of - // `bufs` to disk and returns the portion that wasn't - // written, which we can use to set the write buffer for the next - // round - bufs = tail; + // write to file + // `write_*` only writes at most `slice.len` bytes of + // `bufs` to disk and returns the portion that wasn't + // written, which we can use to set the write buffer for the next + // round + bufs = file.write_vectored(file_slice, bufs)?; + } else { + debug_assert!(!buffer.is_empty()); + buffer = file.write(file_slice, buffer)?; + } + } torrent_write_offset += file_slice.len as u64; total_write_count += file_slice.len; @@ -167,23 +210,30 @@ pub(super) fn read( let mut blocks = Vec::with_capacity(block_count); for i in 0..block_count { let block_len = block_len(len, i); - let mut buf = Vec::new(); - buf.resize(block_len as usize, 0u8); + let mut buf = vec![0u8; block_len as usize]; blocks.push(Arc::new(buf)) } - // convert the blocks to IO slices that the underlying - // systemcall can deal with - let mut iovecs: Vec> = - blocks - .iter_mut() - .map(|b| { - IoVec::from_mut_slice(Arc::get_mut(b).expect( - "cannot get mut ref to buffer only used by this thread", - ).as_mut_slice()) - }) - .collect(); - let mut bufs = iovecs.as_mut_slice(); + cfg_if! { + if #[cfg(target_os = "linux")] { + // convert the blocks to IO slices that the underlying + // systemcall can deal with + let mut iovecs: Vec> = + blocks + .iter_mut() + .map(|b| { + IoVec::from_mut_slice(Arc::get_mut(b).expect( + "cannot get mut ref to buffer only used by this thread", + ).as_mut_slice()) + }) + .collect(); + let mut bufs = iovecs.as_mut_slice(); + } else { + // See the comment above (in the `write` method) if curious what's happening. + let buffer = Vec::with_capacity(len as usize); + let buffer_mut = &mut buffer; + } + } // loop through all files piece overlaps with and read that part of // file @@ -209,7 +259,13 @@ pub(super) fn read( debug_assert!(file_slice.len > 0); // read data - bufs = file.read(file_slice, bufs)?; + cfg_if! { + if #[cfg(target_os = "linux")] { + bufs = file.read_vectored(file_slice, bufs)?; + } else { + buffer_mut = file.read(file_slice, buffer_mut)?; + } + } torrent_read_offset += file_slice.len as u64; total_read_count += file_slice.len; @@ -218,5 +274,19 @@ pub(super) fn read( // we should have read in the whole piece debug_assert_eq!(total_read_count, len); + // we work with one linear buffer, + // so let's copy contents from it to block buffers + cfg_if! { + if #[cfg(target_os = "linux")] { + // preadv did everything for us + } else { + for (i, (block, data)) in blocks.iter_mut().zip(buffer.chunks()).enumerate() { + let block_len = block_len(len, i); + debug_assert_eq!(block_len, data.len()); + block.copy_from_slice(data); + } + } + } + Ok(blocks) } diff --git a/cratetorrent/src/lib.rs b/cratetorrent/src/lib.rs index 7da296c..3043515 100644 --- a/cratetorrent/src/lib.rs +++ b/cratetorrent/src/lib.rs @@ -6,10 +6,7 @@ //! //! # Caveats //! -//! The engine currently only supports Linux. This is expected to change in the -//! future, however. -//! -//! It also lacks most features present in battle-hardened torrent engines, such +//! The engine lacks most features present in battle-hardened torrent engines, such //! as [libtorrent](https://github.com/arvidn/libtorrent). These include: DHT //! for peer exchange, magnet links, stream encryption, UDP trackers, and many //! more. @@ -156,7 +153,6 @@ mod disk; mod download; pub mod engine; pub mod error; -pub mod iovecs; pub mod metainfo; pub mod peer; mod piece_picker; @@ -165,6 +161,10 @@ pub mod storage_info; pub mod torrent; mod tracker; +#[cfg(target_os = "linux")] +pub mod iovecs; + + /// Each torrent gets a randomly assigned ID that is unique within the /// engine. This id is used in engine APIs to interact with torrents. #[derive(Clone, Copy, Debug, PartialEq, Eq, Ord, PartialOrd, Hash)]