Revert Work around potential race in PipeWriter (#10315)

John Luo · web-flow · commit a05d555f115b · 2019-05-17T01:20:52.000-07:00
* Remove Flaky attributes * Revert "Work around potential race in PipeWriter (#10165)" This reverts commit bff1d0e.
diff --git a/src/ProjectTemplates/test/SpaTemplateTest/AngularTemplateTest.cs b/src/ProjectTemplates/test/SpaTemplateTest/AngularTemplateTest.cs
@@ -17,7 +17,6 @@ public AngularTemplateTest(ProjectFactoryFixture projectFactory, BrowserFixture
             : base(projectFactory, browserFixture, output) { }
 
         [Fact]
-        [Flaky("https://github.com/aspnet/AspNetCore-Internal/issues/2422", FlakyOn.All)]
         public Task AngularTemplate_Works()
             => SpaTemplateImplAsync("angularnoauth", "angular", useLocalDb: false, usesAuth: false);
 
diff --git a/src/Servers/Kestrel/Core/src/Internal/Infrastructure/TimingPipeFlusher.cs b/src/Servers/Kestrel/Core/src/Internal/Infrastructure/TimingPipeFlusher.cs
@@ -51,39 +51,35 @@ public ValueTask<FlushResult> FlushAsync(MinDataRate minRate, long count)
 
         public ValueTask<FlushResult> FlushAsync(MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)
         {
-            // https://github.com/dotnet/corefxlab/issues/1334
-            // Pipelines don't support multiple awaiters on flush.
-            lock (_flushLock)
-            {
-                if (_lastFlushTask != null && !_lastFlushTask.IsCompleted)
-                {
-                    _lastFlushTask = AwaitLastFlushAndTimeFlushAsync(_lastFlushTask, minRate, count, outputAborter, cancellationToken);
-                    return new ValueTask<FlushResult>(_lastFlushTask);
-                }
-
-                return TimeFlushAsync(minRate, count, outputAborter, cancellationToken);
-            }
-        }
-
-        private ValueTask<FlushResult> TimeFlushAsync(MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)
-        {
-            var pipeFlushTask = _writer.FlushAsync(cancellationToken);
+            var flushValueTask = _writer.FlushAsync(cancellationToken);
 
             if (minRate != null)
             {
                 _timeoutControl.BytesWrittenToBuffer(minRate, count);
             }
 
-            if (pipeFlushTask.IsCompletedSuccessfully)
+            if (flushValueTask.IsCompletedSuccessfully)
             {
-                return new ValueTask<FlushResult>(pipeFlushTask.Result);
+                return new ValueTask<FlushResult>(flushValueTask.Result);
             }
 
-            _lastFlushTask = TimeFlushAsyncAwaited(pipeFlushTask, minRate, count, outputAborter, cancellationToken);
-            return new ValueTask<FlushResult>(_lastFlushTask);
+            // https://github.com/dotnet/corefxlab/issues/1334
+            // Pipelines don't support multiple awaiters on flush.
+            // While it's acceptable to call PipeWriter.FlushAsync again before the last FlushAsync completes,
+            // it is not acceptable to attach a new continuation (via await, AsTask(), etc..). In this case,
+            // we find previous flush Task which still accounts for any newly committed bytes and await that.
+            lock (_flushLock)
+            {
+                if (_lastFlushTask == null || _lastFlushTask.IsCompleted)
+                {
+                    _lastFlushTask = flushValueTask.AsTask();
+                }
+
+                return TimeFlushAsync(minRate, count, outputAborter, cancellationToken);
+            }
         }
 
-        private async Task<FlushResult> TimeFlushAsyncAwaited(ValueTask<FlushResult> pipeFlushTask, MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)
+        private async ValueTask<FlushResult> TimeFlushAsync(MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)
         {
             if (minRate != null)
             {
@@ -92,7 +88,7 @@ private async Task<FlushResult> TimeFlushAsyncAwaited(ValueTask<FlushResult> pip
 
             try
             {
-                return await pipeFlushTask;
+                return await _lastFlushTask;
             }
             catch (OperationCanceledException ex) when (outputAborter != null)
             {
@@ -115,11 +111,5 @@ private async Task<FlushResult> TimeFlushAsyncAwaited(ValueTask<FlushResult> pip
 
             return default;
         }
-
-        private async Task<FlushResult> AwaitLastFlushAndTimeFlushAsync(Task lastFlushTask, MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)
-        {
-            await lastFlushTask;
-            return await TimeFlushAsync(minRate, count, outputAborter, cancellationToken);
-        }
     }
 }
diff --git a/src/Servers/Kestrel/Core/test/TimingPipeFlusherTests.cs b/src/Servers/Kestrel/Core/test/TimingPipeFlusherTests.cs
diff --git a/src/Servers/Kestrel/Transport.Libuv/test/LibuvOutputConsumerTests.cs b/src/Servers/Kestrel/Transport.Libuv/test/LibuvOutputConsumerTests.cs
@@ -553,15 +553,15 @@ await Task.Run(async () =>
                     Assert.False(task1Waits.IsFaulted);
 
                     // following tasks should wait.
-                    var task2Canceled = outputProducer.WriteDataAsync(fullBuffer, cancellationToken: abortedSource.Token);
+                    var task3Canceled = outputProducer.WriteDataAsync(fullBuffer, cancellationToken: abortedSource.Token);
 
                     // Give time for tasks to percolate
                     await _mockLibuv.OnPostTask;
 
-                    // Second task is not completed
-                    Assert.False(task2Canceled.IsCompleted);
-                    Assert.False(task2Canceled.IsCanceled);
-                    Assert.False(task2Canceled.IsFaulted);
+                    // Third task is not completed
+                    Assert.False(task3Canceled.IsCompleted);
+                    Assert.False(task3Canceled.IsCanceled);
+                    Assert.False(task3Canceled.IsFaulted);
 
                     abortedSource.Cancel();
 
@@ -571,29 +571,29 @@ await Task.Run(async () =>
                         await _libuvThread.PostAsync(cb => cb(0), triggerNextCompleted);
                     }
 
-                    // First task is completed
+                    // First task is  completed
                     Assert.True(task1Waits.IsCompleted);
                     Assert.False(task1Waits.IsCanceled);
                     Assert.False(task1Waits.IsFaulted);
 
-                    // Second task is now canceled
-                    await Assert.ThrowsAsync<OperationCanceledException>(() => task2Canceled);
-                    Assert.True(task2Canceled.IsCanceled);
+                    // A final write guarantees that the error is observed by OutputProducer,
+                    // but doesn't return a canceled/faulted task.
+                    var task4Success = outputProducer.WriteDataAsync(fullBuffer);
+                    Assert.True(task4Success.IsCompleted);
+                    Assert.False(task4Success.IsCanceled);
+                    Assert.False(task4Success.IsFaulted);
 
-                    // A final write can still succeed.
-                    var task3Success = outputProducer.WriteDataAsync(fullBuffer);
+                    // Third task is now canceled
+                    await Assert.ThrowsAsync<OperationCanceledException>(() => task3Canceled);
+                    Assert.True(task3Canceled.IsCanceled);
 
                     await _mockLibuv.OnPostTask;
 
-                    // Complete the 3rd write
+                    // Complete the 4th write
                     while (completeQueue.TryDequeue(out var triggerNextCompleted))
                     {
                         await _libuvThread.PostAsync(cb => cb(0), triggerNextCompleted);
                     }
-
-                    Assert.True(task3Success.IsCompleted);
-                    Assert.False(task3Success.IsCanceled);
-                    Assert.False(task3Success.IsFaulted);;
                 }
             });
         }
diff --git a/src/Servers/Kestrel/perf/Kestrel.Performance/Http1LargeWritingBenchmark.cs b/src/Servers/Kestrel/perf/Kestrel.Performance/Http1LargeWritingBenchmark.cs
diff --git a/src/Servers/Kestrel/perf/Kestrel.Performance/Http1WritingBenchmark.cs b/src/Servers/Kestrel/perf/Kestrel.Performance/Http1WritingBenchmark.cs
@@ -29,7 +29,6 @@ public class Http1WritingBenchmark
         private TestHttp1Connection _http1Connection;
         private DuplexPipe.DuplexPipePair _pair;
         private MemoryPool<byte> _memoryPool;
-        private Task _consumeResponseBodyTask;
 
         private readonly byte[] _writeData = Encoding.ASCII.GetBytes("Hello, World!");
 
@@ -38,7 +37,6 @@ public void GlobalSetup()
         {
             _memoryPool = KestrelMemoryPool.Create();
             _http1Connection = MakeHttp1Connection();
-            _consumeResponseBodyTask = ConsumeResponseBody();
         }
 
         [Params(true, false)]
@@ -127,18 +125,14 @@ private TestHttp1Connection MakeHttp1Connection()
             return http1Connection;
         }
 
