perf(agent-v2): RegistryWatcher on ThreadPool wait + disposal hygiene

Reduce the V2 agent footprint during enrollment without touching any
capability or the deliberate live-tracking cadences (DO 3s poll, IME
100ms poll, download_progress ImmediateUpload — all product USP).

RegistryWatcher: replace the dedicated background thread (new Thread +
WaitHandle.WaitAny on RegNotifyChangeKeyValue) with a ThreadPool wait
(ThreadPool.RegisterWaitForSingleObject), mirroring the proven pattern
in Monitoring/Telemetry/Office/RegistryChangeWatcher. Removes up to
~7-8 dedicated ~1MB thread stacks (Enrollments + CloudDomainJoin +
AadJoin + Provisioning + up to 4x RealmJoin). The public contract is
preserved verbatim: ctor signature, Changed/Error events, IsRunning,
Start()/RequestStop()/Stop() (incl. deferred rethrow of the captured
background failure) and Dispose(). Arming stays synchronous in Start();
Changed is raised outside the lock; the watcher re-arms after each
single-shot notification. All four consumers run unchanged.

Disposal hygiene:
- WmiCollector: dispose each ManagementObject per iteration (native
  WMI/COM memory) via using (obj).
- AgentSelfMetricsCollector: wrap both Process.GetCurrentProcess() in
  using var (handle leak in the 60s metrics loop).

