refactor(client): dedupe peer registration and order eviction by id

The registry stored a registration Instant only to pick the eviction
victim, but the actor id is already monotonic per registration and
refreshed at the same moments, so it orders identically. The
register-refuse-fail sequence was also copy-pasted between the ops
loop and the listener; both now call one register_peer_or_fail.
This commit is contained in:
Michel de Graaf
2026-08-13 15:28:36 +02:00
parent a70c657d31
commit 2a90619a64
3 changed files with 62 additions and 85 deletions
+9 -17
View File
@@ -11,7 +11,6 @@ use std::net::TcpStream;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc::Sender; use std::sync::mpsc::Sender;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use std::time::Instant;
/// Source of unique per-actor ids so terminal-outcome eviction can be made /// Source of unique per-actor ids so terminal-outcome eviction can be made
/// identity-aware (a replaced actor must not evict its replacement). /// identity-aware (a replaced actor must not evict its replacement).
@@ -19,7 +18,7 @@ static NEXT_PEER_ID: AtomicU64 = AtomicU64::new(1);
/// Registered peers keyed by username, each stored with the unique id of the /// Registered peers keyed by username, each stored with the unique id of the
/// actor currently occupying the slot. /// actor currently occupying the slot.
type PeerMap = HashMap<String, (u64, ActorHandle<PeerMessage>, Instant)>; type PeerMap = HashMap<String, (u64, ActorHandle<PeerMessage>)>;
pub struct PeerRegistry { pub struct PeerRegistry {
peers: Arc<Mutex<PeerMap>>, peers: Arc<Mutex<PeerMap>>,
@@ -104,7 +103,7 @@ impl PeerRegistry {
let victim = peers let victim = peers
.iter() .iter()
.filter(|(name, _)| !protected.contains(name.as_str())) .filter(|(name, _)| !protected.contains(name.as_str()))
.min_by_key(|(_, (_, _, registered))| *registered) .min_by_key(|(_, (id, _))| *id)
.map(|(name, _)| name.clone()); .map(|(name, _)| name.clone());
let Some(victim) = victim else { let Some(victim) = victim else {
return Err(format!( return Err(format!(
@@ -112,7 +111,7 @@ impl PeerRegistry {
is busy; refusing {username}" is busy; refusing {username}"
)); ));
}; };
if let Some((_, handle, _)) = peers.remove(&victim) { if let Some((_, handle)) = peers.remove(&victim) {
let _ = handle.stop(); let _ = handle.stop();
debug!( debug!(
"[peer_registry] evicted idle peer {} to admit {}", "[peer_registry] evicted idle peer {} to admit {}",
@@ -131,8 +130,8 @@ impl PeerRegistry {
// become an orphan pinning a pool worker forever. Eviction on the // become an orphan pinning a pool worker forever. Eviction on the
// replaced actor's later shutdown is identity-aware (keyed on its id), // replaced actor's later shutdown is identity-aware (keyed on its id),
// so stopping it here cannot evict this new connection. // so stopping it here cannot evict this new connection.
if let Some((_, old_handle, _)) = if let Some((_, old_handle)) =
peers.insert(username.clone(), (id, handle.clone(), Instant::now())) peers.insert(username.clone(), (id, handle.clone()))
{ {
let _ = old_handle.stop(); let _ = old_handle.stop();
debug!( debug!(
@@ -147,9 +146,7 @@ impl PeerRegistry {
#[must_use] #[must_use]
pub fn get_peer(&self, username: &str) -> Option<ActorHandle<PeerMessage>> { pub fn get_peer(&self, username: &str) -> Option<ActorHandle<PeerMessage>> {
match self.peers.lock_safe() { match self.peers.lock_safe() {
Ok(peers) => { Ok(peers) => peers.get(username).map(|(_, handle)| handle.clone()),
peers.get(username).map(|(_, handle, _)| handle.clone())
}
Err(e) => { Err(e) => {
error!("[peer_registry] get_peer: {}", e); error!("[peer_registry] get_peer: {}", e);
None None
@@ -175,7 +172,7 @@ impl PeerRegistry {
debug!("[peer_registry] Removed peer actor for {}", username); debug!("[peer_registry] Removed peer actor for {}", username);
} }
removed.map(|(_, handle, _)| handle) removed.map(|(_, handle)| handle)
} }
/// Remove and return the actor for `username` only if it is still the actor /// Remove and return the actor for `username` only if it is still the actor
@@ -194,11 +191,8 @@ impl PeerRegistry {
return None; return None;
} }
}; };
if peers if peers.get(username).is_some_and(|(stored, _)| *stored == id) {
.get(username) let removed = peers.remove(username).map(|(_, handle)| handle);
.is_some_and(|(stored, _, _)| *stored == id)
{
let removed = peers.remove(username).map(|(_, handle, _)| handle);
debug!( debug!(
"[peer_registry] Removed peer actor {} for {}", "[peer_registry] Removed peer actor {} for {}",
id, username id, username
@@ -261,7 +255,6 @@ mod tests {
use std::net::{TcpListener, TcpStream}; use std::net::{TcpListener, TcpStream};
use std::sync::Arc; use std::sync::Arc;
use std::sync::atomic::AtomicUsize; use std::sync::atomic::AtomicUsize;
use std::time::Duration;
fn loopback_peer(name: &str) -> (Peer, TcpStream, TcpStream) { fn loopback_peer(name: &str) -> (Peer, TcpStream, TcpStream) {
let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let listener = TcpListener::bind("127.0.0.1:0").unwrap();
@@ -306,7 +299,6 @@ mod tests {
for name in ["a", "b", "c"] { for name in ["a", "b", "c"] {
let (peer, stream, _far_end) = loopback_peer(name); let (peer, stream, _far_end) = loopback_peer(name);
registry.register_peer(peer, Some(stream), None).unwrap(); registry.register_peer(peer, Some(stream), None).unwrap();
std::thread::sleep(Duration::from_millis(5));
} }
assert!(!registry.contains("a"), "the oldest peer must be evicted"); assert!(!registry.contains("a"), "the oldest peer must be evicted");
+46 -33
View File
@@ -199,6 +199,46 @@ impl Client {
} }
} }
pub(crate) fn register_peer_or_fail(
client_context: &Arc<RwLock<ClientContext>>,
peer: Peer,
stream: Option<TcpStream>,
reader: Option<crate::message::MessageReader>,
) -> bool {
let username = peer.username.clone();
let registered = {
let context = match client_context.read_safe() {
Ok(c) => c,
Err(e) => {
error!("[client] register peer read: {}", e);
return false;
}
};
if let Some(ref registry) = context.peer_registry {
let protected = context.protected_peers();
match registry
.register_peer_protected(peer, stream, reader, &protected)
{
Ok(_) => true,
Err(e) => {
warn!(
"[client] refusing peer connection for {:?}: {:?}",
username, e
);
false
}
}
} else {
trace!("PeerRegistry not initialized");
false
}
};
if !registered {
Self::fail_queued_downloads(client_context, &username);
}
registered
}
pub(crate) fn connect_to_peer( pub(crate) fn connect_to_peer(
peer: Peer, peer: Peer,
client_context: Arc<RwLock<ClientContext>>, client_context: Arc<RwLock<ClientContext>>,
@@ -212,39 +252,12 @@ impl Client {
); );
match peer.connection_type { match peer.connection_type {
ConnectionType::P => { ConnectionType::P => {
let username = peer.username; Self::register_peer_or_fail(
&client_context,
let refused = { peer_clone,
let context = match client_context.read_safe() { stream,
Ok(c) => c, None,
Err(e) => { );
error!("[client] connect_to_peer read: {}", e);
return;
}
};
if let Some(ref registry) = context.peer_registry {
let protected = context.protected_peers();
match registry.register_peer_protected(
peer_clone, stream, None, &protected,
) {
Ok(_) => false,
Err(e) => {
warn!(
"[client] refusing peer connection for \
{:?}: {:?}",
username, e
);
true
}
}
} else {
trace!("PeerRegistry not initialized");
false
}
};
if refused {
Self::fail_queued_downloads(&client_context, &username);
}
} }
ConnectionType::F => { ConnectionType::F => {
+7 -35
View File
@@ -12,7 +12,7 @@ use crate::peer::{ConnectionType, DownloadPeer, Peer};
use crate::types::Download; use crate::types::Download;
use crate::utils::lock::RwLockExt; use crate::utils::lock::RwLockExt;
use crate::utils::semaphore::{Permit, Semaphore}; use crate::utils::semaphore::{Permit, Semaphore};
use crate::{DownloadStatus, debug, error, info, trace, warn}; use crate::{DownloadStatus, debug, error, info, trace};
/// How long to wait before accepting again after a failure. /// How long to wait before accepting again after a failure.
/// ///
@@ -151,40 +151,12 @@ fn handle_peer_connection(
} }
stream.set_nodelay(true).ok(); stream.set_nodelay(true).ok();
let registered = { Client::register_peer_or_fail(
let client_context = match context.client_context.read_safe() { &context.client_context,
Ok(c) => c, peer,
Err(e) => { Some(stream),
error!("[listener] handle_peer_connection lock: {}", e); Some(reader),
return false; )
}
};
if let Some(ref registry) = client_context.peer_registry {
let protected = client_context.protected_peers();
match registry.register_peer_protected(
peer.clone(),
Some(stream),
Some(reader),
&protected,
) {
Ok(_) => true,
Err(e) => {
warn!(
"[listener] refusing peer connection for {:?}: {:?}",
peer.username, e
);
false
}
}
} else {
error!("PeerRegistry not initialized");
false
}
};
if !registered {
Client::fail_queued_downloads(&context.client_context, &peer.username);
}
registered
} }
fn handle_file_connection( fn handle_file_connection(