From 87bd0bd3644350018f693911930623c1ed82be55 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Mon, 15 Jun 2026 09:12:00 -0700 Subject: [PATCH 01/14] feat: add peer choke control --- crates/libtortillas/src/peer/actor.rs | 32 +++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/crates/libtortillas/src/peer/actor.rs b/crates/libtortillas/src/peer/actor.rs index d543df1..a8d7a92 100644 --- a/crates/libtortillas/src/peer/actor.rs +++ b/crates/libtortillas/src/peer/actor.rs @@ -508,6 +508,14 @@ impl Message for PeerActor { } } PeerMessages::Request(index, offset, length) => { + if self.peer.choked() { + trace!( + piece_index = index, + offset, length, "Ignoring request from choked peer" + ); + return; + } + debug!( piece_index = index, offset, length, "Peer requested piece data" @@ -525,11 +533,13 @@ impl Message for PeerActor { match data { Some(data) => { + let uploaded_bytes = data.len(); self .stream .send(PeerMessages::Piece(index as u32, offset as u32, data)) .await .expect("Failed to send piece"); + self.peer.increment_bytes_uploaded(uploaded_bytes); } None => { warn!( @@ -682,6 +692,28 @@ pub(crate) mod commands { } } + #[message(derive(Clone, Debug))] + #[instrument(skip(self), fields(peer_addr = %self.stream, peer_id = %self.peer.id.unwrap()))] + pub(crate) async fn set_choked(&mut self, choked: bool) { + if self.peer.choked() == choked { + return; + } + + let message = if choked { + PeerMessages::Choke + } else { + PeerMessages::Unchoke + }; + + if let Err(err) = self.stream.send(message).await { + warn!(?err, choked, "Failed to update peer choke state"); + return; + } + + self.peer.set_choked(choked); + trace!(choked, "Updated peer choke state"); + } + #[message(derive(Clone, Debug))] pub(crate) async fn have_info_dict(&mut self, bitfield: Arc>) { self From 5a6099e1a08454a2a15202e1fd05754b1dee3d02 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Mon, 15 Jun 2026 10:04:00 -0700 Subject: [PATCH 02/14] feat: expose peer transfer stats --- crates/libtortillas/src/peer/actor.rs | 72 ++++++++++++++++++++++++++- 1 file changed, 71 insertions(+), 1 deletion(-) diff --git a/crates/libtortillas/src/peer/actor.rs b/crates/libtortillas/src/peer/actor.rs index a8d7a92..9a7f2d2 100644 --- a/crates/libtortillas/src/peer/actor.rs +++ b/crates/libtortillas/src/peer/actor.rs @@ -22,7 +22,7 @@ use tracing::{Span, debug, info, instrument, trace, warn}; use crate::{ errors::PeerActorError, hashes::InfoHash, - peer::Peer, + peer::{Peer, PeerId}, protocol::{stream::PeerRecv, *}, torrent::{self, TorrentActor}, }; @@ -32,6 +32,34 @@ const MAX_PENDING_MESSAGES: usize = 8; const PEER_KEEPALIVE_TIMEOUT: u64 = 120; const PEER_DISCONNECT_TIMEOUT: u64 = 240; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct PeerStats { + pub(crate) id: PeerId, + pub(crate) interested: bool, + pub(crate) choked: bool, + pub(crate) download_rate: usize, + pub(crate) upload_rate: usize, + pub(crate) bytes_downloaded: usize, + pub(crate) bytes_uploaded: usize, +} + +#[derive(Clone, Copy, Debug)] +struct RateSample { + at: Instant, + bytes_downloaded: usize, + bytes_uploaded: usize, +} + +impl RateSample { + fn new(peer: &Peer) -> Self { + Self { + at: Instant::now(), + bytes_downloaded: peer.bytes_downloaded(), + bytes_uploaded: peer.bytes_uploaded(), + } + } +} + /// The actor that handles all communications with a given peer. pub(crate) struct PeerActor { /// The peers state and statistics @@ -43,6 +71,7 @@ pub(crate) struct PeerActor { pending_block_requests: HashSet<(usize, usize, usize)>, pending_message_requests: VecDeque, + last_rate_sample: RateSample, } impl PeerActor { @@ -306,6 +335,41 @@ impl PeerActor { self.stream.send(msg).await } + + fn snapshot_stats(&mut self) -> Option { + let id = self.peer.id?; + let now = Instant::now(); + let bytes_downloaded = self.peer.bytes_downloaded(); + let bytes_uploaded = self.peer.bytes_uploaded(); + let elapsed_secs = now + .duration_since(self.last_rate_sample.at) + .as_secs() + .max(1) as usize; + + let download_rate = bytes_downloaded.saturating_sub(self.last_rate_sample.bytes_downloaded) + / 1024 + / elapsed_secs; + let upload_rate = + bytes_uploaded.saturating_sub(self.last_rate_sample.bytes_uploaded) / 1024 / elapsed_secs; + + self.peer.set_download_rate(download_rate); + self.peer.set_upload_rate(upload_rate); + self.last_rate_sample = RateSample { + at: now, + bytes_downloaded, + bytes_uploaded, + }; + + Some(PeerStats { + id, + interested: self.peer.interested(), + choked: self.peer.choked(), + download_rate, + upload_rate, + bytes_downloaded, + bytes_uploaded, + }) + } } impl Actor for PeerActor { @@ -340,6 +404,7 @@ impl Actor for PeerActor { .map_err(|e| PeerActorError::SupervisorCommunicationFailed(e.to_string()))?; Ok(Self { + last_rate_sample: RateSample::new(&peer), peer, stream, supervisor, @@ -714,6 +779,11 @@ pub(crate) mod commands { trace!(choked, "Updated peer choke state"); } + #[message] + pub(crate) fn stats(&mut self) -> Option { + self.snapshot_stats() + } + #[message(derive(Clone, Debug))] pub(crate) async fn have_info_dict(&mut self, bitfield: Arc>) { self From 75afc32d4b8d125d42e12fdcd742847366db32a1 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Mon, 15 Jun 2026 11:18:00 -0700 Subject: [PATCH 03/14] feat: add choking peer selection --- crates/libtortillas/src/torrent/choking.rs | 69 ++++++++++++++++++++++ crates/libtortillas/src/torrent/mod.rs | 1 + 2 files changed, 70 insertions(+) create mode 100644 crates/libtortillas/src/torrent/choking.rs diff --git a/crates/libtortillas/src/torrent/choking.rs b/crates/libtortillas/src/torrent/choking.rs new file mode 100644 index 0000000..9a3d357 --- /dev/null +++ b/crates/libtortillas/src/torrent/choking.rs @@ -0,0 +1,69 @@ +use crate::{ + peer::{PeerId, PeerStats}, + torrent::TorrentState, +}; + +pub(crate) const DEFAULT_UPLOAD_SLOTS: usize = 4; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ChokingDecision { + pub(crate) unchoked: Vec, +} + +pub(crate) fn select_unchoked_peers( + peers: &[PeerStats], _torrent_state: TorrentState, upload_slots: usize, +) -> ChokingDecision { + ChokingDecision { + unchoked: peers + .iter() + .filter(|peer| peer.interested) + .take(upload_slots) + .map(|peer| peer.id) + .collect(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn peer_id(value: u8) -> PeerId { + PeerId::from([value; 20]) + } + + fn stats(id: u8) -> PeerStats { + PeerStats { + id: peer_id(id), + interested: true, + choked: true, + download_rate: 0, + upload_rate: 0, + bytes_downloaded: 0, + bytes_uploaded: 0, + } + } + + #[test] + fn selector_only_includes_interested_peers() { + let mut not_interested = stats(2); + not_interested.interested = false; + let peers = [stats(1), not_interested, stats(3)]; + + let decision = select_unchoked_peers(&peers, TorrentState::Downloading, DEFAULT_UPLOAD_SLOTS); + + assert_eq!(decision.unchoked, vec![peer_id(1), peer_id(3)]); + } + + #[test] + fn selector_respects_upload_slot_limit() { + let peers = [stats(1), stats(2), stats(3), stats(4), stats(5)]; + + let decision = select_unchoked_peers(&peers, TorrentState::Downloading, DEFAULT_UPLOAD_SLOTS); + + assert_eq!(decision.unchoked.len(), DEFAULT_UPLOAD_SLOTS); + assert_eq!( + decision.unchoked, + vec![peer_id(1), peer_id(2), peer_id(3), peer_id(4)] + ); + } +} diff --git a/crates/libtortillas/src/torrent/mod.rs b/crates/libtortillas/src/torrent/mod.rs index 0840ea2..25a29d2 100644 --- a/crates/libtortillas/src/torrent/mod.rs +++ b/crates/libtortillas/src/torrent/mod.rs @@ -1,5 +1,6 @@ mod actor; mod block; +mod choking; mod export; mod handle; mod messages; From 852f9279901588398e656267b86caaa58914c59e Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Mon, 15 Jun 2026 13:07:00 -0700 Subject: [PATCH 04/14] feat: rank choking slots by rates --- crates/libtortillas/src/torrent/choking.rs | 58 ++++++++++++++++++++-- 1 file changed, 55 insertions(+), 3 deletions(-) diff --git a/crates/libtortillas/src/torrent/choking.rs b/crates/libtortillas/src/torrent/choking.rs index 9a3d357..89d1557 100644 --- a/crates/libtortillas/src/torrent/choking.rs +++ b/crates/libtortillas/src/torrent/choking.rs @@ -11,18 +11,36 @@ pub(crate) struct ChokingDecision { } pub(crate) fn select_unchoked_peers( - peers: &[PeerStats], _torrent_state: TorrentState, upload_slots: usize, + peers: &[PeerStats], torrent_state: TorrentState, upload_slots: usize, ) -> ChokingDecision { + let mut candidates: Vec<_> = peers + .iter() + .filter(|peer| peer.interested) + .copied() + .collect(); + candidates.sort_by(|left, right| { + rate_for(right, torrent_state) + .cmp(&rate_for(left, torrent_state)) + .then_with(|| left.id.id().cmp(right.id.id())) + }); + ChokingDecision { - unchoked: peers + unchoked: candidates .iter() - .filter(|peer| peer.interested) .take(upload_slots) .map(|peer| peer.id) .collect(), } } +fn rate_for(peer: &PeerStats, torrent_state: TorrentState) -> usize { + match torrent_state { + TorrentState::Downloading => peer.download_rate, + TorrentState::Seeding => peer.upload_rate, + TorrentState::Inactive => 0, + } +} + #[cfg(test)] mod tests { use super::*; @@ -43,6 +61,14 @@ mod tests { } } + fn with_rates(id: u8, download_rate: usize, upload_rate: usize) -> PeerStats { + PeerStats { + download_rate, + upload_rate, + ..stats(id) + } + } + #[test] fn selector_only_includes_interested_peers() { let mut not_interested = stats(2); @@ -66,4 +92,30 @@ mod tests { vec![peer_id(1), peer_id(2), peer_id(3), peer_id(4)] ); } + + #[test] + fn selector_ranks_downloading_peers_by_download_rate() { + let peers = [ + with_rates(1, 30, 500), + with_rates(2, 90, 10), + with_rates(3, 60, 1_000), + ]; + + let decision = select_unchoked_peers(&peers, TorrentState::Downloading, 2); + + assert_eq!(decision.unchoked, vec![peer_id(2), peer_id(3)]); + } + + #[test] + fn selector_ranks_seeding_peers_by_upload_rate() { + let peers = [ + with_rates(1, 100, 20), + with_rates(2, 10, 80), + with_rates(3, 60, 40), + ]; + + let decision = select_unchoked_peers(&peers, TorrentState::Seeding, 2); + + assert_eq!(decision.unchoked, vec![peer_id(2), peer_id(3)]); + } } From 7a1502dbdede6a8844560ac18c57c5e1d53f2907 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Mon, 15 Jun 2026 14:23:00 -0700 Subject: [PATCH 05/14] feat: rotate optimistic unchokes --- crates/libtortillas/src/torrent/choking.rs | 71 +++++++++++++++++++--- 1 file changed, 61 insertions(+), 10 deletions(-) diff --git a/crates/libtortillas/src/torrent/choking.rs b/crates/libtortillas/src/torrent/choking.rs index 89d1557..0037f94 100644 --- a/crates/libtortillas/src/torrent/choking.rs +++ b/crates/libtortillas/src/torrent/choking.rs @@ -8,10 +8,11 @@ pub(crate) const DEFAULT_UPLOAD_SLOTS: usize = 4; #[derive(Clone, Debug, Eq, PartialEq)] pub(crate) struct ChokingDecision { pub(crate) unchoked: Vec, + pub(crate) optimistic: Option, } pub(crate) fn select_unchoked_peers( - peers: &[PeerStats], torrent_state: TorrentState, upload_slots: usize, + peers: &[PeerStats], torrent_state: TorrentState, upload_slots: usize, optimistic_index: usize, ) -> ChokingDecision { let mut candidates: Vec<_> = peers .iter() @@ -24,12 +25,38 @@ pub(crate) fn select_unchoked_peers( .then_with(|| left.id.id().cmp(right.id.id())) }); + if upload_slots == 0 { + return ChokingDecision { + unchoked: Vec::new(), + optimistic: None, + }; + } + + let reserve_optimistic_slot = candidates.len() > upload_slots; + let regular_slots = if reserve_optimistic_slot { + upload_slots.saturating_sub(1) + } else { + upload_slots + }; + + let mut unchoked: Vec<_> = candidates + .iter() + .take(regular_slots) + .map(|peer| peer.id) + .collect(); + + let optimistic = if reserve_optimistic_slot { + let optimistic_candidates = &candidates[regular_slots..]; + let peer = optimistic_candidates[optimistic_index % optimistic_candidates.len()].id; + unchoked.push(peer); + Some(peer) + } else { + None + }; + ChokingDecision { - unchoked: candidates - .iter() - .take(upload_slots) - .map(|peer| peer.id) - .collect(), + unchoked, + optimistic, } } @@ -75,18 +102,22 @@ mod tests { not_interested.interested = false; let peers = [stats(1), not_interested, stats(3)]; - let decision = select_unchoked_peers(&peers, TorrentState::Downloading, DEFAULT_UPLOAD_SLOTS); + let decision = + select_unchoked_peers(&peers, TorrentState::Downloading, DEFAULT_UPLOAD_SLOTS, 0); assert_eq!(decision.unchoked, vec![peer_id(1), peer_id(3)]); + assert_eq!(decision.optimistic, None); } #[test] fn selector_respects_upload_slot_limit() { let peers = [stats(1), stats(2), stats(3), stats(4), stats(5)]; - let decision = select_unchoked_peers(&peers, TorrentState::Downloading, DEFAULT_UPLOAD_SLOTS); + let decision = + select_unchoked_peers(&peers, TorrentState::Downloading, DEFAULT_UPLOAD_SLOTS, 0); assert_eq!(decision.unchoked.len(), DEFAULT_UPLOAD_SLOTS); + assert_eq!(decision.optimistic, Some(peer_id(4))); assert_eq!( decision.unchoked, vec![peer_id(1), peer_id(2), peer_id(3), peer_id(4)] @@ -101,9 +132,10 @@ mod tests { with_rates(3, 60, 1_000), ]; - let decision = select_unchoked_peers(&peers, TorrentState::Downloading, 2); + let decision = select_unchoked_peers(&peers, TorrentState::Downloading, 2, 0); assert_eq!(decision.unchoked, vec![peer_id(2), peer_id(3)]); + assert_eq!(decision.optimistic, Some(peer_id(3))); } #[test] @@ -114,8 +146,27 @@ mod tests { with_rates(3, 60, 40), ]; - let decision = select_unchoked_peers(&peers, TorrentState::Seeding, 2); + let decision = select_unchoked_peers(&peers, TorrentState::Seeding, 2, 0); assert_eq!(decision.unchoked, vec![peer_id(2), peer_id(3)]); + assert_eq!(decision.optimistic, Some(peer_id(3))); + } + + #[test] + fn selector_rotates_optimistic_unchoke() { + let peers = [ + with_rates(1, 100, 0), + with_rates(2, 90, 0), + with_rates(3, 80, 0), + with_rates(4, 70, 0), + ]; + + let first = select_unchoked_peers(&peers, TorrentState::Downloading, 2, 0); + let second = select_unchoked_peers(&peers, TorrentState::Downloading, 2, 1); + + assert_eq!(first.unchoked, vec![peer_id(1), peer_id(2)]); + assert_eq!(first.optimistic, Some(peer_id(2))); + assert_eq!(second.unchoked, vec![peer_id(1), peer_id(3)]); + assert_eq!(second.optimistic, Some(peer_id(3))); } } From d8a8c46480033a864d98b93341f2516a1fe9903a Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Mon, 15 Jun 2026 15:41:00 -0700 Subject: [PATCH 06/14] feat: track choking rotation cadence --- crates/libtortillas/src/torrent/choking.rs | 71 ++++++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/crates/libtortillas/src/torrent/choking.rs b/crates/libtortillas/src/torrent/choking.rs index 0037f94..470862d 100644 --- a/crates/libtortillas/src/torrent/choking.rs +++ b/crates/libtortillas/src/torrent/choking.rs @@ -4,6 +4,56 @@ use crate::{ }; pub(crate) const DEFAULT_UPLOAD_SLOTS: usize = 4; +pub(crate) const OPTIMISTIC_UNCHOKE_ROUNDS: usize = 3; + +#[derive(Clone, Debug)] +pub(crate) struct ChokingScheduler { + upload_slots: usize, + optimistic_unchoke_rounds: usize, + optimistic_index: usize, + rounds_since_optimistic_rotation: usize, +} + +impl Default for ChokingScheduler { + fn default() -> Self { + Self::new(DEFAULT_UPLOAD_SLOTS, OPTIMISTIC_UNCHOKE_ROUNDS) + } +} + +impl ChokingScheduler { + pub(crate) fn new(upload_slots: usize, optimistic_unchoke_rounds: usize) -> Self { + Self { + upload_slots, + optimistic_unchoke_rounds: optimistic_unchoke_rounds.max(1), + optimistic_index: 0, + rounds_since_optimistic_rotation: 0, + } + } + + pub(crate) fn decide( + &mut self, peers: &[PeerStats], torrent_state: TorrentState, + ) -> ChokingDecision { + let decision = select_unchoked_peers( + peers, + torrent_state, + self.upload_slots, + self.optimistic_index, + ); + + if decision.optimistic.is_some() { + self.rounds_since_optimistic_rotation += 1; + if self.rounds_since_optimistic_rotation >= self.optimistic_unchoke_rounds { + self.rounds_since_optimistic_rotation = 0; + self.optimistic_index = self.optimistic_index.wrapping_add(1); + } + } else { + self.rounds_since_optimistic_rotation = 0; + self.optimistic_index = 0; + } + + decision + } +} #[derive(Clone, Debug, Eq, PartialEq)] pub(crate) struct ChokingDecision { @@ -169,4 +219,25 @@ mod tests { assert_eq!(second.unchoked, vec![peer_id(1), peer_id(3)]); assert_eq!(second.optimistic, Some(peer_id(3))); } + + #[test] + fn scheduler_rotates_optimistic_unchoke_after_configured_rounds() { + let peers = [ + with_rates(1, 100, 0), + with_rates(2, 90, 0), + with_rates(3, 80, 0), + with_rates(4, 70, 0), + ]; + let mut scheduler = ChokingScheduler::new(2, 3); + + let first = scheduler.decide(&peers, TorrentState::Downloading); + let second = scheduler.decide(&peers, TorrentState::Downloading); + let third = scheduler.decide(&peers, TorrentState::Downloading); + let fourth = scheduler.decide(&peers, TorrentState::Downloading); + + assert_eq!(first.optimistic, Some(peer_id(2))); + assert_eq!(second.optimistic, Some(peer_id(2))); + assert_eq!(third.optimistic, Some(peer_id(2))); + assert_eq!(fourth.optimistic, Some(peer_id(3))); + } } From 30a8fc4b105ab7f66628d0743c14a07faa154ad3 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Tue, 16 Jun 2026 09:16:00 -0700 Subject: [PATCH 07/14] feat: store torrent choking state --- crates/libtortillas/src/torrent/actor.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/libtortillas/src/torrent/actor.rs b/crates/libtortillas/src/torrent/actor.rs index 8752eff..def83f7 100644 --- a/crates/libtortillas/src/torrent/actor.rs +++ b/crates/libtortillas/src/torrent/actor.rs @@ -23,7 +23,7 @@ use librqbit_utp::UtpSocketUdp; use tokio::sync::oneshot; use tracing::{debug, error, info, instrument, trace, warn}; -use super::util; +use super::{choking::ChokingScheduler, util}; use crate::{ errors::TorrentError, hashes::InfoHash, @@ -102,6 +102,8 @@ pub(crate) struct TorrentActor { pub state: TorrentState, /// Scheduler for managing piece and block requests pub(super) piece_scheduler: PieceScheduler, + /// Scheduler for BEP 3 upload choking decisions. + pub(super) choking_scheduler: ChokingScheduler, pub(super) start_time: Option, /// The number of peers we need to have before we start downloading, defaults @@ -482,6 +484,7 @@ impl Actor for TorrentActor { piece_store, state: TorrentState::default(), piece_scheduler: PieceScheduler::new(piece_count), + choking_scheduler: ChokingScheduler::default(), start_time: None, sufficient_peers: sufficient_peers.unwrap_or(6), autostart: autostart.unwrap_or(true), @@ -770,6 +773,7 @@ mod tests { )), state: TorrentState::Inactive, piece_scheduler, + choking_scheduler: ChokingScheduler::default(), start_time: None, sufficient_peers: 6, autostart: false, From 7e02182a23b842e5f097a99639359acd66e8d3c1 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Tue, 16 Jun 2026 10:29:00 -0700 Subject: [PATCH 08/14] feat: schedule periodic rechokes --- crates/libtortillas/src/torrent/actor.rs | 43 +++++++++++++++++++-- crates/libtortillas/src/torrent/choking.rs | 3 ++ crates/libtortillas/src/torrent/messages.rs | 5 +++ 3 files changed, 48 insertions(+), 3 deletions(-) diff --git a/crates/libtortillas/src/torrent/actor.rs b/crates/libtortillas/src/torrent/actor.rs index def83f7..092a4c6 100644 --- a/crates/libtortillas/src/torrent/actor.rs +++ b/crates/libtortillas/src/torrent/actor.rs @@ -18,12 +18,15 @@ use kameo::{ mailbox::{MailboxReceiver, Signal}, supervision::{RestartPolicy, SupervisionStrategy}, }; -use kameo_actors::scheduler::Scheduler; +use kameo_actors::scheduler::{Scheduler, SetTimeout}; use librqbit_utp::UtpSocketUdp; -use tokio::sync::oneshot; +use tokio::{sync::oneshot, task::AbortHandle}; use tracing::{debug, error, info, instrument, trace, warn}; -use super::{choking::ChokingScheduler, util}; +use super::{ + choking::{ChokingScheduler, RECHOKE_INTERVAL}, + util, +}; use crate::{ errors::TorrentError, hashes::InfoHash, @@ -104,6 +107,7 @@ pub(crate) struct TorrentActor { pub(super) piece_scheduler: PieceScheduler, /// Scheduler for BEP 3 upload choking decisions. pub(super) choking_scheduler: ChokingScheduler, + pub(super) next_rechoke: Option, pub(super) start_time: Option, /// The number of peers we need to have before we start downloading, defaults @@ -237,6 +241,34 @@ impl TorrentActor { state = ?self.state, "Started torrenting process" ); + + self.schedule_next_rechoke().await; + } + + pub(super) async fn schedule_next_rechoke(&mut self) { + if !matches!( + self.state, + TorrentState::Downloading | TorrentState::Seeding + ) { + return; + } + + if let Some(next_rechoke) = self.next_rechoke.take() { + next_rechoke.abort(); + } + + match self + .scheduler + .ask(SetTimeout::new( + self.actor_ref.downgrade(), + RECHOKE_INTERVAL, + super::commands::Rechoke, + )) + .await + { + Ok(next_rechoke) => self.next_rechoke = Some(next_rechoke), + Err(err) => warn!(?err, "Failed to schedule next rechoke"), + } } pub(super) fn send_ready_hooks(&mut self) { @@ -485,6 +517,7 @@ impl Actor for TorrentActor { state: TorrentState::default(), piece_scheduler: PieceScheduler::new(piece_count), choking_scheduler: ChokingScheduler::default(), + next_rechoke: None, start_time: None, sufficient_peers: sufficient_peers.unwrap_or(6), autostart: autostart.unwrap_or(true), @@ -514,6 +547,9 @@ impl Actor for TorrentActor { for tracker in self.trackers.values() { tracker.kill(); } + if let Some(next_rechoke) = self.next_rechoke.take() { + next_rechoke.abort(); + } self.piece_store.kill(); self.scheduler.kill(); @@ -774,6 +810,7 @@ mod tests { state: TorrentState::Inactive, piece_scheduler, choking_scheduler: ChokingScheduler::default(), + next_rechoke: None, start_time: None, sufficient_peers: 6, autostart: false, diff --git a/crates/libtortillas/src/torrent/choking.rs b/crates/libtortillas/src/torrent/choking.rs index 470862d..b178e7c 100644 --- a/crates/libtortillas/src/torrent/choking.rs +++ b/crates/libtortillas/src/torrent/choking.rs @@ -1,9 +1,12 @@ +use std::time::Duration; + use crate::{ peer::{PeerId, PeerStats}, torrent::TorrentState, }; pub(crate) const DEFAULT_UPLOAD_SLOTS: usize = 4; +pub(crate) const RECHOKE_INTERVAL: Duration = Duration::from_secs(10); pub(crate) const OPTIMISTIC_UNCHOKE_ROUNDS: usize = 3; #[derive(Clone, Debug)] diff --git a/crates/libtortillas/src/torrent/messages.rs b/crates/libtortillas/src/torrent/messages.rs index ce67421..13cb0d3 100644 --- a/crates/libtortillas/src/torrent/messages.rs +++ b/crates/libtortillas/src/torrent/messages.rs @@ -235,6 +235,11 @@ pub(crate) mod commands { } } + #[message(derive(Debug, Clone, Copy))] + pub(crate) async fn rechoke(&mut self) { + self.schedule_next_rechoke().await; + } + /// A hook that is called when the torrent is ready to start downloading. /// This is used to implement /// [`Torrent::poll_ready`](crate::torrent::Torrent::poll_ready). From f20397c5c9d49b8f6b077215051a5bef51f30aa3 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Tue, 16 Jun 2026 11:47:00 -0700 Subject: [PATCH 09/14] feat: apply choking decisions --- .../libtortillas/src/torrent/choking_flow.rs | 56 +++++++++++++++++++ crates/libtortillas/src/torrent/messages.rs | 1 + crates/libtortillas/src/torrent/mod.rs | 1 + 3 files changed, 58 insertions(+) create mode 100644 crates/libtortillas/src/torrent/choking_flow.rs diff --git a/crates/libtortillas/src/torrent/choking_flow.rs b/crates/libtortillas/src/torrent/choking_flow.rs new file mode 100644 index 0000000..ab3c4fa --- /dev/null +++ b/crates/libtortillas/src/torrent/choking_flow.rs @@ -0,0 +1,56 @@ +use std::collections::HashSet; + +use tracing::{trace, warn}; + +use super::TorrentActor; +use crate::peer::{ + PeerStats, + commands::{SetChoked, Stats}, +}; + +impl TorrentActor { + pub(super) async fn rechoke_peers(&mut self) { + let peer_stats = self.peer_stats().await; + let decision = self.choking_scheduler.decide(&peer_stats, self.state); + let unchoked: HashSet<_> = decision.unchoked.iter().copied().collect(); + + trace!( + unchoked = decision.unchoked.len(), + optimistic = ?decision.optimistic, + "Applying choking decision" + ); + + for stats in peer_stats { + let choked = !unchoked.contains(&stats.id); + if stats.choked == choked { + continue; + } + + let Some(actor) = self.peers.get(&stats.id) else { + continue; + }; + + if let Err(err) = actor.tell(SetChoked { choked }).await { + warn!(?err, peer_id = %stats.id, choked, "Failed to update peer choke state"); + } + } + } + + async fn peer_stats(&self) -> Vec { + let mut snapshots = Vec::with_capacity(self.peers.len()); + + for (peer_id, actor) in &self.peers { + if !actor.is_alive() { + continue; + } + + match actor.ask(Stats).await { + Ok(Some(stats)) => snapshots.push(stats), + Ok(None) => trace!(%peer_id, "Peer stats unavailable"), + Err(err) => warn!(?err, %peer_id, "Failed to collect peer stats"), + } + } + + snapshots + } +} diff --git a/crates/libtortillas/src/torrent/messages.rs b/crates/libtortillas/src/torrent/messages.rs index 13cb0d3..de8d1f1 100644 --- a/crates/libtortillas/src/torrent/messages.rs +++ b/crates/libtortillas/src/torrent/messages.rs @@ -237,6 +237,7 @@ pub(crate) mod commands { #[message(derive(Debug, Clone, Copy))] pub(crate) async fn rechoke(&mut self) { + self.rechoke_peers().await; self.schedule_next_rechoke().await; } diff --git a/crates/libtortillas/src/torrent/mod.rs b/crates/libtortillas/src/torrent/mod.rs index 25a29d2..ca3bb3c 100644 --- a/crates/libtortillas/src/torrent/mod.rs +++ b/crates/libtortillas/src/torrent/mod.rs @@ -1,6 +1,7 @@ mod actor; mod block; mod choking; +mod choking_flow; mod export; mod handle; mod messages; From c9c655a0d05f26d37b60a737d477ad11ebadf0f9 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Tue, 16 Jun 2026 13:21:00 -0700 Subject: [PATCH 10/14] fix: guard inactive rechoke rounds --- crates/libtortillas/src/torrent/choking.rs | 10 ++++++++++ crates/libtortillas/src/torrent/choking_flow.rs | 17 ++++++++++++++--- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/crates/libtortillas/src/torrent/choking.rs b/crates/libtortillas/src/torrent/choking.rs index b178e7c..df4fc1f 100644 --- a/crates/libtortillas/src/torrent/choking.rs +++ b/crates/libtortillas/src/torrent/choking.rs @@ -177,6 +177,16 @@ mod tests { ); } + #[test] + fn selector_chokes_everyone_without_upload_slots() { + let peers = [stats(1), stats(2), stats(3)]; + + let decision = select_unchoked_peers(&peers, TorrentState::Downloading, 0, 0); + + assert!(decision.unchoked.is_empty()); + assert_eq!(decision.optimistic, None); + } + #[test] fn selector_ranks_downloading_peers_by_download_rate() { let peers = [ diff --git a/crates/libtortillas/src/torrent/choking_flow.rs b/crates/libtortillas/src/torrent/choking_flow.rs index ab3c4fa..986e3e0 100644 --- a/crates/libtortillas/src/torrent/choking_flow.rs +++ b/crates/libtortillas/src/torrent/choking_flow.rs @@ -3,13 +3,24 @@ use std::collections::HashSet; use tracing::{trace, warn}; use super::TorrentActor; -use crate::peer::{ - PeerStats, - commands::{SetChoked, Stats}, +use crate::{ + peer::{ + PeerStats, + commands::{SetChoked, Stats}, + }, + torrent::TorrentState, }; impl TorrentActor { pub(super) async fn rechoke_peers(&mut self) { + if !matches!( + self.state, + TorrentState::Downloading | TorrentState::Seeding + ) { + trace!(state = ?self.state, "Skipping rechoke while torrent is inactive"); + return; + } + let peer_stats = self.peer_stats().await; let decision = self.choking_scheduler.decide(&peer_stats, self.state); let unchoked: HashSet<_> = decision.unchoked.iter().copied().collect(); From bd7ccca2f839d4f0ccbf7fbfcb419a4d481840cd Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Tue, 16 Jun 2026 14:36:00 -0700 Subject: [PATCH 11/14] docs: document choking scheduler --- crates/libtortillas/src/ARCHITECTURE.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/crates/libtortillas/src/ARCHITECTURE.md b/crates/libtortillas/src/ARCHITECTURE.md index 6718098..f152b1e 100644 --- a/crates/libtortillas/src/ARCHITECTURE.md +++ b/crates/libtortillas/src/ARCHITECTURE.md @@ -9,3 +9,11 @@ The library is organized around a small actor hierarchy: Module facades should export stable public types while keeping actor internals private to the crate. Domain types such as torrent state, storage strategy, exported snapshots, tracker model types, and tracker stats live outside actor files so actors can focus on orchestration. + +## Choking + +`TorrentActor` owns the BEP 3 choking scheduler for its swarm. Active torrents run a rechoke round every 10 seconds, collect peer-local transfer stats from `PeerActor`, and keep at most four interested peers unchoked. + +While downloading, regular upload slots are assigned to interested peers with the highest recent download rate. While seeding, regular slots are assigned by recent upload rate. When more interested peers exist than upload slots, one slot is reserved for an optimistic unchoke and rotates every third rechoke round. + +`PeerActor` remains responsible for wire-level enforcement: it sends `Choke` and `Unchoke` messages when the torrent scheduler changes state, ignores piece requests from choked peers, and records uploaded bytes when serving piece data. From cd099dce966468ec4b132beaff2a3b9651a2dd01 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Tue, 16 Jun 2026 16:04:00 -0700 Subject: [PATCH 12/14] test: cover optimistic reset --- crates/libtortillas/src/torrent/choking.rs | 23 ++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/crates/libtortillas/src/torrent/choking.rs b/crates/libtortillas/src/torrent/choking.rs index df4fc1f..b5e351b 100644 --- a/crates/libtortillas/src/torrent/choking.rs +++ b/crates/libtortillas/src/torrent/choking.rs @@ -253,4 +253,27 @@ mod tests { assert_eq!(third.optimistic, Some(peer_id(2))); assert_eq!(fourth.optimistic, Some(peer_id(3))); } + + #[test] + fn scheduler_resets_optimistic_rotation_without_extra_candidates() { + let peers = [with_rates(1, 100, 0), with_rates(2, 90, 0)]; + let mut scheduler = ChokingScheduler::new(4, 3); + + for _ in 0..4 { + let decision = scheduler.decide(&peers, TorrentState::Downloading); + assert_eq!(decision.unchoked, vec![peer_id(1), peer_id(2)]); + assert_eq!(decision.optimistic, None); + } + + let crowded_peers = [ + with_rates(1, 100, 0), + with_rates(2, 90, 0), + with_rates(3, 80, 0), + with_rates(4, 70, 0), + with_rates(5, 60, 0), + ]; + let decision = scheduler.decide(&crowded_peers, TorrentState::Downloading); + + assert_eq!(decision.optimistic, Some(peer_id(4))); + } } From 60ed046a91982914f6c618ad52eab9e0371b244c Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Sun, 28 Jun 2026 20:06:43 -0700 Subject: [PATCH 13/14] fix: address choking review findings --- crates/libtortillas/src/peer/actor.rs | 6 ++---- crates/libtortillas/src/torrent/actor.rs | 13 ++++++++++++- crates/libtortillas/src/torrent/piece_flow.rs | 2 ++ 3 files changed, 16 insertions(+), 5 deletions(-) diff --git a/crates/libtortillas/src/peer/actor.rs b/crates/libtortillas/src/peer/actor.rs index 9a7f2d2..5dc4d2d 100644 --- a/crates/libtortillas/src/peer/actor.rs +++ b/crates/libtortillas/src/peer/actor.rs @@ -496,17 +496,15 @@ impl Message for PeerActor { data_len = data.len(), "Received piece data" ); - self.peer.increment_bytes_downloaded(data.len()); let key = (index as usize, offset as usize, data.len()); // Only send the piece to the supervisor if they request it or it hasn't been // cancelled - if self.pending_block_requests.contains(&key) { - self.pending_block_requests.remove(&key); - + if self.pending_block_requests.remove(&key) { let Some(peer_id) = self.peer.id else { warn!("Received piece from peer without id; ignoring"); return; }; + self.peer.increment_bytes_downloaded(data.len()); let supervisor_msg = torrent::events::IncomingPiece { peer_id, index: index as usize, diff --git a/crates/libtortillas/src/torrent/actor.rs b/crates/libtortillas/src/torrent/actor.rs index 092a4c6..0d19cac 100644 --- a/crates/libtortillas/src/torrent/actor.rs +++ b/crates/libtortillas/src/torrent/actor.rs @@ -242,6 +242,7 @@ impl TorrentActor { "Started torrenting process" ); + self.rechoke_peers().await; self.schedule_next_rechoke().await; } @@ -267,7 +268,17 @@ impl TorrentActor { .await { Ok(next_rechoke) => self.next_rechoke = Some(next_rechoke), - Err(err) => warn!(?err, "Failed to schedule next rechoke"), + Err(err) => { + warn!(?err, "Failed to schedule next rechoke; using local timeout"); + let actor_ref = self.actor_ref.clone(); + let fallback = tokio::spawn(async move { + tokio::time::sleep(RECHOKE_INTERVAL).await; + if let Err(err) = actor_ref.tell(super::commands::Rechoke).await { + warn!(?err, "Failed to run fallback rechoke"); + } + }); + self.next_rechoke = Some(fallback.abort_handle()); + } } } diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index cba8bd8..b760317 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -232,6 +232,8 @@ impl TorrentActor { .await; self.broadcast_to_trackers(Announce).await; info!("Torrenting process completed, switching to seeding mode"); + self.rechoke_peers().await; + self.schedule_next_rechoke().await; } } From f9e8d694698fbf63d0948bf90ca00cf073c9d38c Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Mon, 29 Jun 2026 23:13:33 -0700 Subject: [PATCH 14/14] fix: stabilize choking progress checks --- crates/libtortillas/src/torrent/actor.rs | 55 ++++++++++++------- .../libtortillas/src/torrent/choking_flow.rs | 52 ++++++++++++------ 2 files changed, 72 insertions(+), 35 deletions(-) diff --git a/crates/libtortillas/src/torrent/actor.rs b/crates/libtortillas/src/torrent/actor.rs index 0d19cac..344489a 100644 --- a/crates/libtortillas/src/torrent/actor.rs +++ b/crates/libtortillas/src/torrent/actor.rs @@ -582,14 +582,20 @@ mod tests { use std::{path::PathBuf, time::Duration}; use librqbit_utp::UtpSocket; - use tokio::{fs, time::sleep}; + use tokio::{ + fs, + time::{sleep, timeout}, + }; use tracing::trace; use super::*; use crate::{ metainfo::MetaInfo, testing, - torrent::{BLOCK_SIZE, Torrent, TorrentExport, commands::HasInfoDict}, + torrent::{ + BLOCK_SIZE, Torrent, TorrentExport, + commands::{ExportState, HasInfoDict}, + }, }; #[tokio::test(flavor = "multi_thread")] @@ -722,27 +728,38 @@ mod tests { assert!(torrent.poll_ready().await.is_ok()); - loop { - let mut entries = fs::read_dir(&piece_path).await.unwrap(); - let mut found_piece = false; - while let Some(entry) = entries.next_entry().await.unwrap() { - let path = entry.path(); - if let Some(ext) = path.extension() - && ext == "piece" - { - let metadata = entry.metadata().await.unwrap(); - if metadata.len() == info_dict.piece_length { - found_piece = true; - break; + let wrote_piece_block = timeout(Duration::from_secs(60), async { + loop { + let export = actor.ask(ExportState).await.unwrap(); + let has_persisted_progress = export.bitfield.count_ones() > 0 + || export + .block_map + .iter() + .any(|entry| entry.value().count_ones() > 0); + + if has_persisted_progress { + let mut entries = fs::read_dir(&piece_path).await.unwrap(); + while let Some(entry) = entries.next_entry().await.unwrap() { + let path = entry.path(); + if let Some(ext) = path.extension() + && ext == "piece" + && entry.metadata().await.unwrap().len() > 0 + { + return true; + } } } + + sleep(Duration::from_millis(200)).await; } - if found_piece { - break; - } + }) + .await + .unwrap_or(false); - sleep(Duration::from_millis(200)).await; - } + assert!( + wrote_piece_block, + "timed out waiting for piece storage write" + ); actor.stop_gracefully().await.unwrap(); clear_piece_files(&piece_path).await; diff --git a/crates/libtortillas/src/torrent/choking_flow.rs b/crates/libtortillas/src/torrent/choking_flow.rs index 986e3e0..3b8373a 100644 --- a/crates/libtortillas/src/torrent/choking_flow.rs +++ b/crates/libtortillas/src/torrent/choking_flow.rs @@ -1,16 +1,22 @@ -use std::collections::HashSet; +use std::{collections::HashSet, time::Duration}; +use futures::{StreamExt, stream}; +use kameo::actor::ActorRef; +use tokio::time::timeout; use tracing::{trace, warn}; use super::TorrentActor; use crate::{ peer::{ - PeerStats, + PeerActor, PeerStats, commands::{SetChoked, Stats}, }, torrent::TorrentState, }; +const PEER_STATS_CONCURRENCY: usize = 32; +const PEER_STATS_TIMEOUT: Duration = Duration::from_millis(250); + impl TorrentActor { pub(super) async fn rechoke_peers(&mut self) { if !matches!( @@ -48,20 +54,34 @@ impl TorrentActor { } async fn peer_stats(&self) -> Vec { - let mut snapshots = Vec::with_capacity(self.peers.len()); - - for (peer_id, actor) in &self.peers { - if !actor.is_alive() { - continue; - } - - match actor.ask(Stats).await { - Ok(Some(stats)) => snapshots.push(stats), - Ok(None) => trace!(%peer_id, "Peer stats unavailable"), - Err(err) => warn!(?err, %peer_id, "Failed to collect peer stats"), - } - } + let actor_refs: Vec<(crate::peer::PeerId, ActorRef)> = self + .peers + .iter() + .filter(|(_, actor)| actor.is_alive()) + .map(|(peer_id, actor)| (*peer_id, actor.clone())) + .collect(); - snapshots + stream::iter(actor_refs) + .map(|(peer_id, actor)| async move { + match timeout(PEER_STATS_TIMEOUT, actor.ask(Stats)).await { + Ok(Ok(Some(stats))) => Some(stats), + Ok(Ok(None)) => { + trace!(%peer_id, "Peer stats unavailable"); + None + } + Ok(Err(err)) => { + warn!(?err, %peer_id, "Failed to collect peer stats"); + None + } + Err(_) => { + trace!(%peer_id, "Timed out collecting peer stats"); + None + } + } + }) + .buffer_unordered(PEER_STATS_CONCURRENCY) + .filter_map(std::future::ready) + .collect() + .await } }