Support torrents with excess pieces; increase u32 file sizes

src/dht.rs

@@ -14,20 +14,22 @@ use crate::{
  udp::send_udp,
 };
 
+// TODO: recursive async works weird, rewrite into queue instead
 #[async_recursion]
 pub async fn find_peers(
  peer: PeerInfo,
  peer_id: ByteString,
  info_hash: ByteString,
  min: usize,
+ depth: usize,
 ) -> Result<BTreeSet<PeerInfo>> {
  if min == 0 {
  return Ok(BTreeSet::new());
  }
  trace!("quering dht peer: {:?}, {} left", peer, min);
  let res = timeout(
  // TODO: make configurable
- Duration::from_secs(1),
+ Duration::from_millis(200),
  dht_find_peers(&peer, &peer_id, info_hash.clone()),
  )
  .await??;
@@ -75,9 +77,11 @@ pub async fn find_peers(
  };
 
  debug!(
- "received {} values, {} nodes",
+ "received {}/{} values, {} nodes at depth {}",
  values.clone().unwrap_or_default().len(),
- nodes.clone().unwrap_or_default().len()
+ min,
+ nodes.clone().unwrap_or_default().len(),
+ depth
  );
  let mut found = BTreeSet::new();
  if let Some(vs) = values {
@@ -93,6 +97,7 @@ pub async fn find_peers(
  peer_id.clone(),
  info_hash.clone(),
  min - found.len(),
+ depth + 1,
  ))
  })
  .collect::<FuturesUnordered<_>>();

src/metainfo.rs

