refactor(network): remove unnecessary Mutex for reading and writing types

Author: rustaceanrob

src/network/mod.rs

@@ -11,7 +11,6 @@ use bitcoin::{
 use tokio::{
     io::{AsyncRead, AsyncWrite},
     net::TcpStream,
-    sync::Mutex,
     time::Instant,
 };
 
@@ -36,8 +35,8 @@ pub const RUST_BITCOIN_VERSION: &str = "0.32.4";
 const THIRTY_MINS: u64 = 60 * 30;
 const CONNECTION_TIMEOUT: u64 = 2;
 
-pub(crate) type StreamReader = Mutex<Box<dyn AsyncRead + Send + Unpin>>;
-pub(crate) type StreamWriter = Mutex<Box<dyn AsyncWrite + Send + Unpin>>;
+pub(crate) type StreamReader = Box<dyn AsyncRead + Send + Sync + Unpin>;
+pub(crate) type StreamWriter = Box<dyn AsyncWrite + Send + Sync + Unpin>;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub(crate) struct PeerId(pub(crate) u32);
@@ -134,7 +133,7 @@ impl ConnectionType {
         match timeout {
             Ok(stream) => {
                 let (reader, writer) = stream.into_split();
-                Ok((Mutex::new(Box::new(reader)), Mutex::new(Box::new(writer))))
+                Ok((Box::new(reader), Box::new(writer)))
             }
             Err(_) => Err(PeerError::ConnectionFailed),
         }

src/network/parsers.rs

@@ -24,9 +24,9 @@ impl V1MessageParser {
     }
 
     async fn do_read_message(&mut self) -> Result<Option<NetworkMessage>, PeerReadError> {
-        let mut stream = self.stream.lock().await;
         let mut message_buf = vec![0_u8; 24];
-        let _ = stream
+        let _ = self
+            .stream
             .read_exact(&mut message_buf)
             .await
             .map_err(|_| PeerReadError::ReadBuffer)?;
@@ -42,7 +42,8 @@ impl V1MessageParser {
             return Err(PeerReadError::Deserialization);
         }
         let mut contents_buf = vec![0_u8; header.length as usize];
-        let _ = stream
+        let _ = self
+            .stream
             .read_exact(&mut contents_buf)
             .await
             .map_err(|_| PeerReadError::ReadBuffer)?;
@@ -70,9 +71,9 @@ impl V2MessageParser {
     }
 
     async fn do_read_message(&mut self) -> Result<Option<NetworkMessage>, PeerReadError> {
-        let mut stream = self.stream.lock().await;
         let mut len_buf = [0; 3];
-        let _ = stream
+        let _ = self
+            .stream
             .read_exact(&mut len_buf)
             .await
             .map_err(|_| PeerReadError::ReadBuffer)?;
@@ -81,7 +82,8 @@ impl V2MessageParser {
             return Err(PeerReadError::TooManyMessages);
         }
         let mut response_message = vec![0; message_len];
-        let _ = stream
+        let _ = self
+            .stream
             .read_exact(&mut response_message)
             .await
             .map_err(|_| PeerReadError::ReadBuffer)?;

src/network/peer.rs

@@ -73,21 +73,16 @@ impl Peer {
 
     pub async fn run(
         &mut self,
-        reader: StreamReader,
-        writer: StreamWriter,
+        mut reader: StreamReader,
+        mut writer: StreamWriter,
     ) -> Result<(), PeerError> {
         let start_time = Instant::now();
         let (tx, mut rx) = mpsc::channel(32);
-        let mut lock = writer.lock().await;
-        let writer = lock.deref_mut();
-
         // If a peer signals for V2 we will use it, otherwise just use plaintext.
         let (message_mutex, mut peer_reader) = if self.services.has(ServiceFlags::P2P_V2) {
-            let mut lock = reader.lock().await;
-            let read_lock = lock.deref_mut();
             let handshake_result = tokio::time::timeout(
                 Duration::from_secs(HANDSHAKE_TIMEOUT),
-                self.try_handshake(writer, read_lock),
+                self.try_handshake(&mut writer, &mut reader),
             )
             .await
             .map_err(|_| PeerError::HandshakeFailed)?;
@@ -101,7 +96,6 @@ impl Peer {
             let (decryptor, encryptor) = handshake_result?;
             let message_mutex: MutexMessageGenerator =
                 Mutex::new(Box::new(V2OutboundMessage::new(self.network, encryptor)));
-            drop(lock);
             let reader = Reader::new(V2MessageParser::new(reader, decryptor), tx);
             (message_mutex, reader)
         } else {
@@ -114,7 +108,7 @@ impl Peer {
         let mut message_lock = message_mutex.lock().await;
         let outbound_messages = message_lock.deref_mut();
         let message = outbound_messages.version_message(None)?;
-        self.write_bytes(writer, message).await?;
+        self.write_bytes(&mut writer, message).await?;
         self.message_counter.sent_version();
         let read_handle = tokio::spawn(async move {
             peer_reader
@@ -147,7 +141,7 @@ impl Peer {
                     if let Ok(peer_message) = peer_message {
                         match peer_message {
                             Some(message) => {
-                                match self.handle_peer_message(message, writer, outbound_messages).await {
+                                match self.handle_peer_message(message, &mut writer, outbound_messages).await {
                                     Ok(()) => continue,
                                     Err(e) => {
                                         match e {
@@ -166,7 +160,7 @@ impl Peer {
                 node_message = self.main_thread_recv.recv() => {
                     match node_message {
                         Some(message) => {
-                            match self.main_thread_request(message, writer, outbound_messages).await {
+                            match self.main_thread_request(message, &mut writer, outbound_messages).await {
                                 Ok(()) => continue,
                                 Err(e) => {
                                     match e {