adam/meteoro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fruitless changes

Browse Source
This commit is contained in:
Adam 2025-08-04 00:16:43 +01:00
parent b32f00499a
commit b3f06eeaea
2 changed files with 124 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
src/tracker/tracker.rs
View File

@ -1,5 +1,6 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::net::{SocketAddr, SocketAddrV4, SocketAddrV6}; use std::net::{SocketAddr, SocketAddrV4, SocketAddrV6};
use std::os::linux::raw::stat;
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::thread; use std::thread;
@ -70,7 +71,8 @@ impl Tracker {
pub fn start_on(&mut self, addrs: &[SocketAddr]) -> Result<()> { pub fn start_on(&mut self, addrs: &[SocketAddr]) -> Result<()> {
if addrs.is_empty() { if addrs.is_empty() {
log::error!("No addresses provided for tracker to listen on"); log::warn!("No addresses provided for tracker to listen on");
return Ok(());
} }
// Only one request channel set is necessary as all the server threads send on the same one and the tracker is the only reader // Only one request channel set is necessary as all the server threads send on the same one and the tracker is the only reader
@ -84,14 +86,7 @@ impl Tracker {
let req_chan = req_send.clone(); let req_chan = req_send.clone();
let a = a.clone(); let a = a.clone();
self.server_threads.push( self.server_threads.push(thread::spawn(move || {
thread::Builder::new()
.name(format!(
"{} server thread (ID {})",
env!("CARGO_PKG_NAME"),
server_id
))
.spawn(move || {
let server = UdpServer::new(a, server_id); let server = UdpServer::new(a, server_id);
loop { loop {
match server.start(req_chan.clone(), resp_recv.clone()) { match server.start(req_chan.clone(), resp_recv.clone()) {
@ -101,8 +96,7 @@ impl Tracker {
_ => (), _ => (),
}; };
} }
})?, }));
);
self.response_channels.push(resp_send); self.response_channels.push(resp_send);
@ -121,9 +115,12 @@ impl Tracker {
let start = Instant::now(); let start = Instant::now();
let expired = Tracker::purge_expired_connections(&mut peers.write().unwrap()); let expired = Tracker::purge_expired_connections(&mut peers.write().unwrap());
let purged =
Tracker::purge_expired_swarm_peers(&mut info_hashes.write().unwrap(), &expired); Tracker::purge_expired_swarm_peers(&mut info_hashes.write().unwrap(), &expired);
log::trace!("Garbage collection took {} ns", start.elapsed().as_nanos()) log::debug!("Garbage collected {} expired connections", expired.len());
log::debug!("Garbage collected {} inactive peers from the swarm", purged);
log::debug!("Garbage collection took {} ns", start.elapsed().as_nanos())
} }
}); });
@ -205,12 +202,54 @@ impl Tracker {
let mut info_hashes = self.info_hashes.write().unwrap(); let mut info_hashes = self.info_hashes.write().unwrap();
match info_hashes.get_mut(&announce.info_hash) { match info_hashes.get_mut(&announce.info_hash) {
None => {
// Info hash isn't currently tracked
// No relevant peers in the swarm
info_hashes.insert(
announce.info_hash,
vec![PeerStatus {
connection_id: announce.connection_id,
last_event: announce.event,
last_active: Instant::now(),
}],
);
}
Some(swarm) => { Some(swarm) => {
if announce.event == Event::Stopped { // Insert into swarm if not already present
swarm.retain(|status| { // TODO: sort (?)
status.connection_id != announce.connection_id
let this_swarm_idx: Option<usize> =
match swarm.iter_mut().enumerate().find(|(idx, status)| {
status.connection_id == announce.connection_id
}) {
Some((idx, peer)) => {
// Peer already exists in swarm, update state
peer.last_active = Instant::now();
peer.last_event = announce.event;
Some(idx)
}
None => {
// Peer is not in swarm, add it
swarm.push(PeerStatus {
connection_id: announce.connection_id,
last_event: announce.event,
last_active: Instant::now(),
}); });
} else {
None
}
};
if announce.event == Event::Stopped {
if let Some(idx) = this_swarm_idx {
swarm.remove(idx);
}
}
let peers = self.peers.read().unwrap(); let peers = self.peers.read().unwrap();
for peer_status in swarm { for peer_status in swarm {
@ -219,34 +258,24 @@ impl Tracker {
continue; continue;
} }
// State update for ourselves
if peer_status.connection_id == announce.connection_id { if peer_status.connection_id == announce.connection_id {
if peer_status.last_event != announce.event
&& announce.event != Event::None
{
// Update last event if the event has changed and is not just a `None` for notifying presence
peer_status.last_event = announce.event
} else {
// Peer status has not changed
peer_status.last_active = Instant::now();
}
continue; // don't give the peer their own connection ID's address continue; // don't give the peer their own connection ID's address
} }
// Respect number of peers requested
if n_announce_entries >= n_announce_want { if n_announce_entries >= n_announce_want {
break; break;
} }
// Collect and add the relevant peer's address to the appropriate vector // Collect and add the relevant peer's address to the appropriate vector
if let Some(a) = peers.get(&peer_status.connection_id) { if let Some(metadata) = peers.get(&peer_status.connection_id) {
if a.socket_addr == request.src_addr { if metadata.socket_addr == request.src_addr {
continue; // don't give the peer an expired ID with the same address continue; // don't give the peer an expired ID with the same address
} }
match &a.socket_addr { match &metadata.socket_addr {
SocketAddr::V4(v4) => { SocketAddr::V4(v4) => {
if request.src_addr.is_ipv4() { if request.src_addr.is_ipv4() {
v4_peers.push(*v4); v4_peers.push(*v4);
@ -271,20 +300,6 @@ impl Tracker {
} }
} }
} }
}
None => {
// Info hash isn't currently tracked
// No relevant peers in the swarm
info_hashes.insert(
announce.info_hash,
vec![PeerStatus {
connection_id: announce.connection_id,
last_event: announce.event,
last_active: Instant::now(),
}],
);
}
}; };
let resp_common = AnnounceResponseCommon { let resp_common = AnnounceResponseCommon {
@ -407,18 +422,31 @@ impl Tracker {
return purged; return purged;
} }
fn purge_expired_swarm_peers(info_hashes: &mut InfoHashMap, expired_connections: &[i64]) { fn purge_expired_swarm_peers(
info_hashes: &mut InfoHashMap,
expired_connections: &[i64],
) -> usize {
let mut n_purged: usize = 0;
info_hashes.retain(|_, swarm| { info_hashes.retain(|_, swarm| {
swarm.is_empty() swarm.is_empty()
|| swarm || swarm
.iter() .iter()
.find(|status| { .find(|status| {
expired_connections.contains(&status.connection_id) let cond = expired_connections.contains(&status.connection_id)
|| status.last_active.elapsed() > DEFAULT_ANNOUNCE_INTERVAL * 2 || status.last_active.elapsed() > DEFAULT_ANNOUNCE_INTERVAL * 5
|| status.last_event == Event::Stopped || status.last_event == Event::Stopped;
if cond {
n_purged += 1;
}
cond
}) })
.is_some() .is_some()
}); });
return n_purged;
} }
pub fn should_stop() -> bool { pub fn should_stop() -> bool {

5
src/tracker/udp_server.rs
View File

@ -115,7 +115,10 @@ impl UdpServer {
let n_bytes = match response.response { let n_bytes = match response.response {
UdpResponse::Connect(connect) => connect.write_to_buf(&mut buf), UdpResponse::Connect(connect) => connect.write_to_buf(&mut buf),
UdpResponse::Announce(announce) => match announce { UdpResponse::Announce(announce) => match announce {
AnnounceResponse::V4(v4) => v4.write_to_buf(&mut buf), AnnounceResponse::V4(v4) => {
log::debug!("{:?}", v4);
v4.write_to_buf(&mut buf)
}
AnnounceResponse::V6(v6) => v6.write_to_buf(&mut buf), AnnounceResponse::V6(v6) => v6.write_to_buf(&mut buf),
}, },
UdpResponse::Scrape(scrape) => scrape.write_to_buf(&mut buf), UdpResponse::Scrape(scrape) => scrape.write_to_buf(&mut buf),