huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix ConcurrentPipeWriterTests (#12383)

This commit is contained in:
Stephen Halter 2019-07-24 14:51:31 -07:00 committed by GitHub
parent c54a7209d7
commit 34903da1e9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 141 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

212
src/Servers/Kestrel/Core/test/ConcurrentPipeWriterTests.cs
View File

@ -25,6 +25,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
}; };
var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray); var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray);
// No need to pass in a real sync object since all the calls in this test are passthrough.
var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, new object()); var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, new object());
var memory = concurrentPipeWriter.GetMemory(); var memory = concurrentPipeWriter.GetMemory();
@ -71,24 +73,31 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
new TaskCompletionSource<FlushResult>(TaskCreationOptions.RunContinuationsAsynchronously), new TaskCompletionSource<FlushResult>(TaskCreationOptions.RunContinuationsAsynchronously),
}; };
var sync = new object();
var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray); var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray);
var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, new object()); var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, sync);
var flushTask0 = default(ValueTask<FlushResult>);
var flushTask1 = default(ValueTask<FlushResult>);
var completeTask = default(ValueTask);
var memory = concurrentPipeWriter.GetMemory(); lock (sync)
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount); {
var memory = concurrentPipeWriter.GetMemory();
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount);
concurrentPipeWriter.Advance(memory.Length); concurrentPipeWriter.Advance(memory.Length);
Assert.Equal(1, mockPipeWriter.AdvanceCallCount); Assert.Equal(1, mockPipeWriter.AdvanceCallCount);
var flushTask0 = concurrentPipeWriter.FlushAsync(); flushTask0 = concurrentPipeWriter.FlushAsync();
Assert.Equal(1, mockPipeWriter.FlushCallCount); Assert.Equal(1, mockPipeWriter.FlushCallCount);
Assert.False(flushTask0.IsCompleted); Assert.False(flushTask0.IsCompleted);
// Since the flush was not awaited, the following API calls are queued. // Since the flush was not awaited, the following API calls are queued.
memory = concurrentPipeWriter.GetMemory(); memory = concurrentPipeWriter.GetMemory();
concurrentPipeWriter.Advance(memory.Length); concurrentPipeWriter.Advance(memory.Length);
var flushTask1 = concurrentPipeWriter.FlushAsync(); flushTask1 = concurrentPipeWriter.FlushAsync();
}
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount); Assert.Equal(1, mockPipeWriter.GetMemoryCallCount);
Assert.Equal(1, mockPipeWriter.AdvanceCallCount); Assert.Equal(1, mockPipeWriter.AdvanceCallCount);
@ -99,11 +108,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
mockPipeWriter.FlushTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously); mockPipeWriter.FlushTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
pipeWriterFlushTcsArray[0].SetResult(default); pipeWriterFlushTcsArray[0].SetResult(default);
await mockPipeWriter.FlushTcs.Task.DefaultTimeout(); await mockPipeWriter.FlushTcs.Task.DefaultTimeout();
// Since the flush was not awaited, the following API calls are queued. lock (sync)
memory = concurrentPipeWriter.GetMemory(); {
concurrentPipeWriter.Advance(memory.Length); // Since the flush was not awaited, the following API calls are queued.
var memory = concurrentPipeWriter.GetMemory();
concurrentPipeWriter.Advance(memory.Length);
}
// We do not need to flush the final bytes, since the incomplete flush will pick it up. // We do not need to flush the final bytes, since the incomplete flush will pick it up.
Assert.Equal(2, mockPipeWriter.GetMemoryCallCount); Assert.Equal(2, mockPipeWriter.GetMemoryCallCount);
@ -115,6 +128,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
mockPipeWriter.FlushTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously); mockPipeWriter.FlushTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
pipeWriterFlushTcsArray[1].SetResult(default); pipeWriterFlushTcsArray[1].SetResult(default);
await mockPipeWriter.FlushTcs.Task.DefaultTimeout(); await mockPipeWriter.FlushTcs.Task.DefaultTimeout();
// Even though we only called flush on the ConcurrentPipeWriter twice, the inner PipeWriter was flushed three times. // Even though we only called flush on the ConcurrentPipeWriter twice, the inner PipeWriter was flushed three times.
@ -126,7 +140,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
Assert.False(flushTask1.IsCompleted); Assert.False(flushTask1.IsCompleted);
var completeEx = new Exception(); var completeEx = new Exception();
await concurrentPipeWriter.CompleteAsync(completeEx); lock (sync)
{
completeTask = concurrentPipeWriter.CompleteAsync(completeEx);
}
await completeTask.DefaultTimeout();
// Complete isn't called on the inner PipeWriter until the inner flushes have completed. // Complete isn't called on the inner PipeWriter until the inner flushes have completed.
Assert.Null(mockPipeWriter.CompleteException); Assert.Null(mockPipeWriter.CompleteException);
@ -151,21 +170,29 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
new TaskCompletionSource<FlushResult>(TaskCreationOptions.RunContinuationsAsynchronously), new TaskCompletionSource<FlushResult>(TaskCreationOptions.RunContinuationsAsynchronously),
}; };
var sync = new object();
var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray); var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray);
var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, new object()); var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, sync);
var memory = default(Memory<byte>);
var flushTask0 = default(ValueTask<FlushResult>);
var flushTask1 = default(ValueTask<FlushResult>);
var completeTask = default(ValueTask);
var memory = concurrentPipeWriter.GetMemory(); lock (sync)
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount); {
memory = concurrentPipeWriter.GetMemory();
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount);
concurrentPipeWriter.Advance(memory.Length); concurrentPipeWriter.Advance(memory.Length);
Assert.Equal(1, mockPipeWriter.AdvanceCallCount); Assert.Equal(1, mockPipeWriter.AdvanceCallCount);
var flushTask0 = concurrentPipeWriter.FlushAsync(); flushTask0 = concurrentPipeWriter.FlushAsync();
Assert.Equal(1, mockPipeWriter.FlushCallCount); Assert.Equal(1, mockPipeWriter.FlushCallCount);
Assert.False(flushTask0.IsCompleted); Assert.False(flushTask0.IsCompleted);
// Only GetMemory() is called but not Advance() is not called yet when the first inner flush complets. // Only GetMemory() is called but not Advance() is not called yet when the first inner flush complets.
memory = concurrentPipeWriter.GetMemory(); memory = concurrentPipeWriter.GetMemory();
}
// If the inner flush completes between a call to GetMemory() and Advance(), the outer // If the inner flush completes between a call to GetMemory() and Advance(), the outer
// flush completes, and the next flush will pick up the buffered data. // flush completes, and the next flush will pick up the buffered data.
@ -177,11 +204,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
Assert.Equal(1, mockPipeWriter.AdvanceCallCount); Assert.Equal(1, mockPipeWriter.AdvanceCallCount);
Assert.Equal(1, mockPipeWriter.FlushCallCount); Assert.Equal(1, mockPipeWriter.FlushCallCount);
concurrentPipeWriter.Advance(memory.Length); lock (sync)
memory = concurrentPipeWriter.GetMemory(); {
concurrentPipeWriter.Advance(memory.Length); concurrentPipeWriter.Advance(memory.Length);
memory = concurrentPipeWriter.GetMemory();
concurrentPipeWriter.Advance(memory.Length);
var flushTask1 = concurrentPipeWriter.FlushAsync(); flushTask1 = concurrentPipeWriter.FlushAsync();
}
// Now that we flushed the ConcurrentPipeWriter again, the GetMemory() and Advance() calls are replayed. // Now that we flushed the ConcurrentPipeWriter again, the GetMemory() and Advance() calls are replayed.
// Make sure that MockPipeWriter.SlabMemoryPoolBlockSize matches SlabMemoryPool._blockSize or else // Make sure that MockPipeWriter.SlabMemoryPoolBlockSize matches SlabMemoryPool._blockSize or else
@ -201,7 +231,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
Assert.Equal(2, mockPipeWriter.FlushCallCount); Assert.Equal(2, mockPipeWriter.FlushCallCount);
var completeEx = new Exception(); var completeEx = new Exception();
await concurrentPipeWriter.CompleteAsync(completeEx);
lock (sync)
{
completeTask = concurrentPipeWriter.CompleteAsync(completeEx);
}
await completeTask.DefaultTimeout();
Assert.Same(completeEx, mockPipeWriter.CompleteException); Assert.Same(completeEx, mockPipeWriter.CompleteException);
} }
} }
@ -217,21 +254,28 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
new TaskCompletionSource<FlushResult>(TaskCreationOptions.RunContinuationsAsynchronously), new TaskCompletionSource<FlushResult>(TaskCreationOptions.RunContinuationsAsynchronously),
}; };
var sync = new object();
var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray); var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray);
var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, new object()); var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, sync);
var memory = default(Memory<byte>);
var flushTask0 = default(ValueTask<FlushResult>);
var completeTask = default(ValueTask);
var memory = concurrentPipeWriter.GetMemory(); lock (sync)
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount); {
memory = concurrentPipeWriter.GetMemory();
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount);
concurrentPipeWriter.Advance(memory.Length); concurrentPipeWriter.Advance(memory.Length);
Assert.Equal(1, mockPipeWriter.AdvanceCallCount); Assert.Equal(1, mockPipeWriter.AdvanceCallCount);
var flushTask0 = concurrentPipeWriter.FlushAsync(); flushTask0 = concurrentPipeWriter.FlushAsync();
Assert.Equal(1, mockPipeWriter.FlushCallCount); Assert.Equal(1, mockPipeWriter.FlushCallCount);
Assert.False(flushTask0.IsCompleted); Assert.False(flushTask0.IsCompleted);
// Only GetMemory() is called but not Advance() is not called yet when the first inner flush completes. // Only GetMemory() is called but not Advance() is not called yet when the first inner flush completes.
memory = concurrentPipeWriter.GetMemory(); memory = concurrentPipeWriter.GetMemory();
}
// If the inner flush completes between a call to GetMemory() and Advance(), the outer // If the inner flush completes between a call to GetMemory() and Advance(), the outer
// flush completes, and the next flush will pick up the buffered data. // flush completes, and the next flush will pick up the buffered data.
@ -243,13 +287,19 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
Assert.Equal(1, mockPipeWriter.AdvanceCallCount); Assert.Equal(1, mockPipeWriter.AdvanceCallCount);
Assert.Equal(1, mockPipeWriter.FlushCallCount); Assert.Equal(1, mockPipeWriter.FlushCallCount);
concurrentPipeWriter.Advance(memory.Length);
memory = concurrentPipeWriter.GetMemory();
concurrentPipeWriter.Advance(memory.Length);
// Complete the ConcurrentPipeWriter without flushing any of the queued data.
var completeEx = new Exception(); var completeEx = new Exception();
await concurrentPipeWriter.CompleteAsync(completeEx);
lock (sync)
{
concurrentPipeWriter.Advance(memory.Length);
memory = concurrentPipeWriter.GetMemory();
concurrentPipeWriter.Advance(memory.Length);
// Complete the ConcurrentPipeWriter without flushing any of the queued data.
completeTask = concurrentPipeWriter.CompleteAsync(completeEx);
}
await completeTask.DefaultTimeout();
// Now that we completed the ConcurrentPipeWriter, the GetMemory() and Advance() calls are replayed. // Now that we completed the ConcurrentPipeWriter, the GetMemory() and Advance() calls are replayed.
// Make sure that MockPipeWriter.SlabMemoryPoolBlockSize matches SlabMemoryPool._blockSize or else // Make sure that MockPipeWriter.SlabMemoryPoolBlockSize matches SlabMemoryPool._blockSize or else
@ -272,45 +322,58 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
new TaskCompletionSource<FlushResult>(TaskCreationOptions.RunContinuationsAsynchronously), new TaskCompletionSource<FlushResult>(TaskCreationOptions.RunContinuationsAsynchronously),
}; };
var sync = new object();
var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray); var mockPipeWriter = new MockPipeWriter(pipeWriterFlushTcsArray);
var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, new object()); var concurrentPipeWriter = new ConcurrentPipeWriter(mockPipeWriter, diagnosticPool, sync);
var flushTask0 = default(ValueTask<FlushResult>);
var flushTask1 = default(ValueTask<FlushResult>);
var flushTask2 = default(ValueTask<FlushResult>);
var completeTask = default(ValueTask);
var memory = concurrentPipeWriter.GetMemory(); lock (sync)
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount); {
var memory = concurrentPipeWriter.GetMemory();
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount);
concurrentPipeWriter.Advance(memory.Length); concurrentPipeWriter.Advance(memory.Length);
Assert.Equal(1, mockPipeWriter.AdvanceCallCount); Assert.Equal(1, mockPipeWriter.AdvanceCallCount);
var flushTask0 = concurrentPipeWriter.FlushAsync(); flushTask0 = concurrentPipeWriter.FlushAsync();
Assert.Equal(1, mockPipeWriter.FlushCallCount); Assert.Equal(1, mockPipeWriter.FlushCallCount);
Assert.False(flushTask0.IsCompleted); Assert.False(flushTask0.IsCompleted);
// Since the flush was not awaited, the following API calls are queued.
memory = concurrentPipeWriter.GetMemory();
concurrentPipeWriter.Advance(memory.Length);
var flushTask1 = concurrentPipeWriter.FlushAsync();
Assert.Equal(1, mockPipeWriter.GetMemoryCallCount); // Since the flush was not awaited, the following API calls are queued.
Assert.Equal(1, mockPipeWriter.AdvanceCallCount); memory = concurrentPipeWriter.GetMemory();
Assert.Equal(1, mockPipeWriter.FlushCallCount); concurrentPipeWriter.Advance(memory.Length);
flushTask1 = concurrentPipeWriter.FlushAsync();
Assert.False(flushTask0.IsCompleted); Assert.Equal(1, mockPipeWriter.GetMemoryCallCount);
Assert.False(flushTask1.IsCompleted); Assert.Equal(1, mockPipeWriter.AdvanceCallCount);
Assert.Equal(1, mockPipeWriter.FlushCallCount);
// CancelPendingFlush() does not get queued. Assert.False(flushTask0.IsCompleted);
concurrentPipeWriter.CancelPendingFlush(); Assert.False(flushTask1.IsCompleted);
Assert.Equal(1, mockPipeWriter.CancelPendingFlushCallCount);
// CancelPendingFlush() does not get queued.
concurrentPipeWriter.CancelPendingFlush();
Assert.Equal(1, mockPipeWriter.CancelPendingFlushCallCount);
}
pipeWriterFlushTcsArray[0].SetResult(new FlushResult(isCanceled: true, isCompleted: false)); pipeWriterFlushTcsArray[0].SetResult(new FlushResult(isCanceled: true, isCompleted: false));
Assert.True((await flushTask0.DefaultTimeout()).IsCanceled); Assert.True((await flushTask0.DefaultTimeout()).IsCanceled);
Assert.True((await flushTask1.DefaultTimeout()).IsCanceled); Assert.True((await flushTask1.DefaultTimeout()).IsCanceled);
var flushTask2 = concurrentPipeWriter.FlushAsync(); lock (sync)
{
flushTask2 = concurrentPipeWriter.FlushAsync();
}
Assert.False(flushTask2.IsCompleted); Assert.False(flushTask2.IsCompleted);
pipeWriterFlushTcsArray[1].SetResult(default); pipeWriterFlushTcsArray[1].SetResult(default);
await flushTask2.DefaultTimeout(); await flushTask2.DefaultTimeout();
// We do not need to flush the final bytes, since the incomplete flush will pick it up. // We do not need to flush the final bytes, since the incomplete flush will pick it up.
@ -319,7 +382,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
Assert.Equal(2, mockPipeWriter.FlushCallCount); Assert.Equal(2, mockPipeWriter.FlushCallCount);
var completeEx = new Exception(); var completeEx = new Exception();
await concurrentPipeWriter.CompleteAsync(completeEx);
lock (sync)
{
completeTask = concurrentPipeWriter.CompleteAsync(completeEx);
}
await completeTask.DefaultTimeout();
Assert.Same(completeEx, mockPipeWriter.CompleteException); Assert.Same(completeEx, mockPipeWriter.CompleteException);
} }
} }