feat: lift the client's concurrency ceiling from ~15 peers to 1024
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.
This commit is contained in:
Michel de Graaf
@@ -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.
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
Executable
+37
@@ -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
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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<M> {
|
||||
Stop,
|
||||
}
|
||||
|
||||
/// Actor system that manages actor lifecycle
|
||||
pub struct ActorSystem {
|
||||
thread_pool: Arc<ThreadPool>,
|
||||
}
|
||||
/// 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<ThreadPool>) -> Self {
|
||||
Self { thread_pool }
|
||||
pub const fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
|
||||
/// Spawn a new actor and return its handle
|
||||
pub fn spawn<A: Actor>(&self, mut actor: A) -> ActorHandle<A::Message> {
|
||||
let (sender, receiver) = channel::<ActorMessage<A::Message>>();
|
||||
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<A: Actor>(&self, actor: A) -> ActorHandle<A::Message> {
|
||||
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<A: Actor, F>(
|
||||
&self,
|
||||
mut actor: A,
|
||||
actor: A,
|
||||
init: F,
|
||||
) -> ActorHandle<A::Message>
|
||||
where
|
||||
@@ -96,13 +96,21 @@ impl ActorSystem {
|
||||
let (sender, receiver) = channel::<ActorMessage<A::Message>>();
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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());
|
||||
|
||||
|
||||
@@ -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<DownloadStatus>)> {
|
||||
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<DownloadStatus>,
|
||||
@@ -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<u32> =
|
||||
(0..10_000).map(|_| next_download_token()).collect();
|
||||
|
||||
assert_eq!(
|
||||
tokens.iter().collect::<HashSet<_>>().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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,124 +0,0 @@
|
||||
use std::sync::{Arc, Mutex, mpsc};
|
||||
use std::thread;
|
||||
|
||||
use crate::utils::lock::MutexExt;
|
||||
|
||||
type Job = Box<dyn FnOnce() + Send + 'static>;
|
||||
|
||||
enum Message {
|
||||
NewJob(Job),
|
||||
Terminate,
|
||||
}
|
||||
|
||||
pub struct ThreadPool {
|
||||
workers: Vec<Worker>,
|
||||
sender: Option<mpsc::Sender<Message>>,
|
||||
}
|
||||
|
||||
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<F>(&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<thread::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
impl Worker {
|
||||
fn new(receiver: Arc<Mutex<mpsc::Receiver<Message>>>) -> 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"
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user