dumbgoy/soulseek-rs-modded
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(client): bound the threads a peer flood can create

Browse Source 
One popular search produced a ConnectToPeer for every responder, and
each one cost a 2 MiB wrapper thread, an unbounded 256 KiB actor, a
listener thread per inbound socket, and on failure a 2 MiB thread that
slept for 20 seconds. A container with a pids limit died on one album
search (issues #32, #33).

Four bounds, no behavior change for a quiet session:
- P-type registrations run inline on the ops loop; only F transfers
  keep a worker thread. The dial already ran on the actor thread.
- The broker timeout is a deadline swept by the ops loop instead of a
  sleeping thread per failed dial.
- The registry caps live actors (default 512, Client::set_max_peers).
  At capacity the oldest peer outside the caller's protected set
  (live downloads, uploads, pending serves, queued messages, browses
  in flight) is evicted; if none qualify the registration is refused
  before any thread is spawned.
- Concurrent listener handshakes are bounded by a 128-permit
  semaphore released once the peer-init arrives, so slow handshakes
  cannot multiply threads while transfers stay unthrottled.

A replayed TransferResponse for an InProgress or Completed token is
dropped instead of starting a second transfer thread over the same
.part file.

Fixes #32
Fixes #33
This commit is contained in:
Michel de Graaf
2026-08-13 13:46:18 +02:00
parent fd1d168ca6
commit 9fe1c9984c
9 changed files with 458 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
soulseek-rs-lib/src/client/connection.rs
+13 -2
View File
@@ -50,10 +50,11 @@ impl Client {
) = mpsc::channel();
let mut ctx = self.context.write_safe()?;
let peer_registry = PeerRegistry::new(
let peer_registry = PeerRegistry::with_capacity(
ctx.actor_system.clone(),
sender.clone(),
self.username.clone(),
ctx.max_peers.clone(),
);
ctx.peer_registry = Some(peer_registry);
@@ -175,6 +176,13 @@ impl Client {
Ok(())
}
pub fn set_max_peers(&self, max_peers: usize) {
if let Ok(ctx) = self.context.read_safe() {
ctx.max_peers
.store(max_peers.max(1), std::sync::atomic::Ordering::Relaxed);
}
}
#[allow(dead_code)]
pub fn remove_peer(&self, username: &str) {
let context = match self.context.read_safe() {
@@ -214,7 +222,10 @@ impl Client {
}
};
if let Some(ref registry) = context.peer_registry {
match registry.register_peer(peer_clone, stream, None) {
let protected = context.protected_peers();
match registry.register_peer_protected(
peer_clone, stream, None, &protected,
) {
Ok(_) => (),
Err(e) => {
trace!(
soulseek-rs-lib/src/client/mod.rs
+65 -4
View File
@@ -23,7 +23,7 @@ use std::{
net::TcpStream,
sync::{
Arc, RwLock,
atomic::{AtomicBool, AtomicU32, Ordering},
atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering},
mpsc::{self, Receiver, Sender},
},
thread::{self, sleep},
@@ -58,6 +58,10 @@ pub const DEFAULT_UPLOAD_SLOTS: usize = 10;
/// before giving up and failing the download. Matches the direct-dial timeout.
const BROKER_CONNECT_TIMEOUT: Duration = Duration::from_secs(20);
const DEFAULT_MAX_PEERS: usize = 512;
const BROWSE_PROTECT_WINDOW: Duration = Duration::from_mins(1);
/// Source of non-zero correlation tokens for server-brokered connections.
static NEXT_CONNECT_TOKEN: AtomicU32 = AtomicU32::new(1);
@@ -295,7 +299,8 @@ pub struct ClientContext {
private_messages: Vec<UserMessage>,
/// Correlation tokens for server-brokered (firewalled) connections, mapping
/// a token we sent in a ConnectToPeer to the peer we expect back.
pending_connect_tokens: HashMap<u32, String>,
pending_connect_tokens: HashMap<u32, (String, Instant)>,
max_peers: Arc<AtomicUsize>,
/// Files we share with peers (read-only after connect).
pub shares: Arc<Shares>,
/// The directories the current share index was built from.
@@ -311,6 +316,7 @@ pub struct ClientContext {
pending_serves: HashMap<String, Vec<u32>>,
/// Shared-file listings received from peers we browsed.
browse_results: HashMap<String, Vec<SharedDirectory>>,
pending_browses: HashMap<String, Instant>,
/// Latest snapshot of the public chat-room list (from `RoomList`, code 64).
room_list: Vec<RoomInfo>,
/// Chat-room events awaiting consumption by the client/UI.
@@ -395,6 +401,7 @@ impl ClientContext {
searches: HashMap::new(),
private_messages: Vec::new(),
pending_connect_tokens: HashMap::new(),
max_peers: Arc::new(AtomicUsize::new(DEFAULT_MAX_PEERS)),
shares: Arc::new(Shares::empty()),
shared_directories: Vec::new(),
peer_addresses: HashMap::new(),
@@ -403,6 +410,7 @@ impl ClientContext {
active_uploads: HashMap::new(),
pending_serves: HashMap::new(),
browse_results: HashMap::new(),
pending_browses: HashMap::new(),
room_list: Vec::new(),
room_events: Vec::new(),
room_members: HashMap::new(),
@@ -563,9 +571,17 @@ impl ClientContext {
username: String,
directories: Vec<SharedDirectory>,
) {
self.pending_browses.remove(&username);
self.browse_results.insert(username, directories);
}
pub(crate) fn mark_browse_pending(&mut self, username: &str) {
let now = Instant::now();
self.pending_browses.retain(|_, deadline| *deadline > now);
self.pending_browses
.insert(username.to_string(), now + BROWSE_PROTECT_WINDOW);
}
/// Remove and return the shared-file listing browsed from `username`.
pub fn take_browse_result(
&mut self,
@@ -577,12 +593,57 @@ impl ClientContext {
/// Remember that a server-brokered connection to `username` is pending under
/// `token`; the peer will quote it back in a PierceFirewall.
pub fn add_pending_connect(&mut self, token: u32, username: String) {
self.pending_connect_tokens.insert(token, username);
self.pending_connect_tokens
.insert(token, (username, Instant::now() + BROKER_CONNECT_TIMEOUT));
}
/// Resolve and consume the peer expected for a brokered connection `token`.
pub fn take_pending_connect(&mut self, token: u32) -> Option<String> {
self.pending_connect_tokens.remove(&token)
self.pending_connect_tokens
.remove(&token)
.map(|(username, _)| username)
}
pub fn take_expired_connects(&mut self, now: Instant) -> Vec<String> {
let mut expired = Vec::new();
self.pending_connect_tokens
.retain(|_, (username, deadline)| {
if now >= *deadline {
expired.push(std::mem::take(username));
return false;
}
true
});
expired
}
pub(crate) fn protected_peers(&self) -> HashSet<String> {
let mut protected: HashSet<String> = self
.downloads
.list()
.iter()
.filter(|d| {
matches!(
d.status,
DownloadStatus::Queued
| DownloadStatus::InProgress { .. }
| DownloadStatus::Paused { .. }
)
})
.map(|d| d.username.clone())
.collect();
protected
.extend(self.active_uploads.values().map(|u| u.username.clone()));
protected.extend(self.pending_serves.keys().cloned());
protected.extend(self.pending_peer_messages.keys().cloned());
let now = Instant::now();
protected.extend(
self.pending_browses
.iter()
.filter(|(_, deadline)| **deadline > now)
.map(|(username, _)| username.clone()),
);
protected
}
/// Record a private message received from another user.
soulseek-rs-lib/src/client/operations.rs
+70 -42
View File
@@ -1,12 +1,14 @@
use super::{
Arc, BROKER_CONNECT_TIMEOUT, Client, ClientContext, ClientOperation,
ConnectionType, Download, DownloadPeer, DownloadStatus, Peer, PeerMessage,
Arc, Client, ClientContext, ClientOperation, ConnectionType, Download,
DownloadPeer, DownloadStatus, Duration, Instant, Peer, PeerMessage,
PeerRegistry, Receiver, RwLock, RwLockExt, ServerMessage,
build_search_response, debug, error, info, next_connect_token, sleep,
build_search_response, debug, error, info, mpsc, next_connect_token,
thread, trace, warn,
};
use crate::message::server::MessageFactory;
const CONNECT_SWEEP_INTERVAL: Duration = Duration::from_secs(1);
impl Client {
pub(crate) fn listen_to_client_operations(
reader: Receiver<ClientOperation>,
@@ -14,20 +16,40 @@ impl Client {
own_username: String,
) {
thread::spawn(move || {
for operation in reader {
let mut last_sweep = Instant::now();
loop {
let next = reader.recv_timeout(CONNECT_SWEEP_INTERVAL);
if last_sweep.elapsed() >= CONNECT_SWEEP_INTERVAL {
last_sweep = Instant::now();
Self::sweep_expired_connects(&client_context);
}
let operation = match next {
Ok(operation) => operation,
Err(mpsc::RecvTimeoutError::Timeout) => continue,
Err(mpsc::RecvTimeoutError::Disconnected) => break,
};
match operation {
ClientOperation::ConnectToPeer(peer) => {
let client_context_clone = client_context.clone();
let own_username_clone = own_username.clone();
if matches!(peer.connection_type, ConnectionType::F) {
let client_context_clone = client_context.clone();
let own_username_clone = own_username.clone();
thread::spawn(move || {
thread::spawn(move || {
Self::connect_to_peer(
peer,
client_context_clone,
own_username_clone,
None,
);
});
} else {
Self::connect_to_peer(
peer,
client_context_clone,
own_username_clone,
client_context.clone(),
own_username.clone(),
None,
);
});
}
}
ClientOperation::SearchResult(search_result) => {
trace!("[client] SearchResult {:?}", search_result);
@@ -122,6 +144,28 @@ impl Client {
);
continue;
};
if matches!(
download.status,
DownloadStatus::InProgress { .. }
| DownloadStatus::Completed
) {
debug!(
"[client] transfer token {} already claimed; \
ignoring replayed TransferResponse",
token
);
continue;
}
if let Ok(mut ctx) = client_context.write_safe() {
ctx.update_download_with_status(
token,
DownloadStatus::InProgress {
bytes_downloaded: 0,
total_bytes: download.size,
speed_bytes_per_sec: 0.0,
},
);
}
thread::spawn(move || {
let download_peer = DownloadPeer::new(
@@ -285,17 +329,12 @@ impl Client {
u8::try_from(obfuscation_type).unwrap_or(0),
obfuscated_port,
);
let client_context_clone = client_context.clone();
let own_username_clone = own_username.clone();
thread::spawn(move || {
Self::connect_to_peer(
peer,
client_context_clone,
own_username_clone,
None,
);
});
Self::connect_to_peer(
peer,
client_context.clone(),
own_username.clone(),
None,
);
}
}
ClientOperation::UpdateDownloadTokens(
@@ -751,30 +790,19 @@ impl Client {
ConnectionType::P,
);
let _ = sender.send(ServerMessage::SendMessage(msg));
// Bound the brokered attempt: if no PierceFirewall
// consumes the token, fail the peer's queued
// downloads (so the caller's Receiver unblocks)
// and reclaim the token. A successful pierce
// takes the token first, making this a no-op.
let timeout_ctx = client_context.clone();
let timeout_user = username.clone();
thread::spawn(move || {
sleep(BROKER_CONNECT_TIMEOUT);
let still_pending =
timeout_ctx.write_safe().is_ok_and(|mut c| {
c.take_pending_connect(token).is_some()
});
if still_pending {
Self::fail_queued_downloads(
&timeout_ctx,
&timeout_user,
);
}
});
}
}
}
});
}
fn sweep_expired_connects(client_context: &Arc<RwLock<ClientContext>>) {
let expired = match client_context.write_safe() {
Ok(mut ctx) => ctx.take_expired_connects(Instant::now()),
Err(_) => return,
};
for username in expired {
Self::fail_queued_downloads(client_context, &username);
}
}
}
soulseek-rs-lib/src/client/rooms.rs
+2 -1
View File
@@ -177,7 +177,8 @@ impl Client {
let request =
crate::message::server::MessageFactory::build_get_share_file_list();
let (connected, registry) = {
let ctx = self.context.read_safe()?;
let mut ctx = self.context.write_safe()?;
ctx.mark_browse_pending(username);
(
ctx.peer_registry
.as_ref()
soulseek-rs-lib/src/client/tests.rs
+48
View File
@@ -395,3 +395,51 @@ fn a_clean_disconnect_keeps_queued_uploads_an_error_drops_them() {
thread::sleep(Duration::from_millis(10));
}
}
#[test]
fn a_pending_connect_expires_after_the_broker_timeout() {
let mut context = ClientContext::new();
context.add_pending_connect(7, "ghost".to_string());
assert!(context.take_expired_connects(Instant::now()).is_empty());
let after_deadline =
Instant::now() + BROKER_CONNECT_TIMEOUT + Duration::from_secs(1);
assert_eq!(
context.take_expired_connects(after_deadline),
vec!["ghost".to_string()]
);
assert!(context.take_pending_connect(7).is_none());
}
#[test]
fn protected_peers_covers_downloads_uploads_and_pending_serves() {
let mut context = ClientContext::new();
context.add_download(download(
"downloader",
"song.mp3",
1,
DownloadStatus::Queued,
mpsc::channel().0,
));
context.add_download(download(
"done",
"old.mp3",
2,
DownloadStatus::Completed,
mpsc::channel().0,
));
context
.pending_serves
.insert("waiting".to_string(), vec![9]);
context.mark_browse_pending("browsed");
let protected = context.protected_peers();
assert!(protected.contains("downloader"));
assert!(protected.contains("waiting"));
assert!(protected.contains("browsed"));
assert!(!protected.contains("done"));
context.store_browse_result("browsed".to_string(), Vec::new());
assert!(!context.protected_peers().contains("browsed"));
}