diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DiagnosticsEventPipeProcessor.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DiagnosticsEventPipeProcessor.cs index b07a1742db..b843607503 100644 --- a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DiagnosticsEventPipeProcessor.cs +++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DiagnosticsEventPipeProcessor.cs @@ -43,7 +43,7 @@ Func, CancellationToken, Task> onEventSourceAva public async Task Process(DiagnosticsClient client, TimeSpan duration, CancellationToken token) { //No need to guard against reentrancy here, since the calling pipeline does this already. - IDisposable registration = token.Register(() => _initialized.TrySetCanceled()); + IDisposable registration = token.Register(() => TryCancelCompletionSources(token)); await await Task.Factory.StartNew(async () => { EventPipeEventSource source = null; @@ -82,7 +82,13 @@ await await Task.Factory.StartNew(async () => } catch (DiagnosticsClientException ex) { - throw new InvalidOperationException("Failed to start the event pipe session", ex); + InvalidOperationException wrappingException = new("Failed to start the event pipe session", ex); + TryFailCompletionSourcesReturnFalse(wrappingException); + throw wrappingException; + } + catch (Exception ex) when (TryFailCompletionSourcesReturnFalse(ex)) + { + throw; } finally { @@ -156,5 +162,31 @@ public async ValueTask DisposeAsync() _eventSource?.Dispose(); } + + // Helper method for observing an exception while processing the trace session + // so that session start task completion source can be failed and the exception handler + // does not catch the exception. + private bool TryFailCompletionSourcesReturnFalse(Exception ex) + { + // Use best-effort to set the completion sources to be cancelled or failed. + if (ex is OperationCanceledException canceledException) + { + TryCancelCompletionSources(canceledException.CancellationToken); + } + else + { + _initialized.TrySetException(ex); + _sessionStarted.TrySetException(ex); + } + + // Return false to make the exception handler not handle the exception. + return false; + } + + private void TryCancelCompletionSources(CancellationToken token) + { + _initialized.TrySetCanceled(token); + _sessionStarted.TrySetCanceled(token); + } } } diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs index 0123c2dab8..125a60642b 100644 --- a/src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs +++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs @@ -10,14 +10,12 @@ namespace Microsoft.Diagnostics.Monitoring.EventPipe { - internal abstract class EventSourcePipeline : Pipeline, IEventSourcePipelineInternal where T : EventSourcePipelineSettings + internal abstract class EventSourcePipeline : Pipeline where T : EventSourcePipelineSettings { private readonly Lazy _processor; public DiagnosticsClient Client { get; } public T Settings { get; } - Task IEventSourcePipelineInternal.SessionStarted => _processor.Value.SessionStarted; - protected EventSourcePipeline(DiagnosticsClient client, T settings) { Client = client ?? throw new ArgumentNullException(nameof(client)); @@ -67,16 +65,36 @@ protected override async Task OnStop(CancellationToken token) } } + /// + /// Starts the pipeline and returns a that completes when the pipeline + /// finishes running to completion. + /// + /// The token to monitor for cancellation requests. + /// + /// A task that completes when the event pipe session for the pipeline has started. The inner + /// task completes when the pipeline runs to completion. + /// + /// + /// The will cancel the running on the pipeline if it is signaled + /// before the pipeline runs to completion. + /// + public async Task StartAsync(CancellationToken token) + { + Task runTask = RunAsync(token); + + // Await both the session started or the run task and return when either is completed. + // This works around an issue where the run task may fail but not cancel/fault the session + // started task. Logically, the run task will not successfully complete before the session + // started task. Thus, the combined task completes either when the session started task is + // completed OR the run task has cancelled/failed. + await Task.WhenAny(_processor.Value.SessionStarted, runTask).Unwrap(); + + return runTask; + } + protected virtual Task OnEventSourceAvailable(EventPipeEventSource eventSource, Func stopSessionAsync, CancellationToken token) { return Task.CompletedTask; } } - - internal interface IEventSourcePipelineInternal - { - // Allows tests to know when the event pipe session has started so that the - // target application can start producing events. - Task SessionStarted { get; } - } } diff --git a/src/Microsoft.Diagnostics.NETCore.Client/DiagnosticsIpc/IpcTransport.cs b/src/Microsoft.Diagnostics.NETCore.Client/DiagnosticsIpc/IpcTransport.cs index 150459037f..10d13a100f 100644 --- a/src/Microsoft.Diagnostics.NETCore.Client/DiagnosticsIpc/IpcTransport.cs +++ b/src/Microsoft.Diagnostics.NETCore.Client/DiagnosticsIpc/IpcTransport.cs @@ -83,7 +83,7 @@ public static async Task ConnectAsync(IpcEndpointConfig config, Cancella ".", config.Address, PipeDirection.InOut, - PipeOptions.None, + PipeOptions.Asynchronous, TokenImpersonationLevel.Impersonation); await namedPipe.ConnectAsync(token).ConfigureAwait(false); return namedPipe; diff --git a/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/PipelineTestUtilities.cs b/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/PipelineTestUtilities.cs index 1769288212..bc7992542a 100644 --- a/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/PipelineTestUtilities.cs +++ b/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/PipelineTestUtilities.cs @@ -12,22 +12,68 @@ namespace Microsoft.Diagnostics.Monitoring.EventPipe.UnitTests { internal static class PipelineTestUtilities { - public static async Task ExecutePipelineWithDebugee(ITestOutputHelper outputHelper, Pipeline pipeline, RemoteTestExecution testExecution, TaskCompletionSource waitTaskSource = null) + private static readonly TimeSpan DefaultPipelineRunTimeout = TimeSpan.FromMinutes(1); + + public static async Task ExecutePipelineWithDebugee( + ITestOutputHelper outputHelper, + Pipeline pipeline, + RemoteTestExecution testExecution, + TaskCompletionSource waitTaskSource = null) { - using var cancellation = new CancellationTokenSource(TimeSpan.FromMinutes(1)); + using var cancellation = new CancellationTokenSource(DefaultPipelineRunTimeout); - await ExecutePipelineWithDebugee(outputHelper, pipeline, testExecution, cancellation.Token, waitTaskSource); + await ExecutePipelineWithDebugee( + outputHelper, + pipeline, + testExecution, + cancellation.Token, + waitTaskSource); } - public static async Task ExecutePipelineWithDebugee(ITestOutputHelper outputHelper, Pipeline pipeline, RemoteTestExecution testExecution, CancellationToken token, TaskCompletionSource waitTaskSource = null) + public static async Task ExecutePipelineWithDebugee( + ITestOutputHelper outputHelper, + EventSourcePipeline pipeline, + RemoteTestExecution testExecution, + TaskCompletionSource waitTaskSource = null) + where T : EventSourcePipelineSettings { - Task processingTask = pipeline.RunAsync(token); + using var cancellation = new CancellationTokenSource(DefaultPipelineRunTimeout); - // Wait for event session to be established before telling target app to produce events. - if (pipeline is IEventSourcePipelineInternal eventSourcePipeline) - { - await eventSourcePipeline.SessionStarted; - } + await ExecutePipelineWithDebugee( + outputHelper, + pipeline, + (p, t) => p.StartAsync(t), + testExecution, + cancellation.Token, + waitTaskSource); + } + + public static Task ExecutePipelineWithDebugee( + ITestOutputHelper outputHelper, + Pipeline pipeline, + RemoteTestExecution testExecution, + CancellationToken token, + TaskCompletionSource waitTaskSource = null) + { + return ExecutePipelineWithDebugee( + outputHelper, + pipeline, + (p, t) => Task.FromResult(p.RunAsync(t)), + testExecution, + token, + waitTaskSource); + } + + private static async Task ExecutePipelineWithDebugee( + ITestOutputHelper outputHelper, + TPipeline pipeline, + Func> startPipelineAsync, + RemoteTestExecution testExecution, + CancellationToken token, + TaskCompletionSource waitTaskSource = null) + where TPipeline : Pipeline + { + Task runTask = await startPipelineAsync(pipeline, token); //Begin event production testExecution.SendSignal(); @@ -53,7 +99,7 @@ public static async Task ExecutePipelineWithDebugee(ITestOutputHelper outputHelp await pipeline.StopAsync(token); //After a pipeline is stopped, we should expect the RunTask to eventually finish - await processingTask; + await runTask; } finally {