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

fix: deadlock issue with sending messages

Browse Source 
intoduced a channel and used stream.try_clone
This commit is contained in:
michel
2024-09-10 11:59:51 +02:00
committed by Michel de Graaf
parent 1461e385cf
commit 6281b0c89e
6 changed files with 158 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server.rs
+112 -97
View File
@@ -1,13 +1,13 @@
use crate::dispatcher::MessageDispatcher;
use crate::message::factory::{build_init_message, build_login_message};
use crate::message::MessageReader;
use std::collections::VecDeque;
use crate::message::{Message, MessageReader};
use std::io::{self, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::sync::{Arc, Barrier, Condvar, Mutex};
use std::sync::mpsc;
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, Barrier, Mutex};
use std::thread::{self};
use std::time::Duration;
#[derive(Debug, Clone)]
pub struct ServerAddress {
host: String,
@@ -26,57 +26,69 @@ impl ServerAddress {
pub fn get_port(&self) -> u16 {
self.port
}
// pub fn set_host(&mut self, host: String) {
// self.host = host;
// }
//
// pub fn set_port(&mut self, port: u16) {
// self.port = port;
// }
//
// pub fn set_host_port(&mut self, host: String, port: u16) {
// self.host = host;
// self.port = port;
// }
//
// pub fn get_host_port(&self) -> String {
// format!("{}:{}", self.host, self.port)
// }
}
pub struct Server {
address: ServerAddress,
stream: Arc<Mutex<TcpStream>>,
message_queue: Arc<(Mutex<VecDeque<Vec<u8>>>, Condvar)>,
}
#[derive(Debug, Clone)]
#[derive(Debug)]
pub struct Context {
message_queue: Arc<(Mutex<VecDeque<Vec<u8>>>, Condvar)>,
rooms: Vec<Room>,
message_sender: Arc<Mutex<Sender<Message>>>,
rooms: Rooms,
}
impl Context {
pub fn new(message_queue: Arc<(Mutex<VecDeque<Vec<u8>>>, Condvar)>) -> Self {
pub fn new(message_sender: Arc<Mutex<Sender<Message>>>) -> Self {
Self {
message_queue,
rooms: Vec::new(),
message_sender,
rooms: Rooms::new(),
}
}
pub fn set_rooms(&mut self, rooms: Vec<Room>) {
self.rooms = rooms;
pub fn get_rooms(&mut self) -> &mut Rooms {
&mut self.rooms
}
pub fn queue_message(&self, message: Vec<u8>) {
let (lock, cvar) = &*self.message_queue;
let mut queue = lock.lock().unwrap();
println!("Queueing message: {:?}", message); // Debug log before pushing to queue
queue.push_back(message);
cvar.notify_one();
pub fn queue_message(&self, message: Message) {
self.message_sender.lock().unwrap().send(message).unwrap();
}
}
#[derive(Debug, Clone)]
#[derive(Debug)]
pub struct Rooms {
pub public_rooms: Vec<Room>,
pub owned_private_rooms: Vec<Room>,
pub private_rooms: Vec<Room>,
pub operated_private_rooms: Vec<Room>,
}
impl Rooms {
fn new() -> Self {
Self {
public_rooms: Vec::new(),
owned_private_rooms: Vec::new(),
private_rooms: Vec::new(),
operated_private_rooms: Vec::new(),
}
}
pub fn print(&self) {
println!("Public rooms ({}):", self.public_rooms.len());
for room in &self.public_rooms {
room.print();
}
println!("Owned private rooms ({}):", self.owned_private_rooms.len());
for room in &self.owned_private_rooms {
room.print();
}
println!("Private rooms ({}):", self.private_rooms.len());
for room in &self.private_rooms {
room.print();
}
println!(
"Operated private rooms ({}):",
self.operated_private_rooms.len()
);
for room in &self.operated_private_rooms {
room.print();
}
}
}
#[derive(Debug)]
pub struct Room {
name: String,
number_of_users: i32,
@@ -91,34 +103,38 @@ impl Room {
pub fn set_number_of_users(&mut self, number_of_users: i32) {
self.number_of_users = number_of_users;
}
pub fn print(&self) {
println!(
"Room: {}, Number of users: {}",
self.name, self.number_of_users
);
}
// pub fn get_name(&self) -> &str {
// &self.name
// }
// pub fn get_number_of_users(&self) -> i32 {
// self.number_of_users
// }
}
pub struct Server {
address: ServerAddress,
context: Arc<Mutex<Context>>,
}
impl Server {
/// Create a new instance of Server, returning a Result
pub fn new(address: ServerAddress) -> Result<Self, io::Error> {
println!("Connecting to server at {}:{}", address.host, address.port);
let (message_sender, message_reader): (Sender<Message>, Receiver<Message>) =
mpsc::channel();
// Convert the host and port into a socket address
let socket_address = format!("{}:{}", address.host, address.port)
.to_socket_addrs()?
.next()
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid address"))?;
let context = Arc::new(Mutex::new(Context::new(Arc::new(Mutex::new(
message_sender,
)))));
// Establish the connection with a timeout
let stream = TcpStream::connect_timeout(&socket_address, Duration::from_secs(10))?;
let server = Self { address, context };
// Set the stream to non-blocking mode if needed or configure timeouts
stream.set_read_timeout(Some(Duration::from_secs(5)))?;
stream.set_write_timeout(Some(Duration::from_secs(5)))?;
let message_queue = Arc::new((Mutex::new(VecDeque::new()), Condvar::new())); // Initialize the message queue with Condvar
let server = Self {
address,
stream: Arc::new(Mutex::new(stream)),
message_queue,
};
server.start_read_write_loops();
server.start_read_write_loops(message_reader)?;
Ok(server)
}
@@ -128,25 +144,37 @@ impl Server {
}
/// Start reading and writing loops in separate threads
fn start_read_write_loops(&self) {
// Create a channel to send messages to the write thread
let read_stream = Arc::clone(&self.stream);
let write_stream = Arc::clone(&self.stream);
let message_queue = Arc::clone(&self.message_queue);
fn start_read_write_loops(&self, message_reader: Receiver<Message>) -> Result<(), io::Error> {
let socket_address = format!("{}:{}", self.address.host, self.address.port)
.to_socket_addrs()?
.next()
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid address"))?;
println!(
"Connecting to server at {}:{}",
self.address.host, self.address.port
);
let stream = TcpStream::connect_timeout(&socket_address, Duration::from_secs(10))?;
stream.set_read_timeout(Some(Duration::from_secs(5)))?;
stream.set_write_timeout(Some(Duration::from_secs(5)))?;
let mut read_stream = stream.try_clone()?;
let mut write_stream = stream.try_clone()?;
let barrier = Arc::new(Barrier::new(2));
let read_barrier = barrier.clone();
let write_barrier = barrier.clone();
let mut buffered_reader = MessageReader::new();
let mut dispatcher = MessageDispatcher::new(Context::new(Arc::clone(&message_queue)));
let dispatcher = MessageDispatcher::new(Arc::clone(&self.context));
// Spawn a thread to handle reading from the server
thread::spawn(move || {
read_barrier.wait();
let mut buffered_reader = MessageReader::new();
loop {
match buffered_reader.read_from_socket(&mut read_stream.lock().unwrap()) {
match buffered_reader.read_from_socket(&mut read_stream) {
Ok(_) => {}
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
@@ -173,44 +201,31 @@ impl Server {
}
});
// introduce a channel , get rid of the thread ?
thread::spawn(move || {
write_barrier.wait();
loop {
let (lock, cvar) = &*message_queue;
let mut queue = lock.lock().unwrap();
// Wait for a message to be added to the queue
while queue.is_empty() {
queue = cvar.wait(queue).unwrap();
}
// Pop and send the message
if let Some(message) = queue.pop_front() {
println!("Sending message from queue: {:?}", message); // Debug log when sending
match write_stream.lock().unwrap().write_all(&message) {
Ok(_) => println!("Sent buffered message: {:?}", message),
Err(e) => {
eprintln!("Failed to send buffered message: {}", e);
// Handle error appropriately
break;
}
let message = message_reader.recv().unwrap();
println!("Sending message from queue: {:?}", message); // Debug log when sending
match write_stream.write_all(&message.get_data()) {
Ok(_) => println!("Sent buffered message: {:?}", message),
Err(e) => {
eprintln!("Failed to send buffered message: {}", e);
break;
}
}
}
});
Ok(())
}
fn queue_message(&self, message: Vec<u8>) {
let context = Context::new(self.message_queue.clone());
context.queue_message(message);
fn queue_message(&self, message: Message) {
self.context.lock().unwrap().queue_message(message);
}
pub fn login(&mut self, username: &str, password: &str) -> Result<(), std::io::Error> {
self.queue_message(build_init_message().get_data());
self.queue_message(build_login_message(username, password).get_data());
// self.queue_message(build_shared_folders_message(1, 2).get_data());
self.queue_message(build_init_message());
self.queue_message(build_login_message(username, password));
Ok(())
}
}