From ad7b59ea3c9277164a22a1088e79480e192de25b Mon Sep 17 00:00:00 2001 From: Igor Katson Date: Wed, 21 Aug 2024 15:57:18 +0100 Subject: [PATCH] Remove some args in start() function --- crates/librqbit/src/session.rs | 25 ++++------------ crates/librqbit/src/torrent_state/mod.rs | 37 ++++++++++-------------- 2 files changed, 20 insertions(+), 42 deletions(-) diff --git a/crates/librqbit/src/session.rs b/crates/librqbit/src/session.rs index eab7a5e..aa5261d 100644 --- a/crates/librqbit/src/session.rs +++ b/crates/librqbit/src/session.rs @@ -101,7 +101,7 @@ pub struct Session { peer_id: Id20, dht: Option, persistence: Option>, - bitv_factory: Arc, + pub(crate) bitv_factory: Arc, peer_opts: PeerConnectionOptions, spawner: BlockingSpawner, next_id: AtomicUsize, @@ -117,9 +117,8 @@ pub struct Session { default_storage_factory: Option, reqwest_client: reqwest::Client, - connector: Arc, - - concurrent_initialize_semaphore: Arc, + pub(crate) connector: Arc, + pub(crate) concurrent_initialize_semaphore: Arc, root_span: Option, @@ -1186,14 +1185,7 @@ impl Session { let _ = span.enter(); managed_torrent - .start( - peer_rx, - opts.paused, - self.cancellation_token.child_token(), - self.concurrent_initialize_semaphore.clone(), - self.bitv_factory.clone(), - self.stats.atomic.clone(), - ) + .start(peer_rx, opts.paused) .context("error starting torrent")?; } @@ -1343,14 +1335,7 @@ impl Session { self.tcp_listen_port, handle.info().options.force_tracker_interval, )?; - handle.start( - peer_rx, - false, - self.cancellation_token.child_token(), - self.concurrent_initialize_semaphore.clone(), - self.bitv_factory.clone(), - self.stats.atomic.clone(), - )?; + handle.start(peer_rx, false)?; self.try_update_persistence_metadata(handle).await; Ok(()) } diff --git a/crates/librqbit/src/torrent_state/mod.rs b/crates/librqbit/src/torrent_state/mod.rs index ac7b3b8..c8c5846 100644 --- a/crates/librqbit/src/torrent_state/mod.rs +++ b/crates/librqbit/src/torrent_state/mod.rs @@ -34,10 +34,8 @@ use tracing::debug; use tracing::error_span; use tracing::warn; -use crate::bitv_factory::BitVFactory; use crate::chunk_tracker::ChunkTracker; use crate::session::TorrentId; -use crate::session_stats::atomic::AtomicSessionStats; use crate::spawn_utils::BlockingSpawner; use crate::storage::BoxStorageFactory; use crate::stream_connect::StreamConnector; @@ -210,11 +208,11 @@ impl ManagedTorrent { self: &Arc, peer_rx: Option, start_paused: bool, - live_cancellation_token: CancellationToken, - init_semaphore: Arc, - bitv_factory: Arc, - session_stats: Arc, ) -> anyhow::Result<()> { + let session = self + .session + .upgrade() + .context("session is dead, cannot start torrent")?; let mut g = self.locked.write(); let spawn_fatal_errors_receiver = @@ -295,18 +293,20 @@ impl ManagedTorrent { drop(g); let t = self.clone(); let span = self.info().span.clone(); - let token = live_cancellation_token.clone(); + let token = session.cancellation_token().child_token().clone(); spawn_with_cancel( error_span!(parent: span.clone(), "initialize_and_start"), token.clone(), async move { - let _permit = init_semaphore + let concurrent_init_semaphore = + session.concurrent_initialize_semaphore.clone(); + let _permit = concurrent_init_semaphore .acquire() .await .context("bug: concurrent init semaphore was closed")?; - match init.check(bitv_factory).await { + match init.check(session.bitv_factory.clone()).await { Ok(paused) => { let mut g = t.locked.write(); if let ManagedTorrentState::Initializing(_) = &g.state { @@ -325,8 +325,8 @@ impl ManagedTorrent { let live = TorrentStateLive::new( paused, tx, - live_cancellation_token, - session_stats, + session.cancellation_token().child_token(), + session.stats.atomic.clone(), )?; g.state = ManagedTorrentState::Live(live.clone()); drop(g); @@ -355,13 +355,13 @@ impl ManagedTorrent { let live = TorrentStateLive::new( paused, tx, - live_cancellation_token.clone(), - session_stats, + session.cancellation_token().child_token().clone(), + session.stats.atomic.clone(), )?; g.state = ManagedTorrentState::Live(live.clone()); drop(g); - spawn_fatal_errors_receiver(self, rx, live_cancellation_token); + spawn_fatal_errors_receiver(self, rx, session.cancellation_token().child_token()); spawn_peer_adder(&live, peer_rx); Ok(()) } @@ -377,14 +377,7 @@ impl ManagedTorrent { self.state_change_notify.notify_waiters(); // Recurse. - self.start( - peer_rx, - start_paused, - live_cancellation_token, - init_semaphore, - bitv_factory, - session_stats, - ) + self.start(peer_rx, start_paused) } ManagedTorrentState::None => bail!("bug: torrent is in empty state"), }