-        private async Task ConsumeResponseBody()
+        [IterationCleanup]
+        public void Cleanup()
         {
             var reader = _pair.Application.Input;
-            var readResult = await reader.ReadAsync();
-
-            while (!readResult.IsCompleted)
+            if (reader.TryRead(out var readResult))
             {
                 reader.AdvanceTo(readResult.Buffer.End);
-                readResult = await reader.ReadAsync();
             }
-
-            reader.Complete();
         }
 
         public enum Startup
@@ -151,8 +145,6 @@ public enum Startup
         [GlobalCleanup]
         public void Dispose()
         {
-            _pair.Transport.Output.Complete();
-            _consumeResponseBodyTask.GetAwaiter().GetResult();
             _memoryPool?.Dispose();
         }
     }

Original file line number	Diff line number	Diff line change
`@@ -51,39 +51,35 @@ public ValueTask<FlushResult> FlushAsync(MinDataRate minRate, long count)`
`51`	`51`
`52`	`52`	`public ValueTask<FlushResult> FlushAsync(MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)`
`53`	`53`	`{`
`54`		`- // https://github.com/dotnet/corefxlab/issues/1334`
`55`		`- // Pipelines don't support multiple awaiters on flush.`
`56`		`- lock (_flushLock)`
`57`		`- {`
`58`		`- if (_lastFlushTask != null && !_lastFlushTask.IsCompleted)`
`59`		`- {`
`60`		`- _lastFlushTask = AwaitLastFlushAndTimeFlushAsync(_lastFlushTask, minRate, count, outputAborter, cancellationToken);`
`61`		`- return new ValueTask<FlushResult>(_lastFlushTask);`
`62`		`- }`
`63`		`-`
`64`		`- return TimeFlushAsync(minRate, count, outputAborter, cancellationToken);`
`65`		`- }`
`66`		`- }`
`67`		`-`
`68`		`- private ValueTask<FlushResult> TimeFlushAsync(MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)`
`69`		`- {`
`70`		`- var pipeFlushTask = _writer.FlushAsync(cancellationToken);`
	`54`	`+ var flushValueTask = _writer.FlushAsync(cancellationToken);`
