Replace State tokio::sync::Mutex by std::sync::RwLock

2 files changed, 33 insertions, 23 deletions

diff --git a/mumd/src/network/tcp.rs b/mumd/src/network/tcp.rs
index 1414318..c5eded2 100644
--- a/mumd/src/network/tcp.rs
+++ b/mumd/src/network/tcp.rs
@@ -10,7 +10,7 @@ use mumble_protocol::control::{msgs, ClientControlCodec, ControlCodec, ControlPa
 use mumble_protocol::crypt::ClientCryptState;
 use mumble_protocol::voice::VoicePacket;
 use mumble_protocol::{Clientbound, Serverbound};
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::RwLock};
 use std::convert::{Into, TryInto};
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -79,7 +79,7 @@ impl TcpEventQueue {
 }
 
 pub async fn handle(
-    state: Arc<Mutex<State>>,
+    state: Arc<RwLock<State>>,
     mut connection_info_receiver: watch::Receiver<Option<ConnectionInfo>>,
     crypt_state_sender: mpsc::Sender<ClientCryptState>,
     packet_sender: mpsc::UnboundedSender<ControlPacket<Serverbound>>,
@@ -103,7 +103,7 @@ pub async fn handle(
         .await?;
 
         // Handshake (omitting `Version` message for brevity)
-        let state_lock = state.lock().await;
+        let state_lock = state.read().unwrap();
         let username = state_lock.username().unwrap().to_string();
         let password = state_lock.password().map(|x| x.to_string());
         authenticate(&mut sink, username, password).await?;
@@ -241,7 +241,7 @@ async fn send_voice(
 }
 
 async fn listen(
-    state: Arc<Mutex<State>>,
+    state: Arc<RwLock<State>>,
     mut stream: TcpReceiver,
     crypt_state_sender: mpsc::Sender<ClientCryptState>,
     event_queue: TcpEventQueue,
@@ -260,7 +260,7 @@ async fn listen(
                 // We end up here if the login was rejected. We probably want
                 // to exit before that.
                 warn!("TCP stream gone");
-                state.lock().await.broadcast_phase(StatePhase::Disconnected);
+                state.read().unwrap().broadcast_phase(StatePhase::Disconnected);
                 break;
             }
         };
@@ -299,7 +299,7 @@ async fn listen(
                         .await;
                 }
                 event_queue.resolve(TcpEventData::Connected(Ok(&msg))).await;
-                let mut state = state.lock().await;
+                let mut state = state.write().unwrap();
                 let server = state.server_mut().unwrap();
                 server.parse_server_sync(*msg);
                 match &server.welcome_text {
@@ -323,24 +323,24 @@ async fn listen(
                 }
             }
             ControlPacket::UserState(msg) => {
-                state.lock().await.parse_user_state(*msg);
+                state.write().unwrap().parse_user_state(*msg);
             }
             ControlPacket::UserRemove(msg) => {
-                state.lock().await.remove_client(*msg);
+                state.write().unwrap().remove_client(*msg);
             }
             ControlPacket::ChannelState(msg) => {
                 debug!("Channel state received");
                 state
-                    .lock()
-                    .await
+                    .write()
+                    .unwrap()
                     .server_mut()
                     .unwrap()
                     .parse_channel_state(*msg); //TODO parse initial if initial
             }
             ControlPacket::ChannelRemove(msg) => {
                 state
-                    .lock()
-                    .await
+                    .write()
+                    .unwrap()
                     .server_mut()
                     .unwrap()
                     .parse_channel_remove(*msg);
@@ -356,8 +356,8 @@ async fn listen(
                         ..
                     } => {
                         state
-                            .lock()
-                            .await
+                            .read()
+                            .unwrap()
                             .audio_output()
                             .decode_packet_payload(
                                 VoiceStreamType::TCP,
diff --git a/mumd/src/network/udp.rs b/mumd/src/network/udp.rs
index 8614358..f24d4b4 100644
--- a/mumd/src/network/udp.rs
+++ b/mumd/src/network/udp.rs
@@ -9,7 +9,7 @@ use mumble_protocol::crypt::ClientCryptState;
 use mumble_protocol::ping::{PingPacket, PongPacket};
 use mumble_protocol::voice::VoicePacket;
 use mumble_protocol::Serverbound;
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::RwLock};
 use std::convert::TryFrom;
 use std::net::{Ipv6Addr, SocketAddr};
 use std::rc::Rc;
@@ -29,11 +29,15 @@ type UdpSender = SplitSink<UdpFramed<ClientCryptState>, (VoicePacket<Serverbound
 type UdpReceiver = SplitStream<UdpFramed<ClientCryptState>>;
 
 pub async fn handle(
-    state: Arc<Mutex<State>>,
+    state: Arc<RwLock<State>>,
     mut connection_info_receiver: watch::Receiver<Option<ConnectionInfo>>,
     mut crypt_state_receiver: mpsc::Receiver<ClientCryptState>,
 ) -> Result<(), UdpError> {
+<<<<<<< HEAD
     let receiver = state.lock().await.audio_input().receiver();
+=======
+    let receiver = state.read().unwrap().audio().input_receiver();
+>>>>>>> 48f0d38 (Replace State tokio::sync::Mutex by std::sync::RwLock)
 
     loop {
         let connection_info = 'data: loop {
@@ -49,7 +53,7 @@ pub async fn handle(
         let sink = Arc::new(Mutex::new(sink));
         let source = Arc::new(Mutex::new(source));
 
-        let phase_watcher = state.lock().await.phase_receiver();
+        let phase_watcher = state.read().unwrap().phase_receiver();
         let last_ping_recv = AtomicU64::new(0);
 
         run_until(
@@ -119,7 +123,7 @@ async fn new_crypt_state(
 }
 
 async fn listen(
-    state: Arc<Mutex<State>>,
+    state: Arc<RwLock<State>>,
     source: Arc<Mutex<UdpReceiver>>,
     last_ping_recv: &AtomicU64,
 ) {
@@ -136,8 +140,8 @@ async fn listen(
         match packet {
             VoicePacket::Ping { timestamp } => {
                 state
-                    .lock() //TODO clean up unnecessary lock by only updating phase if it should change
-                    .await
+                    .read()
+                    .unwrap()
                     .broadcast_phase(StatePhase::Connected(VoiceStreamType::UDP));
                 last_ping_recv.store(timestamp, Ordering::Relaxed);
             }
@@ -149,9 +153,15 @@ async fn listen(
                 ..
             } => {
                 state
+<<<<<<< HEAD
                     .lock() //TODO change so that we only have to lock audio and not the whole state
                     .await
                     .audio_output()
+=======
+                    .read()
+                    .unwrap()
+                    .audio()
+>>>>>>> 48f0d38 (Replace State tokio::sync::Mutex by std::sync::RwLock)
                     .decode_packet_payload(VoiceStreamType::UDP, session_id, payload);
             }
         }
@@ -159,7 +169,7 @@ async fn listen(
 }
 
 async fn send_pings(
-    state: Arc<Mutex<State>>,
+    state: Arc<RwLock<State>>,
     sink: Arc<Mutex<UdpSender>>,
     server_addr: SocketAddr,
     last_ping_recv: &AtomicU64,
@@ -173,8 +183,8 @@ async fn send_pings(
         if last_send.is_some() && last_send.unwrap() != last_recv {
             debug!("Sending TCP voice");
             state
-                .lock()
-                .await
+                .read()
+                .unwrap()
                 .broadcast_phase(StatePhase::Connected(VoiceStreamType::TCP));
         }
         match sink