Verified: V2.Core + V2 agent build clean; full V2.Core suite 1264/1264
green, incl. the RegistryWatcher Win32 integration tests (value write,
subtree subkey, repeated successive changes) and the RealmJoin /
AadJoin / Provisioning / TenantId consumer tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Oliver Kieselbach
2026-06-09 14:31:20 +02:00
parent 5586963bcc
commit 993ebbaa67
3 changed files with 189 additions and 137 deletions
@@ -10,12 +10,26 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime
/// <summary> /// <summary>
/// Watches a registry key for changes using RegNotifyChangeKeyValue. /// Watches a registry key for changes using RegNotifyChangeKeyValue.
/// Raises Changed whenever the watched key/subtree changes. /// Raises Changed whenever the watched key/subtree changes.
/// ///
/// Notes: /// Notes:
/// - Notification is for the key scope, not one individual value. /// - Notification is for the key scope, not one individual value.
/// - To track a specific value, read/compare that value in the Changed handler. /// - To track a specific value, read/compare that value in the Changed handler.
/// - Uses a dedicated background thread for robustness. /// - Backed by a <see cref="ThreadPool"/> wait (<see cref="ThreadPool.RegisterWaitForSingleObject"/>)
/// on the async-signalled change event — NOT a dedicated background thread. This is the same
/// lightweight pattern used by <c>Monitoring/Telemetry/Office/RegistryChangeWatcher</c>; it
/// removes one OS thread (~1 MB committed stack) per watcher. The public contract — the ctor,
/// <see cref="Changed"/>/<see cref="Error"/> events, <see cref="Start"/>, <see cref="RequestStop"/>,
/// <see cref="Stop"/> (which rethrows a captured background failure) and <see cref="Dispose"/> —
/// is preserved verbatim.
/// </summary> /// </summary>
/// <remarks>
/// Threading contract: <see cref="Changed"/> is raised on a ThreadPool thread, OUTSIDE the
/// internal lock, so a handler may freely call <see cref="RequestStop"/> (the documented way to
/// stop from inside a handler). RegNotifyChangeKeyValue is single-shot, so the watcher re-arms
/// after each notification — there is, by design, a tiny window between the signal and the
/// re-arm during which an intermediate change is collapsed into the next notification; callers
/// always re-read current state in their handler, so no edge is lost in practice.
/// </remarks>
internal sealed class RegistryWatcher : IDisposable internal sealed class RegistryWatcher : IDisposable
{ {
private readonly UIntPtr _hive; private readonly UIntPtr _hive;
@@ -27,15 +41,18 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime
private readonly object _sync = new object(); private readonly object _sync = new object();
private CancellationTokenSource _cts; private SafeRegistryHandle _keyHandle;
private Thread _thread; private AutoResetEvent _signal;
private RegisteredWaitHandle _registeredWait;
private bool _running;
private bool _stopped;
private Exception _backgroundException; private Exception _backgroundException;
private bool _disposed; private bool _disposed;
public event EventHandler Changed; public event EventHandler Changed;
public event EventHandler<Exception> Error; public event EventHandler<Exception> Error;
/// <param name="trace">Optional trace callback for diagnostic logging (called on every loop iteration).</param> /// <param name="trace">Optional trace callback for diagnostic logging (called on arm / signal / re-arm).</param>
public RegistryWatcher( public RegistryWatcher(
RegistryHive hive, RegistryHive hive,
string subKey, string subKey,
@@ -60,7 +77,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime
{ {
lock (_sync) lock (_sync)
{ {
return _thread != null; return _running;
} }
} }
} }
@@ -69,112 +86,97 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime
{ {
ThrowIfDisposed(); ThrowIfDisposed();
Exception openError;
lock (_sync) lock (_sync)
{ {
if (_thread != null) if (_running)
throw new InvalidOperationException("Watcher is already running."); throw new InvalidOperationException("Watcher is already running.");
_running = true;
_stopped = false;
_backgroundException = null; _backgroundException = null;
_cts = new CancellationTokenSource(); _signal = new AutoResetEvent(false);
_thread = new Thread(WatchThreadMain) // Open the key + arm the first async notification synchronously, so by the time
{ // Start() returns the kernel is already tracking changes on our handle (callers
IsBackground = true, // and tests rely on the watch being live shortly after Start). A RegOpenKeyEx /
Name = $"RegistryWatcher: {_subKey}" // RegNotifyChangeKeyValue failure does NOT throw from Start() — to preserve the
}; // dedicated-thread contract it is captured, surfaced via Error, and rethrown on
// Stop(). Every production caller wraps Start() in try/catch and swallows the
_thread.Start(_cts.Token); // Stop() rethrow, so this keeps their behaviour identical.
openError = OpenKeyAndArm();
if (openError != null)
_backgroundException = openError;
} }
if (openError != null)
RaiseError(openError);
} }
/// <summary> /// <summary>
/// Requests stop without blocking. /// Requests stop without blocking. Safe to call from Changed handlers (it does not dispose
/// Safe to call from Changed handlers. /// the underlying handles — that happens in <see cref="Stop"/>/<see cref="Dispose"/> — it
/// only guarantees no further Changed is raised).
/// </summary> /// </summary>
public void RequestStop() public void RequestStop()
{ {
lock (_sync) lock (_sync)
{ {
_cts?.Cancel(); if (!_running) return;
_stopped = true;
UnregisterWait();
try { _signal?.Set(); } catch { /* releasing a pending wait must not throw */ }
} }
} }
/// <summary> /// <summary>
/// Stops and waits for the watcher thread to exit. /// Stops the watcher and releases its handles. Rethrows a captured background failure
/// Avoid calling from inside Changed; use RequestStop there. /// (other than cancellation) as an <see cref="InvalidOperationException"/>, preserving the
/// original dedicated-thread contract. Avoid calling from inside Changed; use
/// <see cref="RequestStop"/> there.
/// </summary> /// </summary>
public void Stop() public void Stop()
{ {
Thread threadToJoin = null; Exception backgroundException;
CancellationTokenSource ctsToDispose = null;
Exception backgroundException = null;
lock (_sync) lock (_sync)
{ {
if (_thread == null) if (!_running)
return; return;
_cts.Cancel(); _running = false;
threadToJoin = _thread; _stopped = true;
}
threadToJoin.Join(); UnregisterWait();
try { _signal?.Set(); } catch { /* release any pending wait */ }
try { _signal?.Dispose(); } catch { /* dispose must not throw */ }
_signal = null;
try { _keyHandle?.Dispose(); } catch { /* dispose must not throw */ }
_keyHandle = null;
lock (_sync)
{
backgroundException = _backgroundException; backgroundException = _backgroundException;
ctsToDispose = _cts;
_cts = null;
_thread = null;
_backgroundException = null; _backgroundException = null;
} }
ctsToDispose?.Dispose(); _trace?.Invoke("Stopped (cancellation requested)");
if (backgroundException != null && if (backgroundException != null &&
!(backgroundException is OperationCanceledException)) !(backgroundException is OperationCanceledException))
{ {
throw new InvalidOperationException( throw new InvalidOperationException(
"Registry watcher stopped because the background thread failed.", "Registry watcher stopped because the background notification failed.",
backgroundException); backgroundException);
} }
} }
private void WatchThreadMain(object state) /// <summary>
{ /// Opens the watched key and arms the first async notification. Must hold <see cref="_sync"/>.
var cancellationToken = (CancellationToken)state; /// Returns the failure exception (so the caller can capture/surface it without throwing) or
/// <c>null</c> on success.
try /// </summary>
{ private Exception OpenKeyAndArm()
WatchLoop(cancellationToken);
}
catch (OperationCanceledException ex)
{
lock (_sync)
{
_backgroundException = ex;
}
}
catch (Exception ex)
{
lock (_sync)
{
_backgroundException = ex;
}
try
{
Error?.Invoke(this, ex);
}
catch
{
// Never let handler exceptions escape the background thread.
}
}
}
private void WatchLoop(CancellationToken cancellationToken)
{ {
int samDesired = RegistryNativeMethods.GetSamDesired(_view); int samDesired = RegistryNativeMethods.GetSamDesired(_view);
@@ -189,81 +191,127 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime
if (openResult != 0) if (openResult != 0)
{ {
throw new Win32Exception( return new Win32Exception(
openResult, openResult,
$"RegOpenKeyEx failed for '{_subKey}' (view: {_view})."); $"RegOpenKeyEx failed for '{_subKey}' (view: {_view}).");
} }
_keyHandle = keyHandle;
_trace?.Invoke($"RegOpenKeyEx succeeded — handle valid: {!keyHandle.IsInvalid}"); _trace?.Invoke($"RegOpenKeyEx succeeded — handle valid: {!keyHandle.IsInvalid}");
using (keyHandle) return ArmNotify();
using (var changedEvent = new AutoResetEvent(false)) }
using (var cancelEvent = new ManualResetEvent(false))
using (cancellationToken.Register(() => cancelEvent.Set())) /// <summary>
/// Arms one async RegNotifyChangeKeyValue and registers a ThreadPool wait on the change
/// event. Single-shot — re-armed after each notification by <see cref="OnSignalled"/>.
/// Must hold <see cref="_sync"/>. Returns the failure exception or <c>null</c> on success.
/// </summary>
private Exception ArmNotify()
{
if (_disposed || _stopped || _keyHandle == null || _signal == null)
return null;
_trace?.Invoke($"RegNotifyChangeKeyValue (subtree={_watchSubtree}, filter=0x{(uint)_filter:X})");
int notifyResult = RegistryNativeMethods.RegNotifyChangeKeyValue(
_keyHandle,
_watchSubtree,
_filter,
_signal.SafeWaitHandle,
true);
if (notifyResult != 0)
{ {
WaitHandle[] waitHandles = { changedEvent, cancelEvent }; return new Win32Exception(
int iteration = 0; notifyResult,
$"RegNotifyChangeKeyValue failed for '{_subKey}'.");
}
while (!cancellationToken.IsCancellationRequested) UnregisterWait();
_registeredWait = ThreadPool.RegisterWaitForSingleObject(
_signal,
OnSignalled,
state: null,
millisecondsTimeOutInterval: Timeout.Infinite,
executeOnlyOnce: true);
_trace?.Invoke("Armed — waiting for change notification (ThreadPool)");
return null;
}
private void OnSignalled(object state, bool timedOut)
{
// Drop the (now-consumed, executeOnlyOnce) registration reference.
lock (_sync)
{
_registeredWait = null;
}
bool proceed;
lock (_sync)
{
proceed = _running && !_stopped && !_disposed && !timedOut;
}
if (!proceed)
{
_trace?.Invoke("Signalled after stop/dispose — ignoring");
return;
}
_trace?.Invoke("Change notification received — invoking Changed handler");
// Raise Changed OUTSIDE the lock so a handler may call RequestStop() (or post work to
// another thread) without risk of a lock-ordering surprise. Handler exceptions are
// routed to Error, never allowed to escape the ThreadPool callback.
try
{
Changed?.Invoke(this, EventArgs.Empty);
}
catch (Exception ex)
{
RaiseError(ex);
}
// Re-arm for the next change (RegNotifyChangeKeyValue is single-shot).
Exception armError = null;
lock (_sync)
{
if (!_running || _stopped || _disposed)
{ {
iteration++; _trace?.Invoke("Stop requested during Changed — not re-arming");
return;
_trace?.Invoke($"Iteration {iteration}: calling RegNotifyChangeKeyValue (subtree={_watchSubtree}, filter=0x{(uint)_filter:X})");
int notifyResult = RegistryNativeMethods.RegNotifyChangeKeyValue(
keyHandle,
_watchSubtree,
_filter,
changedEvent.SafeWaitHandle,
true);
if (notifyResult != 0)
{
throw new Win32Exception(
notifyResult,
$"RegNotifyChangeKeyValue failed for '{_subKey}'.");
}
_trace?.Invoke($"Iteration {iteration}: waiting for change notification...");
int signaled = WaitHandle.WaitAny(waitHandles);
if (signaled == WaitHandle.WaitTimeout)
{
_trace?.Invoke($"Iteration {iteration}: WaitAny returned WaitTimeout — continuing");
continue;
}
if (signaled == 1)
{
_trace?.Invoke($"Iteration {iteration}: cancel event signaled — exiting loop");
break;
}
_trace?.Invoke($"Iteration {iteration}: change notification received — invoking Changed handler");
try
{
Changed?.Invoke(this, EventArgs.Empty);
}
catch (Exception ex)
{
try
{
Error?.Invoke(this, ex);
}
catch
{
// Ignore secondary error-handler failures.
}
}
_trace?.Invoke($"Iteration {iteration}: Changed handler completed");
} }
_trace?.Invoke("WatchLoop exiting (cancellation requested)"); armError = ArmNotify();
cancellationToken.ThrowIfCancellationRequested(); if (armError != null)
_backgroundException = armError;
} }
if (armError != null)
RaiseError(armError);
}
private void RaiseError(Exception ex)
{
try
{
Error?.Invoke(this, ex);
}
catch
{
// Never let an Error-handler exception escape the ThreadPool callback / Start.
}
}
/// <summary>Unregisters the pending ThreadPool wait without blocking. Must hold <see cref="_sync"/>.</summary>
private void UnregisterWait()
{
// Unregister(null) is non-blocking: it does NOT wait for an in-flight callback, so it is
// safe to call from within OnSignalled itself (mirrors Office RegistryChangeWatcher).
try { _registeredWait?.Unregister(null); } catch { /* best-effort */ }
_registeredWait = null;
} }
public void Dispose() public void Dispose()
@@ -280,7 +328,10 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime
// Dispose must not throw. // Dispose must not throw.
} }
_disposed = true; lock (_sync)
{
_disposed = true;
}
} }
private void ThrowIfDisposed() private void ThrowIfDisposed()
@@ -308,4 +359,4 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Runtime
} }
} }
} }
} }
@@ -28,6 +28,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Gather.Collectors
{ {
int index = 0; int index = 0;
foreach (ManagementObject obj in searcher.Get()) foreach (ManagementObject obj in searcher.Get())
using (obj) // ManagementObject holds native WMI/COM memory — dispose per iteration
{ {
var item = new Dictionary<string, object>(); var item = new Dictionary<string, object>();
foreach (var prop in obj.Properties) foreach (var prop in obj.Properties)
@@ -62,7 +62,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
// Prime the baseline for delta calculations // Prime the baseline for delta calculations
try try
{ {
var proc = Process.GetCurrentProcess(); using var proc = Process.GetCurrentProcess();
_prevCpuTime = proc.TotalProcessorTime; _prevCpuTime = proc.TotalProcessorTime;
_prevWallTime = DateTime.UtcNow; _prevWallTime = DateTime.UtcNow;
} }
@@ -90,7 +90,7 @@ namespace AutopilotMonitor.Agent.V2.Core.Monitoring.Telemetry.Periodic
// --- Process metrics (no WMI, no PerformanceCounter) --- // --- Process metrics (no WMI, no PerformanceCounter) ---
try try
{ {
var proc = Process.GetCurrentProcess(); using var proc = Process.GetCurrentProcess();
proc.Refresh(); // ensure fresh values proc.Refresh(); // ensure fresh values
// CPU %: (delta CPU time) / (delta wall time) / cores * 100 // CPU %: (delta CPU time) / (delta wall time) / cores * 100