From 816838cdf081ad83242eb478aa8261ffa2c87442 Mon Sep 17 00:00:00 2001 From: Igor Katson Date: Thu, 12 Sep 2024 23:59:50 +0100 Subject: [PATCH 1/2] Fastresume: check at least one piece from each file --- .../src/torrent_state/initializing.rs | 55 +++++++++++++++---- 1 file changed, 44 insertions(+), 11 deletions(-) diff --git a/crates/librqbit/src/torrent_state/initializing.rs b/crates/librqbit/src/torrent_state/initializing.rs index 868cea6..ed4e228 100644 --- a/crates/librqbit/src/torrent_state/initializing.rs +++ b/crates/librqbit/src/torrent_state/initializing.rs @@ -100,27 +100,60 @@ impl TorrentStateInitializing { use rand::seq::SliceRandom; - let mut have_pieces = hp - .as_slice() - .iter_ones() - .filter_map(|i| self.shared.lengths.validate_piece_index(i.try_into().ok()?)) - .collect_vec(); - have_pieces.shuffle(&mut rand::thread_rng()); + let mut to_validate = BF::from_boxed_slice( + vec![0u8; self.shared.lengths.piece_bitfield_bytes()].into_boxed_slice(), + ); + let mut queue = hp.as_slice().to_owned(); - // Validate a certain threshold of fastresume pieces with decreasing probability of actual disk reads. - for (tmp_id, hpiece) in have_pieces.iter().enumerate() { + // Validate at least one piece from each file, if we claim we have it. + for fi in self.shared.file_infos.iter() { + let prange = fi.piece_range_usize(); + let offset = prange.start; + for piece_id in hp + .as_slice() + .get(fi.piece_range_usize()) + .into_iter() + .flat_map(|s| s.iter_ones()) + .map(|pid| pid + offset) + .take(1) + { + to_validate.set(piece_id, true); + queue.set(piece_id, false); + } + } + + // For all the remaining pieces we claim we have, validate them with decreasing probability. + let mut queue = queue.iter_ones().collect_vec(); + queue.shuffle(&mut rand::thread_rng()); + for (tmp_id, piece_id) in queue.into_iter().enumerate() { let denom: u32 = (tmp_id + 1).min(50).try_into().unwrap(); - if rand::thread_rng().gen_ratio(1, denom) && fo.check_piece(*hpiece).is_err() { + if rand::thread_rng().gen_ratio(1, denom) { + to_validate.set(piece_id, true); + } + } + + let to_validate_count = to_validate.count_ones(); + for (id, piece_id) in to_validate + .iter_ones() + .filter_map(|id| { + self.shared + .lengths + .validate_piece_index(id.try_into().ok()?) + }) + .enumerate() + { + if fo.check_piece(piece_id).is_err() { return true; } #[allow(clippy::cast_possible_truncation)] let progress = (self.shared.lengths.total_length() as f64 - / have_pieces.len() as f64 - * (tmp_id + 1) as f64) as u64; + / to_validate_count as f64 + * (id + 1) as f64) as u64; let progress = progress.min(self.shared.lengths.total_length()); self.checked_bytes.store(progress, Ordering::Relaxed); } + false }); From 00daae97eda70cf444ad44b27625f71a9f60cadc Mon Sep 17 00:00:00 2001 From: Igor Katson Date: Fri, 13 Sep 2024 00:20:37 +0100 Subject: [PATCH 2/2] Reorder removal so that mmap is removed before storage --- crates/librqbit/src/bitv.rs | 7 +++++++ crates/librqbit/src/session.rs | 16 ++++++++-------- 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/crates/librqbit/src/bitv.rs b/crates/librqbit/src/bitv.rs index f3c0129..38ce994 100644 --- a/crates/librqbit/src/bitv.rs +++ b/crates/librqbit/src/bitv.rs @@ -8,6 +8,7 @@ use bitvec::{ vec::BitVec, view::{AsBits, AsMutBits}, }; +use tracing::trace; pub trait BitV: Send + Sync { fn as_slice(&self) -> &BitSlice; @@ -24,6 +25,12 @@ pub struct MmapBitV { mmap: memmap2::MmapMut, } +impl Drop for MmapBitV { + fn drop(&mut self) { + trace!("dropping MmapBitV, this should unmap the .bitv file") + } +} + impl MmapBitV { pub fn new(file: File) -> anyhow::Result { let mmap = diff --git a/crates/librqbit/src/session.rs b/crates/librqbit/src/session.rs index 0522b7b..0ddf83e 100644 --- a/crates/librqbit/src/session.rs +++ b/crates/librqbit/src/session.rs @@ -1251,14 +1251,6 @@ impl Session { debug!("error pausing torrent before deletion: {e:?}") } - if let Some(p) = self.persistence.as_ref() { - if let Err(e) = p.delete(id).await { - error!(error=?e, "error deleting torrent from persistence database"); - } else { - debug!(?id, "deleted torrent from persistence database") - } - } - let storage = removed .with_state_mut(|s| match s.take() { ManagedTorrentState::Initializing(p) => p.files.take().ok(), @@ -1277,6 +1269,14 @@ impl Session { .map(Ok) .unwrap_or_else(|| removed.shared.storage_factory.create(removed.shared())); + if let Some(p) = self.persistence.as_ref() { + if let Err(e) = p.delete(id).await { + error!(error=?e, "error deleting torrent from persistence database"); + } else { + debug!(?id, "deleted torrent from persistence database") + } + } + match (storage, delete_files) { (Err(e), true) => return Err(e).context("torrent deleted, but could not delete files"), (Ok(storage), true) => {