From a9a49bef8974de0181a4dd79232efa48ef623926 Mon Sep 17 00:00:00 2001 From: LucasDLTG Date: Tue, 21 Jul 2026 14:47:47 +0200 Subject: [PATCH] fix: thread lock in tcp send/receive --- src/dsp408/client/dsp408.rs | 59 +++++++++++++++++++++++-------------- 1 file changed, 37 insertions(+), 22 deletions(-) diff --git a/src/dsp408/client/dsp408.rs b/src/dsp408/client/dsp408.rs index b4ff180..411e1c4 100644 --- a/src/dsp408/client/dsp408.rs +++ b/src/dsp408/client/dsp408.rs @@ -3,7 +3,7 @@ use std::{ fmt, io::{Read, Write}, net::{IpAddr, SocketAddr, TcpStream}, - sync::{Mutex, OnceLock}, + sync::{Arc, Mutex, OnceLock}, time::Duration, }; @@ -33,8 +33,7 @@ fn instances() -> &'static Mutex> { pub struct DSP408 { connection: types::DeviceConnection, - socket: Option, - lock: Mutex<()>, + socket: Arc>>, connected: bool, state: Option, } @@ -62,9 +61,7 @@ impl DSP408 { device_id, }, - socket: None, - - lock: Mutex::new(()), + socket: Arc::new(Mutex::new(None)), connected: false, @@ -72,19 +69,13 @@ impl DSP408 { }) } - fn send_raw(&mut self, data: &[u8]) -> Result<(), DSPError> { - let _guard = self.lock.lock().unwrap(); - - let socket = self.socket.as_mut().ok_or(DSPError::NotConnected)?; - + fn send_raw(socket: &mut TcpStream, data: &[u8]) -> Result<(), DSPError> { socket.write_all(data)?; Ok(()) } - fn recv_raw(&mut self) -> Result, DSPError> { - let socket = self.socket.as_mut().ok_or(DSPError::NotConnected)?; - + fn recv_raw(socket: &mut TcpStream) -> Result, DSPError> { let mut buffer = vec![0u8; RECV_BUFSIZE]; match socket.read(&mut buffer) { @@ -101,10 +92,14 @@ impl DSP408 { } } - pub fn send_recv(&mut self, command: &[u8]) -> Result { - self.send_raw(command)?; + pub fn send_recv(&self, command: &[u8]) -> Result { + let mut socket_guard = self.socket.lock().unwrap(); - let raw = self.recv_raw()?; + let socket = socket_guard.as_mut().ok_or(DSPError::NotConnected)?; + + Self::send_raw(socket, command)?; + + let raw = Self::recv_raw(socket)?; let frame = extract_frame(&raw)?; @@ -117,6 +112,10 @@ impl DSP408 { self.connected } + pub fn connection(&self) -> &types::DeviceConnection { + &self.connection + } + pub fn state(&self) -> Result<&types::DSPState, DSPError> { self.state.as_ref().ok_or(DSPError::NotConnected) } @@ -144,16 +143,31 @@ impl DSP408 { stream.set_read_timeout(Some(SOCKET_TIMEOUT))?; - self.socket = Some(stream); + { + let mut socket = self.socket.lock().unwrap(); + *socket = Some(stream); + } let handshake = commands::build_handshake(); - self.send_raw(&handshake)?; + { + let mut socket = self.socket.lock().unwrap(); + + let socket = socket.as_mut().ok_or(DSPError::NotConnected)?; + + Self::send_raw(socket, &handshake)?; + } let start = std::time::Instant::now(); while start.elapsed() < SOCKET_TIMEOUT { - let raw = self.recv_raw()?; + let raw = { + let mut socket = self.socket.lock().unwrap(); + + let socket = socket.as_mut().ok_or(DSPError::NotConnected)?; + + Self::recv_raw(socket)? + }; let frame = extract_frame(&raw)?; let response = map_response(&frame)?; @@ -198,9 +212,10 @@ impl DSP408 { pub fn disconnect(&mut self) { let key = (self.connection.host, self.connection.device_id); - if let Some(mut socket) = self.socket.take() { - let _ = socket.write_all(&commands::build_disconnect()); + let mut socket_guard = self.socket.lock().unwrap(); + if let Some(mut socket) = socket_guard.take() { + let _ = socket.write_all(&commands::build_disconnect()); let _ = socket.shutdown(std::net::Shutdown::Both); }