one more mutex -> parking_lot

crates/dap/src/client.rs

@@ -152,8 +152,7 @@ impl DebugAdapterClient {
             arguments: Some(serialized_arguments),
         };
         self.transport_delegate
-            .add_pending_request(sequence_id, callback_tx)
-            .await;
+            .add_pending_request(sequence_id, callback_tx);
 
         log::debug!(
             "Client {} send `{}` request with sequence_id: {}",

crates/dap/src/transport.rs

@@ -64,7 +64,7 @@ impl TransportPipe {
     }
 }
 
-type Requests = Arc<Mutex<HashMap<u64, oneshot::Sender<Result<Response>>>>>;
+type Requests = Arc<parking_lot::Mutex<HashMap<u64, oneshot::Sender<Result<Response>>>>>;
 type LogHandlers = Arc<parking_lot::Mutex<SmallVec<[(LogKind, IoHandler); 2]>>>;
 
 pub enum Transport {
@@ -125,7 +125,6 @@ impl Transport {
 
 pub(crate) struct TransportDelegate {
     log_handlers: LogHandlers,
-    current_requests: Requests,
     pending_requests: Requests,
     transport: Transport,
     server_tx: Arc<Mutex<Option<Sender<Message>>>>,
@@ -142,7 +141,6 @@ impl TransportDelegate {
             transport,
             server_tx: Default::default(),
             log_handlers: Default::default(),
-            current_requests: Default::default(),
             pending_requests: Default::default(),
             _tasks: Vec::new(),
         };
@@ -202,7 +200,7 @@ impl TransportDelegate {
                     Ok(()) => {}
                     Err(e) => log::error!("Error handling debugger output: {e}"),
                 }
-                let mut pending_requests = pending_requests.lock().await;
+                let mut pending_requests = pending_requests.lock();
                 pending_requests.drain().for_each(|(_, request)| {
                     request
                         .send(Err(anyhow!("debugger shutdown unexpectedly")))
@@ -228,19 +226,9 @@ impl TransportDelegate {
                 }));
             }
 
-            let current_requests = self.current_requests.clone();
-            let pending_requests = self.pending_requests.clone();
             let log_handler = log_handler.clone();
             self._tasks.push(cx.background_spawn(async move {
-                match Self::handle_input(
-                    params.input,
-                    client_rx,
-                    current_requests,
-                    pending_requests,
-                    log_handler,
-                )
-                .await
-                {
+                match Self::handle_input(params.input, client_rx, log_handler).await {
                     Ok(()) => {}
                     Err(e) => log::error!("Error handling debugger input: {e}"),
                 }
@@ -255,12 +243,12 @@ impl TransportDelegate {
         Ok((server_rx, server_tx))
     }
 
-    pub(crate) async fn add_pending_request(
+    pub(crate) fn add_pending_request(
         &self,
         sequence_id: u64,
         request: oneshot::Sender<Result<Response>>,
     ) {
-        let mut pending_requests = self.pending_requests.lock().await;
+        let mut pending_requests = self.pending_requests.lock();
         pending_requests.insert(sequence_id, request);
     }
 
@@ -316,8 +304,6 @@ impl TransportDelegate {
     async fn handle_input<Stdin>(
         mut server_stdin: Stdin,
         client_rx: Receiver<Message>,
-        current_requests: Requests,
-        pending_requests: Requests,
         log_handlers: Option<LogHandlers>,
     ) -> Result<()>
     where
@@ -326,12 +312,6 @@ impl TransportDelegate {
         let result = loop {
             match client_rx.recv().await {
                 Ok(message) => {
-                    if let Message::Request(request) = &message {
-                        if let Some(sender) = current_requests.lock().await.remove(&request.seq) {
-                            pending_requests.lock().await.insert(request.seq, sender);
-                        }
-                    }
-
                     let command = match &message {
                         Message::Request(request) => Some(request.command.as_str()),
                         Message::Response(response) => Some(response.command.as_str()),
@@ -393,7 +373,8 @@ impl TransportDelegate {
                     return Ok(());
                 }
                 ConnectionResult::Result(Ok(Message::Response(res))) => {
-                    if let Some(tx) = pending_requests.lock().await.remove(&res.request_seq) {
+                    let tx = pending_requests.lock().remove(&res.request_seq);
+                    if let Some(tx) = tx {
                         if let Err(e) = tx.send(Self::process_response(res)) {
                             log::trace!("Did not send response `{:?}` for a cancelled", e);
                         }
@@ -547,17 +528,9 @@ impl TransportDelegate {
             server_tx.close();
         }
 
-        let mut current_requests = self.current_requests.lock().await;
-        let mut pending_requests = self.pending_requests.lock().await;
-
-        current_requests.clear();
-        pending_requests.clear();
-
+        self.pending_requests.lock().clear();
         self.transport.kill().await;
 
-        drop(current_requests);
-        drop(pending_requests);
-
         log::debug!("Shutdown client completed");
 
         anyhow::Ok(())