@@ -16,7 +16,7 @@ pub struct Metainfo {
 
 #[derive(Clone, PartialEq, PartialOrd, Hash)]
 pub struct Info {
- pub piece_length: u32,
+ pub piece_length: u64,
  pub pieces: Vec<PieceHash>,
  pub name: String,
  pub file_info: FileInfo,
@@ -41,9 +41,9 @@ pub enum FileInfo {
 }
 
 impl FileInfo {
- pub fn total_length(&self) -> u32 {
+ pub fn total_length(&self) -> u64 {
  match self {
- FileInfo::Single(path) => path.length,
+ FileInfo::Single(file) => file.length,
  FileInfo::Multi(files) => files.iter().map(|f| f.length).sum(),
  }
  }
@@ -58,7 +58,7 @@ impl FileInfo {
 
 #[derive(Clone, Debug, PartialEq, PartialOrd, Hash)]
 pub struct PathInfo {
- pub length: u32,
+ pub length: u64,
  pub path: PathBuf,
  pub md5_sum: Option<String>,
 }
@@ -88,7 +88,7 @@ impl TryFrom<BencodeValue> for Metainfo {
  None => FileInfo::Single(PathInfo {
  path: PathBuf::from(&name),
  length: match info_dict.get("length") {
- Some(BencodeValue::Int(v)) => *v as u32,
+ Some(BencodeValue::Int(v)) => *v as u64,
  _ => return Err("'length' missing".into()),
  },
  md5_sum: match info_dict.get("md5_sum") {
@@ -100,7 +100,7 @@ impl TryFrom<BencodeValue> for Metainfo {
  let metainfo = Metainfo {
  info: Info {
  piece_length: match info_dict.get("piece length") {
- Some(BencodeValue::Int(v)) => *v as u32,
+ Some(BencodeValue::Int(v)) => *v as u64,
  _ => return Err("'piece length' missing".into()),
  },
  pieces,
@@ -172,7 +172,7 @@ fn parse_files_info(value: &BencodeValue) -> Result<Vec<PathInfo>, String> {
  };
  Ok(PathInfo {
  length: match d.get("length") {
- Some(BencodeValue::Int(v)) => *v as u32,
+ Some(BencodeValue::Int(v)) => *v as u64,
  _ => return Err("'length' missing".into()),
  },
  path,

src/state.rs

@@ -160,15 +160,18 @@ pub fn init_pieces(info: &Info) -> BTreeMap<u32, Piece> {
  .collect::<Vec<_>>();
  let total_len = info.file_info.total_length();
  if info.pieces.len() != (total_len as f64 / info.piece_length as f64).ceil() as usize {
- warn!("total length/piece size/piece count inconsistent");
+ warn!(
+ "total length/piece size/piece count inconsistent: {} info pieces, {} expected",
+ info.pieces.len(),
+ (total_len as f64 / info.piece_length as f64).ceil()
+ );
  }
  info.pieces
  .iter()
  .cloned()
  .enumerate()
- .map(|(i, p)| (i as u32, p))
- .map(|(i, p)| {
- let length = if i == info.pieces.len() as u32 - 1 {
+ .flat_map(|(i, p)| {
+ let length: u64 = if i == info.pieces.len() - 1 {
  total_len % info.piece_length
  } else {
  info.piece_length
@@ -180,7 +183,7 @@ pub fn init_pieces(info: &Info) -> BTreeMap<u32, Piece> {
  .flat_map(|(f_i, f_start)| {
  let f_len = info.file_info.files()[f_i].length;
  let f_end = f_start + f_len;
- let piece_start = i * info.piece_length;
+ let piece_start = i as u64 * info.piece_length;
  let piece_end = piece_start + length;
  let p_start = (f_start as i64).clamp(piece_start as i64, piece_end as i64);
  let p_end = (f_end as i64).clamp(piece_start as i64, piece_end as i64);
@@ -199,21 +202,22 @@ pub fn init_pieces(info: &Info) -> BTreeMap<u32, Piece> {
  }
  })
  .collect();
- // TODO: verify files' location integrity
- if file_locations.iter().map(|f| f.length as u32).sum::<u32>() != length {
- panic!("incorrect file location length");
+ if file_locations.is_empty() {
+  debug!("piece does not map to any files: {:?}", p);
+ return vec![];
  }
- (
- i,
+ // TODO: verify files' location integrity
+ vec![(
+ i as u32,
  Piece {
  hash: p,
- index: i,
- length,
+ index: i as u32,
+ length: length as u32,
  blocks: BTreeMap::new(),
  status: TorrentStatus::Started,
  file_locations,
  },
- )
+ )]
  })
  .collect()
 }

src/torrent.rs

@@ -72,7 +72,7 @@ pub async fn download_torrent(
  TrackerResponseSuccess {
  // set peers discovered via DHT
  peers,
- // should never poll again, use rely on DHT
+ // should never poll again, rely on DHT
  interval: i64::MAX,
  ..Default::default()
  }
@@ -196,7 +196,7 @@ async fn discover_peers(
 
  let handles = dht_peers
  .into_iter()
- .map(|p| spawn(find_peers(p, peer_id.clone(), info_hash.to_vec(), min_p)))
+ .map(|p| spawn(find_peers(p, peer_id.clone(), info_hash.to_vec(), min_p, 0)))
  .collect::<FuturesUnordered<_>>();
  for h in handles {
  match h.await {

src/tracker.rs

@@ -16,12 +16,12 @@ use crate::{
 pub struct TrackerRequest {
  pub info_hash: ByteString,
  pub peer_id: ByteString,
- pub port: i64,
- pub uploaded: i64,
- pub downloaded: i64,
- pub left: i64,
- pub compact: i64,
- pub no_peer_id: i64,
+ pub port: u64,
+ pub uploaded: u64,
+ pub downloaded: u64,
+ pub left: u64,
+ pub compact: u64,
+ pub no_peer_id: u64,
  pub event: Option<TrackerEvent>,
  pub ip: Option<ByteString>,
  pub numwant: Option<i64>,
@@ -40,7 +40,7 @@ impl TrackerRequest {
  TrackerRequest {
  info_hash,
  peer_id,
- port: port as i64,
+ port: port as u64,
  // TODO
  uploaded: 0,
  // TODO