diff --git a/crates/project/src/lsp_store.rs b/crates/project/src/lsp_store.rs index fee9666ec7..593c1dd66b 100644 --- a/crates/project/src/lsp_store.rs +++ b/crates/project/src/lsp_store.rs @@ -14703,8 +14703,13 @@ impl PartialEq for LanguageServerPromptRequest { #[derive(Clone, Debug, PartialEq)] pub enum LanguageServerLogType { Log(MessageType), - Trace { verbose_info: Option }, - Rpc { received: bool }, + Trace { + verbose_info: Option, + }, + Rpc { + received: bool, + elapsed: Option, + }, } impl LanguageServerLogType { @@ -14731,14 +14736,19 @@ impl LanguageServerLogType { verbose_info: verbose_info.to_owned(), }) } - Self::Rpc { received } => { + Self::Rpc { received, elapsed } => { let kind = if *received { proto::rpc_message::Kind::Received } else { proto::rpc_message::Kind::Sent }; let kind = kind as i32; - proto::language_server_log::LogType::Rpc(proto::RpcMessage { kind }) + let elapsed_nanos = + elapsed.map(|elapsed| u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)); + proto::language_server_log::LogType::Rpc(proto::RpcMessage { + kind, + elapsed_nanos, + }) } } } @@ -14765,6 +14775,7 @@ impl LanguageServerLogType { rpc_message::Kind::Received => true, rpc_message::Kind::Sent => false, }, + elapsed: message.elapsed_nanos.map(Duration::from_nanos), }, } } diff --git a/crates/project/src/lsp_store/log_store.rs b/crates/project/src/lsp_store/log_store.rs index 0cfe3c14cf..417bf82b70 100644 --- a/crates/project/src/lsp_store/log_store.rs +++ b/crates/project/src/lsp_store/log_store.rs @@ -1,4 +1,8 @@ -use std::{collections::VecDeque, sync::Arc}; +use std::{ + collections::VecDeque, + sync::Arc, + time::{Duration, Instant}, +}; use collections::HashMap; use futures::{StreamExt, channel::mpsc}; @@ -7,16 +11,16 @@ use gpui::{ }; use lsp::{ IoKind, LanguageServer, LanguageServerId, LanguageServerName, LanguageServerSelector, - MessageType, TraceValue, + MessageType, RequestId, TraceValue, }; use rpc::proto; +use serde::Deserialize; use settings::WorktreeId; use crate::{LanguageServerLogType, LspStore, Project, ProjectItem as _}; -const SEND_LINE: &str = "\n// Send:"; -const RECEIVE_LINE: &str = "\n// Receive:"; const MAX_STORED_LOG_ENTRIES: usize = 2000; +const MAX_PENDING_REQUESTS: usize = MAX_STORED_LOG_ENTRIES; pub fn init(on_headless_host: bool, cx: &mut App) -> Entity { let log_store = cx.new(|cx| LogStore::new(on_headless_host, cx)); @@ -43,7 +47,7 @@ pub struct LogStore { on_headless_host: bool, projects: HashMap, ProjectState>, pub language_servers: HashMap, - io_tx: mpsc::UnboundedSender<(LanguageServerId, IoKind, String)>, + io_tx: mpsc::UnboundedSender<(LanguageServerId, IoKind, String, Instant)>, } struct ProjectState { @@ -188,14 +192,127 @@ impl LanguageServerKind { pub struct LanguageServerRpcState { pub rpc_messages: VecDeque, last_message_kind: Option, + request_tracker: RpcRequestTracker, } -#[derive(Debug, Copy, Clone, PartialEq, Eq)] +#[derive(Debug, Default)] +struct RpcRequestTracker { + pending_requests: HashMap, +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct PendingRequestKey { + kind: MessageKind, + id: RequestId, +} + +#[derive(Deserialize)] +struct RpcEnvelope<'a> { + id: Option, + method: Option<&'a str>, +} + +#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)] enum MessageKind { Send, Receive, } +impl MessageKind { + fn opposite(self) -> Self { + match self { + Self::Send => Self::Receive, + Self::Receive => Self::Send, + } + } +} + +impl RpcRequestTracker { + fn observe( + &mut self, + kind: MessageKind, + message: &str, + observed_at: Instant, + ) -> Option { + let envelope = serde_json::from_str::(message).ok()?; + let id = envelope.id?; + if envelope.method.is_some() { + self.insert(PendingRequestKey { kind, id }, observed_at); + None + } else { + self.pending_requests + .remove(&PendingRequestKey { + kind: kind.opposite(), + id, + }) + .and_then(|started_at| observed_at.checked_duration_since(started_at)) + } + } + + fn insert(&mut self, key: PendingRequestKey, observed_at: Instant) { + if self.pending_requests.len() >= MAX_PENDING_REQUESTS + && !self.pending_requests.contains_key(&key) + && let Some(oldest_key) = self + .pending_requests + .iter() + .min_by_key(|(_, started_at)| **started_at) + .map(|(key, _)| key.clone()) + { + self.pending_requests.remove(&oldest_key); + } + self.pending_requests.insert(key, observed_at); + } +} + +#[cfg(feature = "test-support")] +#[derive(Default)] +pub struct TestRpcRequestTracker(RpcRequestTracker); + +#[cfg(feature = "test-support")] +impl TestRpcRequestTracker { + pub fn new() -> Self { + Self::default() + } + + pub fn observe( + &mut self, + received: bool, + message: &str, + observed_at: Instant, + ) -> Option { + let kind = if received { + MessageKind::Receive + } else { + MessageKind::Send + }; + self.0.observe(kind, message, observed_at) + } + + pub fn pending_request_count(&self) -> usize { + self.0.pending_requests.len() + } + + pub fn max_pending_requests() -> usize { + MAX_PENDING_REQUESTS + } +} + +enum RpcTiming { + ObservedAt(Instant), + Forwarded(Option), +} + +fn format_duration(duration: Duration) -> String { + let seconds = duration.as_secs_f64(); + if seconds < 0.001 { + format!("{:.0}µs", seconds * 1_000_000.0) + } else if seconds < 1.0 { + format!("{:.1}ms", seconds * 1_000.0) + } else { + format!("{seconds:.2}s") + } +} + #[derive(Clone, Copy, Debug, Default, PartialEq)] pub enum LogKind { Rpc, @@ -227,10 +344,10 @@ impl LogStore { io_tx, }; cx.spawn(async move |log_store, cx| { - while let Some((server_id, io_kind, message)) = io_rx.next().await { + while let Some((server_id, io_kind, message, observed_at)) = io_rx.next().await { if let Some(log_store) = log_store.upgrade() { log_store.update(cx, |log_store, cx| { - log_store.on_io(server_id, io_kind, &message, cx); + log_store.on_io(server_id, io_kind, &message, observed_at, cx); }); } } @@ -331,13 +448,19 @@ impl LogStore { cx, ); } - crate::LanguageServerLogType::Rpc { received } => { + crate::LanguageServerLogType::Rpc { received, elapsed } => { let kind = if *received { MessageKind::Receive } else { MessageKind::Send }; - log_store.add_language_server_rpc(*id, kind, message, cx); + log_store.add_language_server_rpc( + *id, + kind, + message, + RpcTiming::Forwarded(*elapsed), + cx, + ); } } } @@ -400,8 +523,9 @@ impl LogStore { let io_tx = self.io_tx.clone(); let server_id = server.server_id(); server_state.io_logs_subscription = Some(server.on_io(move |io_kind, message| { + let observed_at = Instant::now(); io_tx - .unbounded_send((server_id, io_kind, message.to_string())) + .unbounded_send((server_id, io_kind, message.to_string(), observed_at)) .ok(); })); } @@ -519,6 +643,7 @@ impl LogStore { language_server_id: LanguageServerId, kind: MessageKind, message: &str, + timing: RpcTiming, cx: &mut Context<'_, Self>, ) { let store_logs = !self.on_headless_host; @@ -529,43 +654,54 @@ impl LogStore { return; }; + let elapsed = match timing { + RpcTiming::ObservedAt(observed_at) => { + state.request_tracker.observe(kind, message, observed_at) + } + RpcTiming::Forwarded(elapsed) => elapsed, + }; + let received = kind == MessageKind::Receive; - let rpc_log_lines = &mut state.rpc_messages; - if state.last_message_kind != Some(kind) { + let direction = if received { "Receive" } else { "Send" }; + let mut header = None; + if state.last_message_kind != Some(kind) || elapsed.is_some() { + header = Some(match elapsed { + Some(elapsed) => format!("\n// {direction} (took {}):", format_duration(elapsed)), + None => format!("\n// {direction}:"), + }); + } + state.last_message_kind = Some(kind); + + if store_logs { + let rpc_log_lines = &mut state.rpc_messages; while rpc_log_lines.len() + 1 >= MAX_STORED_LOG_ENTRIES { rpc_log_lines.pop_front(); } - let line_before_message = match kind { - MessageKind::Send => SEND_LINE, - MessageKind::Receive => RECEIVE_LINE, - }; - if store_logs { - rpc_log_lines.push_back(RpcMessage { - message: line_before_message.to_string(), - }); - } - // Do not send a synthetic message over the wire, it will be derived from the actual RPC message - cx.emit(Event::NewServerLogEntry { - id: language_server_id, - kind: LanguageServerLogType::Rpc { received }, - text: line_before_message.to_string(), + let message = message.trim(); + rpc_log_lines.push_back(RpcMessage { + message: match &header { + Some(header) => format!("{header}\n{message}"), + None => message.to_owned(), + }, }); } - while rpc_log_lines.len() + 1 >= MAX_STORED_LOG_ENTRIES { - rpc_log_lines.pop_front(); - } - - if store_logs { - rpc_log_lines.push_back(RpcMessage { - message: message.trim().to_owned(), + if let Some(header) = header { + // Do not send a synthetic message over the wire, it will be derived from the actual RPC message + cx.emit(Event::NewServerLogEntry { + id: language_server_id, + kind: LanguageServerLogType::Rpc { + received, + elapsed: None, + }, + text: header, }); } self.emit_event( Event::NewServerLogEntry { id: language_server_id, - kind: LanguageServerLogType::Rpc { received }, + kind: LanguageServerLogType::Rpc { received, elapsed }, text: message.to_owned(), }, cx, @@ -614,6 +750,7 @@ impl LogStore { .get_or_insert_with(|| LanguageServerRpcState { rpc_messages: VecDeque::with_capacity(MAX_STORED_LOG_ENTRIES), last_message_kind: None, + request_tracker: RpcRequestTracker::default(), }); Some(rpc_state) } @@ -641,6 +778,7 @@ impl LogStore { language_server_id: LanguageServerId, io_kind: IoKind, message: &str, + observed_at: Instant, cx: &mut Context, ) -> Option<()> { let is_received = match io_kind { @@ -658,7 +796,13 @@ impl LogStore { MessageKind::Send }; - self.add_language_server_rpc(language_server_id, kind, message, cx); + self.add_language_server_rpc( + language_server_id, + kind, + message, + RpcTiming::ObservedAt(observed_at), + cx, + ); cx.notify(); Some(()) } diff --git a/crates/project/tests/integration/lsp_store.rs b/crates/project/tests/integration/lsp_store.rs index 100042f5d9..34f81fdbbe 100644 --- a/crates/project/tests/integration/lsp_store.rs +++ b/crates/project/tests/integration/lsp_store.rs @@ -1,11 +1,17 @@ -use std::path::Path; +use std::{ + path::Path, + time::{Duration, Instant}, +}; use fs::FakeFs; use futures::StreamExt; use gpui::TestAppContext; use language::{CodeLabel, FakeLspAdapter, HighlightId, rust_lang}; use lsp::Uri; -use project::{Project, lsp_store::*}; +use project::{ + Project, + lsp_store::{log_store::TestRpcRequestTracker, *}, +}; use serde_json::json; use util::path; @@ -90,6 +96,133 @@ async fn test_removing_invisible_worktree_cleans_reused_lsp_bookkeeping(cx: &mut }); } +#[test] +fn test_rpc_request_tracker_distinguishes_request_directions() { + let mut tracker = TestRpcRequestTracker::new(); + let started_at = Instant::now(); + + assert_eq!( + tracker.observe( + false, + r#"{"jsonrpc":"2.0","id":1,"method":"textDocument/hover"}"#, + started_at, + ), + None + ); + assert_eq!( + tracker.observe( + true, + r#"{"jsonrpc":"2.0","id":1,"method":"workspace/configuration"}"#, + started_at + Duration::from_millis(10), + ), + None + ); + assert_eq!( + tracker.observe( + false, + r#"{"jsonrpc":"2.0","id":1,"result":[]}"#, + started_at + Duration::from_millis(30), + ), + Some(Duration::from_millis(20)) + ); + assert_eq!( + tracker.observe( + true, + r#"{"jsonrpc":"2.0","id":1,"result":null}"#, + started_at + Duration::from_millis(50), + ), + Some(Duration::from_millis(50)) + ); +} + +#[test] +fn test_rpc_request_tracker_decodes_ids_and_times_cancelled_requests() { + let mut tracker = TestRpcRequestTracker::new(); + let started_at = Instant::now(); + + tracker.observe( + true, + r#"{"jsonrpc":"2.0","id":"foo\u002fbar","method":"workspace/configuration"}"#, + started_at, + ); + assert_eq!( + tracker.observe( + false, + r#"{"jsonrpc":"2.0","id":"foo/bar","result":[]}"#, + started_at + Duration::from_millis(25), + ), + Some(Duration::from_millis(25)) + ); + + tracker.observe( + false, + r#"{"jsonrpc":"2.0","id":7,"method":"textDocument/hover"}"#, + started_at, + ); + tracker.observe( + false, + r#"{"jsonrpc":"2.0","method":"$/cancelRequest","params":{"id":7}}"#, + started_at + Duration::from_millis(1), + ); + assert_eq!(tracker.pending_request_count(), 1); + assert_eq!( + tracker.observe( + true, + r#"{"jsonrpc":"2.0","id":7,"error":{"code":-32800,"message":"Request was cancelled"}}"#, + started_at + Duration::from_millis(10), + ), + Some(Duration::from_millis(10)) + ); + assert_eq!(tracker.pending_request_count(), 0); +} + +#[test] +fn test_rpc_request_tracker_bounds_unanswered_requests() { + let mut tracker = TestRpcRequestTracker::new(); + let started_at = Instant::now(); + let max_pending_requests = TestRpcRequestTracker::max_pending_requests(); + + for id in 0..=max_pending_requests { + tracker.observe( + false, + &format!(r#"{{"jsonrpc":"2.0","id":{id},"method":"textDocument/hover"}}"#), + started_at + Duration::from_nanos(id as u64), + ); + } + + assert_eq!(tracker.pending_request_count(), max_pending_requests); + assert_eq!( + tracker.observe( + true, + r#"{"jsonrpc":"2.0","id":0,"result":null}"#, + started_at + Duration::from_secs(1), + ), + None + ); + assert!( + tracker + .observe( + true, + r#"{"jsonrpc":"2.0","id":1,"result":null}"#, + started_at + Duration::from_secs(1), + ) + .is_some() + ); +} + +#[test] +fn test_rpc_log_duration_proto_roundtrip() { + let log_type = LanguageServerLogType::Rpc { + received: true, + elapsed: Some(Duration::from_micros(1234)), + }; + + assert_eq!( + LanguageServerLogType::from_proto(log_type.to_proto()), + log_type + ); +} + #[test] fn test_glob_literal_prefix() { assert_eq!(glob_literal_prefix(Path::new("**/*.js")), Path::new("")); diff --git a/crates/proto/proto/lsp.proto b/crates/proto/proto/lsp.proto index cf7c2b9766..2e87e1ec12 100644 --- a/crates/proto/proto/lsp.proto +++ b/crates/proto/proto/lsp.proto @@ -757,6 +757,7 @@ message TraceMessage { message RpcMessage { Kind kind = 1; + optional uint64 elapsed_nanos = 2; enum Kind { RECEIVED = 0;