use std::collections::VecDeque;
use bitvec::{
prelude::{BitBox, Msb0},
vec::BitVec,
};
use lava_torrent::torrent::v1::Torrent;
#[cfg(test)]
use rand::SeedableRng;
use rand::{RngExt, rngs::SmallRng};
use slotmap::SecondaryMap;
use crate::{buf_pool::Buffer, event_loop::ConnectionId};
pub const SUBPIECE_SIZE: i32 = 16_384;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Subpiece {
pub index: i32,
pub offset: i32,
pub size: i32,
pub timed_out: bool,
}
pub struct PieceSelector {
downloaded_pieces: BitBox<u8, Msb0>,
allocated_pieces: BitBox<u8, Msb0>,
completed_pieces: BitBox<u8, Msb0>,
interesting_peer_pieces: SecondaryMap<ConnectionId, BitBox<u8, Msb0>>,
last_piece_length: u32,
piece_length: u32,
rng_gen: SmallRng,
}
impl PieceSelector {
pub fn new(torrent_info: &Torrent) -> Self {
let completed_pieces: BitBox<u8, Msb0> =
BitVec::repeat(false, torrent_info.pieces.len()).into();
let allocated_pieces = completed_pieces.clone();
let hashing_pieces = completed_pieces.clone();
let piece_length = torrent_info.piece_length;
let mut last_piece_length = torrent_info.length % piece_length;
if last_piece_length == 0 {
last_piece_length = piece_length;
}
Self {
downloaded_pieces: completed_pieces,
allocated_pieces,
completed_pieces: hashing_pieces,
last_piece_length: last_piece_length as u32,
piece_length: piece_length as u32,
interesting_peer_pieces: Default::default(),
#[cfg(not(test))]
rng_gen: rand::make_rng(),
#[cfg(test)]
rng_gen: SmallRng::seed_from_u64(0xbeefdead),
}
}
pub(crate) fn set_completed_bitfield(&mut self, completed_pieces: BitBox<u8, Msb0>) {
assert_eq!(self.completed_pieces.len(), completed_pieces.len());
self.completed_pieces = completed_pieces.clone();
self.downloaded_pieces = completed_pieces;
}
pub fn next_piece(
&mut self,
connection_id: ConnectionId,
endgame_mode: &mut bool,
) -> Option<i32> {
let interesting_pieces = self.interesting_peer_pieces.get(connection_id)?;
let pickable = !self.downloaded_pieces.clone() & interesting_pieces;
let first_pickable = pickable.first_one();
let unallocated_pickable = !self.allocated_pieces.clone() & pickable;
if unallocated_pickable.not_any() {
let pickable = first_pickable?;
log::debug!("Peer {connection_id:?} is entering endgame mode");
*endgame_mode = true;
return Some(pickable as i32);
}
let procentage_left =
self.downloaded_pieces.count_zeros() as f32 / self.downloaded_pieces.len() as f32;
if procentage_left > 0.95 {
for _ in 0..5 {
let index =
(self.rng_gen.random::<f32>() * self.downloaded_pieces.len() as f32) as usize;
if unallocated_pickable[index] {
*endgame_mode = false;
return Some(index as i32);
}
}
log::warn!("Random piece selection failed");
let available_index = unallocated_pickable.first_one()?;
*endgame_mode = false;
Some(available_index as i32)
} else {
let mut count = vec![0; unallocated_pickable.len()];
for available in unallocated_pickable.iter_ones() {
for peer_pieces in self.interesting_peer_pieces.values() {
if peer_pieces[available] {
count[available] += 1;
}
}
}
let (rarest_index, _) = count
.into_iter()
.enumerate()
.filter(|(_pos, count)| count > &0)
.min_by_key(|(_pos, val)| *val)?;
*endgame_mode = false;
Some(rarest_index as i32)
}
}
#[inline]
pub fn bitfield_received(&self, connection_id: ConnectionId) -> bool {
self.interesting_peer_pieces.contains_key(connection_id)
}
pub fn peer_bitfield(
&mut self,
connection_id: ConnectionId,
peer_pieces: BitBox<u8, Msb0>,
) -> bool {
let not_completed = !self.downloaded_pieces.clone();
let interesting_pieces = peer_pieces & not_completed;
let is_interesting = interesting_pieces.any();
self.interesting_peer_pieces
.insert(connection_id, interesting_pieces);
is_interesting
}
pub fn update_peer_piece_intrest(
&mut self,
connection_id: ConnectionId,
piece_index: usize,
) -> bool {
let is_interesting = !self.downloaded_pieces[piece_index];
let entry = self
.interesting_peer_pieces
.entry(connection_id)
.expect("peer must remain in primary map");
entry
.and_modify(|pieces| pieces.set(piece_index, is_interesting))
.or_insert_with(|| {
let mut all_pieces: BitBox<u8, Msb0> =
BitVec::repeat(false, self.downloaded_pieces.len()).into();
all_pieces.set(piece_index, is_interesting);
all_pieces
});
is_interesting
}
pub fn interesting_peer_pieces(
&self,
connection_id: ConnectionId,
) -> Option<&BitBox<u8, Msb0>> {
self.interesting_peer_pieces.get(connection_id)
}
#[inline]
pub fn mark_complete(&mut self, index: usize) {
debug_assert!(!self.completed_pieces[index]);
debug_assert!(self.downloaded_pieces[index]);
self.completed_pieces.set(index, true);
self.allocated_pieces.set(index, false);
for interesting_pieces in self.interesting_peer_pieces.values_mut() {
interesting_pieces.set(index, false);
}
}
#[inline]
pub fn mark_downloaded(&mut self, index: usize) {
debug_assert!(!self.downloaded_pieces[index]);
debug_assert!(!self.completed_pieces[index]);
self.downloaded_pieces.set(index, true);
}
#[inline]
pub fn mark_not_downloaded(&mut self, index: usize) {
debug_assert!(self.downloaded_pieces[index]);
debug_assert!(!self.completed_pieces[index]);
self.downloaded_pieces.set(index, false);
}
#[inline]
pub fn mark_allocated(&mut self, index: i32, connection_id: ConnectionId) {
let index = index as usize;
self.allocated_pieces.set(index, true);
let interesting_pieces = &mut self.interesting_peer_pieces.get_mut(connection_id).unwrap();
let old = interesting_pieces.replace(index, false);
debug_assert!(old);
}
#[inline]
pub fn mark_not_allocated(&mut self, index: i32, connection_id: ConnectionId) {
let index = index as usize;
debug_assert!(self.allocated_pieces[index]);
self.allocated_pieces.set(index, false);
self.update_peer_piece_intrest(connection_id, index);
}
#[inline]
pub fn completed_all(&self) -> bool {
self.completed_pieces.all()
}
#[inline]
pub fn downloaded_clone(&self) -> BitBox<u8, Msb0> {
self.downloaded_pieces.clone()
}
#[inline]
pub fn has_downloaded(&self, index: usize) -> bool {
self.downloaded_pieces[index]
}
#[inline]
pub fn is_complete(&self, index: usize) -> bool {
self.completed_pieces[index]
}
#[inline]
pub fn is_allocated(&self, index: usize) -> bool {
self.allocated_pieces[index]
}
#[inline]
pub fn total_completed(&self) -> usize {
self.completed_pieces.count_ones()
}
#[inline]
pub fn total_allocated(&self) -> usize {
self.allocated_pieces.count_ones()
}
#[inline]
pub fn piece_len(&self, index: i32) -> u32 {
if index == (self.downloaded_pieces.len() as i32 - 1) {
self.last_piece_length
} else {
self.piece_length
}
}
#[inline]
pub fn avg_piece_length(&self) -> u32 {
self.piece_length
}
#[inline]
pub fn avg_num_subpieces(&self) -> u32 {
self.piece_length / SUBPIECE_SIZE as u32
}
}
#[derive(Debug)]
pub struct DownloadedPiece {
pub index: usize,
pub conn_id: ConnectionId,
pub hash_matched: bool,
pub buffer: Buffer,
}
#[derive(Debug)]
pub struct Piece {
pub index: i32,
pub completed_subpieces: BitBox,
pub last_subpiece_length: i32,
pub piece_data: Buffer,
pub ref_count: u8,
}
impl Piece {
pub fn new(index: i32, lenght: u32, piece_view: Buffer) -> Self {
assert!(lenght > 0, "Piece lenght must be non zero");
let last_subpiece_length = if lenght as i32 % SUBPIECE_SIZE == 0 {
SUBPIECE_SIZE
} else {
lenght as i32 % SUBPIECE_SIZE
};
let subpieces =
(lenght / SUBPIECE_SIZE as u32) + u32::from(last_subpiece_length != SUBPIECE_SIZE);
let completed_subpieces: BitBox = (0..subpieces).map(|_| false).collect();
Self {
index,
completed_subpieces,
last_subpiece_length,
piece_data: piece_view,
ref_count: 0,
}
}
pub fn allocate_remaining_subpieces(&mut self) -> VecDeque<Subpiece> {
let mut deque = VecDeque::with_capacity(self.completed_subpieces.len());
let last_subpiece_index = self.completed_subpieces.len() - 1;
let mut last_is_last_index = false;
for subpiece_index in self.completed_subpieces.iter_zeros() {
deque.push_back(Subpiece {
index: self.index,
offset: SUBPIECE_SIZE * subpiece_index as i32,
size: SUBPIECE_SIZE,
timed_out: false,
});
last_is_last_index = subpiece_index == last_subpiece_index;
}
if last_is_last_index {
let last_subpiece = deque.back_mut().unwrap();
last_subpiece.size = self.last_subpiece_length;
}
self.ref_count += 1;
deque
}
pub fn into_buffer(self) -> Buffer {
self.piece_data
}
pub fn on_subpiece(&mut self, index: i32, begin: i32, data: &[u8]) {
debug_assert_eq!(self.index, index);
let subpiece_index = begin / SUBPIECE_SIZE;
if self.completed_subpieces[subpiece_index as usize] {
return;
}
log::trace!("Subpiece index received: {subpiece_index}",);
let last_subpiece = subpiece_index == self.last_subpiece_index();
if last_subpiece {
debug_assert_eq!(data.len() as i32, self.last_subpiece_length);
} else {
debug_assert_eq!(data.len() as i32, SUBPIECE_SIZE);
}
let begin = begin as usize;
self.piece_data.raw_mut_slice()[begin..(begin + data.len())].copy_from_slice(data);
self.completed_subpieces.set(subpiece_index as usize, true);
}
#[inline]
pub fn last_subpiece_index(&self) -> i32 {
self.completed_subpieces.len() as i32 - 1
}
#[inline]
pub fn is_complete(&self) -> bool {
self.completed_subpieces.all()
}
}