Fix processor to cancel/fail session started task source.

src/Microsoft.Diagnostics.Monitoring.EventPipe/DiagnosticsEventPipeProcessor.cs

@@ -43,7 +43,7 @@ Func<EventPipeEventSource, Func<Task>, 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);
+        }
     }
 }

src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs

@@ -10,14 +10,12 @@
 
 namespace Microsoft.Diagnostics.Monitoring.EventPipe
 {
-    internal abstract class EventSourcePipeline<T> : Pipeline, IEventSourcePipelineInternal where T : EventSourcePipelineSettings
+    internal abstract class EventSourcePipeline<T> : Pipeline where T : EventSourcePipelineSettings
     {
         private readonly Lazy<DiagnosticsEventPipeProcessor> _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)
             }
         }
 
+        /// <summary>
+        /// Starts the pipeline and returns a <see cref="Task"/> that completes when the pipeline
+        /// finishes running to completion.
+        /// </summary>
+        /// <param name="token">The token to monitor for cancellation requests.</param>
+        /// <returns>
+        /// A task that completes when the event pipe session for the pipeline has started. The inner
+        /// task completes when the pipeline runs to completion.
+        /// </returns>
+        /// <remarks>
+        /// The <paramref name="token"/> will cancel the running on the pipeline if it is signaled
+        /// before the pipeline runs to completion.
+        /// </remarks>
+        public async Task<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<Task> 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; }
-    }
 }

src/Microsoft.Diagnostics.NETCore.Client/DiagnosticsIpc/IpcTransport.cs

@@ -83,7 +83,7 @@ public static async Task<Stream> ConnectAsync(IpcEndpointConfig config, Cancella
                     ".",
                     config.Address,
                     PipeDirection.InOut,
-                    PipeOptions.None,
+                    PipeOptions.Asynchronous,
                     TokenImpersonationLevel.Impersonation);
                 await namedPipe.ConnectAsync(token).ConfigureAwait(false);
                 return namedPipe;

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<object> waitTaskSource = null)
+        private static readonly TimeSpan DefaultPipelineRunTimeout = TimeSpan.FromMinutes(1);
+
+        public static async Task ExecutePipelineWithDebugee(
+            ITestOutputHelper outputHelper,
+            Pipeline pipeline,
+            RemoteTestExecution testExecution,
+            TaskCompletionSource<object> 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<object> waitTaskSource = null)
+        public static async Task ExecutePipelineWithDebugee<T>(
+            ITestOutputHelper outputHelper,
+            EventSourcePipeline<T> pipeline,
+            RemoteTestExecution testExecution,
+            TaskCompletionSource<object> 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<object> waitTaskSource = null)
+        {
+            return ExecutePipelineWithDebugee(
+                outputHelper,
+                pipeline,
+                (p, t) => Task.FromResult(p.RunAsync(t)),
+                testExecution,
+                token,
+                waitTaskSource);
+        }
+
+        private static async Task ExecutePipelineWithDebugee<TPipeline>(
+            ITestOutputHelper outputHelper,
+            TPipeline pipeline,
+            Func<TPipeline, CancellationToken, Task<Task>> startPipelineAsync,
+            RemoteTestExecution testExecution,
+            CancellationToken token,
+            TaskCompletionSource<object> 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
             {