`71`	`55`
`72`	`56`	`if (minRate != null)`
`73`	`57`	`{`
`74`	`58`	`_timeoutControl.BytesWrittenToBuffer(minRate, count);`
`75`	`59`	`}`
`76`	`60`
`77`		`- if (pipeFlushTask.IsCompletedSuccessfully)`
	`61`	`+ if (flushValueTask.IsCompletedSuccessfully)`
`78`	`62`	`{`
`79`		`- return new ValueTask<FlushResult>(pipeFlushTask.Result);`
	`63`	`+ return new ValueTask<FlushResult>(flushValueTask.Result);`
`80`	`64`	`}`
`81`	`65`
`82`		`- _lastFlushTask = TimeFlushAsyncAwaited(pipeFlushTask, minRate, count, outputAborter, cancellationToken);`
`83`		`- return new ValueTask<FlushResult>(_lastFlushTask);`
	`66`	`+ // https://github.com/dotnet/corefxlab/issues/1334`
	`67`	`+ // Pipelines don't support multiple awaiters on flush.`
	`68`	`+ // While it's acceptable to call PipeWriter.FlushAsync again before the last FlushAsync completes,`
	`69`	`+ // it is not acceptable to attach a new continuation (via await, AsTask(), etc..). In this case,`
	`70`	`+ // we find previous flush Task which still accounts for any newly committed bytes and await that.`
	`71`	`+ lock (_flushLock)`
	`72`	`+ {`
	`73`	`+ if (_lastFlushTask == null \|\| _lastFlushTask.IsCompleted)`
	`74`	`+ {`
	`75`	`+ _lastFlushTask = flushValueTask.AsTask();`
	`76`	`+ }`
	`77`	`+`
	`78`	`+ return TimeFlushAsync(minRate, count, outputAborter, cancellationToken);`
	`79`	`+ }`
