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

Merge branch 'develop': fd-exhaustion fixes

Browse Source
This commit is contained in:
Michel de Graaf
2026-08-10 09:22:50 +02:00
parent b8cdb2c3fe 0bbb559f68
commit f24a7f6369
17 changed files with 442 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3
View File
@@ -2569,6 +2569,9 @@ dependencies = [
[[package]]
name = "soulseek-rs-lib"
version = "13.0.0"
dependencies = [
"libc",
]
[[package]]
name = "stable_deref_trait"
soulseek-rs-lib/Cargo.toml
+3
View File
@@ -18,3 +18,6 @@ path = "src/lib.rs"
workspace = true
[dependencies]
[target.'cfg(unix)'.dependencies]
libc = "0.2.189"
soulseek-rs-lib/src/actor/mod.rs
+29 -8
View File
@@ -87,9 +87,34 @@ impl ActorSystem {
/// 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
F: FnOnce(&mut A, ActorHandle<A::Message>) + Send + 'static,
{
match self.try_spawn_with_handle(actor, init) {
Ok(handle) => handle,
Err(e) => {
error!("[actor_system] failed to spawn actor thread: {}", e);
// A handle nobody serves: sends fail, which callers of the
// infallible spawn already tolerate.
let (sender, _) = channel();
ActorHandle { sender }
}
}
}
/// Like [`Self::spawn_with_handle`], but hands a thread-spawn failure back
/// to the caller instead of returning a handle no actor will ever serve.
/// Storing such a dead handle is how a registry entry becomes a zombie
/// that claims its username for the rest of the process.
#[allow(clippy::unused_self)] // method for symmetry with the spawn family
pub(crate) fn try_spawn_with_handle<A: Actor, F>(
&self,
mut actor: A,
init: F,
) -> std::io::Result<ActorHandle<A::Message>>
where
F: FnOnce(&mut A, ActorHandle<A::Message>) + Send + 'static,
{
@@ -97,7 +122,7 @@ impl ActorSystem {
let handle = ActorHandle { sender };
let handle_for_init = handle.clone();
let spawned = thread::Builder::new()
thread::Builder::new()
.name("soulseek-actor".to_string())
.stack_size(ACTOR_STACK_SIZE)
.spawn(move || {
@@ -105,13 +130,9 @@ impl ActorSystem {
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
Ok(handle)
}
fn run_actor_loop<A: Actor>(
soulseek-rs-lib/src/actor/peer_actor.rs
+84 -1
View File
@@ -88,12 +88,23 @@ pub struct PeerActor {
/// larger than the send buffer — a full share listing is easily megabytes —
/// only partly fits; the rest drains on later ticks.
pending_write: Vec<u8>,
/// When bytes last moved on this connection, in either direction. A peer
/// that goes quiet is eventually reaped: nothing else ever closes an idle
/// control connection, and each one holds a descriptor and a thread for
/// the life of the process.
last_activity: Instant,
}
/// Cap on queued outbound bytes. A peer that stops reading must not be able to
/// grow this without bound; dropping that connection is the right answer.
const MAX_PENDING_WRITE: usize = 32 * 1024 * 1024;
/// How long a connected peer may stay silent before its control connection is
/// closed. Transfers run on their own sockets, and a peer that wants us again
/// simply reconnects — every serious client reaps idle peers this way. Five
/// minutes comfortably outlives a search's chatter and queue updates.
const IDLE_DISCONNECT: Duration = Duration::from_mins(5);
impl PeerActor {
#[must_use]
pub fn new(
@@ -128,6 +139,7 @@ impl PeerActor {
id,
serving_tokens: std::collections::HashSet::new(),
pending_write: Vec::new(),
last_activity: Instant::now(),
}
}
@@ -460,7 +472,9 @@ impl PeerActor {
};
match self.reader.read_from_socket(stream) {
Ok(()) => {}
Ok(()) => {
self.last_activity = Instant::now();
}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
if let Ok(peer_lock) = self.peer.read_safe() {
@@ -609,6 +623,9 @@ impl PeerActor {
}
};
self.pending_write.drain(..written);
if written > 0 {
self.last_activity = Instant::now();
}
if let Err(e) = outcome {
error!(
@@ -876,9 +893,75 @@ impl Actor for PeerActor {
self.flush_pending_write();
self.process_read();
self.flush_pending_write();
if self.stream.is_some()
&& self.last_activity.elapsed() > IDLE_DISCONNECT
{
debug!(
"[peer:{}] idle for {}s, closing",
self.peer_username(),
IDLE_DISCONNECT.as_secs()
);
self.disconnect();
}
}
}
ConnectionState::Disconnected => {}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::peer::{ConnectionType, Peer};
use std::net::{TcpListener, TcpStream};
use std::sync::mpsc::Receiver;
/// A connected inbound actor over a real loopback socket, plus the far
/// end (kept alive so reads see silence, not EOF).
fn connected_actor() -> (PeerActor, Receiver<ClientOperation>, TcpStream) {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let stream = TcpStream::connect(addr).unwrap();
stream.set_nonblocking(true).unwrap();
let (far_end, _) = listener.accept().unwrap();
let (tx, rx) = std::sync::mpsc::channel();
let peer = Peer::new(
"bob".to_string(),
ConnectionType::P,
"127.0.0.1".to_string(),
u32::from(addr.port()),
None,
0,
0,
0,
);
let mut actor =
PeerActor::new(peer, Some(stream), None, tx, "me".to_string(), 7);
actor.on_start();
(actor, rx, far_end)
}
#[test]
fn tick_reaps_a_peer_idle_past_the_deadline() {
let (mut actor, rx, _far_end) = connected_actor();
actor.tick();
assert!(actor.stream.is_some(), "a fresh connection stays open");
actor.last_activity = Instant::now()
.checked_sub(IDLE_DISCONNECT + Duration::from_secs(1))
.unwrap();
actor.tick();
assert!(actor.stream.is_none(), "an idle stream must be closed");
match rx.try_recv() {
Ok(ClientOperation::PeerDisconnected(7, username, None)) => {
assert_eq!(username, "bob");
}
other => panic!("expected a clean PeerDisconnected, got {other:?}"),
}
}
}
soulseek-rs-lib/src/actor/peer_registry.rs
+60 -5
View File
@@ -60,15 +60,23 @@ impl PeerRegistry {
id,
);
let handle =
self.actor_system.spawn_with_handle(actor, |actor, handle| {
actor.set_self_handle(handle);
});
// Take the map lock before the actor exists: a peer that dies
// instantly (refused dial, immediate hangup) reports its terminal
// outcome to the client loop, whose eviction takes this same lock.
// With the insert racing the spawn, eviction could run first, find
// nothing, and the entry inserted afterwards became a permanent
// zombie claiming the username.
let mut peers = self
.peers
.lock_safe()
.map_err(|e| format!("peer registry lock poisoned: {e}"))?;
let handle = self
.actor_system
.try_spawn_with_handle(actor, |actor, handle| {
actor.set_self_handle(handle);
})
.map_err(|e| format!("failed to spawn peer actor thread: {e}"))?;
// Stop any actor already registered under this username so it does not
// become an orphan pinning a pool worker forever. Eviction on the
// replaced actor's later shutdown is identity-aware (keyed on its id),
@@ -233,4 +241,51 @@ mod tests {
let _ = handle.unwrap().stop();
assert!(!registry.contains("bob"));
}
// A dial that is refused reports its terminal outcome almost instantly —
// possibly while register_peer is still between spawn and insert. The
// registry holds its lock across both, so the eviction that follows must
// always find the entry; a miss left a permanent zombie claiming the
// username.
#[test]
fn refused_dial_does_not_leave_a_zombie_entry() {
use crate::client::ClientOperation;
let system = Arc::new(ActorSystem::new());
let (tx, rx) = std::sync::mpsc::channel();
let registry = PeerRegistry::new(system, tx, "me".to_string());
// A port with nothing listening behind it: bind, learn it, drop it.
let port = {
let probe = TcpListener::bind("127.0.0.1:0").unwrap();
probe.local_addr().unwrap().port()
};
let peer = Peer::new(
"ghost".to_string(),
ConnectionType::P,
"127.0.0.1".to_string(),
u32::from(port),
None,
0,
0,
0,
);
registry.register_peer(peer, None, None).unwrap();
// Play the client ops loop: take the terminal outcome, evict by id.
match rx.recv_timeout(std::time::Duration::from_secs(10)) {
Ok(ClientOperation::PeerConnectFailed(id, username)) => {
assert_eq!(username, "ghost");
if let Some(handle) = registry.remove_peer_if(&username, id) {
let _ = handle.stop();
}
}
other => panic!("expected PeerConnectFailed, got {other:?}"),
}
assert!(
!registry.contains("ghost"),
"a refused dial must not leave a registry entry"
);
}
}
soulseek-rs-lib/src/client/connection.rs
+33 -1
View File
@@ -2,7 +2,7 @@ use super::{
Arc, Client, ClientContext, ClientOperation, ConnectionType, DownloadPeer,
DownloadStatus, Listen, Peer, PeerRegistry, Receiver, Result, RwLock,
RwLockExt, Sender, ServerActor, ServerMessage, Shares, SoulseekRs,
TcpStream, error, info, mpsc, thread, trace,
TcpStream, error, info, mpsc, thread, trace, warn,
};
/// Ceiling on the wait for a login verdict. Generous enough for a slow server
@@ -10,8 +10,40 @@ use super::{
const LOGIN_RESPONSE_TIMEOUT: std::time::Duration =
std::time::Duration::from_secs(45);
/// Soft open-file limit below which a session should expect "too many open
/// files": every peer holds a socket, and one busy search talks to hundreds
/// of them.
const COMFORTABLE_OPEN_FILES: u64 = 1024;
/// Say up front when the process cannot hold a busy session's sockets, so an
/// operator reads a warning with a number in it instead of an EMFILE storm an
/// hour later. The embedding process owns its rlimits; the library only looks.
#[cfg(unix)]
fn warn_if_open_file_limit_is_low() {
// SAFETY: getrlimit only fills the struct it is handed.
let limit = unsafe {
let mut limit = std::mem::zeroed::<libc::rlimit>();
if libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) != 0 {
return;
}
limit.rlim_cur
};
if limit < COMFORTABLE_OPEN_FILES {
warn!(
"only {limit} open files allowed; a busy session wants \
{COMFORTABLE_OPEN_FILES} or more — expect \"too many open \
files\" under load"
);
}
}
#[cfg(not(unix))]
fn warn_if_open_file_limit_is_low() {}
impl Client {
pub fn connect(&mut self) -> Result<()> {
warn_if_open_file_limit_is_low();
let (sender, message_reader): (
Sender<ClientOperation>,
Receiver<ClientOperation>,
soulseek-rs-lib/src/client/mod.rs
+1 -1
View File
@@ -248,7 +248,7 @@ pub enum ClientOperation {
query: String,
},
/// A peer queued one of our shared files; `requester_key` is the registry
/// key of the peer actor (may carry a `:direct` suffix).
/// key of the peer actor — the peer's username.
QueueUpload {
requester_key: String,
filename: String,
soulseek-rs-lib/src/client/operations.rs
+18 -15
View File
@@ -1,4 +1,3 @@
use super::uploads::downloader_of;
use super::{
Arc, BROKER_CONNECT_TIMEOUT, Client, ClientContext, ClientOperation,
ConnectionType, Download, DownloadPeer, DownloadStatus, Peer, PeerMessage,
@@ -75,6 +74,14 @@ impl Client {
let _ = handle.stop();
}
}
// Only an error is evidence the peer is gone. A clean
// close — our idle reaper, or a remote client tidying
// an idle socket while it waits in our queue — must
// not throw away everything that peer has queued: the
// connection comes back, the queue cannot. A departed
// peer wedging uploads shut is still covered: its
// erroring transfer fails within the socket deadlines
// and the error branch here frees its slots.
if let Some(error) = error {
warn!(
"[client] Peer {} disconnected with error: {:?}",
@@ -85,12 +92,11 @@ impl Client {
&username,
None,
);
Self::release_upload_slots(
&client_context,
&username,
);
}
// A peer that leaves must not keep occupying an
// upload slot: with a slot cap, a couple of
// departed peers would otherwise wedge uploads
// shut for the rest of the session.
Self::release_upload_slots(&client_context, &username);
}
ClientOperation::DownloadFromPeer(token, peer, allowed) => {
let maybe_download = match client_context.read_safe() {
@@ -440,10 +446,11 @@ impl Client {
}
}
ClientOperation::PeerConnected(username) => {
// An outbound control connection just handshook.
// Flush any downloads that were queued for this
// peer while we were still connecting. Collect
// under a read guard, then act without it held.
// A control connection just came up — one we dialled,
// or an inbound one the listener registered. Flush any
// downloads that were queued for this peer while it
// was unreachable. Collect under a read guard, then
// act without it held.
let (registry, files): (
Option<PeerRegistry>,
Vec<String>,
@@ -547,8 +554,6 @@ impl Client {
filename,
} => {
// The peer served next may not be this one.
let downloader =
downloader_of(&requester_key).to_string();
match client_context.write_safe() {
Ok(mut ctx) => {
let Some(file) = ctx.shares.get(&filename)
@@ -563,7 +568,6 @@ impl Client {
let real_path = file.real_path.clone();
ctx.enqueue_upload(
&requester_key,
&downloader,
&filename,
real_path,
size,
@@ -580,12 +584,11 @@ impl Client {
requester_key,
filename,
} => {
let downloader = downloader_of(&requester_key);
let (registry, place) = match client_context.read_safe()
{
Ok(ctx) => (
ctx.peer_registry.clone(),
ctx.place_in_queue(downloader, &filename),
ctx.place_in_queue(&requester_key, &filename),
),
Err(_) => continue,
};
soulseek-rs-lib/src/client/tests.rs
+68
View File
@@ -411,3 +411,71 @@ fn upload_speed_is_reported_only_while_running() {
"unexpected rate {rate}"
);
}
// A clean disconnect — the idle reaper, or a remote client tidying an idle
// socket while it waits in our queue — must keep the peer's queued uploads;
// only an error disconnect is evidence the peer is gone.
#[test]
fn a_clean_disconnect_keeps_queued_uploads_an_error_drops_them() {
let client = Client::new("test-user", "test-password");
let (ops, ops_rx) = mpsc::channel();
Client::listen_to_client_operations(
ops_rx,
client.context.clone(),
"me".to_string(),
);
client.context.write().unwrap().enqueue_upload(
"amy",
"@@share\\f.mp3",
std::path::PathBuf::from("/tmp/f.mp3"),
4096,
);
ops.send(ClientOperation::PeerDisconnected(
1,
"amy".to_string(),
None,
))
.unwrap();
// The loop is serial, so once this fence op is visible the disconnect
// before it has been handled.
ops.send(ClientOperation::OwnPrivileges(7)).unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
while client.own_privilege_seconds() != Some(7) {
assert!(Instant::now() < deadline, "ops loop never caught up");
thread::sleep(Duration::from_millis(10));
}
assert_eq!(
client
.context
.read()
.unwrap()
.place_in_queue("amy", "@@share\\f.mp3"),
Some(1),
"a clean disconnect must keep the queue"
);
ops.send(ClientOperation::PeerDisconnected(
1,
"amy".to_string(),
Some(crate::error::SoulseekRs::NotConnected),
))
.unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
loop {
let place = client
.context
.read()
.unwrap()
.place_in_queue("amy", "@@share\\f.mp3");
if place.is_none() {
break;
}
assert!(
Instant::now() < deadline,
"an error disconnect must drop the queue"
);
thread::sleep(Duration::from_millis(10));
}
}
soulseek-rs-lib/src/client/upload_queue.rs
+3 -8
View File
@@ -31,9 +31,7 @@ pub struct UploadOffer {
/// A file a peer has asked us for and is waiting on.
#[derive(Debug, Clone)]
pub struct QueuedUpload {
/// Registry key of the asking peer's actor, which may carry a `:direct`
/// suffix — not the same string as `downloader`.
pub requester_key: String,
/// The asking peer's username — also the registry key of its actor.
pub downloader: String,
pub virtual_path: String,
pub real_path: PathBuf,
@@ -98,7 +96,6 @@ impl ClientContext {
/// for a file we do not have never reaches the queue.
pub fn enqueue_upload(
&mut self,
requester_key: &str,
downloader: &str,
filename: &str,
real_path: std::path::PathBuf,
@@ -106,7 +103,6 @@ impl ClientContext {
) {
self.upload_seq += 1;
self.upload_queue.push(QueuedUpload {
requester_key: requester_key.to_string(),
downloader: downloader.to_string(),
virtual_path: filename.to_string(),
real_path,
@@ -134,7 +130,7 @@ impl ClientContext {
let job = self.upload_queue.remove(index);
let token = next_token();
offers.push(UploadOffer {
requester_key: job.requester_key,
requester_key: job.downloader.clone(),
token,
virtual_path: job.virtual_path.clone(),
size: job.size,
@@ -259,7 +255,6 @@ mod tests {
fn queued(user: &str, privileged: bool, seq: u64) -> QueuedUpload {
QueuedUpload {
requester_key: user.to_string(),
downloader: user.to_string(),
virtual_path: format!("@@share\\{user}.mp3"),
real_path: PathBuf::from("/tmp/x.mp3"),
@@ -365,7 +360,7 @@ mod context_tests {
}
fn ask(ctx: &mut ClientContext, user: &str, file: &str) {
ctx.enqueue_upload(user, user, file, PathBuf::from("/tmp/x"), 4096);
ctx.enqueue_upload(user, file, PathBuf::from("/tmp/x"), 4096);
}
/// Who the pump offered to, by the order it offered them.
soulseek-rs-lib/src/client/uploads.rs
-8
View File
@@ -5,14 +5,6 @@ use super::{
use crate::types::UploadStatus;
use std::sync::atomic::{AtomicBool, AtomicU64};
/// The peer behind a registry key, which may carry a `:direct` suffix the
/// server never sees.
pub fn downloader_of(requester_key: &str) -> &str {
requester_key
.strip_suffix(":direct")
.unwrap_or(requester_key)
}
impl Client {
/// Whether the server listed `username` as privileged. Privileged peers are
/// served before everyone else.
soulseek-rs-lib/src/peer/download_peer.rs
+35 -3
View File
@@ -231,6 +231,19 @@ impl DownloadPeer {
)
.map_err(DownloadError::ConnectionFailed)?;
Self::apply_transfer_deadlines(&stream)?;
Ok(stream)
}
/// Bound every stall on a transfer socket. This must cover streams the
/// listener accepted as well as ones we dialled: an accepted F connection
/// used to arrive here with no read timeout at all, so a peer that sent a
/// token and went silent parked the transfer thread — holding the socket
/// and the open `.part` file — for the rest of the session.
fn apply_transfer_deadlines(
stream: &TcpStream,
) -> Result<(), DownloadError> {
stream
.set_read_timeout(Some(Duration::from_secs(30)))
.map_err(DownloadError::ConnectionFailed)?;
@@ -240,8 +253,7 @@ impl DownloadPeer {
stream
.set_nodelay(true)
.map_err(DownloadError::ConnectionFailed)?;
Ok(stream)
Ok(())
}
fn perform_handshake(
@@ -481,7 +493,10 @@ impl DownloadPeer {
}
let mut stream = match stream {
Some(s) => s,
Some(s) => {
Self::apply_transfer_deadlines(&s)?;
s
}
None => self.establish_connection()?,
};
@@ -622,6 +637,23 @@ mod tests {
let _ = std::fs::remove_dir_all(dir);
}
// Streams the listener accepted arrive with no deadlines; without these a
// peer that sends a token and goes silent parks the transfer thread —
// socket and open .part file included — for the rest of the session.
#[test]
fn supplied_streams_get_transfer_deadlines() {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let stream =
std::net::TcpStream::connect(listener.local_addr().unwrap())
.unwrap();
DownloadPeer::apply_transfer_deadlines(&stream).unwrap();
let secs = std::time::Duration::from_secs;
assert_eq!(stream.read_timeout().unwrap(), Some(secs(30)));
assert_eq!(stream.write_timeout().unwrap(), Some(secs(5)));
}
#[test]
fn test_establish_connection_invalid_address() {
let download_peer = DownloadPeer::new(
soulseek-rs-lib/src/peer/listen.rs
+13 -2
View File
@@ -178,7 +178,7 @@ fn handle_file_connection(
let failure_token = download.as_ref().map(|d| d.token);
let download_peer = DownloadPeer::new(
format!("{}:direct", peer.username),
peer.username.clone(),
peer.host.clone(),
peer.port,
token,
@@ -325,8 +325,13 @@ fn handle_incoming_connection(
init_data.username, init_data.connection_type, init_data.token
);
// Register under the plain username, the same key outbound connections
// use. A suffixed key ("user:direct") put the same peer in the registry
// twice — two sockets, two threads — and made every "already connected?"
// check miss, so an inbound searcher still got a second, outbound
// connection dialled at them.
let peer = Peer::new(
format!("{}:direct", init_data.username),
init_data.username.clone(),
init_data.connection_type.clone(),
peer_ip.clone(),
peer_port.into(),
@@ -339,6 +344,12 @@ fn handle_incoming_connection(
match init_data.connection_type {
ConnectionType::P => {
handle_peer_connection(peer, stream, reader, &context);
// Inbound peers don't self-announce, so nudge the client to flush
// anything queued for this now-connected peer — same as the
// PierceFirewall path.
let _ = context
.client_sender
.send(ClientOperation::PeerConnected(init_data.username));
}
ConnectionType::F => handle_file_connection(
soulseek-rs-lib/src/peer/upload_peer.rs
+11
View File
@@ -42,6 +42,17 @@ pub fn serve_file(
})?;
let mut stream =
TcpStream::connect_timeout(&socket, Duration::from_secs(20))?;
// A downloader that connects and then goes silent must cost seconds, not
// the rest of the session: without deadlines a wedged `read_exact` (or a
// zero-window `write_all`) holds this thread, the socket, the open file,
// and — because the upload stays InProgress — an upload slot, forever.
// Thirty seconds of total silence is a dead transfer, not a slow one: even
// a slow reader drains some buffer within that. The write deadline is
// deliberately looser than the download side's 5s: bulk data flows this
// way, and a congested but living peer may hold the send buffer full far
// longer than a peer draining our tiny control frames ever would.
stream.set_read_timeout(Some(Duration::from_secs(30)))?;
stream.set_write_timeout(Some(Duration::from_secs(30)))?;
stream.set_nodelay(true).ok();
// PeerInit(F) + the 4-byte token in a single write so they coalesce.
soulseek-rs-lib/tests/e2e.rs
+3 -3
View File
@@ -734,14 +734,14 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() {
.recv_timeout(Duration::from_secs(5))
.expect("mock uploader P connection");
// The listener registers an incoming peer under "<username>:direct"; give
// The listener registers an incoming peer under its plain username; give
// that registration a moment to complete before queuing the download.
std::thread::sleep(Duration::from_millis(1500));
let (_download, status_rx) = client
.download(
filename.to_string(),
"e2e_mockpeer:direct".to_string(),
"e2e_mockpeer".to_string(),
size,
download_dir.display().to_string(),
)
@@ -811,7 +811,7 @@ fn an_interrupted_download_resumes_from_its_partial_file() {
let (_download, status_rx) = client
.download(
filename.to_string(),
"e2e_resume_peer:direct".to_string(),
"e2e_resume_peer".to_string(),
size,
download_dir.display().to_string(),
)
soulseek-rs/src/daemon/mod.rs
-77
View File
@@ -44,15 +44,6 @@ pub const TOKEN_NAME: &str = "daemon.token";
/// loses little, rare enough not to churn the disk on a busy queue.
const SAVE_INTERVAL: Duration = Duration::from_secs(5);
/// Open files the daemon asks the OS for at startup.
///
/// The client holds a socket per peer and will talk to a thousand of them
/// during a busy search, plus transfers, the share index, and control
/// connections. A shell hands its children a soft limit of 256 on macOS, which
/// a single search blows through — and a one-shot command never noticed
/// because it exited before it mattered. A resident process has to ask.
const WANTED_OPEN_FILES: u64 = 8192;
#[must_use]
pub fn socket_path() -> Option<PathBuf> {
crate::persist::paths::state_dir().map(|dir| dir.join(SOCKET_NAME))
@@ -161,21 +152,6 @@ pub fn run(
let token = load_or_create_token(&state_dir.join(TOKEN_NAME))
.map_err(|e| CliError::usage(format!("cannot read the token: {e}")))?;
// Before anything opens a socket: running out of file descriptors mid
// search leaves the client unable to accept peers *or* control
// connections, and the only symptom is everything quietly failing.
match raise_open_file_limit() {
Ok(limit) if limit < WANTED_OPEN_FILES => ctx.out.warn(&format!(
"only {limit} open files allowed; a busy search wants \
{WANTED_OPEN_FILES}. Raise it with `ulimit -n {WANTED_OPEN_FILES}` \
before starting the daemon."
)),
Ok(_) => {}
Err(e) => ctx
.out
.warn(&format!("could not check the open-file limit: {e}")),
}
// Bind before logging in: a second daemon should fail in a millisecond,
// not after a full login handshake against the server.
let mut listeners = Vec::new();
@@ -272,59 +248,6 @@ pub fn run(
Ok(())
}
/// Ask the OS for enough open files to run a busy session, and report what we
/// actually got.
///
/// Raising our own soft limit is what a service does: the limit a shell
/// inherits is sized for a shell. The hard limit is the ceiling, and asking
/// past it fails outright rather than clamping, so this asks for the smaller
/// of what we want and what we are allowed.
#[cfg(unix)]
fn raise_open_file_limit() -> std::io::Result<u64> {
// SAFETY: every call takes a pointer to a `rlimit` this function owns, and
// none of them retains it.
unsafe {
let mut limit = std::mem::zeroed::<libc::rlimit>();
if libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) != 0 {
return Err(std::io::Error::last_os_error());
}
if limit.rlim_cur >= WANTED_OPEN_FILES {
return Ok(limit.rlim_cur);
}
let previous = limit.rlim_cur;
let ceiling = if limit.rlim_max == libc::RLIM_INFINITY {
WANTED_OPEN_FILES
} else {
limit.rlim_max.min(WANTED_OPEN_FILES)
};
// macOS rejects an unlimited *hard* limit for open files, so raising
// the soft one while leaving `rlim_max` as it came back fails with
// EINVAL. Naming both is what it will accept.
for candidate in [
libc::rlimit {
rlim_cur: ceiling,
rlim_max: limit.rlim_max,
},
libc::rlimit {
rlim_cur: ceiling,
rlim_max: ceiling,
},
] {
if libc::setrlimit(libc::RLIMIT_NOFILE, &raw const candidate) == 0 {
return Ok(ceiling);
}
}
Ok(previous)
}
}
#[cfg(not(unix))]
const fn raise_open_file_limit() -> std::io::Result<u64> {
Ok(WANTED_OPEN_FILES)
}
/// Say what is now possible, once, on stderr.
///
/// The token goes here rather than only into a file because the moment you
soulseek-rs/src/run.rs
+78
View File
@@ -16,6 +16,23 @@ use std::time::Duration;
use std::{env, io};
pub fn run(mut cli: Cli, out: &Out) -> CliResult {
// Before anything opens a socket: every entry point that logs in — the
// TUI, `serve`, one-shot commands, the daemon — holds a socket per peer,
// and the soft limit a shell hands its children (256 on macOS) is gone
// after one busy search. The raise lived in the daemon first; the other
// entry points run the same peer machinery and crashed the same way.
match raise_open_file_limit() {
Ok(limit) if limit < WANTED_OPEN_FILES => out.warn(&format!(
"only {limit} open files allowed; a busy search wants \
{WANTED_OPEN_FILES}. Raise it with `ulimit -n \
{WANTED_OPEN_FILES}` first."
)),
Ok(_) => {}
Err(e) => {
out.warn(&format!("could not check the open-file limit: {e}"));
}
}
let config_path = config_path(&cli);
let file_config = match &config_path {
Some(path) => crate::persist::config::FileConfig::load(path)
@@ -557,3 +574,64 @@ fn persist_credentials(
eprintln!("⚠️ Could not store password in keychain: {e}");
}
}
/// Open files a session asks the OS for at startup.
///
/// The client holds a socket per peer and will talk to a thousand of them
/// during a busy search, plus transfers, the share index, and control
/// connections. A shell hands its children a soft limit of 256 on macOS,
/// which a single search blows through.
const WANTED_OPEN_FILES: u64 = 8192;
/// Ask the OS for enough open files to run a busy session, and report what we
/// actually got.
///
/// Raising our own soft limit is what a service does: the limit a shell
/// inherits is sized for a shell. The hard limit is the ceiling, and asking
/// past it fails outright rather than clamping, so this asks for the smaller
/// of what we want and what we are allowed.
#[cfg(unix)]
fn raise_open_file_limit() -> std::io::Result<u64> {
// SAFETY: every call takes a pointer to a `rlimit` this function owns, and
// none of them retains it.
unsafe {
let mut limit = std::mem::zeroed::<libc::rlimit>();
if libc::getrlimit(libc::RLIMIT_NOFILE, &raw mut limit) != 0 {
return Err(std::io::Error::last_os_error());
}
if limit.rlim_cur >= WANTED_OPEN_FILES {
return Ok(limit.rlim_cur);
}
let previous = limit.rlim_cur;
let ceiling = if limit.rlim_max == libc::RLIM_INFINITY {
WANTED_OPEN_FILES
} else {
limit.rlim_max.min(WANTED_OPEN_FILES)
};
// macOS rejects an unlimited *hard* limit for open files, so raising
// the soft one while leaving `rlim_max` as it came back fails with
// EINVAL. Naming both is what it will accept.
for candidate in [
libc::rlimit {
rlim_cur: ceiling,
rlim_max: limit.rlim_max,
},
libc::rlimit {
rlim_cur: ceiling,
rlim_max: ceiling,
},
] {
if libc::setrlimit(libc::RLIMIT_NOFILE, &raw const candidate) == 0 {
return Ok(ceiling);
}
}
Ok(previous)
}
}
#[cfg(not(unix))]
const fn raise_open_file_limit() -> std::io::Result<u64> {
Ok(WANTED_OPEN_FILES)
}