From 5a71098a257c2346088c442cbdd6f784c632e83d Mon Sep 17 00:00:00 2001 From: Christian Findlay <16697547+MelbourneDeveloper@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:14:30 +1100 Subject: [PATCH] Stop releases memory steps its target holds; a capture ends with its stream A direct runtime reads the live heap, so a memory step can wait on its target indefinitely. Two places let that wedge an editor: - memory/stop answered while a step on the session was still in flight. An editor's interval autopilot whose capture was in flight at stop stayed busy until that request answered, and never captured the next session. - the .NET engine waited out its 90 s deadline after the event stream had ended, because the target took the session with it. A target that died mid-capture therefore held its step for 90 s, and then failed as a generic "Failed" instead of PROCESS_NOT_FOUND. Stop now ends every step in flight on its session the way cancellation does (dropping the future terminates its tools, the target keeps running), waits for each to answer, then releases the session ([PROF-MEMORY-STOP]). The engine stops waiting the moment the stream ends, stops only a session still streaming, and reports any failure of a command whose target is gone as ProcessNotFound ([PROF-DOTNET-HEAP-CAPTURE], [PROF-DOTNET-GC]). The memory steps move out of service.rs into service/memory_steps.rs, which brings service.rs under the 500-line limit. --- .../tests/dotnet_held_e2e.rs | 228 ++++++++++++++++++ .../lsp-profiler-dotnet/tests/support/mod.rs | 4 + crates/lsp-profiler/src/memory/gate.rs | 81 +++++++ crates/lsp-profiler/src/memory/mod.rs | 4 + crates/lsp-profiler/src/memory/session.rs | 43 ++++ crates/lsp-profiler/src/service.rs | 151 +----------- .../lsp-profiler/src/service/memory_steps.rs | 191 +++++++++++++++ docs/specs/LSP-PROFILER-SPEC.md | 10 +- .../FailureTests.cs | 82 +++++++ .../SessionRunTests.cs | 107 ++++++++ dotnet/LspProfiler.DotNet.Engine/Errors.cs | 15 +- dotnet/LspProfiler.DotNet.Engine/GcCommand.cs | 7 +- dotnet/LspProfiler.DotNet.Engine/GcWatch.cs | 55 ++++- .../LspProfiler.DotNet.Engine/HeapCapture.cs | 4 +- dotnet/LspProfiler.DotNet.Engine/Program.cs | 9 +- 15 files changed, 833 insertions(+), 158 deletions(-) create mode 100644 crates/lsp-profiler-dotnet/tests/dotnet_held_e2e.rs create mode 100644 crates/lsp-profiler/src/memory/gate.rs create mode 100644 crates/lsp-profiler/src/service/memory_steps.rs create mode 100644 dotnet/LspProfiler.DotNet.Engine.Tests/FailureTests.cs create mode 100644 dotnet/LspProfiler.DotNet.Engine.Tests/SessionRunTests.cs diff --git a/crates/lsp-profiler-dotnet/tests/dotnet_held_e2e.rs b/crates/lsp-profiler-dotnet/tests/dotnet_held_e2e.rs new file mode 100644 index 0000000..006bd98 --- /dev/null +++ b/crates/lsp-profiler-dotnet/tests/dotnet_held_e2e.rs @@ -0,0 +1,228 @@ +//! Memory steps their target holds, end to end. Tests [PROF-MEMORY-STOP]. +//! +//! A frozen target never answers the engine, so a snapshot or a diff of it +//! waits for as long as the target stays frozen: the kernel's own stop state +//! (`kill -STOP`, confirmed by `ps`) holds the step, never a timer. Stopping +//! the session must still answer, release every step it holds and leave the +//! target alive; stopping one session must leave another's steps alone. The +//! guard only bounds a regression, which would otherwise hang. + +#![cfg(unix)] +#![allow( + clippy::allow_attributes, + clippy::expect_used, + clippy::unwrap_used, + clippy::panic, + clippy::indexing_slicing, + missing_docs +)] + +#[path = "../../../test-support/stopped_process.rs"] +mod stopped_process; +mod support; + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use lsp_profiler::wire::methods; +use lsp_profiler::{ProfilerError, ProfilerService}; +use serde_json::{json, Value}; +use support::{call, completed, service, Fixture}; +use tokio::task::JoinHandle; + +const FIXTURE: &str = "HotLeakApp"; +/// Bounds a regression: a released step answers at once, a held one never does. +const GUARD: Duration = Duration::from_secs(60); +/// The code every memory failure that is not an attach failure answers with ([PROF-ERRORS]). +const OTHER: i64 = -32000; + +type Step = JoinHandle>; + +/// A tracked fixture frozen in the kernel's stop state, and its session. +struct Held { + fixture: Fixture, + frozen: stopped_process::Stopped, + id: String, +} + +impl Held { + /// Track a fresh fixture, then freeze it. + async fn start(service: &ProfilerService) -> Self { + let fixture = Fixture::start(FIXTURE); + let started = call( + service, + methods::MEMORY_START, + json!({ "pid": fixture.pid }), + ) + .await; + let id = started["memory_session_id"] + .as_str() + .expect("session id") + .to_owned(); + let frozen = stopped_process::suspend(fixture.pid, GUARD).await; + Self { + fixture, + frozen, + id, + } + } + + /// Let the target run again. + fn thaw(self) -> Fixture { + self.frozen.resume(); + self.fixture + } + + fn session(&self) -> Value { + json!({ "memory_session_id": self.id }) + } + + /// Where the session's graphs live ([PROF-DOTNET-MEMORY]). + fn graphs(&self, dir: &Path) -> PathBuf { + dir.join(format!("heap-{}", self.id)) + } + + /// Start `method`; settles once the step has made the session's graph + /// directory, so it is past its session check and waits on the target. + async fn in_flight( + &self, + service: &Arc, + dir: &Path, + method: &'static str, + ) -> Step { + let (service, session) = (Arc::clone(service), self.session()); + let step = tokio::spawn(async move { service.handle(method, session).await }); + let graphs = self.graphs(dir); + tokio::time::timeout(GUARD, async { + while !graphs.exists() { + tokio::task::yield_now().await; + } + }) + .await + .expect("the step reaches its frozen target"); + step + } + + /// The target outlived everything the profiler did to it. + fn assert_alive(&mut self) { + assert!( + self.fixture + .child + .try_wait() + .expect("read the target's state") + .is_none(), + "stopping a session never ends its target" + ); + } +} + +/// The step's answer, which a released step gives at once. +async fn answer(step: Step) -> Result { + tokio::time::timeout(GUARD, step) + .await + .expect("the released step answers") + .expect("the step's task completes") +} + +/// A step stop released: refused as stopped, naming its session. +fn assert_released(answer: Result, id: &str) { + let error = answer.expect_err("a released step is refused, never completed"); + assert_eq!(error.code(), OTHER, "{error}"); + assert_eq!( + error.to_string(), + format!("memory session {id} stopped before the step finished") + ); +} + +/// Stop answers `Ack`; afterwards the session's graphs are gone and every step is refused. +async fn assert_stopped(service: &ProfilerService, dir: &Path, held: &Held) { + let stopped = call(service, methods::MEMORY_STOP, held.session()).await; + assert_eq!(completed(&stopped)["outcome"]["kind"], "Ack", "{stopped}"); + assert!( + !held.graphs(dir).exists(), + "stop deletes the session's graphs" + ); + let later = service + .handle(methods::MEMORY_SNAPSHOT, held.session()) + .await; + let error = later.expect_err("a stopped session refuses every step"); + assert_eq!(error.code(), OTHER, "{error}"); + assert_eq!( + error.to_string(), + format!("unknown memory session: {}", held.id) + ); +} + +/// The thawed target is whole: a new session captures its heap at once. +async fn assert_capturable(service: &ProfilerService, target: &Fixture) { + let started = call(service, methods::MEMORY_START, json!({ "pid": target.pid })).await; + let session = json!({ "memory_session_id": started["memory_session_id"] }); + let snapshot = call(service, methods::MEMORY_SNAPSHOT, session.clone()).await; + assert_eq!( + completed(&snapshot)["outcome"]["kind"], + "Snapshot", + "{snapshot}" + ); + let stopped = call(service, methods::MEMORY_STOP, session).await; + assert_eq!(completed(&stopped)["outcome"]["kind"], "Ack", "{stopped}"); +} + +async fn stop_releases(method: &'static str) { + let dir = tempfile::tempdir().expect("tempdir"); + let service = Arc::new(service(dir.path().to_path_buf())); + let mut held = Held::start(&service).await; + let step = held.in_flight(&service, dir.path(), method).await; + assert!(!step.is_finished(), "a frozen target holds the {method}"); + + assert_stopped(&service, dir.path(), &held).await; + + assert_released(answer(step).await, &held.id); + held.assert_alive(); + assert_capturable(&service, &held.thaw()).await; + service.shutdown().await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn stop_releases_a_snapshot_its_frozen_target_holds() { + stop_releases(methods::MEMORY_SNAPSHOT).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn stop_releases_a_diff_its_frozen_target_holds() { + stop_releases(methods::MEMORY_DIFF).await; +} + +/// Each session's stop releases its own steps only. +#[tokio::test(flavor = "multi_thread")] +async fn stopping_one_session_leaves_another_sessions_held_step_held() { + let dir = tempfile::tempdir().expect("tempdir"); + let service = Arc::new(service(dir.path().to_path_buf())); + let mut first = Held::start(&service).await; + let mut second = Held::start(&service).await; + let first_step = first + .in_flight(&service, dir.path(), methods::MEMORY_SNAPSHOT) + .await; + let second_step = second + .in_flight(&service, dir.path(), methods::MEMORY_DIFF) + .await; + + assert_stopped(&service, dir.path(), &first).await; + assert_released(answer(first_step).await, &first.id); + assert!( + !second_step.is_finished(), + "the other session's frozen target still holds its step" + ); + assert!( + second.graphs(dir.path()).exists(), + "the other session keeps its graphs" + ); + + assert_stopped(&service, dir.path(), &second).await; + assert_released(answer(second_step).await, &second.id); + first.assert_alive(); + second.assert_alive(); + assert_capturable(&service, &first.thaw()).await; + assert_capturable(&service, &second.thaw()).await; + service.shutdown().await; +} diff --git a/crates/lsp-profiler-dotnet/tests/support/mod.rs b/crates/lsp-profiler-dotnet/tests/support/mod.rs index 28963d3..309ef6a 100644 --- a/crates/lsp-profiler-dotnet/tests/support/mod.rs +++ b/crates/lsp-profiler-dotnet/tests/support/mod.rs @@ -155,6 +155,10 @@ pub async fn call(service: &ProfilerService, method: &str, params: Value) -> Val /// the whole session, so its entry point is never hot, and the worker's hot /// function holds most of the share. Counting the idle thread caps that share /// at half, since the runtime samples both threads every tick. +#[allow( + dead_code, + reason = "only the CPU suites sample; the held-step suite's crate never calls this" +)] pub fn assert_only_busy_threads_count(report: &Value, entry_point: &str, hot_function: &str) { let functions = report["hot_functions"].as_array().expect("hot_functions"); let named = |prefix: &str| { diff --git a/crates/lsp-profiler/src/memory/gate.rs b/crates/lsp-profiler/src/memory/gate.rs new file mode 100644 index 0000000..9389f71 --- /dev/null +++ b/crates/lsp-profiler/src/memory/gate.rs @@ -0,0 +1,81 @@ +//! The steps in flight on one memory session, and stopping them. Implements +//! [PROF-MEMORY-STOP]. +//! +//! A step can wait on its target indefinitely: a direct runtime reads the live +//! heap, which a frozen, paused or dying program never yields. Stopping the +//! session must not wait on that. It ends every step in flight the way +//! cancellation does, by dropping the step's future, which terminates each tool +//! the step started ([PROF-HOST-CANCEL]), then waits until every one has +//! answered, so nothing touches the session after stop removes its state. + +use std::future::Future; +use std::sync::Arc; + +use tokio::sync::{watch, OwnedRwLockReadGuard, RwLock}; + +use super::MemoryError; + +/// Admits steps to one session until it stops. +#[derive(Debug, Clone)] +pub(crate) struct StepGate { + /// Held shared by every step in flight, and exclusively by a stop. + steps: Arc>, + /// True once the session is stopping. + stopping: Arc>, +} + +/// One admitted step: a stop waits for it, and ends it once the stop begins. +#[derive(Debug)] +pub(crate) struct StepPass { + /// Keeps a stop waiting until this step has answered. + _held: OwnedRwLockReadGuard<()>, + /// Observes the session starting to stop. + stopping: watch::Receiver, + /// The session the step belongs to, for the error that ends it. + session_id: String, +} + +impl StepGate { + /// A gate admitting steps. + pub(crate) fn new() -> Self { + Self { + steps: Arc::new(RwLock::new(())), + stopping: Arc::new(watch::channel(false).0), + } + } + + /// Admit a step on `session_id`, or refuse it once the session is stopping. + pub(crate) async fn admit(&self, session_id: &str) -> Result { + let held = Arc::clone(&self.steps).read_owned().await; + let stopping = self.stopping.subscribe(); + if *stopping.borrow() { + return Err(MemoryError::Stopped(session_id.to_owned())); + } + Ok(StepPass { + _held: held, + stopping, + session_id: session_id.to_owned(), + }) + } + + /// End every step in flight and wait until each one has answered. + pub(crate) async fn stop(&self) { + let _ = self.stopping.send_replace(true); + drop(self.steps.write().await); + } +} + +impl StepPass { + /// Run `step` until it answers or its session starts stopping, whichever is first. + pub(crate) async fn run>( + mut self, + step: impl Future>, + ) -> Result { + let session_id = self.session_id.clone(); + tokio::select! { + biased; + _ = self.stopping.wait_for(|stopping| *stopping) => Err(MemoryError::Stopped(session_id).into()), + answer = step => answer, + } + } +} diff --git a/crates/lsp-profiler/src/memory/mod.rs b/crates/lsp-profiler/src/memory/mod.rs index a82037c..e3a0474 100644 --- a/crates/lsp-profiler/src/memory/mod.rs +++ b/crates/lsp-profiler/src/memory/mod.rs @@ -10,6 +10,7 @@ pub mod compare; pub mod diagnostics; +mod gate; pub mod heapprofile; pub mod leaks; pub mod session; @@ -58,6 +59,9 @@ pub enum MemoryError { /// No such memory session. #[error("unknown memory session: {0}")] UnknownSession(String), + /// The session stopped before the step answered ([PROF-MEMORY-STOP]). + #[error("memory session {0} stopped before the step finished")] + Stopped(String), /// The runtime does not support the requested step. #[error("{0} is not supported by this runtime")] Unsupported(String), diff --git a/crates/lsp-profiler/src/memory/session.rs b/crates/lsp-profiler/src/memory/session.rs index 52b3343..b9e72b4 100644 --- a/crates/lsp-profiler/src/memory/session.rs +++ b/crates/lsp-profiler/src/memory/session.rs @@ -16,6 +16,7 @@ use tracing::info; use super::compare::{compare, from_changes}; use super::diagnostics::{allocation_diagnostics, leak_diagnostics, MemoryHotspotConfig}; +use super::gate::{StepGate, StepPass}; use super::leaks::LeakTracker; use super::timeline::{MemoryTimeline, TREND_POINTS}; use super::{heapprofile, MemoryError, MemoryPayload}; @@ -46,6 +47,8 @@ struct MemorySession { timeline: MemoryTimeline, /// Monotonic counter that gives snapshots session-unique identifiers. snapshot_count: u64, + /// Admits steps until the session stops ([PROF-MEMORY-STOP]). + gate: StepGate, } /// How results are scored and where exports go. @@ -110,6 +113,7 @@ impl MemorySessionManager { leaks: LeakTracker::default(), timeline: MemoryTimeline::default(), snapshot_count: 0, + gate: StepGate::new(), }; let mut sessions = self.sessions.lock().await; let _ = sessions.insert(context.session_id.clone(), session); @@ -133,6 +137,45 @@ impl MemorySessionManager { Ok((session.context.clone(), Arc::clone(&session.runtime))) } + /// The context and runtime of a session, and the step's admission to it. + /// + /// # Errors + /// + /// Returns [`MemoryError::UnknownSession`] for an unknown ID and + /// [`MemoryError::Stopped`] once the session is stopping. + pub(crate) async fn enter( + &self, + session_id: &str, + ) -> Result<(MemoryContext, Arc, StepPass), MemoryError> { + let (context, runtime, gate) = { + let sessions = self.sessions.lock().await; + let session = sessions + .get(session_id) + .ok_or_else(|| MemoryError::UnknownSession(session_id.to_owned()))?; + let runtime = Arc::clone(&session.runtime); + (session.context.clone(), runtime, session.gate.clone()) + }; + Ok((context, runtime, gate.admit(session_id).await?)) + } + + /// End every step in flight on a session and wait until each has answered + /// ([PROF-MEMORY-STOP]). An unknown session has none. + pub(crate) async fn stop_steps(&self, session_id: &str) { + let gate = self + .sessions + .lock() + .await + .get(session_id) + .map(|session| session.gate.clone()); + if let Some(gate) = gate { + gate.stop().await; + info!( + session_id, + "memory session stopping: every step in flight has answered" + ); + } + } + /// Close a session. pub async fn close(&self, session_id: &str) { let _ = self.sessions.lock().await.remove(session_id); diff --git a/crates/lsp-profiler/src/service.rs b/crates/lsp-profiler/src/service.rs index 7999473..726b374 100644 --- a/crates/lsp-profiler/src/service.rs +++ b/crates/lsp-profiler/src/service.rs @@ -20,26 +20,22 @@ use crate::diagnostics::DiagnosticStyle; use crate::error::ProfilerError; use crate::memory::diagnostics::MemoryHotspotConfig; use crate::memory::session::{MemorySessionManager, MemorySettings}; -use crate::memory::{MemoryAction, MemoryError}; use crate::presets::{resolve, SamplingRequest}; use crate::report::build_report; -use crate::runtime::{MemoryBackend, Runtime, TypeQuery}; +use crate::runtime::Runtime; use crate::session::CpuSessionManager; use crate::wire::{ methods, CooperativeAttachParams, CooperativeScript, CooperativeScriptParams, - KillProcessParams, KillProcessResult, MemoryIngestParams, MemoryResult, MemorySessionParams, - MemoryStartParams, MemoryStartResult, MemoryStep, MemoryTypeParams, ProcessList, - ProcessListParams, ProfileReport, ProfileSessionList, ProfileSessionParams, - ProfilerCapabilities, StartProfileParams, StartProfileResult, + KillProcessParams, KillProcessResult, ProcessList, ProcessListParams, ProfileReport, + ProfileSessionList, ProfileSessionParams, ProfilerCapabilities, StartProfileParams, + StartProfileResult, }; use crate::{cooperative, process}; +mod memory_steps; + /// How often live progress is pushed while a CPU session runs. const PROGRESS_INTERVAL: Duration = Duration::from_secs(1); -/// Default retention-graph node budget. -const DEFAULT_GRAPH_LIMIT: u32 = 200; -/// Default retention-graph depth. -const DEFAULT_GRAPH_DEPTH: u32 = 6; /// Delivers server-to-client notifications for the host's transport. #[async_trait] @@ -396,125 +392,6 @@ impl ProfilerService { Ok(started) } - /// Begin memory tracking. Implements [PROF-MEMORY-START]. - /// - /// # Errors - /// - /// Returns an error when no runtime with memory support recognises the process. - pub async fn memory_start( - &self, - request: MemoryStartParams, - ) -> Result { - let pid = pid_of(request.pid)?; - // Refuse a PID no runtime recognises before any session exists, so a - // refused start never leaves a session for a process that is not there. - let _owner = self.recognise(pid).await?; - let runtime = self.memory_runtime(request.runtime.as_deref(), pid).await?; - let backend = memory_backend(runtime.as_ref())?; - let depth = request.traceback_depth.and_then(|d| u32::try_from(d).ok()); - let context = self - .inner - .memory - .open(Arc::clone(&runtime), pid, depth) - .await; - let action = backend.start(&context).await?; - Ok(MemoryStartResult { - memory_session_id: context.session_id.clone(), - runtime: runtime.id().to_owned(), - mode: backend.capabilities().mode, - step: self.to_step(&context.session_id, action).await?, - }) - } - - /// Dispatch a parameter-free memory operation and normalize its next step. - async fn memory_step( - &self, - method: &str, - request: MemorySessionParams, - ) -> Result { - let (context, runtime) = self.inner.memory.lookup(&request.memory_session_id).await?; - let backend = memory_backend(runtime.as_ref())?; - let action = match method { - methods::MEMORY_SNAPSHOT => backend.snapshot(&context).await?, - methods::MEMORY_DIFF => backend.diff(&context).await?, - methods::MEMORY_GC_COLLECT => backend.gc_collect(&context).await?, - methods::MEMORY_STOP => backend.stop(&context).await?, - other => { - return Err(ProfilerError::InvalidParams(format!( - "unknown method {other}" - ))) - } - }; - let step = self.to_step(&context.session_id, action).await?; - if method == methods::MEMORY_STOP && matches!(step, MemoryStep::Completed { .. }) { - self.inner.memory.close(&context.session_id).await; - } - Ok(step) - } - - /// Run a bounded type query for references or live-object inspection. - async fn memory_type_step( - &self, - method: &str, - request: MemoryTypeParams, - ) -> Result { - let (context, runtime) = self.inner.memory.lookup(&request.memory_session_id).await?; - let backend = memory_backend(runtime.as_ref())?; - let query = TypeQuery { - type_name: request.type_name, - limit: request - .limit - .and_then(|l| u32::try_from(l).ok()) - .unwrap_or(DEFAULT_GRAPH_LIMIT), - max_depth: request - .max_depth - .and_then(|d| u32::try_from(d).ok()) - .unwrap_or(DEFAULT_GRAPH_DEPTH), - }; - let action = if method == methods::MEMORY_REFERENCES { - backend.references(&context, &query).await? - } else { - backend.objects_by_type(&context, &query).await? - }; - self.to_step(&context.session_id, action).await - } - - /// Score courier output. Implements [PROF-MEMORY-INGEST]. - /// - /// # Errors - /// - /// Returns an error for an unknown session or unrecognised output. - pub async fn memory_ingest( - &self, - request: MemoryIngestParams, - ) -> Result { - Ok(self - .inner - .memory - .ingest(&request.memory_session_id, &request.output) - .await?) - } - - /// Convert runtime instructions into evaluation or a scored completed result. - async fn to_step( - &self, - session_id: &str, - action: MemoryAction, - ) -> Result { - Ok(match action { - MemoryAction::Evaluate { - script, - output_marker, - } => MemoryStep::Evaluate { - script, - output_marker, - }, - MemoryAction::Payload(payload) => MemoryStep::Completed { - result: self.inner.memory.apply(session_id, payload).await?, - }, - }) - } - /// Stop every CPU session. Call on shutdown. pub async fn shutdown(&self) { self.inner.cpu.stop_all().await; @@ -612,20 +489,4 @@ impl ProfilerService { ) -> Result, ProfilerError> { self.pick(id, pid, |r| r.cpu().is_some()).await } - - /// Select a runtime with memory support using the standard resolution order. - async fn memory_runtime( - &self, - id: Option<&str>, - pid: u32, - ) -> Result, ProfilerError> { - self.pick(id, pid, |r| r.memory().is_some()).await - } -} - -/// Require memory support after runtime selection, with a classified error. -fn memory_backend(runtime: &dyn Runtime) -> Result<&dyn MemoryBackend, ProfilerError> { - runtime.memory().ok_or_else(|| { - MemoryError::Unsupported(format!("memory profiling of {}", runtime.display_name())).into() - }) } diff --git a/crates/lsp-profiler/src/service/memory_steps.rs b/crates/lsp-profiler/src/service/memory_steps.rs new file mode 100644 index 0000000..8d6e303 --- /dev/null +++ b/crates/lsp-profiler/src/service/memory_steps.rs @@ -0,0 +1,191 @@ +//! The memory half of the JSON-RPC surface. Implements [PROF-MEMORY-START], +//! [PROF-MEMORY-STEP], [PROF-MEMORY-INGEST] and [PROF-MEMORY-STOP]. +//! +//! Every step but `stop` runs under its session's admission, so `stop` can end +//! the steps in flight before it releases the session. + +use std::sync::Arc; + +use super::{pid_of, ProfilerService}; +use crate::error::ProfilerError; +use crate::memory::{MemoryAction, MemoryError}; +use crate::runtime::{MemoryBackend, MemoryContext, Runtime, TypeQuery}; +use crate::wire::{ + methods, MemoryIngestParams, MemoryResult, MemorySessionParams, MemoryStartParams, + MemoryStartResult, MemoryStep, MemoryTypeParams, +}; + +/// Default retention-graph node budget. +const DEFAULT_GRAPH_LIMIT: u32 = 200; +/// Default retention-graph depth. +const DEFAULT_GRAPH_DEPTH: u32 = 6; + +impl ProfilerService { + /// Begin memory tracking. Implements [PROF-MEMORY-START]. + /// + /// # Errors + /// + /// Returns an error when no runtime with memory support recognises the process. + pub async fn memory_start( + &self, + request: MemoryStartParams, + ) -> Result { + let pid = pid_of(request.pid)?; + // Refuse a PID no runtime recognises before any session exists, so a + // refused start never leaves a session for a process that is not there. + let _owner = self.recognise(pid).await?; + let runtime = self.memory_runtime(request.runtime.as_deref(), pid).await?; + let backend = memory_backend(runtime.as_ref())?; + let depth = request.traceback_depth.and_then(|d| u32::try_from(d).ok()); + let context = self + .inner + .memory + .open(Arc::clone(&runtime), pid, depth) + .await; + let action = backend.start(&context).await?; + Ok(MemoryStartResult { + memory_session_id: context.session_id.clone(), + runtime: runtime.id().to_owned(), + mode: backend.capabilities().mode, + step: self.to_step(&context.session_id, action).await?, + }) + } + + /// Dispatch a parameter-free memory operation and normalize its next step. + pub(super) async fn memory_step( + &self, + method: &str, + request: MemorySessionParams, + ) -> Result { + if method == methods::MEMORY_STOP { + return self.memory_stop(&request.memory_session_id).await; + } + let (context, runtime, pass) = self.inner.memory.enter(&request.memory_session_id).await?; + let backend = memory_backend(runtime.as_ref())?; + pass.run(async { + let action = parameterless_step(backend, method, &context).await?; + self.to_step(&context.session_id, action).await + }) + .await + } + + /// End every step in flight, then release the session. Implements + /// [PROF-MEMORY-STOP]. + async fn memory_stop(&self, session_id: &str) -> Result { + self.inner.memory.stop_steps(session_id).await; + let (context, runtime) = self.inner.memory.lookup(session_id).await?; + let backend = memory_backend(runtime.as_ref())?; + let step = self + .to_step(session_id, backend.stop(&context).await?) + .await?; + if matches!(step, MemoryStep::Completed { .. }) { + self.inner.memory.close(session_id).await; + } + Ok(step) + } + + /// Run a bounded type query for references or live-object inspection. + pub(super) async fn memory_type_step( + &self, + method: &str, + request: MemoryTypeParams, + ) -> Result { + let (context, runtime, pass) = self.inner.memory.enter(&request.memory_session_id).await?; + let backend = memory_backend(runtime.as_ref())?; + let query = type_query(request); + pass.run(async { + let action = if method == methods::MEMORY_REFERENCES { + backend.references(&context, &query).await? + } else { + backend.objects_by_type(&context, &query).await? + }; + self.to_step(&context.session_id, action).await + }) + .await + } + + /// Score courier output. Implements [PROF-MEMORY-INGEST]. + /// + /// # Errors + /// + /// Returns an error for an unknown session or unrecognised output. + pub async fn memory_ingest( + &self, + request: MemoryIngestParams, + ) -> Result { + Ok(self + .inner + .memory + .ingest(&request.memory_session_id, &request.output) + .await?) + } + + /// Convert runtime instructions into evaluation or a scored completed result. + async fn to_step( + &self, + session_id: &str, + action: MemoryAction, + ) -> Result { + Ok(match action { + MemoryAction::Evaluate { + script, + output_marker, + } => MemoryStep::Evaluate { + script, + output_marker, + }, + MemoryAction::Payload(payload) => MemoryStep::Completed { + result: self.inner.memory.apply(session_id, payload).await?, + }, + }) + } + + /// Select a runtime with memory support using the standard resolution order. + async fn memory_runtime( + &self, + id: Option<&str>, + pid: u32, + ) -> Result, ProfilerError> { + self.pick(id, pid, |r| r.memory().is_some()).await + } +} + +/// Run the parameter-free step `method` on `backend`. +async fn parameterless_step( + backend: &dyn MemoryBackend, + method: &str, + context: &MemoryContext, +) -> Result { + Ok(match method { + methods::MEMORY_SNAPSHOT => backend.snapshot(context).await?, + methods::MEMORY_DIFF => backend.diff(context).await?, + methods::MEMORY_GC_COLLECT => backend.gc_collect(context).await?, + other => { + return Err(ProfilerError::InvalidParams(format!( + "unknown method {other}" + ))) + } + }) +} + +/// The type query a references or objects request asks for, with the default budgets. +fn type_query(request: MemoryTypeParams) -> TypeQuery { + TypeQuery { + type_name: request.type_name, + limit: request + .limit + .and_then(|l| u32::try_from(l).ok()) + .unwrap_or(DEFAULT_GRAPH_LIMIT), + max_depth: request + .max_depth + .and_then(|d| u32::try_from(d).ok()) + .unwrap_or(DEFAULT_GRAPH_DEPTH), + } +} + +/// Require memory support after runtime selection, with a classified error. +fn memory_backend(runtime: &dyn Runtime) -> Result<&dyn MemoryBackend, ProfilerError> { + runtime.memory().ok_or_else(|| { + MemoryError::Unsupported(format!("memory profiling of {}", runtime.display_name())).into() + }) +} diff --git a/docs/specs/LSP-PROFILER-SPEC.md b/docs/specs/LSP-PROFILER-SPEC.md index c1c7dd1..7c40294 100644 --- a/docs/specs/LSP-PROFILER-SPEC.md +++ b/docs/specs/LSP-PROFILER-SPEC.md @@ -423,6 +423,12 @@ Runtimes use one of two modes: The `snapshot`, `diff`, `references`, `objectsByType`, `gcCollect` and `stop` commands each return a `MemoryStep`. +### [PROF-MEMORY-STOP] Stop + +A step can wait on its target indefinitely: a direct runtime reads the live heap, which a frozen, paused or dying program never yields. `stop` never waits on that. It first ends every step in flight on the session, the way cancellation does ([PROF-HOST-CANCEL]): the step's future is dropped, which terminates every tool it started and leaves the target running. Each such step answers `OTHER`, "memory session `` stopped before the step finished". `stop` answers only once every one of them has answered, then the runtime releases the session (the .NET runtime deletes its graphs) and the session closes. A step that arrives while the session is stopping is refused the same way; one that arrives after is an unknown session. Stopping one session never touches another's steps. + +An editor tracking on a timer ([PROF-UI-AUTOPILOT]) relies on this: a capture still in flight when tracking stops answers at once, so it never holds the next session's captures back. + ### [PROF-MEMORY-INGEST] Ingest `memory.ingest { memory_session_id, output }` parses courier output with the runtime's `parse`. It scores the payload ([PROF-MEMORY-OUTCOME]) and returns a `MemoryResult`. These are errors: @@ -750,6 +756,8 @@ Thread samples are grouped into ticks: A snapshot is an EventPipe session on `Microsoft-Windows-DotNETRuntime` at Verbose with the keywords `GC`, `Loader`, `Type`, `GCHeapDump`, `GCHeapCollect` and `GCHeapAndTypeNames`, plus rundown. Enabling them makes the runtime perform a full collection and report every surviving object (`GCBulkNode`), reference (`GCBulkEdge`, in node order), root (`GCBulkRootEdge`, `GCBulkRootStaticVar`) and type (`TypeBulkType`); module load and rundown events give each type's assembly path. The engine waits for the induced collection's `GCStop` and the `GCHeapStats` after it (generation sizes), then stops the session. No pause, no dump and no debugger permission are involved, which keeps the capture identical on macOS, Linux and Windows. +The engine stops waiting as soon as the event stream ends, whatever it was waiting for: a stream that ends took its session with it, because the target exited, and nothing is left to stop. Waiting on would only sit out the deadline. Any failure of a command whose target is no longer running is `ProcessNotFound` ("process `` is not running"), whatever broke first: a stream cut off mid-read, or a session that ended before its collection. + The runtime emits the whole heap in one burst during that collection, faster than any reader drains it, so the session buffer must hold the burst. It is 1024 MB, which holds about fourteen million objects. The 256 MB default lost events from about six million. If EventPipe still reports lost events, the snapshot fails (`Failed`) with a message naming how many events were dropped and the buffer size, and no graph is kept. Lost edges cut live objects off from their roots, so a partial graph would report a leaking heap as nearly empty. ### [PROF-DOTNET-HEAP-GRAPH] Heap graph @@ -787,7 +795,7 @@ Graphs live under `/heap-/snapshot-.heapgraph.json`. A g ### [PROF-DOTNET-GC] Induced collection -An EventPipe session with the GC keyword plus `GCHeapCollect` makes the runtime perform a full blocking collection. The engine waits for the collection's `GCStop` and the `GCHeapStats` after it (two seconds of grace), then reports the generation sizes. `observed: false` means no collection completed before the timeout. +An EventPipe session with the GC keyword plus `GCHeapCollect` makes the runtime perform a full blocking collection. The engine waits for the collection's `GCStop` and the `GCHeapStats` after it (two seconds of grace), then reports the generation sizes. `observed: false` means no collection completed before the timeout. A stream that ends first, because the target exited, is `ProcessNotFound`, not an unobserved collection ([PROF-DOTNET-HEAP-CAPTURE]). ### [PROF-DOTNET-PERMISSIONS] Permissions diff --git a/dotnet/LspProfiler.DotNet.Engine.Tests/FailureTests.cs b/dotnet/LspProfiler.DotNet.Engine.Tests/FailureTests.cs new file mode 100644 index 0000000..3a6cc7c --- /dev/null +++ b/dotnet/LspProfiler.DotNet.Engine.Tests/FailureTests.cs @@ -0,0 +1,82 @@ +// Unit tests for the failure a command reports. Implements [PROF-ERRORS]: a +// command whose target is gone fails PROCESS_NOT_FOUND whatever broke first, +// because the target exiting is the cause; every other failure keeps its kind +// and its own message. +using LspProfiler.DotNet.Engine; +using Microsoft.Diagnostics.NETCore.Client; +using Xunit; + +namespace LspProfiler.DotNet.Engine.Tests; + +public sealed class FailureTests +{ + /// No process has this id: Windows ids are multiples of four and no Unix pid_max reaches it. + private const int Gone = int.MaxValue; + + [Theory] + [InlineData("end-of-stream")] + [InlineData("broken-pipe")] + [InlineData("no-collection")] + [InlineData("stream-ended")] + [InlineData("server-not-available")] + [InlineData("unsupported-command")] + [InlineData("access-denied")] + [InlineData("client")] + public void AnyFailureOfACommandWhoseTargetIsGoneIsProcessNotFound(string cause) + { + var failure = Errors.Describe(Raised(cause), Gone); + + Assert.Equal("failure", failure.Kind); + Assert.Equal(ErrorKind.ProcessNotFound, failure.ErrorKind); + Assert.Equal($"process {Gone} is not running", failure.Message); + } + + [Theory] + [InlineData("end-of-stream", "Failed")] + [InlineData("broken-pipe", "Failed")] + [InlineData("no-collection", "Failed")] + [InlineData("stream-ended", "Failed")] + [InlineData("server-not-available", "UnsupportedRuntime")] + [InlineData("unsupported-command", "UnsupportedRuntime")] + [InlineData("access-denied", "PermissionDenied")] + [InlineData("client", "AttachFailed")] + public void AFailureWhileTheTargetRunsKeepsItsKindAndMessage(string cause, string kind) + { + var error = Raised(cause); + + var failure = Errors.Describe(error, Environment.ProcessId); + + Assert.Equal("failure", failure.Kind); + Assert.Equal(kind, failure.ErrorKind.ToString()); + Assert.Equal(error.Message, failure.Message); + } + + [Theory] + [InlineData("end-of-stream", "Failed")] + [InlineData("server-not-available", "UnsupportedRuntime")] + [InlineData("access-denied", "PermissionDenied")] + [InlineData("client", "AttachFailed")] + public void AFailureOfACommandWithNoTargetKeepsItsKindAndMessage(string cause, string kind) + { + var error = Raised(cause); + + var failure = Errors.Describe(error, null); + + Assert.Equal(kind, failure.ErrorKind.ToString()); + Assert.Equal(error.Message, failure.Message); + } + + /// What each way of breaking throws, as the engine's commands meet it. + private static Exception Raised(string cause) => cause switch + { + "end-of-stream" => new EndOfStreamException("Unable to read beyond the end of the stream."), + "broken-pipe" => new IOException("Pipe is broken."), + "no-collection" => new TimeoutException("the runtime did not complete the heap snapshot collection"), + "stream-ended" => new InvalidOperationException("the event stream ended before the collection completed"), + "server-not-available" => new ServerNotAvailableException("Process 7 not running compatible .NET runtime."), + "unsupported-command" => new UnsupportedCommandException("unsupported"), + "access-denied" => new UnauthorizedAccessException("Access to the path is denied."), + "client" => new DiagnosticsClientException("connect failed"), + _ => throw new ArgumentOutOfRangeException(nameof(cause), cause, "no such cause"), + }; +} diff --git a/dotnet/LspProfiler.DotNet.Engine.Tests/SessionRunTests.cs b/dotnet/LspProfiler.DotNet.Engine.Tests/SessionRunTests.cs new file mode 100644 index 0000000..61b7e92 --- /dev/null +++ b/dotnet/LspProfiler.DotNet.Engine.Tests/SessionRunTests.cs @@ -0,0 +1,107 @@ +// Unit tests for how an EventPipe session's run ends. Implements +// [PROF-DOTNET-HEAP-CAPTURE] and [PROF-DOTNET-GC]: the run ends when the watcher +// sees its collection, when the stream ends because the target took the session +// with it, or at the deadline, and only a session still streaming is stopped. +using LspProfiler.DotNet.Engine; +using Xunit; + +namespace LspProfiler.DotNet.Engine.Tests; + +public sealed class SessionRunTests +{ + /// Long enough that nothing here reaches it unless the run waits for a watcher that cannot finish. + private static readonly TimeSpan Bounded = TimeSpan.FromSeconds(5); + + [Fact] + public void AStreamThatEndsBeforeTheWatcherEndsTheRunWithoutStoppingTheSession() + { + using var done = new ManualResetEvent(false); + var stops = new Stops(); + + var end = SessionRun.Await(done, Task.CompletedTask, stops.Stop, Bounded); + + Assert.Equal(SessionEnd.StreamEnded, end); + Assert.Equal(0, stops.Count); + Assert.False(done.WaitOne(0)); + } + + [Fact] + public void AStreamThatFailsBeforeTheWatcherEndsTheRunWithoutStoppingTheSession() + { + using var done = new ManualResetEvent(false); + var stops = new Stops(); + var broken = Task.FromException(new EndOfStreamException("the target exited")); + + var end = SessionRun.Await(done, broken, stops.Stop, Bounded); + + Assert.Equal(SessionEnd.StreamEnded, end); + Assert.Equal(0, stops.Count); + Assert.True(broken.IsFaulted); + } + + [Fact] + public void AWatcherDoneWhileStreamingStopsTheSessionOnceAndDrainsIt() + { + using var done = new ManualResetEvent(true); + var stream = new TaskCompletionSource(); + var stops = new Stops(stream); + + var end = SessionRun.Await(done, stream.Task, stops.Stop, Bounded); + + Assert.Equal(SessionEnd.Watched, end); + Assert.Equal(1, stops.Count); + Assert.True(stream.Task.IsCompletedSuccessfully); + } + + [Fact] + public void AWatcherDoneAfterTheStreamEndedLeavesNothingToStop() + { + using var done = new ManualResetEvent(true); + var stops = new Stops(); + + var end = SessionRun.Await(done, Task.CompletedTask, stops.Stop, Bounded); + + Assert.Equal(SessionEnd.Watched, end); + Assert.Equal(0, stops.Count); + } + + [Fact] + public void ADeadlineWithNeitherStopsTheStreamingSessionAndSaysSo() + { + using var done = new ManualResetEvent(false); + var stream = new TaskCompletionSource(); + var stops = new Stops(stream); + + var end = SessionRun.Await(done, stream.Task, stops.Stop, TimeSpan.Zero); + + Assert.Equal(SessionEnd.DeadlinePassed, end); + Assert.Equal(1, stops.Count); + Assert.True(stream.Task.IsCompletedSuccessfully); + } + + [Fact] + public void AStoppedSessionWhoseStreamNeverEndsIsATimeout() + { + using var done = new ManualResetEvent(true); + var endless = new TaskCompletionSource(); + var stops = new Stops(); + + var error = Assert.Throws(() => SessionRun.Await(done, endless.Task, stops.Stop, TimeSpan.Zero)); + + Assert.Equal(1, stops.Count); + Assert.Contains("did not end after the session was stopped", error.Message, StringComparison.Ordinal); + Assert.False(endless.Task.IsCompleted); + } + + /// Counts stops; stopping ends when given, as stopping a live session does. + private sealed class Stops(TaskCompletionSource? stream = null) + { + public int Count { get; private set; } + + public void Stop() + { + Count++; + stream?.TrySetResult(); + } + } +} diff --git a/dotnet/LspProfiler.DotNet.Engine/Errors.cs b/dotnet/LspProfiler.DotNet.Engine/Errors.cs index a32fa20..224d200 100644 --- a/dotnet/LspProfiler.DotNet.Engine/Errors.cs +++ b/dotnet/LspProfiler.DotNet.Engine/Errors.cs @@ -16,12 +16,22 @@ internal static class Errors _ => ErrorKind.Failed, }; + /// + /// The failure a command reports for . When the process it works + /// on, , is gone, that is the cause whatever broke first: a stream + /// cut off mid-read or a session that ended early ([PROF-DOTNET-HEAP-CAPTURE]). + /// + public static Failure Describe(Exception error, int? target) => + target is { } pid && !IsRunning(pid) + ? NotRunning(pid) + : new("failure", Classify(error), error.Message); + /// A client for , or the reason there cannot be one. public static (DiagnosticsClient? Client, Failure? Failure) Connect(int pid) { if (!IsRunning(pid)) { - return (null, new Failure("failure", ErrorKind.ProcessNotFound, $"process {pid} is not running")); + return (null, NotRunning(pid)); } // [PROF-DOTNET-ENGINE] Resolve only the requested endpoint. Enumerating published // processes can fail when an unrelated /proc entry disappears during attachment. @@ -47,6 +57,9 @@ public static (DiagnosticsClient? Client, Failure? Failure) Connect(int pid) ? new Failure("failure", ErrorKind.UnsupportedRuntime, $"process {pid} has no .NET diagnostics endpoint") : null; + private static Failure NotRunning(int pid) => + new("failure", ErrorKind.ProcessNotFound, $"process {pid} is not running"); + private static bool IsRunning(int pid) { try diff --git a/dotnet/LspProfiler.DotNet.Engine/GcCommand.cs b/dotnet/LspProfiler.DotNet.Engine/GcCommand.cs index 8f1b0d0..1e24c73 100644 --- a/dotnet/LspProfiler.DotNet.Engine/GcCommand.cs +++ b/dotnet/LspProfiler.DotNet.Engine/GcCommand.cs @@ -32,7 +32,12 @@ public static int Run(int pid) using var source = new EventPipeEventSource(session.EventStream); using var watch = new GcWatch(); watch.Attach(source); - SessionRun.Until(session, source, watch, Deadline); + // A stream that ended took the target with it: that is a failure, not an + // unobserved collection ([PROF-DOTNET-GC]). + if (SessionRun.Until(session, source, watch, Deadline) == SessionEnd.StreamEnded && !watch.Collected) + { + throw SessionRun.Unobserved(SessionEnd.StreamEnded, "the collection"); + } var generations = watch.Generations(); Emit.Message(new GcDone("gc", watch.Collected, generations.Sum(), generations)); return 0; diff --git a/dotnet/LspProfiler.DotNet.Engine/GcWatch.cs b/dotnet/LspProfiler.DotNet.Engine/GcWatch.cs index c15eb6c..6c048b7 100644 --- a/dotnet/LspProfiler.DotNet.Engine/GcWatch.cs +++ b/dotnet/LspProfiler.DotNet.Engine/GcWatch.cs @@ -74,18 +74,59 @@ private void OnStats(GCHeapStatsTraceData stats) } } -/// Runs an event source until the watcher is done or the deadline passes, then stops the session. +/// How a watched session ended. +internal enum SessionEnd +{ + /// The watcher saw what it waits for. + Watched, + + /// The event stream ended first, so the target took the session with it. + StreamEnded, + + /// Neither happened before the deadline. + DeadlinePassed, +} + +/// Runs an event source until the watcher is done, the stream ends or the deadline passes. internal static class SessionRun { - public static void Until(Microsoft.Diagnostics.NETCore.Client.EventPipeSession session, EventPipeEventSource source, GcWatch watch, TimeSpan deadline) + public static SessionEnd Until(Microsoft.Diagnostics.NETCore.Client.EventPipeSession session, EventPipeEventSource source, GcWatch watch, TimeSpan deadline) { var processing = Task.Run(source.Process); - var finished = watch.Done.Wait(deadline); - Emit.Trace(finished ? "watch done; stopping session" : "deadline passed; stopping session"); - session.Stop(); - if (!processing.Wait(deadline)) + var end = Await(watch.Done.WaitHandle, processing, session.Stop, deadline); + Emit.Trace(processing.IsFaulted + ? $"session ended: {end}; the stream failed: {processing.Exception?.GetBaseException().Message}" + : $"session ended: {end}"); + return end; + } + + /// + /// Wait for or the end of the stream + /// reads, whichever comes first. A stream that ended took its session with it: the target + /// exited, and waiting on would only sit out the deadline. Only a session still streaming + /// is stopped, then drained. + /// + public static SessionEnd Await(WaitHandle done, Task processing, Action stop, TimeSpan deadline) + { + var end = WaitHandle.WaitAny([done, ((IAsyncResult)processing).AsyncWaitHandle], deadline) switch { - throw new TimeoutException("the event stream did not end after the session was stopped"); + 0 => SessionEnd.Watched, + 1 => SessionEnd.StreamEnded, + _ => SessionEnd.DeadlinePassed, + }; + if (!processing.IsCompleted) + { + stop(); + if (!processing.Wait(deadline)) + { + throw new TimeoutException("the event stream did not end after the session was stopped"); + } } + return end; } + + /// Why was not seen, as the command fails with it. + public static Exception Unobserved(SessionEnd end, string collection) => end == SessionEnd.StreamEnded + ? new InvalidOperationException($"the event stream ended before {collection} completed") + : new TimeoutException($"the runtime did not complete {collection}"); } diff --git a/dotnet/LspProfiler.DotNet.Engine/HeapCapture.cs b/dotnet/LspProfiler.DotNet.Engine/HeapCapture.cs index c66e010..c3cff20 100644 --- a/dotnet/LspProfiler.DotNet.Engine/HeapCapture.cs +++ b/dotnet/LspProfiler.DotNet.Engine/HeapCapture.cs @@ -43,11 +43,11 @@ public static HeapGraph Capture(DiagnosticsClient client, int bufferMB) watch.Attach(source); events.Attach(source); Emit.Trace("session started; processing events"); - SessionRun.Until(session, source, watch, Deadline); + var end = SessionRun.Until(session, source, watch, Deadline); Emit.Trace($"event stream ended; lost={source.EventsLost} collected={watch.Collected} nodes={events.NodeCount} edges={events.EdgeCount} roots={events.RootCount} types={events.TypeCount}"); if (!watch.Collected) { - throw new TimeoutException("the runtime did not complete the heap snapshot collection"); + throw SessionRun.Unobserved(end, "the heap snapshot collection"); } if (source.EventsLost > 0) { diff --git a/dotnet/LspProfiler.DotNet.Engine/Program.cs b/dotnet/LspProfiler.DotNet.Engine/Program.cs index c892c53..3c348c9 100644 --- a/dotnet/LspProfiler.DotNet.Engine/Program.cs +++ b/dotnet/LspProfiler.DotNet.Engine/Program.cs @@ -24,7 +24,8 @@ public static int Run(string[] args) } catch (Exception error) when (error is not OutOfMemoryException) { - return Emit.Fail(Errors.Classify(error), error.Message); + Emit.Message(Errors.Describe(error, options.Target)); + return 1; } } @@ -73,5 +74,11 @@ public string Text(string name) => public int Int(string name, int fallback) => _values.ContainsKey(name) ? Int(name) : fallback; public bool Flag(string name) => _values.ContainsKey(name); + + /// The process a command works on: its --pid, when it names one. + public int? Target => + _values.TryGetValue("pid", out var text) && int.TryParse(text, System.Globalization.NumberStyles.Integer, System.Globalization.CultureInfo.InvariantCulture, out var pid) + ? pid + : null; } }