`84`	`80`	`}`
`85`	`81`
`86`		`- private async Task<FlushResult> TimeFlushAsyncAwaited(ValueTask<FlushResult> pipeFlushTask, MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)`
	`82`	`+ private async ValueTask<FlushResult> TimeFlushAsync(MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)`
`87`	`83`	`{`
`88`	`84`	`if (minRate != null)`
`89`	`85`	`{`
`@@ -92,7 +88,7 @@ private async Task<FlushResult> TimeFlushAsyncAwaited(ValueTask<FlushResult> pip`
`92`	`88`
`93`	`89`	`try`
`94`	`90`	`{`
`95`		`- return await pipeFlushTask;`
	`91`	`+ return await _lastFlushTask;`
`96`	`92`	`}`
`97`	`93`	`catch (OperationCanceledException ex) when (outputAborter != null)`
`98`	`94`	`{`
`@@ -115,11 +111,5 @@ private async Task<FlushResult> TimeFlushAsyncAwaited(ValueTask<FlushResult> pip`
`115`	`111`
`116`	`112`	`return default;`
`117`	`113`	`}`
`118`		`-`
`119`		`- private async Task<FlushResult> AwaitLastFlushAndTimeFlushAsync(Task lastFlushTask, MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)`
`120`		`- {`
`121`		`- await lastFlushTask;`
`122`		`- return await TimeFlushAsync(minRate, count, outputAborter, cancellationToken);`
`123`		`- }`
`124`	`114`	`}`
`125`	`115`	`}`
Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,6 @@ public class Http1WritingBenchmark`
`29`	`29`	`private TestHttp1Connection _http1Connection;`
`30`	`30`	`private DuplexPipe.DuplexPipePair _pair;`
`31`	`31`	`private MemoryPool<byte> _memoryPool;`
`32`		`- private Task _consumeResponseBodyTask;`
`33`	`32`
`34`	`33`	`private readonly byte[] _writeData = Encoding.ASCII.GetBytes("Hello, World!");`
`35`	`34`
`@@ -38,7 +37,6 @@ public void GlobalSetup()`
`38`	`37`	`{`
`39`	`38`	`_memoryPool = KestrelMemoryPool.Create();`
`40`	`39`	`_http1Connection = MakeHttp1Connection();`
`41`		`- _consumeResponseBodyTask = ConsumeResponseBody();`
`42`	`40`	`}`
`43`	`41`
`44`	`42`	`[Params(true, false)]`
`@@ -127,18 +125,14 @@ private TestHttp1Connection MakeHttp1Connection()`
`127`	`125`	`return http1Connection;`
`128`	`126`	`}`
`129`	`127`
`130`		`- private async Task ConsumeResponseBody()`
	`128`	`+ [IterationCleanup]`
	`129`	`+ public void Cleanup()`
`131`	`130`	`{`
`132`	`131`	`var reader = _pair.Application.Input;`
`133`		`- var readResult = await reader.ReadAsync();`
`134`		`-`
`135`		`- while (!readResult.IsCompleted)`
	`132`	`+ if (reader.TryRead(out var readResult))`
`136`	`133`	`{`
`137`	`134`	`reader.AdvanceTo(readResult.Buffer.End);`
`138`		`- readResult = await reader.ReadAsync();`
`139`	`135`	`}`
`140`		`-`
`141`		`- reader.Complete();`
`142`	`136`	`}`
`143`	`137`
`144`	`138`	`public enum Startup`
`@@ -151,8 +145,6 @@ public enum Startup`
`151`	`145`	`[GlobalCleanup]`
`152`	`146`	`public void Dispose()`
`153`	`147`	`{`
`154`		`- _pair.Transport.Output.Complete();`
`155`		`- _consumeResponseBodyTask.GetAwaiter().GetResult();`
`156`	`148`	`_memoryPool?.Dispose();`
`157`	`149`	`}`
`158`	`150`	`}`