From 5b4ef30f08d72b2cb9eb200242003dbe61f42b35 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Sun, 26 Jul 2026 17:01:03 +0200 Subject: [PATCH] feat: lift the client's concurrency ceiling from ~15 peers to 1024 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The client had never been driven at scale. A stress harness that mixes searches, downloads, uploads and browses against a local soulfind found four ceilings, three of which lose work silently rather than merely being slow. Peers were capped at one per core. ActorSystem ran each actor's whole lifetime as one job on a fixed-size ThreadPool, and actor loops block until stopped, so the pool acted as a semaphore: past the cap an actor never started, never connected and never errored, and uploads stopped working entirely. The pool had exactly one consumer and every job it received was non-terminating, so it is gone; actors get their own thread. TransferResponse raced the bookkeeping it depends on. It was sent straight back to the peer while UpdateDownloadTokens was still queued for the client loop — but the response is what invites the peer's file connection, and that connection is matched by token. Under load the client loop lost, the file connection matched nothing, and the download sat on Queued forever. The response now goes out after the token is recorded. Download tokens were md5(filename) truncated to 20 bits, and the store deletes every entry matching a token, so two downloads sharing one destroyed each other — which fetching the same filename from two peers does every time. They now come from a counter. Port 0 from GetPeerAddress means the server does not know where a user listens. It was cached and dialled, and the upload waiting on it was dropped. It is no longer cached or served against; the connect attempt stays, because its failure is what makes an unreachable peer fall back to server brokering. Mixed profile at 1024 peers: 512/512 downloads, 256/256 uploads, 400-850 MiB/s in ~7s, against 71/96 downloads at 7.9 MiB/s and a 60s timeout before. --- benchmarks/README.md | 76 ++ benchmarks/stress-baseline.json | 23 + scripts/stress.sh | 37 + soulseek-rs-lib/examples/stress.rs | 1236 ++++++++++++++++++++ soulseek-rs-lib/src/actor/mod.rs | 101 +- soulseek-rs-lib/src/actor/peer_actor.rs | 23 +- soulseek-rs-lib/src/actor/peer_registry.rs | 4 +- soulseek-rs-lib/src/client/downloads.rs | 37 +- soulseek-rs-lib/src/client/mod.rs | 22 +- soulseek-rs-lib/src/client/operations.rs | 46 +- soulseek-rs-lib/src/utils/mod.rs | 1 - soulseek-rs-lib/src/utils/thread_pool.rs | 124 -- 12 files changed, 1539 insertions(+), 191 deletions(-) create mode 100644 benchmarks/README.md create mode 100644 benchmarks/stress-baseline.json create mode 100755 scripts/stress.sh create mode 100644 soulseek-rs-lib/examples/stress.rs delete mode 100644 soulseek-rs-lib/src/utils/thread_pool.rs diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 0000000..f8d7b71 --- /dev/null +++ b/benchmarks/README.md @@ -0,0 +1,76 @@ +# Concurrency benchmark + +Answers one question: how much simultaneous Soulseek activity can the client +absorb before it starts losing work? + +```sh +scripts/stress.sh # run the default profile +scripts/stress.sh --save # ...and store it as the new baseline if it wins +``` + +A run takes about 7 seconds. It spawns a local soulfind, surrounds one real +`Client` with a swarm of mock peers, and drives searches, downloads, uploads and +browses **at the same time** — the mixing is the point, since every ceiling found +so far only appeared under a mixed load. + +## The score + +One number out of 100, weighted across five dimensions: + +| Dimension | Weight | Measures | +|------------|--------|-------------------------------------------------| +| downloads | 0.30 | completed / started | +| uploads | 0.20 | served / requested | +| searches | 0.20 | queries that got at least one result | +| browses | 0.15 | browse requests answered | +| throughput | 0.15 | aggregate MiB/s against `TARGET_MBPS` (500) | + +The four functional dimensions should stay at 100%: anything less means work was +silently dropped, which is the failure mode that matters. Throughput is the term +with headroom, so the score still moves when transfer handling regresses. + +`--save` only overwrites `stress-baseline.json` when the run beats it, so the +baseline ratchets upward. Without `--save` a run just prints the delta. + +## Tuning the load + +Every knob is an env var; the defaults are in `Config::from_env`. + +```sh +STRESS_PEERS=2048 STRESS_FILE_KB=8192 scripts/stress.sh +``` + +`STRESS_PEERS` `STRESS_SEARCHES` `STRESS_DOWNLOADS` `STRESS_LEECHERS` +`STRESS_BROWSES` `STRESS_FILE_KB` `STRESS_TIMEOUT` + +## Reading a failure + +`downloads completed 480/512` is the headline, but the two lines under it say +where to look: + +- **`download states`** — a straggler on `Queued` never got its transfer + negotiated; one on `InProgress` is merely slow. These are different bugs. +- **`mocks never online`** — mock peers the harness could not bring up. That is + a harness/server limit, not the client, and it is reported separately so it + cannot quietly shrink a denominator and flatter the score. + +`peak threads` counts the whole process — client *and* mock swarm — so treat it +as a trend, not a client measurement. + +`LOG_LEVEL=debug scripts/stress.sh` correlates a stuck peer's name against the +client's own trace, which is how each ceiling below was actually found. + +## Ceilings found and removed + +| Symptom | Cause | +|----------------------------------------------------|-------------------------------------------------------------------------------| +| Hard stop at ~15 peers; later peers never start | Every `PeerActor` held a `ThreadPool` worker for life, so the pool capped peers at one per core | +| Downloads stuck on `Queued` under load | `TransferResponse` was sent before the client recorded the peer's token, so the file connection raced its own bookkeeping | +| One download vanishing per few hundred | Token was `md5(filename)`, and the store deletes *every* entry with a token — same filename from two peers destroyed both | +| Uploads dropped after the peer accepted | Server answered `GetPeerAddress` with port 0; the client cached it and dialled it | + +## Requirements + +A soulfind binary (`SOULFIND_BIN`, defaulted in `scripts/stress.sh`), or +`SOULSEEK_TEST_SERVER=host:port` to point at a running server. The script raises +the descriptor limit — a swarm holds a lot of sockets open. diff --git a/benchmarks/stress-baseline.json b/benchmarks/stress-baseline.json new file mode 100644 index 0000000..2ce9a57 --- /dev/null +++ b/benchmarks/stress-baseline.json @@ -0,0 +1,23 @@ +{ + "score": 97.39, + "parts": { + "downloads": 1.0000, + "uploads": 1.0000, + "searches": 1.0000, + "browses": 1.0000, + "throughput": 0.8257 + }, + "load": { + "peers": 1024, + "searches": 128, + "downloads": 512, + "leechers": 256, + "browses": 128, + "file_bytes": 4194304 + }, + "observed": { + "downloads_completed": 512, + "uploads_completed": 256, + "bytes": 3221225472 + } +} diff --git a/scripts/stress.sh b/scripts/stress.sh new file mode 100755 index 0000000..36896ba --- /dev/null +++ b/scripts/stress.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +# Run the concurrency stress benchmark against a local soulfind. +# +# scripts/stress.sh # run with the default load profile +# scripts/stress.sh --save # store the run as the new baseline if it wins +# STRESS_PEERS=128 scripts/stress.sh +# +# Knobs: STRESS_PEERS STRESS_SEARCHES STRESS_DOWNLOADS STRESS_LEECHERS +# STRESS_BROWSES STRESS_FILE_KB STRESS_TIMEOUT +set -euo pipefail + +cd "$(dirname "$0")/.." + +# soulfind is a D binary linked against Homebrew's SQLite. +export SOULFIND_BIN="${SOULFIND_BIN:-$HOME/src/soulseek_download/soulfind/bin/soulfind}" +export DYLD_LIBRARY_PATH="${DYLD_LIBRARY_PATH:-/opt/homebrew/opt/sqlite/lib}" + +if [ ! -x "$SOULFIND_BIN" ]; then + echo "stress: no soulfind at $SOULFIND_BIN — set SOULFIND_BIN" >&2 + exit 2 +fi + +# Stale databases from a killed run make the next login behave oddly. +rm -f "${TMPDIR:-/tmp}"/soulfind-stress-*.db + +# Every mock peer holds sockets open; the default 256 descriptors is nowhere near +# enough for a swarm. +ulimit -n 20000 2>/dev/null || true + +for arg in "$@"; do + case "$arg" in + --save) export STRESS_SAVE=1 ;; + *) echo "stress: unknown argument $arg" >&2; exit 2 ;; + esac +done + +exec cargo run --release --quiet -p soulseek-rs-lib --example stress diff --git a/soulseek-rs-lib/examples/stress.rs b/soulseek-rs-lib/examples/stress.rs new file mode 100644 index 0000000..18ee654 --- /dev/null +++ b/soulseek-rs-lib/examples/stress.rs @@ -0,0 +1,1236 @@ +//! Concurrency stress harness for the soulseek-rs client. +//! +//! Spins up a local soulfind, surrounds one real `Client` with a swarm of +//! mock peers, and drives searches, downloads, uploads and browses at the same +//! time. Prints a metrics table and a single 0-100 score, and compares that +//! score against a stored baseline so regressions are visible. +//! +//! Run it with `scripts/stress.sh` (sets SOULFIND_BIN and release mode). +//! +//! Knobs (all env vars, defaults in `Config::from_env`): +//! STRESS_PEERS mock seeders online, each answering searches +//! STRESS_SEARCHES searches fired at once +//! STRESS_DOWNLOADS downloads started at once +//! STRESS_LEECHERS mock peers pulling uploads from us at once +//! STRESS_BROWSES browses fired at once +//! STRESS_FILE_KB size of every transferred file +//! STRESS_TIMEOUT seconds before the run is called off +//! STRESS_SAVE=1 store this run as the new baseline + +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use soulseek_rs::message::Message; +use soulseek_rs::message::peer::{ + FileEntry, SharedDirectory, build_file_search_response, + build_shared_file_list, +}; +use soulseek_rs::message::server::MessageFactory; +use soulseek_rs::{Client, ClientSettings, ClientVersion, PeerAddress}; + +/// Files matching this tag are what every mock seeder answers searches for. +const TAG: &str = "zqxstress"; + +/// Aggregate transfer rate (MiB/s) that scores full marks. Set well above any +/// real network link: the point is to keep the client, not the wire, as the +/// limiting factor, so the score still moves when transfer handling regresses. +const TARGET_MBPS: f64 = 500.0; + +const BASELINE: &str = "benchmarks/stress-baseline.json"; + +// --------------------------------------------------------------------------- +// Config +// --------------------------------------------------------------------------- + +struct Config { + peers: usize, + searches: usize, + downloads: usize, + leechers: usize, + browses: usize, + file_bytes: usize, + timeout: Duration, + save: bool, +} + +fn env_usize(key: &str, default: usize) -> usize { + std::env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +impl Config { + fn from_env() -> Self { + Self { + peers: env_usize("STRESS_PEERS", 1024), + searches: env_usize("STRESS_SEARCHES", 128), + downloads: env_usize("STRESS_DOWNLOADS", 512), + leechers: env_usize("STRESS_LEECHERS", 256), + browses: env_usize("STRESS_BROWSES", 128), + file_bytes: env_usize("STRESS_FILE_KB", 4096) * 1024, + timeout: Duration::from_secs(env_usize("STRESS_TIMEOUT", 60) as u64), + save: std::env::var("STRESS_SAVE").is_ok_and(|v| v == "1"), + } + } +} + +// --------------------------------------------------------------------------- +// Metrics +// --------------------------------------------------------------------------- + +#[derive(Default)] +struct Metrics { + peers_online: AtomicU64, + searches_issued: AtomicU64, + searches_answered: AtomicU64, + results_total: AtomicU64, + downloads_issued: AtomicU64, + downloads_completed: AtomicU64, + bytes_downloaded: AtomicU64, + uploads_requested: AtomicU64, + uploads_completed: AtomicU64, + bytes_uploaded: AtomicU64, + browses_issued: AtomicU64, + browses_answered: AtomicU64, + peer_connect_failures: AtomicU64, + /// Mock peers that never got online (server refused the login, no free + /// port, ...). Counted separately so a harness-side limit shows up as a + /// number instead of silently shrinking a denominator. + mocks_offline: AtomicU64, +} + +impl Metrics { + fn bump(counter: &AtomicU64) { + counter.fetch_add(1, Ordering::Relaxed); + } + fn get(counter: &AtomicU64) -> u64 { + counter.load(Ordering::Relaxed) + } +} + +fn ratio(done: u64, total: u64) -> f64 { + if total == 0 { + 1.0 + } else { + done as f64 / total as f64 + } +} + +// --------------------------------------------------------------------------- +// soulfind +// --------------------------------------------------------------------------- + +struct TestServer { + host: String, + port: u16, + child: Option, + db: Option, +} + +impl TestServer { + fn spawn() -> Option { + if let Ok(addr) = std::env::var("SOULSEEK_TEST_SERVER") { + let (host, port) = addr.rsplit_once(':')?; + let port = port.parse().ok()?; + wait_until_listening(host, port, Duration::from_secs(2))?; + return Some(Self { + host: host.to_string(), + port, + child: None, + db: None, + }); + } + + let bin = PathBuf::from(std::env::var("SOULFIND_BIN").ok()?); + let port = free_port()?; + let db = + std::env::temp_dir().join(format!("soulfind-stress-{port}.db")); + let _ = std::fs::remove_file(&db); + + let mut child = Command::new(&bin) + .arg("-p") + .arg(port.to_string()) + .arg("-d") + .arg(&db) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .ok()?; + + if wait_until_listening("127.0.0.1", port, Duration::from_secs(5)) + .is_none() + { + let _ = child.kill(); + let _ = child.wait(); + return None; + } + + Some(Self { + host: "127.0.0.1".to_string(), + port, + child: Some(child), + db: Some(db), + }) + } + + fn addr(&self) -> String { + format!("{}:{}", self.host, self.port) + } +} + +impl Drop for TestServer { + fn drop(&mut self) { + if let Some(mut child) = self.child.take() { + let _ = child.kill(); + let _ = child.wait(); + } + if let Some(db) = self.db.take() { + let _ = std::fs::remove_file(&db); + } + } +} + +fn free_port() -> Option { + TcpListener::bind("127.0.0.1:0") + .ok()? + .local_addr() + .ok() + .map(|a| a.port()) +} + +fn wait_until_listening( + host: &str, + port: u16, + timeout: Duration, +) -> Option<()> { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if TcpStream::connect((host, port)).is_ok() { + return Some(()); + } + std::thread::sleep(Duration::from_millis(50)); + } + None +} + +// --------------------------------------------------------------------------- +// Raw peer protocol helpers +// --------------------------------------------------------------------------- + +fn peer_init_bytes(username: &str, conn_type: &str, token: u32) -> Vec { + let mut m = Message::new(); + m.write_int8(1) + .write_string(username) + .write_string(conn_type) + .write_int32(token); + m.get_buffer() +} + +fn read_framed(stream: &mut TcpStream) -> std::io::Result { + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf)?; + let len = u32::from_le_bytes(len_buf) as usize; + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload)?; + let mut data = len_buf.to_vec(); + data.extend_from_slice(&payload); + Ok(Message::new_with_data(data)) +} + +fn connect_retry(addr: &str, timeout: Duration) -> std::io::Result { + let deadline = Instant::now() + timeout; + loop { + match TcpStream::connect(addr) { + Ok(stream) => return Ok(stream), + Err(e) if Instant::now() >= deadline => return Err(e), + Err(_) => std::thread::sleep(Duration::from_millis(20)), + } + } +} + +/// Log in to the server and drain up to the login response, leaving the user +/// online for as long as the returned stream lives. +fn login_raw(server_addr: &str, username: &str) -> std::io::Result { + let mut srv = connect_retry(server_addr, Duration::from_secs(10))?; + srv.set_read_timeout(Some(Duration::from_secs(20)))?; + srv.write_all( + &MessageFactory::build_login_message( + username, + "pw", + ClientVersion::default(), + ) + .get_buffer(), + )?; + srv.flush()?; + loop { + let msg = read_framed(&mut srv)?; + if msg.get_message_code() == 1 { + return Ok(srv); + } + } +} + +// --------------------------------------------------------------------------- +// Mock seeder: answers searches, serves browses, uploads files to the client +// --------------------------------------------------------------------------- + +struct Seeder { + name: String, + server_addr: String, + client_addr: String, + files: Vec<(String, u64)>, + content: Arc>, + metrics: Arc, + stop: Arc, + login_delay: Duration, +} + +impl Seeder { + fn run(self) { + // soulfind logs every login through SQLite, and a thousand of them + // arriving in the same millisecond starves it badly enough that the + // client under test cannot log in either. Real swarms do not arrive in + // lockstep; spreading the herd keeps the server out of the measurement. + std::thread::sleep(self.login_delay); + let Ok(mut srv) = login_raw(&self.server_addr, &self.name) else { + Metrics::bump(&self.metrics.mocks_offline); + return; + }; + let Some(port) = free_port() else { + Metrics::bump(&self.metrics.mocks_offline); + return; + }; + let Ok(listener) = TcpListener::bind(("127.0.0.1", port)) else { + Metrics::bump(&self.metrics.mocks_offline); + return; + }; + if srv + .write_all( + &MessageFactory::build_set_wait_port_message(port).get_buffer(), + ) + .is_err() + { + return; + } + let _ = srv.flush(); + Metrics::bump(&self.metrics.peers_online); + + // Inbound P connections (the client dialing us to queue a download or + // browse our shares) each get their own handler thread, exactly as a + // real peer would. + let inbound = Arc::new(self.clone_state()); + let stop = self.stop.clone(); + std::thread::spawn(move || { + let _ = listener.set_nonblocking(true); + while !stop.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, _)) => { + let state = inbound.clone(); + std::thread::spawn(move || { + let _ = stream.set_nonblocking(false); + state.serve_connection(stream, true); + }); + } + Err(_) => std::thread::sleep(Duration::from_millis(10)), + } + } + }); + + // Server connection: watch for relayed searches and answer the ones + // that match our tag by dialling the searcher directly. + let _ = srv.set_read_timeout(Some(Duration::from_millis(500))); + let state = Arc::new(self.clone_state()); + while !self.stop.load(Ordering::Relaxed) { + let Ok(mut msg) = read_framed(&mut srv) else { + continue; + }; + if msg.get_message_code() != 26 { + continue; + } + msg.set_pointer(8); + let _searcher = msg.read_string(); + let token = msg.read_int32(); + let query = msg.read_string(); + if !query.contains(TAG) { + continue; + } + state.answer_search(token); + } + } + + fn clone_state(&self) -> SeederState { + SeederState { + name: self.name.clone(), + client_addr: self.client_addr.clone(), + files: self.files.clone(), + content: self.content.clone(), + metrics: self.metrics.clone(), + stop: self.stop.clone(), + conn: Mutex::new(None), + } + } +} + +struct SeederState { + name: String, + client_addr: String, + files: Vec<(String, u64)>, + content: Arc>, + metrics: Arc, + stop: Arc, + /// The one control connection we hold open to the client, as a real peer + /// does. Dialling afresh per search response would be both unrealistic and + /// a self-inflicted load: it makes the harness, not the client, the thing + /// under strain. + conn: Mutex>, +} + +impl SeederState { + /// Push a `FileSearchResponse` over our control connection, opening it on + /// first use and reusing it thereafter. + fn answer_search(self: &Arc, token: u32) { + let entries: Vec = self + .files + .iter() + .map(|(name, size)| FileEntry { + name, + size: *size, + attribs: &[], + }) + .collect(); + let response = + build_file_search_response(&self.name, token, &entries, 1, 0); + + let Ok(mut guard) = self.conn.lock() else { + return; + }; + if guard.is_none() { + let Ok(mut p) = + connect_retry(&self.client_addr, Duration::from_secs(10)) + else { + Metrics::bump(&self.metrics.peer_connect_failures); + return; + }; + if p.write_all(&peer_init_bytes(&self.name, "P", 0)).is_err() { + Metrics::bump(&self.metrics.peer_connect_failures); + return; + } + // Read side runs on its own thread so browses and queued downloads + // arriving on this connection are still served. + let Ok(read_half) = p.try_clone() else { + Metrics::bump(&self.metrics.peer_connect_failures); + return; + }; + let state = self.clone(); + std::thread::spawn(move || { + state.serve_connection(read_half, false); + }); + *guard = Some(p); + } + + if let Some(p) = guard.as_mut() + && p.write_all(&response.get_buffer()).is_err() + { + Metrics::bump(&self.metrics.peer_connect_failures); + *guard = None; + } + } + + /// Write a reply on the connection `p` belongs to. When `p` is the read + /// half of the shared control connection, the write half lives behind + /// `conn` and every writer must take that lock — two threads writing the + /// same socket would interleave and corrupt the framing. + fn reply(&self, local: &mut TcpStream, shared: bool, bytes: &[u8]) -> bool { + if !shared { + return local.write_all(bytes).and_then(|()| local.flush()).is_ok(); + } + let Ok(mut guard) = self.conn.lock() else { + return false; + }; + let Some(p) = guard.as_mut() else { + return false; + }; + p.write_all(bytes).and_then(|()| p.flush()).is_ok() + } + + /// Handle one P connection for its lifetime: browses, queued downloads and + /// the F-connection file transfers they lead to. `expect_init` marks a + /// connection the client dialled (so it owns its own write half); anything + /// else is our shared control connection. + fn serve_connection(&self, mut p: TcpStream, expect_init: bool) { + let shared = !expect_init; + let _ = p.set_read_timeout(Some(Duration::from_millis(500))); + if expect_init { + // The client announces itself first; anything else is a stray dial. + let Ok(init) = read_framed(&mut p) else { + return; + }; + if init.get_message_code() != 1 { + return; + } + } + + while !self.stop.load(Ordering::Relaxed) { + let mut msg = match read_framed(&mut p) { + Ok(m) => m, + Err(ref e) if is_timeout(e) => continue, + Err(_) => return, + }; + match msg.get_message_code() { + // GetShareFileList: reply with everything we hold. + 4 => { + let dir = SharedDirectory { + name: format!("stress\\{}", self.name), + files: self.files.clone(), + }; + let list = + build_shared_file_list(std::slice::from_ref(&dir)); + if !self.reply(&mut p, shared, &list.get_buffer()) { + return; + } + } + // QueueUpload: offer the transfer, then stream it once allowed. + 43 => { + msg.set_pointer(8); + let requested = msg.read_string(); + let Some((name, size)) = + self.files.iter().find(|(n, _)| *n == requested) + else { + continue; + }; + let token = next_token(); + let mut tr = Message::new(); + tr.write_int32(40) + .write_int32(1) + .write_int32(token) + .write_string(name) + .write_int64(*size); + if !self.reply(&mut p, shared, &tr.get_buffer()) { + return; + } + if !self.wait_for_allow(&mut p, token) { + return; + } + let content = self.content.clone(); + let addr = self.client_addr.clone(); + let name = self.name.clone(); + std::thread::spawn(move || { + let _ = + serve_file_over_f(&addr, &name, token, &content); + }); + } + _ => {} + } + } + } + + /// Wait for the client's `TransferResponse` allowing `token`. + fn wait_for_allow(&self, p: &mut TcpStream, token: u32) -> bool { + let deadline = Instant::now() + Duration::from_secs(30); + while Instant::now() < deadline && !self.stop.load(Ordering::Relaxed) { + match read_framed(p) { + Ok(mut msg) if msg.get_message_code() == 41 => { + msg.set_pointer(8); + if msg.read_int32() == token { + return true; + } + } + Ok(_) => {} + Err(ref e) if is_timeout(e) => {} + Err(_) => return false, + } + } + false + } +} + +fn is_timeout(e: &std::io::Error) -> bool { + matches!( + e.kind(), + std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut + ) +} + +static NEXT_TOKEN: AtomicU64 = AtomicU64::new(0x1000); + +fn next_token() -> u32 { + NEXT_TOKEN.fetch_add(1, Ordering::Relaxed) as u32 +} + +/// Open an F connection to the downloader's listener and stream the file. +fn serve_file_over_f( + downloader_addr: &str, + username: &str, + token: u32, + content: &[u8], +) -> std::io::Result<()> { + let mut f = connect_retry(downloader_addr, Duration::from_secs(10))?; + f.set_read_timeout(Some(Duration::from_secs(30)))?; + let mut init = peer_init_bytes(username, "F", token); + init.extend_from_slice(&token.to_le_bytes()); + f.write_all(&init)?; + f.flush()?; + + let mut start = [0u8; 8]; + f.read_exact(&mut start)?; + let offset = u64::from_le_bytes(start) as usize; + if offset < content.len() { + f.write_all(&content[offset..])?; + } + f.flush()?; + std::thread::sleep(Duration::from_millis(300)); + Ok(()) +} + +// --------------------------------------------------------------------------- +// Mock leecher: pulls an upload out of the client under test +// --------------------------------------------------------------------------- + +struct Leecher { + name: String, + server_addr: String, + client_addr: String, + filename: String, + metrics: Arc, + /// The run's global deadline. Every blocking wait honours it so a stuck + /// transfer costs the iteration nothing beyond the budget already spent. + deadline: Instant, + login_delay: Duration, + /// Raised once the client is up. Leechers come online with the seeders so + /// the server has really registered their listening port by the time they + /// ask for a file — a real peer is online long before it queues anything. + start: Arc, +} + +impl Leecher { + fn run(self) { + std::thread::sleep(self.login_delay); + Metrics::bump(&self.metrics.uploads_requested); + let Ok(mut srv) = login_raw(&self.server_addr, &self.name) else { + Metrics::bump(&self.metrics.mocks_offline); + return; + }; + let Some(port) = free_port() else { + Metrics::bump(&self.metrics.mocks_offline); + return; + }; + let Ok(listener) = TcpListener::bind(("127.0.0.1", port)) else { + Metrics::bump(&self.metrics.mocks_offline); + return; + }; + // The client dials this port for the F leg, so the server must know it. + if srv + .write_all( + &MessageFactory::build_set_wait_port_message(port).get_buffer(), + ) + .is_err() + { + return; + } + let _ = srv.flush(); + + while !self.start.load(Ordering::Relaxed) { + if Instant::now() >= self.deadline { + return; + } + std::thread::sleep(Duration::from_millis(20)); + } + + if let Err(e) = self.pull(&listener) { + eprintln!("[leecher {}] {e}", self.name); + } + } + + fn pull(&self, listener: &TcpListener) -> std::io::Result<()> { + let mut p = connect_retry(&self.client_addr, Duration::from_secs(10))?; + p.set_read_timeout(Some(Duration::from_secs(30)))?; + p.write_all(&peer_init_bytes(&self.name, "P", 0))?; + p.write_all( + &MessageFactory::build_queue_upload_message(&self.filename) + .get_buffer(), + )?; + p.flush()?; + + // The client offers the transfer; accept it so it starts streaming. + let token = loop { + if Instant::now() >= self.deadline { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "client never offered the queued upload", + )); + } + match read_framed(&mut p) { + Ok(mut msg) if msg.get_message_code() == 40 => { + msg.set_pointer(8); + let _direction = msg.read_int32(); + break msg.read_int32(); + } + Ok(_) => {} + Err(ref e) if is_timeout(e) => {} + Err(e) => return Err(e), + } + }; + let mut ok = Message::new(); + ok.write_int32(41).write_int32(token).write_int8(1); + p.write_all(&ok.get_buffer())?; + p.flush()?; + + // Accept the client's inbound F connection and drain the file. The + // client also dials our P port, so keep accepting until the PeerInit + // announces "F" — otherwise we block reading a transfer token that a + // control connection is never going to send. + let mut f = accept_file_connection(listener, self.deadline)?; + f.write_all(&0u64.to_le_bytes())?; + f.flush()?; + + let mut sink = Vec::new(); + let received = f.read_to_end(&mut sink)? as u64; + self.metrics + .bytes_uploaded + .fetch_add(received, Ordering::Relaxed); + if received > 0 { + Metrics::bump(&self.metrics.uploads_completed); + } + Ok(()) + } +} + +/// Accept connections until one announces itself as an F (file) connection, +/// and return it positioned just past its PeerInit and transfer token. +fn accept_file_connection( + listener: &TcpListener, + deadline: Instant, +) -> std::io::Result { + listener.set_nonblocking(true)?; + let mut held = Vec::new(); + loop { + if Instant::now() >= deadline { + return Err(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "no inbound F connection", + )); + } + let Ok((mut stream, _)) = listener.accept() else { + std::thread::sleep(Duration::from_millis(10)); + continue; + }; + stream.set_nonblocking(false)?; + stream.set_read_timeout(Some(Duration::from_secs(30)))?; + let Ok(mut init) = read_framed(&mut stream) else { + continue; + }; + if init.get_message_code() != 1 { + continue; + } + init.set_pointer(5); + let _username = init.read_string(); + if init.read_string() != "F" { + // A control connection: hold it open so the client does not see a + // reset, and keep waiting for the transfer. + held.push(stream); + continue; + } + let mut token_buf = [0u8; 4]; + stream.read_exact(&mut token_buf)?; + drop(held); + return Ok(stream); + } +} + +// --------------------------------------------------------------------------- +// Scoring and baseline +// --------------------------------------------------------------------------- + +struct Score { + total: f64, + parts: Vec<(&'static str, f64, f64)>, // name, achieved 0-1, weight +} + +fn compute_score(m: &Metrics, elapsed: Duration) -> Score { + let bytes = + Metrics::get(&m.bytes_downloaded) + Metrics::get(&m.bytes_uploaded); + let mbps = (bytes as f64 / 1_048_576.0) / elapsed.as_secs_f64().max(0.001); + + let parts = vec![ + ( + "downloads", + ratio( + Metrics::get(&m.downloads_completed), + Metrics::get(&m.downloads_issued), + ), + 0.30, + ), + ( + "uploads", + ratio( + Metrics::get(&m.uploads_completed), + Metrics::get(&m.uploads_requested), + ), + 0.20, + ), + ( + "searches", + ratio( + Metrics::get(&m.searches_answered), + Metrics::get(&m.searches_issued), + ), + 0.20, + ), + ( + "browses", + ratio( + Metrics::get(&m.browses_answered), + Metrics::get(&m.browses_issued), + ), + 0.15, + ), + ("throughput", (mbps / TARGET_MBPS).min(1.0), 0.15), + ]; + + let total = 100.0 * parts.iter().map(|(_, v, w)| v * w).sum::(); + Score { total, parts } +} + +fn read_baseline(path: &Path) -> Option { + let text = std::fs::read_to_string(path).ok()?; + let idx = text.find("\"score\"")?; + text[idx + 7..] + .trim_start() + .trim_start_matches(':') + .trim_start() + .split(|c: char| !c.is_ascii_digit() && c != '.') + .find(|s| !s.is_empty())? + .parse() + .ok() +} + +fn write_baseline(path: &Path, score: &Score, cfg: &Config, m: &Metrics) { + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + let parts = score + .parts + .iter() + .map(|(n, v, _)| format!(" \"{n}\": {v:.4}")) + .collect::>() + .join(",\n"); + let body = format!( + "{{\n \"score\": {:.2},\n \"parts\": {{\n{}\n }},\n \"load\": {{\n \ + \"peers\": {},\n \"searches\": {},\n \"downloads\": {},\n \ + \"leechers\": {},\n \"browses\": {},\n \"file_bytes\": {}\n }},\n \ + \"observed\": {{\n \"downloads_completed\": {},\n \ + \"uploads_completed\": {},\n \"bytes\": {}\n }}\n}}\n", + score.total, + parts, + cfg.peers, + cfg.searches, + cfg.downloads, + cfg.leechers, + cfg.browses, + cfg.file_bytes, + Metrics::get(&m.downloads_completed), + Metrics::get(&m.uploads_completed), + Metrics::get(&m.bytes_downloaded) + Metrics::get(&m.bytes_uploaded), + ); + let _ = std::fs::write(path, body); +} + +/// Live thread count for this process, so the harness can show whether load is +/// being absorbed or simply turned into threads. +fn thread_count() -> usize { + Command::new("ps") + .args(["-M", &std::process::id().to_string()]) + .output() + .ok() + .map_or(0, |o| String::from_utf8_lossy(&o.stdout).lines().count()) + .saturating_sub(1) +} + +// --------------------------------------------------------------------------- +// The run +// --------------------------------------------------------------------------- + +fn main() { + let cfg = Config::from_env(); + let Some(server) = TestServer::spawn() else { + eprintln!( + "stress: no soulfind server. Set SOULFIND_BIN (or SOULSEEK_TEST_SERVER)." + ); + std::process::exit(2); + }; + let server_addr = server.addr(); + let metrics = Arc::new(Metrics::default()); + let stop = Arc::new(AtomicBool::new(false)); + + // Shares the client under test offers to mock leechers. + let run_id = std::process::id(); + let share_dir = + std::env::temp_dir().join(format!("soulseek-stress-share-{run_id}")); + let download_dir = + std::env::temp_dir().join(format!("soulseek-stress-dl-{run_id}")); + let _ = std::fs::remove_dir_all(&share_dir); + let _ = std::fs::remove_dir_all(&download_dir); + std::fs::create_dir_all(&share_dir).expect("share dir"); + std::fs::create_dir_all(&download_dir).expect("download dir"); + + let payload: Vec = + (0..cfg.file_bytes).map(|i| (i % 251) as u8).collect(); + // Peers ask for the backslash-separated virtual path (root dir name first), + // not the basename — anything else is silently unknown to the share index. + let share_root = share_dir + .file_name() + .expect("share dir name") + .to_string_lossy() + .into_owned(); + let shared_names: Vec = (0..cfg.leechers.max(1)) + .map(|i| format!("{share_root}\\{TAG}_own_{i}.bin")) + .collect(); + for name in &shared_names { + let basename = name.rsplit('\\').next().expect("basename"); + std::fs::write(share_dir.join(basename), &payload) + .expect("write share"); + } + + let client_port = free_port().expect("client port"); + let client_name = format!("stress_client_{run_id}"); + let client_addr = format!("127.0.0.1:{client_port}"); + + println!( + "stress: {} peers, {} searches, {} downloads, {} leechers, {} browses, \ + {} KiB files", + cfg.peers, + cfg.searches, + cfg.downloads, + cfg.leechers, + cfg.browses, + cfg.file_bytes / 1024 + ); + + // --- mock seeders ----------------------------------------------------- + let content = Arc::new(payload); + let mut seeder_names = Vec::new(); + for i in 0..cfg.peers { + let name = format!("stress_seed_{run_id}_{i}"); + seeder_names.push(name.clone()); + let files = + vec![(format!("share\\{TAG}_{i}.bin"), cfg.file_bytes as u64)]; + let seeder = Seeder { + name, + server_addr: server_addr.clone(), + client_addr: client_addr.clone(), + files, + content: content.clone(), + metrics: metrics.clone(), + stop: stop.clone(), + login_delay: Duration::from_micros(1500 * i as u64), + }; + std::thread::spawn(move || seeder.run()); + } + + // --- mock leechers, brought online alongside the seeders --------------- + // They wait on `start` before queueing anything, so the server has really + // registered their listening port by the time we ask it for their address. + let start = Arc::new(AtomicBool::new(false)); + let run_deadline_hint = + Instant::now() + cfg.timeout + Duration::from_secs(40); + let mut leechers = Vec::new(); + for i in 0..cfg.leechers { + let leecher = Leecher { + name: format!("stress_leech_{run_id}_{i}"), + server_addr: server_addr.clone(), + client_addr: client_addr.clone(), + filename: shared_names[i % shared_names.len()].clone(), + metrics: metrics.clone(), + deadline: run_deadline_hint, + login_delay: Duration::from_micros(1500 * i as u64), + start: start.clone(), + }; + leechers.push(std::thread::spawn(move || leecher.run())); + } + + // Let the swarm register its wait ports before the client starts. + let deadline = Instant::now() + Duration::from_secs(30); + while Metrics::get(&metrics.peers_online) < cfg.peers as u64 + && Instant::now() < deadline + { + std::thread::sleep(Duration::from_millis(50)); + } + println!( + "stress: {}/{} seeders online", + Metrics::get(&metrics.peers_online), + cfg.peers + ); + + // --- client under test ------------------------------------------------ + let mut client = Client::with_settings(ClientSettings { + username: client_name, + password: "pw".to_string(), + server_address: PeerAddress::new(server.host.clone(), server.port), + enable_listen: true, + listen_port: client_port, + shared_directories: vec![share_dir.display().to_string()], + version: ClientVersion::default(), + }); + client.connect().expect("client connect"); + assert!(client.login().expect("client login"), "client login failed"); + std::thread::sleep(Duration::from_secs(1)); + + let client = Arc::new(client); + let started = Instant::now(); + let run_deadline = started + cfg.timeout; + let peak_threads = Arc::new(AtomicU64::new(0)); + + // Sample thread count while the run is in flight. + { + let peak = peak_threads.clone(); + let stop = stop.clone(); + std::thread::spawn(move || { + while !stop.load(Ordering::Relaxed) { + let n = thread_count() as u64; + peak.fetch_max(n, Ordering::Relaxed); + std::thread::sleep(Duration::from_millis(200)); + } + }); + } + + let mut workers = Vec::new(); + + // --- concurrent searches --------------------------------------------- + let queries: Vec = + (0..cfg.searches).map(|k| format!("{TAG}_q{k}")).collect(); + for query in queries.clone() { + let client = client.clone(); + let metrics = metrics.clone(); + workers.push(std::thread::spawn(move || { + Metrics::bump(&metrics.searches_issued); + let _ = client.search(&query, Duration::from_secs(3)); + })); + } + + // --- concurrent uploads: release the leechers already standing by ----- + start.store(true, Ordering::Relaxed); + workers.extend(leechers); + + // --- concurrent browses ---------------------------------------------- + for name in seeder_names.iter().take(cfg.browses) { + let client = client.clone(); + let metrics = metrics.clone(); + let name = name.clone(); + workers.push(std::thread::spawn(move || { + Metrics::bump(&metrics.browses_issued); + if client.browse_user(&name).is_err() { + return; + } + let deadline = run_deadline; + while Instant::now() < deadline { + if client.take_browse_result(&name).is_some() { + Metrics::bump(&metrics.browses_answered); + return; + } + std::thread::sleep(Duration::from_millis(100)); + } + })); + } + + // --- concurrent downloads -------------------------------------------- + // Collect distinct (peer, file) hits from the searches, then fire every + // download at once — that simultaneity is the point of the exercise. + let hit_deadline = started + cfg.timeout / 3; + let mut hits: HashMap = HashMap::new(); + let mut answered = 0; + let mut results_seen = 0u64; + // Keep collecting until there is enough to download *and* every search has + // been answered, so search recall is measured even on download-free runs. + while Instant::now() < hit_deadline + && (hits.len() < cfg.downloads || answered < queries.len()) + { + answered = 0; + results_seen = 0; + for query in &queries { + let results = client.get_search_results(query); + if !results.is_empty() { + answered += 1; + } + for result in results { + results_seen += result.files.len() as u64; + if let Some(file) = result.files.first() { + hits.entry(result.username.clone()) + .or_insert_with(|| (file.name.clone(), file.size)); + } + } + } + std::thread::sleep(Duration::from_millis(100)); + } + metrics + .searches_answered + .store(answered as u64, Ordering::Relaxed); + metrics.results_total.store(results_seen, Ordering::Relaxed); + println!( + "stress: {} distinct peers answered searches, firing {} downloads", + hits.len(), + hits.len().min(cfg.downloads) + ); + + let completed = Arc::new(Mutex::new(Vec::new())); + for (peer, (filename, size)) in hits.into_iter().take(cfg.downloads) { + let client = client.clone(); + let metrics = metrics.clone(); + let dir = download_dir.display().to_string(); + let completed = completed.clone(); + workers.push(std::thread::spawn(move || { + Metrics::bump(&metrics.downloads_issued); + let Ok((_download, status_rx)) = + client.download(filename.clone(), peer.clone(), size, dir) + else { + return; + }; + let deadline = run_deadline; + while Instant::now() < deadline { + match status_rx.recv_timeout(Duration::from_millis(500)) { + Ok(soulseek_rs::DownloadStatus::Completed) => { + Metrics::bump(&metrics.downloads_completed); + metrics + .bytes_downloaded + .fetch_add(size, Ordering::Relaxed); + if let Ok(mut c) = completed.lock() { + c.push(peer); + } + return; + } + Ok( + soulseek_rs::DownloadStatus::Failed(_) + | soulseek_rs::DownloadStatus::TimedOut, + ) => return, + _ => {} + } + } + })); + } + + // --- wait for the run to settle -------------------------------------- + for worker in workers { + let remaining = run_deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + let _ = worker.join(); + } + let elapsed = started.elapsed(); + stop.store(true, Ordering::Relaxed); + + // --- report ----------------------------------------------------------- + let score = compute_score(&metrics, elapsed); + let bytes = Metrics::get(&metrics.bytes_downloaded) + + Metrics::get(&metrics.bytes_uploaded); + + println!("\n--- results ({:.1}s) ---", elapsed.as_secs_f64()); + println!( + " seeders online {}/{}", + Metrics::get(&metrics.peers_online), + cfg.peers + ); + println!( + " searches answered {}/{}", + Metrics::get(&metrics.searches_answered), + Metrics::get(&metrics.searches_issued) + ); + println!( + " search results {}", + Metrics::get(&metrics.results_total) + ); + println!( + " downloads completed {}/{}", + Metrics::get(&metrics.downloads_completed), + Metrics::get(&metrics.downloads_issued) + ); + println!( + " uploads completed {}/{}", + Metrics::get(&metrics.uploads_completed), + Metrics::get(&metrics.uploads_requested) + ); + println!( + " browses answered {}/{}", + Metrics::get(&metrics.browses_answered), + Metrics::get(&metrics.browses_issued) + ); + println!( + " peer dial failures {}", + Metrics::get(&metrics.peer_connect_failures) + ); + println!( + " mocks never online {} (harness-side, not the client)", + Metrics::get(&metrics.mocks_offline) + ); + println!( + " transferred {:.1} MiB ({:.1} MiB/s)", + bytes as f64 / 1_048_576.0, + bytes as f64 / 1_048_576.0 / elapsed.as_secs_f64().max(0.001) + ); + println!( + " peak threads {} (client + mock swarm, same process)", + peak_threads.load(Ordering::Relaxed) + ); + + // Where the stragglers got stuck is the whole diagnostic: a download parked + // in Queued never got its transfer negotiated, while one in InProgress is + // simply slow. + let mut by_status: HashMap<&'static str, usize> = HashMap::new(); + for download in client.get_all_downloads() { + let label = match download.status { + soulseek_rs::DownloadStatus::Queued => "Queued", + soulseek_rs::DownloadStatus::InProgress { .. } => "InProgress", + soulseek_rs::DownloadStatus::Completed => "Completed", + soulseek_rs::DownloadStatus::Failed(_) => "Failed", + soulseek_rs::DownloadStatus::TimedOut => "TimedOut", + soulseek_rs::DownloadStatus::Paused { .. } => "Paused", + }; + *by_status.entry(label).or_default() += 1; + } + if !by_status.is_empty() { + let mut parts: Vec = + by_status.iter().map(|(k, v)| format!("{k}={v}")).collect(); + parts.sort(); + println!(" download states {}", parts.join(" ")); + } + let stuck: Vec = client + .get_all_downloads() + .iter() + .filter(|d| matches!(d.status, soulseek_rs::DownloadStatus::Queued)) + .map(|d| d.username.clone()) + .collect(); + if !stuck.is_empty() { + println!(" stuck peers {}", stuck.join(" ")); + } + + println!("\n--- score ---"); + for (name, value, weight) in &score.parts { + println!(" {name:<12} {:>6.1}% x{weight:.2}", value * 100.0); + } + println!(" TOTAL {:.2}/100", score.total); + + let baseline_path = PathBuf::from(BASELINE); + if let Some(prev) = read_baseline(&baseline_path) { + let delta = score.total - prev; + let sign = if delta >= 0.0 { "+" } else { "" }; + println!(" baseline {prev:.2} ({sign}{delta:.2})"); + if cfg.save && score.total > prev { + write_baseline(&baseline_path, &score, &cfg, &metrics); + println!(" baseline updated to {:.2}", score.total); + } else if cfg.save { + println!(" baseline kept (run did not beat it)"); + } + } else { + write_baseline(&baseline_path, &score, &cfg, &metrics); + println!(" baseline created at {:.2}", score.total); + } + + let _ = std::fs::remove_dir_all(&share_dir); + let _ = std::fs::remove_dir_all(&download_dir); +} diff --git a/soulseek-rs-lib/src/actor/mod.rs b/soulseek-rs-lib/src/actor/mod.rs index f409a22..9bb1623 100644 --- a/soulseek-rs-lib/src/actor/mod.rs +++ b/soulseek-rs-lib/src/actor/mod.rs @@ -1,9 +1,8 @@ -use std::sync::Arc; use std::sync::mpsc::{Receiver, Sender, channel}; +use std::thread; use std::time::{Duration, Instant}; -use crate::trace; -use crate::utils::thread_pool::ThreadPool; +use crate::{error, trace}; pub mod peer_actor; pub mod peer_registry; @@ -58,36 +57,37 @@ pub(crate) enum ActorMessage { Stop, } -/// Actor system that manages actor lifecycle -pub struct ActorSystem { - thread_pool: Arc, -} +/// Stack size for an actor thread. An actor loop parses messages into +/// heap-allocated buffers and never recurses, so it needs a fraction of the +/// 2 MiB default — and at a thousand concurrent peers that difference is the +/// difference between 2 GiB of reserved address space and 256 MiB. +const ACTOR_STACK_SIZE: usize = 256 * 1024; + +/// Actor system that manages actor lifecycle. +/// +/// Each actor owns a thread for its whole lifetime. That is deliberate: an +/// actor loop blocks in `recv_timeout` until it is stopped, so running actors +/// on a fixed-size pool would make the pool a semaphore capping the client at +/// one peer per core — and every peer past the cap would silently never start. +#[derive(Default)] +pub struct ActorSystem; impl ActorSystem { #[must_use] - pub const fn new(thread_pool: Arc) -> Self { - Self { thread_pool } + pub const fn new() -> Self { + Self } /// Spawn a new actor and return its handle - pub fn spawn(&self, mut actor: A) -> ActorHandle { - let (sender, receiver) = channel::>(); - let handle = ActorHandle { sender }; - - self.thread_pool.execute(move || { - actor.on_start(); - Self::run_actor_loop(&mut actor, receiver); - actor.on_stop(); - }); - - handle + pub fn spawn(&self, actor: A) -> ActorHandle { + self.spawn_with_handle(actor, |_, _| {}) } /// Spawn a new actor with initialization callback and return its handle /// The callback receives the actor handle before `on_start` is called pub fn spawn_with_handle( &self, - mut actor: A, + actor: A, init: F, ) -> ActorHandle where @@ -96,13 +96,21 @@ impl ActorSystem { let (sender, receiver) = channel::>(); let handle = ActorHandle { sender }; let handle_for_init = handle.clone(); + let mut actor = actor; - self.thread_pool.execute(move || { - init(&mut actor, handle_for_init); - actor.on_start(); - Self::run_actor_loop(&mut actor, receiver); - actor.on_stop(); - }); + let spawned = thread::Builder::new() + .name("soulseek-actor".to_string()) + .stack_size(ACTOR_STACK_SIZE) + .spawn(move || { + init(&mut actor, handle_for_init); + actor.on_start(); + Self::run_actor_loop(&mut actor, receiver); + actor.on_stop(); + }); + + if let Err(e) = spawned { + error!("[actor_system] failed to spawn actor thread: {}", e); + } handle } @@ -153,6 +161,7 @@ impl ActorSystem { #[cfg(test)] mod tests { use super::*; + use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; struct CounterActor { @@ -177,8 +186,7 @@ mod tests { #[test] fn test_actor_system() { - let thread_pool = Arc::new(ThreadPool::new(4)); - let system = ActorSystem::new(thread_pool.clone()); + let system = ActorSystem::new(); let count = Arc::new(AtomicUsize::new(0)); let actor = CounterActor { @@ -201,8 +209,39 @@ mod tests { // Give actor time to process the stop message std::thread::sleep(Duration::from_millis(100)); + } - // Drop the thread pool explicitly to wait for all threads - drop(thread_pool); + // Actors must not be capped by a worker count: every peer runs one, and a + // cap silently strands every peer past it. Far more actors than cores must + // all make progress concurrently. + #[test] + fn far_more_actors_than_cores_all_run() { + let system = ActorSystem::new(); + let count = Arc::new(AtomicUsize::new(0)); + + let handles: Vec<_> = (0..256) + .map(|_| { + system.spawn(CounterActor { + count: count.clone(), + }) + }) + .collect(); + for handle in &handles { + handle.send(1).unwrap(); + } + + let deadline = Instant::now() + Duration::from_secs(10); + while count.load(Ordering::SeqCst) < 256 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + } + assert_eq!( + count.load(Ordering::SeqCst), + 256, + "every actor should have processed its message" + ); + + for handle in &handles { + handle.stop().unwrap(); + } } } diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index 5c5d393..37a4cf8 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -295,6 +295,14 @@ impl PeerActor { } } + /// A peer offers us a file. Record the peer's transfer token with the + /// client, and let *that* step send the TransferResponse. + /// + /// The response is what invites the peer's file connection, and that + /// connection is matched by the peer's token — so replying here would race + /// the client loop that records the token. Under load the client loop is + /// the one that loses, the file connection finds no download, and the + /// transfer is dropped with the download stuck on Queued forever. fn handle_transfer_request(&self, transfer: Transfer) { let username = self.peer_username(); debug!("[peer:{}] TransferRequest for {}", username, transfer.token); @@ -302,7 +310,7 @@ impl PeerActor { if let Err(e) = self.client_channel .send(ClientOperation::UpdateDownloadTokens( - transfer.clone(), + transfer, username.clone(), )) { @@ -311,19 +319,6 @@ impl PeerActor { username, e ); } - - let transfer_response = - MessageFactory::build_transfer_response_message(transfer); - - if let Some(ref handle) = self.self_handle - && let Err(e) = - handle.send(PeerMessage::SendMessage(transfer_response)) - { - error!( - "[peer:{}] Failed to send TransferResponse message: {}", - username, e - ); - } } fn handle_transfer_response( diff --git a/soulseek-rs-lib/src/actor/peer_registry.rs b/soulseek-rs-lib/src/actor/peer_registry.rs index 7c7338d..5f6413f 100644 --- a/soulseek-rs-lib/src/actor/peer_registry.rs +++ b/soulseek-rs-lib/src/actor/peer_registry.rs @@ -215,14 +215,12 @@ mod tests { use super::PeerRegistry; use crate::actor::ActorSystem; use crate::peer::{ConnectionType, Peer}; - use crate::utils::thread_pool::ThreadPool; use std::net::{TcpListener, TcpStream}; use std::sync::Arc; #[test] fn remove_peer_if_respects_actor_identity() { - let pool = Arc::new(ThreadPool::new(2)); - let system = Arc::new(ActorSystem::new(pool)); + let system = Arc::new(ActorSystem::new()); let (tx, _rx) = std::sync::mpsc::channel(); let registry = PeerRegistry::new(system, tx, "me".to_string()); diff --git a/soulseek-rs-lib/src/client/downloads.rs b/soulseek-rs-lib/src/client/downloads.rs index 19bcf39..076d4d6 100644 --- a/soulseek-rs-lib/src/client/downloads.rs +++ b/soulseek-rs-lib/src/client/downloads.rs @@ -1,6 +1,7 @@ use super::{ Arc, Client, ClientContext, Download, DownloadMetadata, DownloadStatus, - Receiver, Result, RwLock, RwLockExt, Sender, error, info, md5, mpsc, + Receiver, Result, RwLock, RwLockExt, Sender, error, info, mpsc, + next_download_token, }; impl Client { @@ -52,9 +53,8 @@ impl Client { } /// Remove every download for `username`/`filename` regardless of status. - /// Call this before re-issuing [`Client::download`] for a failed download, - /// otherwise the stale entry (whose md5-derived token equals the retry's) - /// shadows the fresh one and its completion is misrouted. + /// Useful before re-issuing [`Client::download`] for a failed download, so + /// the list does not accumulate dead entries for the same file. /// /// Returns whether anything was removed. #[must_use] @@ -94,8 +94,7 @@ impl Client { ) -> Result<(Download, Receiver)> { info!("[client] Downloading {} from {}", filename, username); - let hash = md5::md5(&filename); - let token = u32::from_str_radix(&hash[0..5], 16)?; + let token = next_download_token(); let (download_sender, download_receiver): ( Sender, @@ -189,3 +188,29 @@ impl Client { } } } + +#[cfg(test)] +mod tests { + use super::next_download_token; + use std::collections::HashSet; + + // The download store is keyed by token and removes *every* entry matching + // one, so two downloads sharing a token means queueing one silently evicts + // the other — which is what fetching the same filename from two peers does + // if the token depends on the filename. + #[test] + fn download_tokens_are_unique_and_below_the_upload_range() { + let tokens: Vec = + (0..10_000).map(|_| next_download_token()).collect(); + + assert_eq!( + tokens.iter().collect::>().len(), + tokens.len(), + "every download must get its own token" + ); + assert!( + tokens.iter().all(|t| *t < 0x8000_0000), + "download tokens must not reach into the upload range" + ); + } +} diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 1066de5..0e9a762 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -19,7 +19,7 @@ use crate::{ }, shares::Shares, types::{Download, Search, SearchResult}, - utils::{lock::RwLockExt, md5, thread_pool::ThreadPool}, + utils::{lock::RwLockExt, md5}, }; use std::{ collections::{HashMap, HashSet}, @@ -65,13 +65,25 @@ fn next_connect_token() -> u32 { } /// Upload tokens are minted in the high half of the space so they never collide -/// with download tokens (md5-derived, always < 2^20). +/// with download tokens, which are always < 2^31. static NEXT_UPLOAD_TOKEN: AtomicU32 = AtomicU32::new(0x8000_0000); fn next_upload_token() -> u32 { NEXT_UPLOAD_TOKEN.fetch_add(1, Ordering::Relaxed) } +/// Source of download tokens, kept in the low half of the space. +/// +/// A counter, not a hash of the filename: the download store is keyed by token +/// and removes every entry matching one, so any two downloads sharing a token +/// destroy each other — and the same filename from two peers is an ordinary +/// thing to queue. +static NEXT_DOWNLOAD_TOKEN: AtomicU32 = AtomicU32::new(1); + +fn next_download_token() -> u32 { + NEXT_DOWNLOAD_TOKEN.fetch_add(1, Ordering::Relaxed) % 0x8000_0000 +} + /// A file we have agreed to serve to a peer, awaiting their TransferResponse. struct UploadJob { downloader: String, @@ -386,11 +398,7 @@ impl ClientContext { impl ClientContext { #[must_use] pub fn new() -> Self { - let max_threads = - thread::available_parallelism().map_or(8, std::num::NonZero::get); - - let thread_pool = Arc::new(ThreadPool::new(max_threads)); - let actor_system = Arc::new(ActorSystem::new(thread_pool)); + let actor_system = Arc::new(ActorSystem::new()); Self { peer_registry: None, diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index 63bf1b1..93942f1 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -316,9 +316,26 @@ impl Client { obfuscated_port ); - // Cache the address for the serve/search paths, - // and collect any uploads waiting on it. - let waiting_serves = + // Port 0 is the server saying it does not know + // where this user listens (no SetWaitPort yet, + // or firewalled). It is not an address: caching + // it poisons every later lookup, and an upload + // dialled at it dies with "Can't assign + // requested address" and is dropped on the + // floor. Leave those uploads queued for a later + // resolution instead. + // + // The connect attempt further down still runs: + // its failure is exactly what makes an + // unreachable peer fall back to the + // server-brokered path. + let waiting_serves = if port == 0 { + warn!( + "[client] server reports no listening port for {}", + username + ); + Vec::new() + } else { match client_context.write_safe() { Ok(mut ctx) => { ctx.cache_peer_address( @@ -331,7 +348,8 @@ impl Client { .unwrap_or_default() } Err(_) => Vec::new(), - }; + } + }; for token in waiting_serves { Self::spawn_serve( &client_context, @@ -432,7 +450,7 @@ impl Client { context.add_download(Download { username: username.clone(), - filename: transfer.filename, + filename: transfer.filename.clone(), token: transfer.token, size: transfer.size, download_directory: download @@ -444,6 +462,24 @@ impl Client { }); context.remove_download(old_token); } + + // Only now invite the file connection: it is + // matched by this token, which is recorded as + // of the line above. Answering any earlier + // races the peer's connection against our own + // bookkeeping. + let registry = context.peer_registry.clone(); + drop(context); + if let Some(registry) = registry { + let response = + crate::message::server::MessageFactory::build_transfer_response_message( + transfer, + ); + let _ = registry.send_to_peer( + &username, + PeerMessage::SendMessage(response), + ); + } } ClientOperation::UploadFailed( username, diff --git a/soulseek-rs-lib/src/utils/mod.rs b/soulseek-rs-lib/src/utils/mod.rs index d684ac7..6b42e19 100644 --- a/soulseek-rs-lib/src/utils/mod.rs +++ b/soulseek-rs-lib/src/utils/mod.rs @@ -3,7 +3,6 @@ pub mod logger; pub mod lock; pub mod md5; pub mod path; -pub mod thread_pool; pub mod zlib; // Re-export commonly used items diff --git a/soulseek-rs-lib/src/utils/thread_pool.rs b/soulseek-rs-lib/src/utils/thread_pool.rs deleted file mode 100644 index 20fe59f..0000000 --- a/soulseek-rs-lib/src/utils/thread_pool.rs +++ /dev/null @@ -1,124 +0,0 @@ -use std::sync::{Arc, Mutex, mpsc}; -use std::thread; - -use crate::utils::lock::MutexExt; - -type Job = Box; - -enum Message { - NewJob(Job), - Terminate, -} - -pub struct ThreadPool { - workers: Vec, - sender: Option>, -} - -impl ThreadPool { - #[must_use] - pub fn new(size: usize) -> Self { - assert!(size > 0); - - let (sender, receiver) = mpsc::channel(); - let receiver = Arc::new(Mutex::new(receiver)); - let mut workers = Vec::with_capacity(size); - - for _ in 0..size { - workers.push(Worker::new(Arc::clone(&receiver))); - } - - Self { - workers, - sender: Some(sender), - } - } - - pub fn execute(&self, f: F) - where - F: FnOnce() + Send + 'static, - { - let job = Box::new(f); - if let Some(ref sender) = self.sender { - let _ = sender.send(Message::NewJob(job)); - } - } -} - -impl Drop for ThreadPool { - fn drop(&mut self) { - if let Some(sender) = self.sender.take() { - for _ in &self.workers { - let _ = sender.send(Message::Terminate); - } - } - - // Ignore join errors: a worker whose job panicked returns Err here, but - // Drop must never itself panic (that would abort the process). - for worker in &mut self.workers { - if let Some(thread) = worker.thread.take() { - let _ = thread.join(); - } - } - } -} - -struct Worker { - thread: Option>, -} - -impl Worker { - fn new(receiver: Arc>>) -> Self { - let thread = thread::spawn(move || { - loop { - let message = match receiver.lock_safe() { - Ok(rx) => rx.recv(), - Err(_) => break, - }; - match message { - // Contain a panicking job so it kills only that job, not the - // worker: the pool has a fixed number of workers and a lost - // one is never replaced. The lock is already released here, - // so a panic cannot poison the shared receiver. - Ok(Message::NewJob(job)) => { - let _ = std::panic::catch_unwind( - std::panic::AssertUnwindSafe(job), - ); - } - Ok(Message::Terminate) | Err(_) => break, - } - } - }); - - Self { - thread: Some(thread), - } - } -} - -#[cfg(test)] -mod tests { - use super::ThreadPool; - use std::sync::mpsc; - use std::time::Duration; - - // A single panicking job must not permanently kill its worker: subsequent - // jobs still need to run. Otherwise one malformed network message could - // leak a pool worker until the whole pool is exhausted (a DoS). - #[test] - fn pool_survives_a_panicking_job() { - let pool = ThreadPool::new(1); - pool.execute(|| panic!("boom")); - - let (tx, rx) = mpsc::channel(); - pool.execute(move || { - let _ = tx.send(42); - }); - - assert_eq!( - rx.recv_timeout(Duration::from_secs(5)).ok(), - Some(42), - "worker died after a panicking job" - ); - } -}