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

maybe_transmit_haves: need to optimise

Browse source
This commit is contained in:
Igor Katson 2024-05-03 16:28:02 +01:00
parent 70c2c38329
commit 7ed7e277c0
1 changed files with 7 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -57,7 +57,6 @@ use anyhow::{bail, Context};
use backoff::backoff::Backoff; use backoff::backoff::Backoff;
use buffers::{ByteBuf, ByteBufOwned}; use buffers::{ByteBuf, ByteBufOwned};
use clone_to_owned::CloneToOwned; use clone_to_owned::CloneToOwned;
use futures::{stream::FuturesUnordered, StreamExt};
use librqbit_core::{ use librqbit_core::{
hash_id::Id20, hash_id::Id20,
lengths::{ChunkInfo, Lengths, ValidPieceIndex}, lengths::{ChunkInfo, Lengths, ValidPieceIndex},
@ -554,8 +553,6 @@ impl TorrentStateLive {
} }
fn maybe_transmit_haves(&self, index: ValidPieceIndex) { fn maybe_transmit_haves(&self, index: ValidPieceIndex) {
let mut futures = Vec::new();
for pe in self.peers.states.iter() { for pe in self.peers.states.iter() {
match &pe.value().state.get() { match &pe.value().state.get() {
PeerState::Live(live) => { PeerState::Live(live) => {
@ -572,42 +569,17 @@ impl TorrentStateLive {
continue; continue;
} }
let tx = live.tx.downgrade(); if live
futures.push(async move { .tx
if let Some(tx) = tx.upgrade() { .send(WriterRequest::Message(Message::Have(index.get())))
if tx .is_err()
.send(WriterRequest::Message(Message::Have(index.get()))) {
.is_err() // whatever
{ }
// whatever
}
}
});
} }
_ => continue, _ => continue,
} }
} }
if futures.is_empty() {
trace!("no peers to transmit Have={} to, saving some work", index);
return;
}
let mut unordered: FuturesUnordered<_> = futures.into_iter().collect();
// We don't want to remember this task as there may be too many.
self.spawn(
error_span!(
parent: self.meta.span.clone(),
"transmit_haves",
piece = index.get(),
count = unordered.len()
),
async move {
while unordered.next().await.is_some() {}
Ok(())
},
);
} }
pub(crate) fn add_peer_if_not_seen(&self, addr: SocketAddr) -> anyhow::Result<bool> { pub(crate) fn add_peer_if_not_seen(&self, addr: SocketAddr) -> anyhow::Result<bool> {