Fix race in AsyncProcessOutputReader (#2180)

The process.WaitForExit call doesn't guarantee that the last calls of
ProcessOnOutputDataReceived and ProcessOnErrorDataReceived are processed.
The subsequent call of reader.StopRead detaches the *DataReceived events,
which leaves the output/error queues incomplete.
This race leads to flakiness of various tests (e.g.,
AllSetupAndCleanupMethodRunsForSpecificBenchmark).

Author: AndreyAkinshin

src/BenchmarkDotNet/Loggers/AsyncProcessOutputReader.cs

@@ -10,9 +10,12 @@ namespace BenchmarkDotNet.Loggers
     internal class AsyncProcessOutputReader : IDisposable
     {
         private readonly Process process;
-        private readonly ConcurrentQueue<string> output, error;
-        private readonly bool logOutput, readStandardError;
         private readonly ILogger logger;
+        private readonly bool logOutput, readStandardError;
+
+        private static readonly TimeSpan FinishEventTimeout = TimeSpan.FromMilliseconds(500);
+        private readonly AutoResetEvent outputFinishEvent, errorFinishEvent;
+        private readonly ConcurrentQueue<string> output, error;
 
         private long status;
 
@@ -28,6 +31,8 @@ internal AsyncProcessOutputReader(Process process, bool logOutput = false, ILogg
             this.process = process;
             output = new ConcurrentQueue<string>();
             error = new ConcurrentQueue<string>();
+            outputFinishEvent = new AutoResetEvent(false);
+            errorFinishEvent = new AutoResetEvent(false);
             status = (long)Status.Created;
             this.logOutput = logOutput;
             this.logger = logger;
@@ -39,6 +44,9 @@ public void Dispose()
             Interlocked.Exchange(ref status, (long)Status.Disposed);
 
             Detach();
+
+            outputFinishEvent.Dispose();
+            errorFinishEvent.Dispose();
         }
 
         internal void BeginRead()
@@ -72,6 +80,10 @@ internal void StopRead()
             if (Interlocked.CompareExchange(ref status, (long)Status.Stopped, (long)Status.Started) != (long)Status.Started)
                 throw new InvalidOperationException("Only a started reader can be stopped");
 
+            outputFinishEvent.WaitOne(FinishEventTimeout);
+            if (readStandardError)
+                errorFinishEvent.WaitOne(FinishEventTimeout);
+
             Detach();
         }
 
@@ -103,34 +115,44 @@ private void Detach()
 
         private void ProcessOnOutputDataReceived(object sender, DataReceivedEventArgs e)
         {
-            if (!string.IsNullOrEmpty(e.Data))
+            if (e.Data != null)
             {
-                output.Enqueue(e.Data);
-
-                if (logOutput)
+                if (!string.IsNullOrEmpty(e.Data))
                 {
-                    lock (this) // #2125
+                    output.Enqueue(e.Data);
+
+                    if (logOutput)
                     {
-                        logger.WriteLine(e.Data);
+                        lock (this) // #2125
+                        {
+                            logger.WriteLine(e.Data);
+                        }
                     }
                 }
             }
+            else // 'e.Data == null' means EOF
+                outputFinishEvent.Set();
         }
 
         private void ProcessOnErrorDataReceived(object sender, DataReceivedEventArgs e)
         {
-            if (!string.IsNullOrEmpty(e.Data))
+            if (e.Data != null)
             {
-                error.Enqueue(e.Data);
-
-                if (logOutput)
+                if (!string.IsNullOrEmpty(e.Data))
                 {
-                    lock (this) // #2125
+                    error.Enqueue(e.Data);
+
+                    if (logOutput)
                     {
-                        logger.WriteLineError(e.Data);
+                        lock (this) // #2125
+                        {
+                            logger.WriteLineError(e.Data);
+                        }
                     }
                 }
             }
+            else // 'e.Data == null' means EOF
+                errorFinishEvent.Set();
         }
 
         private T ReturnIfStopped<T>(Func<T> getter)
@@ -146,4 +168,4 @@ private enum Status : long
             Disposed
         }
     }
-}
+}