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" - ); - } -}