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

Merge branch 'refactor/lock-poison-safety'

Browse Source
This commit is contained in:
michel
2026-04-26 12:51:49 +02:00
parent e1ee3ac493 2722552e35
commit 4775ede762
10 changed files with 473 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
soulseek-rs-lib/src/actor/peer_actor.rs
+100 -39
View File
@@ -9,6 +9,7 @@ use crate::message::server::MessageFactory;
use crate::message::{Handlers, Message, MessageReader, MessageType}; use crate::message::{Handlers, Message, MessageReader, MessageType};
use crate::peer::Peer; use crate::peer::Peer;
use crate::types::{Download, SearchResult, Transfer}; use crate::types::{Download, SearchResult, Transfer};
use crate::utils::lock::RwLockExt;
use crate::{debug, error, trace, warn}; use crate::{debug, error, trace, warn};
use std::io::{self, Error, Write}; use std::io::{self, Error, Write};
@@ -80,6 +81,26 @@ impl PeerActor {
self.self_handle = Some(handle); self.self_handle = Some(handle);
} }
fn peer_username(&self) -> String {
match self.peer.read_safe() {
Ok(p) => p.username.clone(),
Err(e) => {
error!("[peer_actor] peer lock poisoned: {}", e);
"<unknown>".to_string()
}
}
}
fn peer_snapshot(&self) -> Option<Peer> {
match self.peer.read_safe() {
Ok(p) => Some(p.clone()),
Err(e) => {
error!("[peer_actor] peer lock poisoned: {}", e);
None
}
}
}
fn initialize_dispatcher(&mut self) { fn initialize_dispatcher(&mut self) {
let (dispatcher_sender, dispatcher_receiver) = let (dispatcher_sender, dispatcher_receiver) =
std::sync::mpsc::channel::<PeerMessage>(); std::sync::mpsc::channel::<PeerMessage>();
@@ -135,23 +156,35 @@ impl PeerActor {
self.send_message(message); self.send_message(message);
} }
PeerMessage::FileSearchResult(file_search) => { PeerMessage::FileSearchResult(file_search) => {
self.client_channel if let Err(e) = self
.client_channel
.send(ClientOperation::SearchResult(file_search)) .send(ClientOperation::SearchResult(file_search))
.unwrap(); {
error!(
"[peer:{}] failed to forward search result: {}",
self.peer_username(),
e
);
}
} }
PeerMessage::TransferRequest(transfer) => { PeerMessage::TransferRequest(transfer) => {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
debug!( debug!(
"[peer:{}] TransferRequest for {}", "[peer:{}] TransferRequest for {}",
username, transfer.token username, transfer.token
); );
self.client_channel if let Err(e) = self.client_channel.send(
.send(ClientOperation::UpdateDownloadTokens( ClientOperation::UpdateDownloadTokens(
transfer.clone(), transfer.clone(),
username.clone(), username.clone(),
)) ),
.unwrap(); ) {
error!(
"[peer:{}] failed to send UpdateDownloadTokens: {}",
username, e
);
}
let transfer_response = let transfer_response =
MessageFactory::build_transfer_response_message(transfer); MessageFactory::build_transfer_response_message(transfer);
@@ -171,7 +204,7 @@ impl PeerActor {
allowed, allowed,
reason, reason,
} => { } => {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
debug!( debug!(
"[peer:{}] transfer response token: {} allowed: {}", "[peer:{}] transfer response token: {} allowed: {}",
username, token, allowed username, token, allowed
@@ -189,17 +222,25 @@ impl PeerActor {
"[peer:{}] Transfer allowed, ready to connect with token {:}", "[peer:{}] Transfer allowed, ready to connect with token {:}",
username, token username, token
); );
self.client_channel let Some(peer_snapshot) = self.peer_snapshot() else {
.send(ClientOperation::DownloadFromPeer( return;
};
if let Err(e) = self.client_channel.send(
ClientOperation::DownloadFromPeer(
token, token,
self.peer.read().unwrap().clone(), peer_snapshot,
allowed, allowed,
)) ),
.unwrap(); ) {
error!(
"[peer:{}] failed to send DownloadFromPeer: {}",
username, e
);
}
} }
} }
PeerMessage::PlaceInQueueResponse { filename, place } => { PeerMessage::PlaceInQueueResponse { filename, place } => {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
debug!( debug!(
"[peer:{}] Place in queue response - file: {}, place: {}", "[peer:{}] Place in queue response - file: {}, place: {}",
username, filename, place username, filename, place
@@ -208,10 +249,15 @@ impl PeerActor {
PeerMessage::SetUsername(username) => { PeerMessage::SetUsername(username) => {
trace!( trace!(
"[peer:{}] SetUsername: {}", "[peer:{}] SetUsername: {}",
self.peer.read().unwrap().username, self.peer_username(),
username username
); );
self.peer.write().unwrap().username = username; match self.peer.write_safe() {
Ok(mut p) => p.username = username,
Err(e) => {
error!("[peer_actor] SetUsername write: {}", e)
}
}
} }
PeerMessage::QueueUpload(filename) => { PeerMessage::QueueUpload(filename) => {
let message = let message =
@@ -229,9 +275,15 @@ impl PeerActor {
self.process_read(); self.process_read();
} }
PeerMessage::UploadFailed(username, filename) => { PeerMessage::UploadFailed(username, filename) => {
self.client_channel if let Err(e) = self
.client_channel
.send(ClientOperation::UploadFailed(username, filename)) .send(ClientOperation::UploadFailed(username, filename))
.unwrap(); {
error!(
"[peer_actor] failed to forward UploadFailed: {}",
e
);
}
} }
} }
} }
@@ -251,14 +303,15 @@ impl PeerActor {
Ok(()) => {} Ok(()) => {}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {} Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {}
Err(ref e) if e.kind() == io::ErrorKind::TimedOut => { Err(ref e) if e.kind() == io::ErrorKind::TimedOut => {
let peer_lock = self.peer.read().unwrap(); if let Ok(peer_lock) = self.peer.read_safe() {
debug!( debug!(
"Read operation timed out for peer actor {:}:{:}", "Read operation timed out for peer actor {:}:{:}",
peer_lock.host, peer_lock.port peer_lock.host, peer_lock.port
); );
}
} }
Err(e) => { Err(e) => {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
error!( error!(
"[peer:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.", "[peer:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.",
username, username,
@@ -274,7 +327,7 @@ impl PeerActor {
} }
fn extract_and_process_messages(&mut self) { fn extract_and_process_messages(&mut self) {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
let mut extracted_count = 0; let mut extracted_count = 0;
loop { loop {
match self.reader.extract_message() { match self.reader.extract_message() {
@@ -315,6 +368,7 @@ impl PeerActor {
} }
fn send_message(&mut self, message: Message) { fn send_message(&mut self, message: Message) {
let username = self.peer_username();
let stream = match self.stream.as_mut() { let stream = match self.stream.as_mut() {
Some(s) => s, Some(s) => s,
None => { None => {
@@ -323,7 +377,6 @@ impl PeerActor {
} }
}; };
let username = self.peer.read().unwrap().username.clone();
trace!( trace!(
"[peer:{}] ➡ {:?}", "[peer:{}] ➡ {:?}",
username, username,
@@ -356,7 +409,7 @@ impl PeerActor {
} }
fn disconnect_with_error(&mut self, error: Error) { fn disconnect_with_error(&mut self, error: Error) {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
debug!("[peer:{}] disconnect", username); debug!("[peer:{}] disconnect", username);
self.stream.take(); self.stream.take();
@@ -368,7 +421,7 @@ impl PeerActor {
} }
} }
fn disconnect(&mut self) { fn disconnect(&mut self) {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
debug!("[peer:{}] disconnect", username); debug!("[peer:{}] disconnect", username);
self.stream.take(); self.stream.take();
@@ -382,11 +435,13 @@ impl PeerActor {
} }
fn initiate_connection(&mut self) -> bool { fn initiate_connection(&mut self) -> bool {
let peer = self.peer.read().unwrap(); let (username, host, port) = match self.peer.read_safe() {
let username = peer.username.clone(); Ok(peer) => (peer.username.clone(), peer.host.clone(), peer.port),
let host = peer.host.clone(); Err(e) => {
let port = peer.port; error!("[peer_actor] initiate_connection peer lock: {}", e);
drop(peer); return false;
}
};
let socket_addr = let socket_addr =
format!("{}:{}", host, port).parse::<std::net::SocketAddr>(); format!("{}:{}", host, port).parse::<std::net::SocketAddr>();
@@ -438,7 +493,7 @@ impl PeerActor {
return; return;
}; };
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
if since.elapsed() > Duration::from_secs(20) { if since.elapsed() > Duration::from_secs(20) {
error!("[peer:{}] Connection timeout after 20 seconds", username); error!("[peer:{}] Connection timeout after 20 seconds", username);
@@ -467,10 +522,16 @@ impl PeerActor {
} }
fn on_connection_established(&mut self) { fn on_connection_established(&mut self) {
let peer = self.peer.read().unwrap(); let (username, token) = match self.peer.read_safe() {
let username = peer.username.clone(); Ok(peer) => (peer.username.clone(), peer.token),
let token = peer.token; Err(e) => {
drop(peer); error!(
"[peer_actor] on_connection_established peer lock: {}",
e
);
return;
}
};
let Some(ref mut stream) = self.stream else { let Some(ref mut stream) = self.stream else {
return; return;
@@ -520,7 +581,7 @@ impl Actor for PeerActor {
} }
fn on_stop(&mut self) { fn on_stop(&mut self) {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer_username();
trace!("[peer:{}] actor stopping", username); trace!("[peer:{}] actor stopping", username);
self.disconnect(); self.disconnect();
} }
soulseek-rs-lib/src/actor/peer_registry.rs
+41 -11
View File
@@ -1,9 +1,10 @@
use crate::actor::peer_actor::{PeerActor, PeerMessage}; use crate::actor::peer_actor::{PeerActor, PeerMessage};
use crate::actor::{ActorHandle, ActorSystem}; use crate::actor::{ActorHandle, ActorSystem};
use crate::client::ClientOperation; use crate::client::ClientOperation;
use crate::debug;
use crate::message::MessageReader; use crate::message::MessageReader;
use crate::peer::Peer; use crate::peer::Peer;
use crate::utils::lock::MutexExt;
use crate::{debug, error};
use std::collections::HashMap; use std::collections::HashMap;
use std::net::TcpStream; use std::net::TcpStream;
@@ -44,22 +45,36 @@ impl PeerRegistry {
actor.set_self_handle(handle); actor.set_self_handle(handle);
}); });
let mut peers = self.peers.lock().unwrap(); let mut peers = self
.peers
.lock_safe()
.map_err(|e| format!("peer registry lock poisoned: {}", e))?;
peers.insert(username.clone(), handle.clone()); peers.insert(username.clone(), handle.clone());
Ok(handle) Ok(handle)
} }
pub fn get_peer(&self, username: &str) -> Option<ActorHandle<PeerMessage>> { pub fn get_peer(&self, username: &str) -> Option<ActorHandle<PeerMessage>> {
let peers = self.peers.lock().unwrap(); match self.peers.lock_safe() {
peers.get(username).cloned() Ok(peers) => peers.get(username).cloned(),
Err(e) => {
error!("[peer_registry] get_peer: {}", e);
None
}
}
} }
pub fn remove_peer( pub fn remove_peer(
&self, &self,
username: &str, username: &str,
) -> Option<ActorHandle<PeerMessage>> { ) -> Option<ActorHandle<PeerMessage>> {
let mut peers = self.peers.lock().unwrap(); let mut peers = match self.peers.lock_safe() {
Ok(p) => p,
Err(e) => {
error!("[peer_registry] remove_peer: {}", e);
return None;
}
};
let handle = peers.remove(username); let handle = peers.remove(username);
if handle.is_some() { if handle.is_some() {
@@ -70,18 +85,33 @@ impl PeerRegistry {
} }
pub fn get_all_usernames(&self) -> Vec<String> { pub fn get_all_usernames(&self) -> Vec<String> {
let peers = self.peers.lock().unwrap(); match self.peers.lock_safe() {
peers.keys().cloned().collect() Ok(peers) => peers.keys().cloned().collect(),
Err(e) => {
error!("[peer_registry] get_all_usernames: {}", e);
Vec::new()
}
}
} }
pub fn count(&self) -> usize { pub fn count(&self) -> usize {
let peers = self.peers.lock().unwrap(); match self.peers.lock_safe() {
peers.len() Ok(peers) => peers.len(),
Err(e) => {
error!("[peer_registry] count: {}", e);
0
}
}
} }
pub fn contains(&self, username: &str) -> bool { pub fn contains(&self, username: &str) -> bool {
let peers = self.peers.lock().unwrap(); match self.peers.lock_safe() {
peers.contains_key(username) Ok(peers) => peers.contains_key(username),
Err(e) => {
error!("[peer_registry] contains: {}", e);
false
}
}
} }
pub fn send_to_peer( pub fn send_to_peer(
soulseek-rs-lib/src/actor/server_actor.rs
+27 -11
View File
@@ -17,6 +17,7 @@ use crate::message::{Handlers, MessageType};
use crate::message::{Message, MessageReader}; use crate::message::{Message, MessageReader};
use crate::peer::ConnectionType; use crate::peer::ConnectionType;
use crate::peer::Peer; use crate::peer::Peer;
use crate::utils::lock::RwLockExt;
use std::io::{self, Error, Write}; use std::io::{self, Error, Write};
use std::net::{TcpStream, ToSocketAddrs}; use std::net::{TcpStream, ToSocketAddrs};
@@ -338,9 +339,12 @@ impl ServerActor {
self.dispatcher_receiver = Some(dispatcher_receiver); self.dispatcher_receiver = Some(dispatcher_receiver);
self.dispatcher_sender = Some(dispatcher_sender.clone()); self.dispatcher_sender = Some(dispatcher_sender.clone());
self.client_channel if let Err(e) = self
.client_channel
.send(ClientOperation::SetServerSender(dispatcher_sender.clone())) .send(ClientOperation::SetServerSender(dispatcher_sender.clone()))
.unwrap(); {
error!("[server] failed to send SetServerSender: {}", e);
}
let mut handlers = Handlers::new(); let mut handlers = Handlers::new();
@@ -401,7 +405,7 @@ impl ServerActor {
} }
{ {
logged_in = context.read().unwrap().logged_in logged_in = context.read_safe()?.logged_in
} }
if logged_in.is_some() { if logged_in.is_some() {
@@ -409,7 +413,8 @@ impl ServerActor {
} }
} }
if logged_in.unwrap() { let logged_in = logged_in.ok_or(SoulseekRs::Timeout)?;
if logged_in {
info!("Logged in as {}", username); info!("Logged in as {}", username);
self.queue_message(MessageFactory::build_shared_folders_message( self.queue_message(MessageFactory::build_shared_folders_message(
1, 499, 1, 499,
@@ -425,7 +430,7 @@ impl ServerActor {
} }
} }
Ok(logged_in.unwrap()) Ok(logged_in)
} }
pub fn file_search(&mut self, token: u32, query: &str) { pub fn file_search(&mut self, token: u32, query: &str) {
@@ -455,12 +460,18 @@ impl ServerActor {
Some(ClientOperation::ConnectToPeer(peer.clone())) Some(ClientOperation::ConnectToPeer(peer.clone()))
} }
ConnectionType::D => None, ConnectionType::D => None,
} { } && let Err(e) = self.client_channel.send(op)
self.client_channel.send(op).unwrap(); {
error!("[server] failed to send ConnectToPeer: {}", e);
} }
} }
ServerMessage::LoginStatus(message) => { ServerMessage::LoginStatus(message) => {
self.context.write().unwrap().logged_in = Some(message); match self.context.write_safe() {
Ok(mut ctx) => ctx.logged_in = Some(message),
Err(e) => {
error!("[server] LoginStatus write: {}", e)
}
}
} }
ServerMessage::PierceFirewall(token) => { ServerMessage::PierceFirewall(token) => {
self.send_message( self.send_message(
@@ -525,9 +536,14 @@ impl ServerActor {
break; break;
} }
if let Some(logged_in) = let logged_in = match context.read_safe() {
context.read().unwrap().logged_in Ok(ctx) => ctx.logged_in,
{ Err(e) => {
let _ = response.send(Err(e));
break;
}
};
if let Some(logged_in) = logged_in {
let result = if logged_in { let result = if logged_in {
Ok(true) Ok(true)
} else { } else {
soulseek-rs-lib/src/client.rs
+237 -92
View File
@@ -8,7 +8,7 @@ use crate::{
error::{Result, SoulseekRs}, error::{Result, SoulseekRs},
peer::{ConnectionType, DownloadPeer, NewPeer, Peer, listen::Listen}, peer::{ConnectionType, DownloadPeer, NewPeer, Peer, listen::Listen},
types::{Download, Search, SearchResult}, types::{Download, Search, SearchResult},
utils::{md5, thread_pool::ThreadPool}, utils::{lock::RwLockExt, md5, thread_pool::ThreadPool},
}; };
use std::{ use std::{
collections::HashMap, collections::HashMap,
@@ -336,13 +336,13 @@ impl Client {
} }
} }
pub fn connect(&mut self) { pub fn connect(&mut self) -> Result<()> {
let (sender, message_reader): ( let (sender, message_reader): (
Sender<ClientOperation>, Sender<ClientOperation>,
Receiver<ClientOperation>, Receiver<ClientOperation>,
) = mpsc::channel(); ) = mpsc::channel();
let mut ctx = self.context.write().unwrap(); let mut ctx = self.context.write_safe()?;
ctx.sender = Some(sender.clone()); ctx.sender = Some(sender.clone());
let peer_registry = let peer_registry =
PeerRegistry::new(ctx.actor_system.clone(), sender.clone()); PeerRegistry::new(ctx.actor_system.clone(), sender.clone());
@@ -385,6 +385,8 @@ impl Client {
self.context.clone(), self.context.clone(),
self.username.clone(), self.username.clone(),
); );
Ok(())
} }
pub fn login(&self) -> Result<bool> { pub fn login(&self) -> Result<bool> {
@@ -408,7 +410,13 @@ impl Client {
#[allow(dead_code)] #[allow(dead_code)]
pub fn remove_peer(&self, username: &str) { pub fn remove_peer(&self, username: &str) {
let context = self.context.read().unwrap(); let context = match self.context.read_safe() {
Ok(c) => c,
Err(e) => {
error!("[client] remove_peer: {}", e);
return;
}
};
if let Some(ref registry) = context.peer_registry if let Some(ref registry) = context.peer_registry
&& let Some(handle) = registry.remove_peer(username) && let Some(handle) = registry.remove_peer(username)
{ {
@@ -437,7 +445,7 @@ impl Client {
let token = u32::from_str_radix(&hash[0..5], 16)?; let token = u32::from_str_radix(&hash[0..5], 16)?;
// Initialize new search with query string as key // Initialize new search with query string as key
self.context.write().unwrap().searches.insert( self.context.write_safe()?.searches.insert(
query.to_string(), query.to_string(),
Search { Search {
token, token,
@@ -476,21 +484,21 @@ impl Client {
pub fn get_search_results_count(&self, search_key: &str) -> usize { pub fn get_search_results_count(&self, search_key: &str) -> usize {
self.context self.context
.read() .read_safe()
.unwrap() .ok()
.searches .and_then(|ctx| {
.get(search_key) ctx.searches.get(search_key).map(|s| s.results.len())
.map(|s| s.results.len()) })
.unwrap_or(0) .unwrap_or(0)
} }
pub fn get_search_results(&self, search_key: &str) -> Vec<SearchResult> { pub fn get_search_results(&self, search_key: &str) -> Vec<SearchResult> {
self.context self.context
.read() .read_safe()
.unwrap() .ok()
.searches .and_then(|ctx| {
.get(search_key) ctx.searches.get(search_key).map(|s| s.results.clone())
.map(|s| s.results.clone()) })
.unwrap_or_default() .unwrap_or_default()
} }
@@ -505,15 +513,27 @@ impl Client {
} }
pub fn get_all_searches(&self) -> HashMap<String, Search> { pub fn get_all_searches(&self) -> HashMap<String, Search> {
self.context.read().unwrap().searches.clone() self.context
.read_safe()
.map(|ctx| ctx.searches.clone())
.unwrap_or_default()
} }
pub fn get_all_downloads(&self) -> Vec<Download> { pub fn get_all_downloads(&self) -> Vec<Download> {
self.context.read().unwrap().get_downloads().clone() self.context
.read_safe()
.map(|ctx| ctx.get_downloads().clone())
.unwrap_or_default()
} }
pub fn pause_download(&self, username: &str, filename: &str) -> bool { pub fn pause_download(&self, username: &str, filename: &str) -> bool {
let mut context = self.context.write().unwrap(); let mut context = match self.context.write_safe() {
Ok(c) => c,
Err(e) => {
error!("[client] pause_download: {}", e);
return false;
}
};
let Some(download) = let Some(download) =
context.get_download_by_file_mut(username, filename) context.get_download_by_file_mut(username, filename)
else { else {
@@ -539,7 +559,13 @@ impl Client {
} }
pub fn resume_download(&self, username: &str, filename: &str) -> bool { pub fn resume_download(&self, username: &str, filename: &str) -> bool {
let mut context = self.context.write().unwrap(); let mut context = match self.context.write_safe() {
Ok(c) => c,
Err(e) => {
error!("[client] resume_download: {}", e);
return false;
}
};
let Some(download) = let Some(download) =
context.get_download_by_file_mut(username, filename) context.get_download_by_file_mut(username, filename)
else { else {
@@ -569,10 +595,15 @@ impl Client {
username: &str, username: &str,
filename: &str, filename: &str,
) -> bool { ) -> bool {
self.context match self.context.write_safe() {
.write() Ok(mut ctx) => {
.unwrap() ctx.remove_queued_download_by_file(username, filename)
.remove_queued_download_by_file(username, filename) }
Err(e) => {
error!("[client] remove_queued_download: {}", e);
false
}
}
} }
pub fn download( pub fn download(
@@ -602,7 +633,7 @@ impl Client {
sender: download_sender, sender: download_sender,
}; };
let mut context = self.context.write().unwrap(); let mut context = self.context.write_safe()?;
context.add_download(download.clone()); context.add_download(download.clone());
let download_initiated = let download_initiated =
@@ -619,8 +650,7 @@ impl Client {
if !download_initiated { if !download_initiated {
let _ = download.sender.send(DownloadStatus::Failed); let _ = download.sender.send(DownloadStatus::Failed);
self.context self.context
.write() .write_safe()?
.unwrap()
.update_download_with_status(token, DownloadStatus::Failed); .update_download_with_status(token, DownloadStatus::Failed);
} }
@@ -632,25 +662,38 @@ impl Client {
username: &str, username: &str,
filename: Option<&str>, filename: Option<&str>,
) { ) {
let context = client_context.read().unwrap(); let failed_downloads: Vec<_> = match client_context.read_safe() {
let failed_downloads: Vec<_> = context Ok(context) => context
.get_downloads() .get_downloads()
.iter() .iter()
.filter(|download| { .filter(|download| {
download.username == username download.username == username
&& (filename.is_none_or(|f| download.filename == *f)) && (filename.is_none_or(|f| download.filename == *f))
}) })
.map(|download| { .map(|download| {
let _ = download.sender.send(DownloadStatus::Failed); let _ = download.sender.send(DownloadStatus::Failed);
download.token download.token
}) })
.collect(); .collect(),
drop(context); Err(e) => {
error!("[client] process_failed_uploads read: {}", e);
return;
}
};
failed_downloads.iter().for_each(|token| { failed_downloads.iter().for_each(|token| {
let mut context = client_context.write().unwrap(); match client_context.write_safe() {
context.update_download_with_status(*token, DownloadStatus::Failed); Ok(mut context) => {
context.remove_download(*token); context.update_download_with_status(
*token,
DownloadStatus::Failed,
);
context.remove_download(*token);
}
Err(e) => {
error!("[client] process_failed_uploads write: {}", e);
}
}
}); });
} }
@@ -683,8 +726,18 @@ impl Client {
"[client] SearchResult {:?}", "[client] SearchResult {:?}",
search_result search_result
); );
let mut context = let mut context = match client_context
client_context.write().unwrap(); .write_safe()
{
Ok(c) => c,
Err(e) => {
error!(
"[client] SearchResult write: {}",
e
);
continue;
}
};
let result_token = search_result.token; let result_token = search_result.token;
// Find the search with matching token // Find the search with matching token
@@ -699,7 +752,16 @@ impl Client {
username, username,
error, error,
) => { ) => {
let context = client_context.read().unwrap(); let context = match client_context.read_safe() {
Ok(c) => c,
Err(e) => {
error!(
"[client] PeerDisconnected read: {}",
e
);
continue;
}
};
if let Some(ref registry) = if let Some(ref registry) =
context.peer_registry context.peer_registry
&& let Some(handle) = && let Some(handle) =
@@ -731,12 +793,19 @@ impl Client {
peer, peer,
allowed, allowed,
) => { ) => {
let maybe_download = { let maybe_download = match client_context
let client_context = .read_safe()
client_context.read().unwrap(); {
client_context Ok(ctx) => ctx
.get_download_by_token(token) .get_download_by_token(token)
.cloned() .cloned(),
Err(e) => {
error!(
"[client] DownloadFromPeer read: {}",
e
);
continue;
}
}; };
let own_username = own_username.clone(); let own_username = own_username.clone();
let client_context_clone = let client_context_clone =
@@ -777,7 +846,10 @@ impl Client {
filename, filename,
)) => { )) => {
let _ = download.sender.send(DownloadStatus::Completed); let _ = download.sender.send(DownloadStatus::Completed);
client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Completed); match client_context_clone.write_safe() {
Ok(mut ctx) => ctx.update_download_with_status(download.token, DownloadStatus::Completed),
Err(e) => error!("[client] download complete write: {}", e),
}
info!( info!(
"Successfully downloaded {} bytes to {}", "Successfully downloaded {} bytes to {}",
download.size, download.size,
@@ -786,7 +858,10 @@ impl Client {
} }
Err(e) => { Err(e) => {
let _ = download.sender.send(DownloadStatus::Failed); let _ = download.sender.send(DownloadStatus::Failed);
client_context_clone.write().unwrap().update_download_with_status(download.token, DownloadStatus::Failed); match client_context_clone.write_safe() {
Ok(mut ctx) => ctx.update_download_with_status(download.token, DownloadStatus::Failed),
Err(e) => error!("[client] download failed write: {}", e),
}
error!( error!(
"Failed to download file '{}' from {}:{} (token: {}) - Error: {}", "Failed to download file '{}' from {}:{} (token: {}) - Error: {}",
filename, filename,
@@ -814,34 +889,61 @@ impl Client {
}; };
} }
ClientOperation::NewPeer(new_peer) => { ClientOperation::NewPeer(new_peer) => {
let peer_exists = client_context let peer_exists = match client_context
.read() .read_safe()
.unwrap() {
.peer_registry Ok(ctx) => ctx
.as_ref() .peer_registry
.map(|r| r.contains(&new_peer.username)) .as_ref()
.unwrap_or(false); .map(|r| r.contains(&new_peer.username))
.unwrap_or(false),
Err(e) => {
error!("[client] NewPeer read: {}", e);
continue;
}
};
if peer_exists { if peer_exists {
debug!( debug!(
"Already connected to {}", "Already connected to {}",
new_peer.username new_peer.username
); );
} else if let Some(server_sender) = } else {
&client_context let send_result = client_context
.read() .read_safe()
.unwrap() .ok()
.server_sender .and_then(|ctx| {
{ ctx.server_sender.as_ref().map(
server_sender |s| {
.send(ServerMessage::GetPeerAddress( s.send(
new_peer.username.clone(), ServerMessage::GetPeerAddress(
)) new_peer
.unwrap(); .username
.clone(),
),
)
},
)
});
if let Some(Err(e)) = send_result {
error!(
"[client] NewPeer send GetPeerAddress: {}",
e
);
}
} }
let addr = let addr = match new_peer.tcp_stream.peer_addr()
new_peer.tcp_stream.peer_addr().unwrap(); {
Ok(a) => a,
Err(e) => {
error!(
"[client] NewPeer peer_addr: {}",
e
);
continue;
}
};
let host = addr.ip().to_string(); let host = addr.ip().to_string();
let port: u32 = addr.port().into(); let port: u32 = addr.port().into();
@@ -879,13 +981,22 @@ impl Client {
obfuscated_port obfuscated_port
); );
let peer_exists = client_context let peer_exists = match client_context
.read() .read_safe()
.unwrap() {
.peer_registry Ok(ctx) => ctx
.as_ref() .peer_registry
.map(|r| r.contains(&username)) .as_ref()
.unwrap_or(false); .map(|r| r.contains(&username))
.unwrap_or(false),
Err(e) => {
error!(
"[client] GetPeerAddressResponse read: {}",
e
);
continue;
}
};
if peer_exists { if peer_exists {
// don't know if i should update? and or reconnect the peer // don't know if i should update? and or reconnect the peer
@@ -933,8 +1044,18 @@ impl Client {
transfer, transfer,
username, username,
) => { ) => {
let mut context = let mut context = match client_context
client_context.write().unwrap(); .write_safe()
{
Ok(c) => c,
Err(e) => {
error!(
"[client] UpdateDownloadTokens write: {}",
e
);
continue;
}
};
let download_to_update = context let download_to_update = context
.get_downloads() .get_downloads()
@@ -981,9 +1102,18 @@ impl Client {
); );
} }
ClientOperation::SetServerSender(sender) => { ClientOperation::SetServerSender(sender) => {
client_context.write().unwrap().server_sender = match client_context.write_safe() {
Some(sender); Ok(mut ctx) => {
debug!("[client] Server sender initialized"); ctx.server_sender = Some(sender);
debug!(
"[client] Server sender initialized"
);
}
Err(e) => error!(
"[client] SetServerSender write: {}",
e
),
}
} }
} }
} }
@@ -1013,7 +1143,13 @@ impl Client {
ConnectionType::P => { ConnectionType::P => {
let username = peer.username.clone(); let username = peer.username.clone();
let context = client_context.read().unwrap(); let context = match client_context.read_safe() {
Ok(c) => c,
Err(e) => {
error!("[client] connect_to_peer read: {}", e);
return;
}
};
if let Some(ref registry) = context.peer_registry { if let Some(ref registry) = context.peer_registry {
match registry.register_peer(peer_clone, stream, None) { match registry.register_peer(peer_clone, stream, None) {
Ok(_) => (), Ok(_) => (),
@@ -1054,13 +1190,16 @@ impl Client {
filename, download.size filename, download.size
); );
let _ = download.sender.send(DownloadStatus::Completed); let _ = download.sender.send(DownloadStatus::Completed);
client_context match client_context.write_safe() {
.write() Ok(mut ctx) => ctx.update_download_with_status(
.unwrap()
.update_download_with_status(
download.token, download.token,
DownloadStatus::Completed, DownloadStatus::Completed,
); ),
Err(e) => error!(
"[client] connect_to_peer F write: {}",
e
),
}
} }
Err(e) => { Err(e) => {
trace!("[client] failed to download: {}", e); trace!("[client] failed to download: {}", e);
@@ -1079,7 +1218,13 @@ impl Client {
) { ) {
debug!("Piercing firewall for peer: {:?}", peer); debug!("Piercing firewall for peer: {:?}", peer);
let context = client_context.read().unwrap(); let context = match client_context.read_safe() {
Ok(c) => c,
Err(e) => {
error!("[client] pierce_firewall read: {}", e);
return;
}
};
if let Some(server_sender) = &context.server_sender { if let Some(server_sender) = &context.server_sender {
if let Some(token) = peer.token { if let Some(token) = peer.token {
match server_sender.send(ServerMessage::PierceFirewall(token)) { match server_sender.send(ServerMessage::PierceFirewall(token)) {
soulseek-rs-lib/src/error.rs
+5
View File
@@ -18,6 +18,8 @@ pub enum SoulseekRs {
NotConnected, NotConnected,
/// Compression/decompression error /// Compression/decompression error
CompressionError(String), CompressionError(String),
/// A lock was poisoned by a panic in another thread
LockPoisoned,
} }
impl fmt::Display for SoulseekRs { impl fmt::Display for SoulseekRs {
@@ -39,6 +41,9 @@ impl fmt::Display for SoulseekRs {
SoulseekRs::CompressionError(msg) => { SoulseekRs::CompressionError(msg) => {
write!(f, "Compression error: {}", msg) write!(f, "Compression error: {}", msg)
} }
SoulseekRs::LockPoisoned => {
write!(f, "Lock poisoned by panicking thread")
}
} }
} }
} }
soulseek-rs-lib/src/peer/listen.rs
+22 -8
View File
@@ -9,6 +9,7 @@ use crate::client::{ClientContext, ClientOperation};
use crate::message::{Message, MessageReader}; use crate::message::{Message, MessageReader};
use crate::peer::{ConnectionType, DownloadPeer, Peer}; use crate::peer::{ConnectionType, DownloadPeer, Peer};
use crate::types::Download; use crate::types::Download;
use crate::utils::lock::RwLockExt;
use crate::{DownloadStatus, debug, error, info, trace}; use crate::{DownloadStatus, debug, error, info, trace};
const PEER_INIT_MESSAGE_CODE: u8 = 1; const PEER_INIT_MESSAGE_CODE: u8 = 1;
@@ -82,7 +83,13 @@ fn extract_download_from_buffer(
username, token username, token
); );
let context = client_context.read().unwrap(); let context = match client_context.read_safe() {
Ok(c) => c,
Err(e) => {
error!("[listener] client context lock: {}", e);
return None;
}
};
let download = context.get_download_by_token(token).cloned(); let download = context.get_download_by_token(token).cloned();
if download.is_none() { if download.is_none() {
@@ -104,7 +111,13 @@ fn handle_peer_connection(
_peer_ip: &str, _peer_ip: &str,
_peer_port: u16, _peer_port: u16,
) { ) {
let client_context = context.client_context.read().unwrap(); let client_context = match context.client_context.read_safe() {
Ok(c) => c,
Err(e) => {
error!("[listener] handle_peer_connection lock: {}", e);
return;
}
};
if let Some(ref registry) = client_context.peer_registry { if let Some(ref registry) = client_context.peer_registry {
match registry.register_peer(peer.clone(), Some(stream), Some(reader)) { match registry.register_peer(peer.clone(), Some(stream), Some(reader)) {
Ok(_) => (), Ok(_) => (),
@@ -158,14 +171,15 @@ fn handle_file_connection(
) { ) {
Ok((download, filename)) => { Ok((download, filename)) => {
let _ = download.sender.send(DownloadStatus::Completed); let _ = download.sender.send(DownloadStatus::Completed);
context match context.client_context.write_safe() {
.client_context Ok(mut ctx) => ctx.update_download_with_status(
.write()
.unwrap()
.update_download_with_status(
download.token, download.token,
DownloadStatus::Completed, DownloadStatus::Completed,
); ),
Err(e) => {
error!("[listener] handle_file_connection write: {}", e)
}
}
info!( info!(
"Successfully downloaded {} bytes to {}", "Successfully downloaded {} bytes to {}",
download.size, filename download.size, filename
soulseek-rs-lib/src/utils/lock.rs
+28
View File
@@ -0,0 +1,28 @@
use std::sync::{Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
use crate::error::{Result, SoulseekRs};
pub trait RwLockExt<T> {
fn read_safe(&self) -> Result<RwLockReadGuard<'_, T>>;
fn write_safe(&self) -> Result<RwLockWriteGuard<'_, T>>;
}
impl<T> RwLockExt<T> for RwLock<T> {
fn read_safe(&self) -> Result<RwLockReadGuard<'_, T>> {
self.read().map_err(|_| SoulseekRs::LockPoisoned)
}
fn write_safe(&self) -> Result<RwLockWriteGuard<'_, T>> {
self.write().map_err(|_| SoulseekRs::LockPoisoned)
}
}
pub trait MutexExt<T> {
fn lock_safe(&self) -> Result<MutexGuard<'_, T>>;
}
impl<T> MutexExt<T> for Mutex<T> {
fn lock_safe(&self) -> Result<MutexGuard<'_, T>> {
self.lock().map_err(|_| SoulseekRs::LockPoisoned)
}
}
soulseek-rs-lib/src/utils/mod.rs
+1
View File
@@ -1,5 +1,6 @@
#[macro_use] #[macro_use]
pub mod logger; pub mod logger;
pub mod lock;
pub mod md5; pub mod md5;
pub mod thread_pool; pub mod thread_pool;
pub mod zlib; pub mod zlib;
soulseek-rs-lib/src/utils/thread_pool.rs
+6 -1
View File
@@ -2,6 +2,8 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, mpsc}; use std::sync::{Arc, Mutex, mpsc};
use std::thread; use std::thread;
use crate::utils::lock::MutexExt;
type Job = Box<dyn FnOnce() + Send + 'static>; type Job = Box<dyn FnOnce() + Send + 'static>;
enum Message { enum Message {
@@ -91,7 +93,10 @@ impl Worker {
) -> Worker { ) -> Worker {
let thread = thread::spawn(move || { let thread = thread::spawn(move || {
loop { loop {
let message = receiver.lock().unwrap().recv(); let message = match receiver.lock_safe() {
Ok(rx) => rx.recv(),
Err(_) => break,
};
match message { match message {
Ok(Message::NewJob(job)) => { Ok(Message::NewJob(job)) => {
// let active = // let active =
soulseek-rs/src/main.rs
+6 -2
View File
@@ -92,7 +92,9 @@ fn main() -> Result<()> {
}; };
let mut client = Client::with_settings(settings); let mut client = Client::with_settings(settings);
client.connect(); client.connect().map_err(|e| {
color_eyre::eyre::eyre!("Failed to connect: {}", e)
})?;
client.login().map_err(|e| { client.login().map_err(|e| {
color_eyre::eyre::eyre!("Failed to login: {}", e) color_eyre::eyre::eyre!("Failed to login: {}", e)
})?; })?;
@@ -144,7 +146,9 @@ fn search_and_download(config: SearchConfig) -> Result<()> {
}; };
let mut client = Client::with_settings(settings); let mut client = Client::with_settings(settings);
client.connect(); client
.connect()
.map_err(|e| color_eyre::eyre::eyre!("Failed to connect: {}", e))?;
client client
.login() .login()
.map_err(|e| color_eyre::eyre::eyre!("Failed to login: {}", e))?; .map_err(|e| color_eyre::eyre::eyre!("Failed to login: {}", e))?;