leyoda/rqbit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Cleanups

Browse source
This commit is contained in:
Igor Katson 2024-08-21 16:20:46 +01:00
parent ad5f62b9b9
commit 3067ad21d5
No known key found for this signature in database
GPG key ID: B4EC22B66D61A3F5
3 changed files with 19 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

12
crates/librqbit/src/torrent_state/live/mod.rs
View file

@ -214,12 +214,16 @@ impl TorrentStateLive {
paused: TorrentStatePaused, paused: TorrentStatePaused,
fatal_errors_tx: tokio::sync::oneshot::Sender<anyhow::Error>, fatal_errors_tx: tokio::sync::oneshot::Sender<anyhow::Error>,
cancellation_token: CancellationToken, cancellation_token: CancellationToken,
session_stats: Arc<AtomicSessionStats>,
) -> anyhow::Result<Arc<Self>> { ) -> anyhow::Result<Arc<Self>> {
let (peer_queue_tx, peer_queue_rx) = unbounded_channel(); let (peer_queue_tx, peer_queue_rx) = unbounded_channel();
let session = paused
let down_speed_estimator = SpeedEstimator::new(5); .info
let up_speed_estimator = SpeedEstimator::new(5); .session
.upgrade()
.context("session is dead, cannot start torrent")?;
let session_stats = session.stats.atomic.clone();
let down_speed_estimator = SpeedEstimator::default();
let up_speed_estimator = SpeedEstimator::default();
let have_bytes = paused.chunk_tracker.get_hns().have_bytes; let have_bytes = paused.chunk_tracker.get_hns().have_bytes;
let lengths = *paused.chunk_tracker.get_lengths(); let lengths = *paused.chunk_tracker.get_lengths();

19
crates/librqbit/src/torrent_state/mod.rs
View file

@ -218,6 +218,7 @@ impl ManagedTorrent {
.upgrade() .upgrade()
.context("session is dead, cannot start torrent")?; .context("session is dead, cannot start torrent")?;
let mut g = self.locked.write(); let mut g = self.locked.write();
let cancellation_token = session.cancellation_token().child_token();
let spawn_fatal_errors_receiver = let spawn_fatal_errors_receiver =
|state: &Arc<Self>, |state: &Arc<Self>,
@ -297,7 +298,7 @@ impl ManagedTorrent {
drop(g); drop(g);
let t = self.clone(); let t = self.clone();
let span = self.shared().span.clone(); let span = self.shared().span.clone();
let token = session.cancellation_token().child_token().clone(); let token = cancellation_token.clone();
spawn_with_cancel( spawn_with_cancel(
error_span!(parent: span.clone(), "initialize_and_start"), error_span!(parent: span.clone(), "initialize_and_start"),
@ -326,12 +327,7 @@ impl ManagedTorrent {
} }
let (tx, rx) = tokio::sync::oneshot::channel(); let (tx, rx) = tokio::sync::oneshot::channel();
let live = TorrentStateLive::new( let live = TorrentStateLive::new(paused, tx, cancellation_token)?;
paused,
tx,
session.cancellation_token().child_token(),
session.stats.atomic.clone(),
)?;
g.state = ManagedTorrentState::Live(live.clone()); g.state = ManagedTorrentState::Live(live.clone());
drop(g); drop(g);
@ -356,16 +352,11 @@ impl ManagedTorrent {
ManagedTorrentState::Paused(_) => { ManagedTorrentState::Paused(_) => {
let paused = g.state.take().assert_paused(); let paused = g.state.take().assert_paused();
let (tx, rx) = tokio::sync::oneshot::channel(); let (tx, rx) = tokio::sync::oneshot::channel();
let live = TorrentStateLive::new( let live = TorrentStateLive::new(paused, tx, cancellation_token.clone())?;
paused,
tx,
session.cancellation_token().child_token().clone(),
session.stats.atomic.clone(),
)?;
g.state = ManagedTorrentState::Live(live.clone()); g.state = ManagedTorrentState::Live(live.clone());
drop(g); drop(g);
spawn_fatal_errors_receiver(self, rx, session.cancellation_token().child_token()); spawn_fatal_errors_receiver(self, rx, cancellation_token);
spawn_peer_adder(&live, peer_rx); spawn_peer_adder(&live, peer_rx);
Ok(()) Ok(())
} }

6
crates/librqbit_core/src/speed_estimator.rs
View file

@ -19,6 +19,12 @@ pub struct SpeedEstimator {
time_remaining_millis: AtomicU64, time_remaining_millis: AtomicU64,
} }
impl Default for SpeedEstimator {
fn default() -> Self {
Self::new(5)
}
}
impl SpeedEstimator { impl SpeedEstimator {
pub fn new(window_seconds: usize) -> Self { pub fn new(window_seconds: usize) -> Self {
assert!(window_seconds > 1); assert!(window_seconds > 1);