chore: debug logging

This commit is contained in:
Michel de Graaf
2025-11-11 14:53:40 +01:00
parent a4aee59e1e
commit a1b88f58bd
7 changed files with 148 additions and 91 deletions
+4 -20
View File
@@ -101,44 +101,29 @@ impl ActorSystem {
}; };
let handle_for_init = handle.clone(); let handle_for_init = handle.clone();
// Start the actor event loop on the thread pool
self.thread_pool.execute(move || { self.thread_pool.execute(move || {
trace!("[actor_system] Actor task starting on thread pool");
init(&mut actor, handle_for_init); init(&mut actor, handle_for_init);
trace!("[actor_system] Actor init callback completed");
actor.on_start(); actor.on_start();
trace!(
"[actor_system] Actor on_start completed, starting event loop"
);
Self::run_actor_loop(&mut actor, receiver); Self::run_actor_loop(&mut actor, receiver);
trace!("[actor_system] Actor event loop ended");
actor.on_stop(); actor.on_stop();
trace!("[actor_system] Actor on_stop completed");
}); });
handle handle
} }
/// Run the actor's message processing loop
fn run_actor_loop<A: Actor>( fn run_actor_loop<A: Actor>(
actor: &mut A, actor: &mut A,
receiver: Receiver<ActorMessage<A::Message>>, receiver: Receiver<ActorMessage<A::Message>>,
) { ) {
trace!("[actor_system] run_actor_loop STARTED");
let tick_interval = Duration::from_millis(100); let tick_interval = Duration::from_millis(100);
let mut last_tick = Instant::now(); let mut last_tick = Instant::now();
let mut message_count = 0; let mut message_count = 0;
let mut tick_count = 0; let mut tick_count = 0;
loop { loop {
// Try to receive messages with timeout for periodic ticks
match receiver.recv_timeout(tick_interval) { match receiver.recv_timeout(tick_interval) {
Ok(ActorMessage::UserMessage(msg)) => { Ok(ActorMessage::UserMessage(msg)) => {
message_count += 1; message_count += 1;
trace!(
"[actor_system] Processing message #{}",
message_count
);
actor.handle(msg); actor.handle(msg);
} }
Ok(ActorMessage::Stop) => { Ok(ActorMessage::Stop) => {
@@ -153,15 +138,15 @@ impl ActorSystem {
"[actor_system] Received explicit Tick message #{}", "[actor_system] Received explicit Tick message #{}",
tick_count tick_count
); );
actor.tick(); actor.tick();
} }
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
// Periodic tick
if last_tick.elapsed() >= tick_interval { if last_tick.elapsed() >= tick_interval {
tick_count += 1; tick_count += 1;
if tick_count % 10 == 0 { // if tick_count % 10 == 0 {
trace!("[actor_system] Periodic tick #{} (every 1s log)", tick_count); // trace!("[actor_system] Periodic tick #{} (every 1s log)", tick_count);
} // }
actor.tick(); actor.tick();
last_tick = Instant::now(); last_tick = Instant::now();
} }
@@ -170,7 +155,6 @@ impl ActorSystem {
trace!( trace!(
"[actor_system] Channel disconnected, breaking loop" "[actor_system] Channel disconnected, breaking loop"
); );
// Channel closed, actor should stop
break; break;
} }
} }
+32 -18
View File
@@ -11,35 +11,29 @@ use crate::peer::Peer;
use crate::types::{Download, FileSearchResult, Transfer}; use crate::types::{Download, FileSearchResult, Transfer};
use crate::{debug, error, trace, warn}; use crate::{debug, error, trace, warn};
use std::io::{self, Write}; use std::io::{self, Error, Write};
use std::net::TcpStream; use std::net::TcpStream;
use std::sync::mpsc::{Receiver, Sender}; use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
/// Messages that can be sent to a PeerActor
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub enum PeerMessage { pub enum PeerMessage {
/// Send a raw message to the peer
SendMessage(Message), SendMessage(Message),
/// Handle a file search result
FileSearchResult(FileSearchResult), FileSearchResult(FileSearchResult),
/// Handle a transfer request
TransferRequest(Transfer), TransferRequest(Transfer),
/// Handle a transfer response UploadFailed(String, String),
TransferResponse { TransferResponse {
token: u32, token: u32,
allowed: bool, allowed: bool,
reason: Option<String>, reason: Option<String>,
}, },
/// Handle a place in queue response PlaceInQueueResponse {
PlaceInQueueResponse { filename: String, place: u32 }, filename: String,
/// Set the peer's username place: u32,
},
SetUsername(String), SetUsername(String),
/// Queue an upload for a file
QueueUpload(String), QueueUpload(String),
/// Request a transfer for a download
RequestTransfer(Download), RequestTransfer(Download),
/// Internal: process read operations
ProcessRead, ProcessRead,
} }
@@ -52,6 +46,7 @@ pub struct PeerActor {
self_handle: Option<ActorHandle<PeerMessage>>, self_handle: Option<ActorHandle<PeerMessage>>,
dispatcher: Option<MessageDispatcher<PeerMessage>>, dispatcher: Option<MessageDispatcher<PeerMessage>>,
dispatcher_receiver: Option<Receiver<PeerMessage>>, dispatcher_receiver: Option<Receiver<PeerMessage>>,
queued_filenames: Vec<String>,
} }
impl PeerActor { impl PeerActor {
@@ -70,6 +65,7 @@ impl PeerActor {
self_handle: None, self_handle: None,
dispatcher: None, dispatcher: None,
dispatcher_receiver: None, dispatcher_receiver: None,
queued_filenames: Vec::new(),
} }
} }
@@ -206,6 +202,7 @@ impl PeerActor {
self.peer.write().unwrap().username = username; self.peer.write().unwrap().username = username;
} }
PeerMessage::QueueUpload(filename) => { PeerMessage::QueueUpload(filename) => {
self.queued_filenames.push(filename.clone());
let message = let message =
MessageFactory::build_queue_upload_message(&filename); MessageFactory::build_queue_upload_message(&filename);
self.send_message(message); self.send_message(message);
@@ -220,6 +217,11 @@ impl PeerActor {
PeerMessage::ProcessRead => { PeerMessage::ProcessRead => {
self.process_read(); self.process_read();
} }
PeerMessage::UploadFailed(username, filename) => {
self.client_channel
.send(ClientOperation::UploadFailed(username, filename))
.unwrap();
}
} }
} }
@@ -256,7 +258,7 @@ impl PeerActor {
"[peer_actor:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.", "[peer_actor:{}] Error reading from peer: {} (kind: {:?}). Disconnecting.",
username, e, e.kind() username, e, e.kind()
); );
self.disconnect(); self.disconnect_with_error(e);
return; return;
} }
} }
@@ -296,7 +298,7 @@ impl PeerActor {
"[peer_actor:{}] Error extracting message: {}. Disconnecting peer.", "[peer_actor:{}] Error extracting message: {}. Disconnecting peer.",
username, e username, e
); );
self.disconnect(); self.disconnect_with_error(e);
return; return;
} }
Ok(None) => { Ok(None) => {
@@ -338,7 +340,7 @@ impl PeerActor {
"[peer_actor:{}] Error writing message: {}. Disconnecting.", "[peer_actor:{}] Error writing message: {}. Disconnecting.",
username, e username, e
); );
self.disconnect(); self.disconnect_with_error(e);
return; return;
} }
@@ -347,10 +349,22 @@ impl PeerActor {
"[peer_actor:{}] Error flushing stream: {}. Disconnecting.", "[peer_actor:{}] Error flushing stream: {}. Disconnecting.",
username, e username, e
); );
self.disconnect(); self.disconnect_with_error(e);
} }
} }
fn disconnect_with_error(&mut self, error: Error) {
let username = self.peer.read().unwrap().username.clone();
debug!("[peer_actor:{}] disconnect", username);
self.stream.take();
if let Err(e) = self.client_channel.send(
ClientOperation::PeerDisconnected(username, Some(error.into())),
) {
error!("Failed to send disconnect notification: {}", e);
}
}
/// Disconnect the peer and notify the client /// Disconnect the peer and notify the client
fn disconnect(&mut self) { fn disconnect(&mut self) {
let username = self.peer.read().unwrap().username.clone(); let username = self.peer.read().unwrap().username.clone();
@@ -360,7 +374,7 @@ impl PeerActor {
if let Err(e) = self if let Err(e) = self
.client_channel .client_channel
.send(ClientOperation::PeerDisconnected(username)) .send(ClientOperation::PeerDisconnected(username, None))
{ {
error!("Failed to send disconnect notification: {}", e); error!("Failed to send disconnect notification: {}", e);
} }
@@ -383,7 +397,7 @@ impl Actor for PeerActor {
// Trigger initial read // Trigger initial read
if let Some(ref handle) = self.self_handle { if let Some(ref handle) = self.self_handle {
match handle.send(PeerMessage::ProcessRead) { match handle.send(PeerMessage::ProcessRead) {
Ok(_) => {}, Ok(_) => {}
Err(e) => error!( Err(e) => error!(
"[peer_actor:{}] FAILED to send ProcessRead: {}", "[peer_actor:{}] FAILED to send ProcessRead: {}",
username, e username, e
-1
View File
@@ -113,7 +113,6 @@ impl PeerRegistry {
handle.send(message) handle.send(message)
} }
/// Queue an upload for a peer
pub fn queue_upload( pub fn queue_upload(
&self, &self,
username: &str, username: &str,
+53 -3
View File
@@ -25,11 +25,13 @@ use std::{
use crate::{debug, error, info, trace}; use crate::{debug, error, info, trace};
const DEFALT_LISTEN_PORT: u32 = 2234; const DEFALT_LISTEN_PORT: u32 = 2234;
#[derive(Debug)]
pub enum ClientOperation { pub enum ClientOperation {
NewPeer(NewPeer), NewPeer(NewPeer),
ConnectToPeer(Peer), ConnectToPeer(Peer),
SearchResult(FileSearchResult), SearchResult(FileSearchResult),
PeerDisconnected(String), PeerDisconnected(String, Option<SoulseekRs>),
PierceFireWall(Peer), PierceFireWall(Peer),
DownloadFromPeer(u32, Peer, bool), DownloadFromPeer(u32, Peer, bool),
UpdateDownloadTokens(Transfer, String), UpdateDownloadTokens(Transfer, String),
@@ -40,6 +42,7 @@ pub enum ClientOperation {
obfuscation_type: u32, obfuscation_type: u32,
obfuscated_port: u16, obfuscated_port: u16,
}, },
UploadFailed(String, String),
} }
pub struct ClientContext { pub struct ClientContext {
pub peer_registry: Option<PeerRegistry>, pub peer_registry: Option<PeerRegistry>,
@@ -228,6 +231,7 @@ impl Client {
size: u64, size: u64,
download_directory: String, download_directory: String,
) -> Result<crate::types::DownloadResult> { ) -> Result<crate::types::DownloadResult> {
info!("[client] Downloading {} from {}", filename, username);
let start_time = Instant::now(); let start_time = Instant::now();
let hash = md5::md5(&filename); let hash = md5::md5(&filename);
@@ -280,6 +284,34 @@ impl Client {
}) })
} }
fn process_failed_uploads(
client_context: Arc<RwLock<ClientContext>>,
username: &str,
filename: Option<&str>,
) {
let context = client_context.read().unwrap();
let download_tokens = &context.download_tokens;
let failed_downloads: Vec<_> = download_tokens
.iter()
.filter(|(_, download)| {
download.username == username
&& (filename.is_none_or(|f| download.filename == *f))
})
.map(|(token, download)| {
download.sender.send(DownloadStatus::Failed).unwrap();
*token
})
.collect();
failed_downloads.iter().for_each(|token| {
client_context
.write()
.unwrap()
.download_tokens
.remove(token);
});
}
fn listen_to_client_operations( fn listen_to_client_operations(
reader: Receiver<ClientOperation>, reader: Receiver<ClientOperation>,
client_context: Arc<RwLock<ClientContext>>, client_context: Arc<RwLock<ClientContext>>,
@@ -289,6 +321,7 @@ impl Client {
if let Ok(operation) = reader.recv() { if let Ok(operation) = reader.recv() {
match operation { match operation {
ClientOperation::ConnectToPeer(peer) => { ClientOperation::ConnectToPeer(peer) => {
trace!("[client] ConnectToPeer {:?}", peer);
Self::connect_to_peer( Self::connect_to_peer(
peer, peer,
client_context.clone(), client_context.clone(),
@@ -304,7 +337,7 @@ impl Client {
.search_results .search_results
.push(file_search.clone()); .push(file_search.clone());
} }
ClientOperation::PeerDisconnected(username) => { ClientOperation::PeerDisconnected(username, error) => {
let context = client_context.read().unwrap(); let context = client_context.read().unwrap();
if let Some(ref registry) = context.peer_registry { if let Some(ref registry) = context.peer_registry {
if let Some(handle) = if let Some(handle) =
@@ -313,6 +346,17 @@ impl Client {
let _ = handle.stop(); let _ = handle.stop();
} }
} }
if let Some(error) = error {
error!(
"[client] Peer {} disconnected with error: {:?}",
username, error
);
Self::process_failed_uploads(
client_context.clone(),
&username,
None,
);
}
} }
ClientOperation::PierceFireWall(peer) => { ClientOperation::PierceFireWall(peer) => {
Self::pierce_firewall( Self::pierce_firewall(
@@ -522,6 +566,13 @@ impl Client {
context.download_tokens.remove(&key); context.download_tokens.remove(&key);
} }
} }
ClientOperation::UploadFailed(username, filename) => {
Self::process_failed_uploads(
client_context.clone(),
&username,
Some(&filename),
);
}
} }
} }
}); });
@@ -544,7 +595,6 @@ impl Client {
ConnectionType::P => { ConnectionType::P => {
let username = peer.username.clone(); let username = peer.username.clone();
// Establish TCP connection or use provided stream
let mut tcp_stream = match stream { let mut tcp_stream = match stream {
Some(s) => s, Some(s) => s,
None => { None => {
-1
View File
@@ -3,7 +3,6 @@ use std::{error::Error, fmt};
/// Custom error type for the Soulseek download library /// Custom error type for the Soulseek download library
#[derive(Debug)] #[derive(Debug)]
pub enum SoulseekRs { pub enum SoulseekRs {
/// Network-related errors (I/O, connection failures)
NetworkError(std::io::Error), NetworkError(std::io::Error),
/// Authentication failed during login /// Authentication failed during login
AuthenticationFailed, AuthenticationFailed,
+4 -2
View File
@@ -6,14 +6,16 @@ fn main() {
PeerAddress::new(String::from("server.slsknet.org"), 2242), PeerAddress::new(String::from("server.slsknet.org"), 2242),
String::from("xxxxxx"), String::from("xxxxxx"),
String::from("xxxxxx"), String::from("xxxxxx"),
true, false,
None, None,
); );
client.connect(); client.connect();
match client.login() { match client.login() {
Ok(_) => { Ok(_) => {
match client.search("michel test file", Duration::from_secs(10)) { match client
.search("Eternity Alex Kassian", Duration::from_secs(10))
{
Ok(results) => { Ok(results) => {
if let Some(result) = if let Some(result) =
results.iter().find(|r| !r.files.is_empty()) results.iter().find(|r| !r.files.is_empty())
+19 -10
View File
@@ -341,21 +341,30 @@ impl Server {
if let Ok(operation) = server_channel.recv() { if let Ok(operation) = server_channel.recv() {
match operation { match operation {
ServerOperation::ConnectToPeer(peer) => { ServerOperation::ConnectToPeer(peer) => {
debug!( trace!(
"[server] ConnectToPeer {} - ConnectionType {}", "[server] → ConnectToPeer {}",
peer.username, peer.connection_type peer.username
); );
if let Some(op) = match peer.connection_type { if let Some(op) = match peer.connection_type {
ConnectionType::P => { ConnectionType::P | ConnectionType::F => {
Some(ClientOperation::ConnectToPeer(peer)) Some(ClientOperation::ConnectToPeer(
} peer.clone(),
ConnectionType::F => { ))
Some(ClientOperation::ConnectToPeer(peer))
} }
ConnectionType::D => None, ConnectionType::D => None,
} { } {
client_channel.send(op).unwrap(); trace!(
"[server] → ConnectToPeer {} - op {:?}",
peer.username,
op
);
match client_channel.send(op) {
Ok(_) => {}
Err(e) => {
error!("[server] Failed to send: {}", e)
}
}
} }
} }
ServerOperation::LoginStatus(message) => { ServerOperation::LoginStatus(message) => {