diff --git a/corefxlab.sln b/corefxlab.sln index de1eb1b7374..1c6f1910c34 100644 --- a/corefxlab.sln +++ b/corefxlab.sln @@ -2,7 +2,7 @@ Microsoft Visual Studio Solution File, Format Version 12.00 # Visual Studio 15 VisualStudioVersion = 15.0.27130.2020 MinimumVisualStudioVersion = 10.0.40219.1 -Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{5E7EB061-B9BC-4DA2-B5E5-859AA7C67695}" +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "MultiSegmentBytesReaderNumbers", "MultiSegmentBytesReaderNumbers", "{5E7EB061-B9BC-4DA2-B5E5-859AA7C67695}" ProjectSection(SolutionItems) = preProject global.json = global.json NuGet.Config = NuGet.Config diff --git a/samples/System.IO.Pipelines.Samples/CompressionSample.cs b/samples/System.IO.Pipelines.Samples/CompressionSample.cs index d368d53c789..eb804188306 100644 --- a/samples/System.IO.Pipelines.Samples/CompressionSample.cs +++ b/samples/System.IO.Pipelines.Samples/CompressionSample.cs @@ -34,7 +34,7 @@ public Task Run() // Wrap the console in a pipeline writer - var outputPipe = new Pipe(options); + var outputPipe = new ResetablePipe(options); outputPipe.Reader.CopyToEndAsync(Console.OpenStandardOutput()); // Copy from the file reader to the console writer diff --git a/samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpClientHandler.cs b/samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpClientHandler.cs index eb9be8551c9..d1c332c121e 100644 --- a/samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpClientHandler.cs +++ b/samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpClientHandler.cs @@ -243,7 +243,7 @@ private static async Task ProduceResponse(ConnectionState state, IPipeConnection } } - private static void WriteHeaders(HttpHeaders headers, IPipeWriter buffer) + private static void WriteHeaders(HttpHeaders headers, PipeWriter buffer) { foreach (var header in headers) { diff --git a/samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpContent.cs b/samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpContent.cs index 6d2d2b1b6ac..822d9597375 100644 --- a/samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpContent.cs +++ b/samples/System.IO.Pipelines.Samples/HttpClient/PipelineHttpContent.cs @@ -10,9 +10,9 @@ namespace System.IO.Pipelines.Samples { public class PipelineHttpContent : HttpContent { - private readonly IPipeReader _output; + private readonly PipeReader _output; - public PipelineHttpContent(IPipeReader output) + public PipelineHttpContent(PipeReader output) { _output = output; } diff --git a/samples/System.IO.Pipelines.Samples/HttpServer/HttpConnection.cs b/samples/System.IO.Pipelines.Samples/HttpServer/HttpConnection.cs index 99a135edbd3..29e4a3b42f3 100644 --- a/samples/System.IO.Pipelines.Samples/HttpServer/HttpConnection.cs +++ b/samples/System.IO.Pipelines.Samples/HttpServer/HttpConnection.cs @@ -17,8 +17,8 @@ public partial class HttpConnection private static readonly byte[] _chunkedEndBytes = Encoding.UTF8.GetBytes("0\r\n\r\n"); private static readonly byte[] _endChunkBytes = Encoding.ASCII.GetBytes("\r\n"); - private readonly IPipeReader _input; - private readonly IPipeWriter _output; + private readonly PipeReader _input; + private readonly PipeWriter _output; private readonly IHttpApplication _application; public RequestHeaderDictionary RequestHeaders => _parser.RequestHeaders; @@ -41,7 +41,7 @@ public partial class HttpConnection private HttpRequestParser _parser = new HttpRequestParser(); - public HttpConnection(IHttpApplication application, IPipeReader input, IPipeWriter output) + public HttpConnection(IHttpApplication application, PipeReader input, PipeWriter output) { _application = application; _input = input; @@ -50,9 +50,9 @@ public HttpConnection(IHttpApplication application, IPipeReader input, _responseBody = new HttpResponseStream(this); } - public IPipeReader Input => _input; + public PipeReader Input => _input; - public IPipeWriter Output => _output; + public PipeWriter Output => _output; public HttpRequestStream RequestBody { get; set; } @@ -194,12 +194,12 @@ public Task WriteAsync(Span data) return FlushAsync(buffer); } - public async Task FlushAsync(IPipeWriter buffer) + public async Task FlushAsync(PipeWriter buffer) { await buffer.FlushAsync(); } - private void WriteBeginResponseHeaders(IPipeWriter buffer) + private void WriteBeginResponseHeaders(PipeWriter buffer) { if (HasStarted) { @@ -217,7 +217,7 @@ private void WriteBeginResponseHeaders(IPipeWriter buffer) ResponseHeaders.CopyTo(_autoChunk, buffer); } - private void WriteEndResponse(IPipeWriter buffer) + private void WriteEndResponse(PipeWriter buffer) { buffer.Write(_chunkedEndBytes); } diff --git a/samples/System.IO.Pipelines.Samples/SampleBase/RawHttpClientSampleBase.cs b/samples/System.IO.Pipelines.Samples/SampleBase/RawHttpClientSampleBase.cs index 9374403ff81..0c84ed71d16 100644 --- a/samples/System.IO.Pipelines.Samples/SampleBase/RawHttpClientSampleBase.cs +++ b/samples/System.IO.Pipelines.Samples/SampleBase/RawHttpClientSampleBase.cs @@ -39,7 +39,7 @@ public async Task Run() protected abstract Task GetConnection(); - private async Task CopyCompletedAsync(IPipeReader input, IPipeWriter output) + private async Task CopyCompletedAsync(PipeReader input, PipeWriter output) { var result = await input.ReadAsync(); var inputBuffer = result.Buffer; diff --git a/samples/System.IO.Pipelines.Samples/SampleBase/RawHttpServerSampleBase.cs b/samples/System.IO.Pipelines.Samples/SampleBase/RawHttpServerSampleBase.cs index 313310a07fa..8db3cb1eb67 100644 --- a/samples/System.IO.Pipelines.Samples/SampleBase/RawHttpServerSampleBase.cs +++ b/samples/System.IO.Pipelines.Samples/SampleBase/RawHttpServerSampleBase.cs @@ -65,7 +65,7 @@ protected async Task ProcessConnection(IPipeConnection connection) // Writing directly to pooled buffers var output = connection.Output; - var formatter = new OutputFormatter(output, SymbolTable.InvariantUtf8); + var formatter = new OutputFormatter(output, SymbolTable.InvariantUtf8); formatter.Append("HTTP/1.1 200 OK"); formatter.Append("\r\nContent-Length: 13"); formatter.Append("\r\nContent-Type: text/plain"); diff --git a/src/System.IO.Pipelines.Compression/CompressionPipelineExtensions.cs b/src/System.IO.Pipelines.Compression/CompressionPipelineExtensions.cs index f6e95ce9b8b..068e8d2d47b 100644 --- a/src/System.IO.Pipelines.Compression/CompressionPipelineExtensions.cs +++ b/src/System.IO.Pipelines.Compression/CompressionPipelineExtensions.cs @@ -10,82 +10,82 @@ namespace System.IO.Pipelines.Compression { public static class CompressionPipelineExtensions { - public static IPipeReader DeflateDecompress(this IPipeReader reader, PipeOptions options) + public static PipeReader DeflateDecompress(this PipeReader reader, PipeOptions options) { var inflater = new ReadableDeflateTransform(ZLibNative.Deflate_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = inflater.Execute(reader, pipe.Writer); return pipe.Reader; } - public static IPipeReader DeflateCompress(this IPipeReader reader, PipeOptions options, CompressionLevel compressionLevel) + public static PipeReader DeflateCompress(this PipeReader reader, PipeOptions options, CompressionLevel compressionLevel) { var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.Deflate_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = deflater.Execute(reader, pipe.Writer); return pipe.Reader; } - public static IPipeReader GZipDecompress(this IPipeReader reader, PipeOptions options) + public static PipeReader GZipDecompress(this PipeReader reader, PipeOptions options) { var inflater = new ReadableDeflateTransform(ZLibNative.GZip_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = inflater.Execute(reader, pipe.Writer); return pipe.Reader; } - public static IPipeWriter GZipCompress(this IPipeWriter writer, PipeOptions options, CompressionLevel compressionLevel) + public static PipeWriter GZipCompress(this PipeWriter writer, PipeOptions options, CompressionLevel compressionLevel) { var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = deflater.Execute(pipe.Reader, writer); return pipe.Writer; } - public static IPipeReader GZipCompress(this IPipeReader reader, PipeOptions options, CompressionLevel compressionLevel) + public static PipeReader GZipCompress(this PipeReader reader, PipeOptions options, CompressionLevel compressionLevel) { var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = deflater.Execute(reader, pipe.Writer); return pipe.Reader; } - public static IPipeReader CreateDeflateDecompressReader(PipeOptions options, IPipeReader reader) + public static PipeReader CreateDeflateDecompressReader(PipeOptions options, PipeReader reader) { var inflater = new ReadableDeflateTransform(ZLibNative.Deflate_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = inflater.Execute(reader, pipe.Writer); return pipe.Reader; } - public static IPipeReader CreateDeflateCompressReader(PipeOptions options, IPipeReader reader, CompressionLevel compressionLevel) + public static PipeReader CreateDeflateCompressReader(PipeOptions options, PipeReader reader, CompressionLevel compressionLevel) { var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.Deflate_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = deflater.Execute(reader, pipe.Writer); return pipe.Reader; } - public static IPipeReader CreateGZipDecompressReader(PipeOptions options, IPipeReader reader) + public static PipeReader CreateGZipDecompressReader(PipeOptions options, PipeReader reader) { var inflater = new ReadableDeflateTransform(ZLibNative.GZip_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = inflater.Execute(reader, pipe.Writer); return pipe.Reader; } - public static IPipeWriter CreateGZipCompressWriter(PipeOptions options, IPipeWriter writer, CompressionLevel compressionLevel) + public static PipeWriter CreateGZipCompressWriter(PipeOptions options, PipeWriter writer, CompressionLevel compressionLevel) { var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = deflater.Execute(pipe.Reader, writer); return pipe.Writer; } - public static IPipeReader CreateGZipCompressReader(PipeOptions options, IPipeReader reader, CompressionLevel compressionLevel) + public static PipeReader CreateGZipCompressReader(PipeOptions options, PipeReader reader, CompressionLevel compressionLevel) { var deflater = new WritableDeflateTransform(compressionLevel, ZLibNative.GZip_DefaultWindowBits); - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = deflater.Execute(reader, pipe.Writer); return pipe.Reader; } @@ -99,7 +99,7 @@ public WritableDeflateTransform(CompressionLevel compressionLevel, int bits) _deflater = new Deflater(compressionLevel, bits); } - public async Task Execute(IPipeReader reader, IPipeWriter writer) + public async Task Execute(PipeReader reader, PipeWriter writer) { List handles = new List(); @@ -210,7 +210,7 @@ public ReadableDeflateTransform(int bits) _inflater = new Inflater(bits); } - public async Task Execute(IPipeReader reader, IPipeWriter writer) + public async Task Execute(PipeReader reader, PipeWriter writer) { List handles = new List(); diff --git a/src/System.IO.Pipelines.Extensions/PipelineReaderExtensions.cs b/src/System.IO.Pipelines.Extensions/PipelineReaderExtensions.cs index eda8d8162d1..2fe680c2d9a 100644 --- a/src/System.IO.Pipelines.Extensions/PipelineReaderExtensions.cs +++ b/src/System.IO.Pipelines.Extensions/PipelineReaderExtensions.cs @@ -9,7 +9,7 @@ namespace System.IO.Pipelines { public static class PipelineReaderExtensions { - public static ValueTask ReadAsync(this IPipeReader input, ArraySegment destination) + public static ValueTask ReadAsync(this PipeReader input, ArraySegment destination) { while (true) { @@ -44,7 +44,7 @@ public static ValueTask ReadAsync(this IPipeReader input, ArraySegment(input.ReadAsyncAwaited(destination)); } - public static async Task CopyToAsync(this IPipeReader input, IPipeWriter output) + public static async Task CopyToAsync(this PipeReader input, PipeWriter output) { while (true) { @@ -75,7 +75,7 @@ public static async Task CopyToAsync(this IPipeReader input, IPipeWriter output) } } - private static async Task ReadAsyncAwaited(this IPipeReader input, ArraySegment destination) + private static async Task ReadAsyncAwaited(this PipeReader input, ArraySegment destination) { while (true) { @@ -99,12 +99,12 @@ private static async Task ReadAsyncAwaited(this IPipeReader input, ArraySeg } } - public static Task CopyToAsync(this IPipeReader input, Stream stream) + public static Task CopyToAsync(this PipeReader input, Stream stream) { return input.CopyToAsync(stream, 4096, CancellationToken.None); } - public static async Task CopyToAsync(this IPipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken) + public static async Task CopyToAsync(this PipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken) { // TODO: Use bufferSize argument while (!cancellationToken.IsCancellationRequested) diff --git a/src/System.IO.Pipelines.Extensions/ReadWriteExtensions.cs b/src/System.IO.Pipelines.Extensions/ReadWriteExtensions.cs index 91ee0f55cad..0083b2dc238 100644 --- a/src/System.IO.Pipelines.Extensions/ReadWriteExtensions.cs +++ b/src/System.IO.Pipelines.Extensions/ReadWriteExtensions.cs @@ -98,7 +98,7 @@ static void WriteLittleEndian<[Primitive]T>(this Span buffer, T value) whe WriteMachineEndian(buffer, ref value); } - public static async Task> ReadToEndAsync(this IPipeReader input) + public static async Task> ReadToEndAsync(this PipeReader input) { while (true) { @@ -158,7 +158,7 @@ private static T ReadMultiLittle<[Primitive]T>(ReadOnlyBuffer buffer, int /// Reads a structure of type T out of a buffer of bytes. /// [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static void WriteBigEndian<[Primitive]T>(this IPipeWriter buffer, T value) where T : struct + public static void WriteBigEndian<[Primitive]T>(this PipeWriter buffer, T value) where T : struct { int len = Unsafe.SizeOf(); buffer.GetMemory(len).Span.WriteBigEndian(value); @@ -169,7 +169,7 @@ public static void WriteBigEndian<[Primitive]T>(this IPipeWriter buffer, T value /// Reads a structure of type T out of a buffer of bytes. /// [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static void WriteLittleEndian<[Primitive]T>(this IPipeWriter buffer, T value) where T : struct + public static void WriteLittleEndian<[Primitive]T>(this PipeWriter buffer, T value) where T : struct { int len = Unsafe.SizeOf(); buffer.GetMemory(len).Span.WriteLittleEndian(value); diff --git a/src/System.IO.Pipelines.Extensions/StreamExtensions.cs b/src/System.IO.Pipelines.Extensions/StreamExtensions.cs index 1fb05c324ec..e3316aeee3b 100644 --- a/src/System.IO.Pipelines.Extensions/StreamExtensions.cs +++ b/src/System.IO.Pipelines.Extensions/StreamExtensions.cs @@ -11,19 +11,19 @@ namespace System.IO.Pipelines public static class StreamExtensions { /// - /// Copies the content of a into a . + /// Copies the content of a into a . /// /// /// /// /// - public static Task CopyToAsync(this Stream stream, IPipeWriter writer, CancellationToken cancellationToken = default) + public static Task CopyToAsync(this Stream stream, PipeWriter writer, CancellationToken cancellationToken = default) { // 81920 is the default bufferSize, there is not stream.CopyToAsync overload that takes only a cancellationToken return stream.CopyToAsync(new PipelineWriterStream(writer), bufferSize: 81920, cancellationToken: cancellationToken); } - public static async Task CopyToEndAsync(this Stream stream, IPipeWriter writer, CancellationToken cancellationToken = default) + public static async Task CopyToEndAsync(this Stream stream, PipeWriter writer, CancellationToken cancellationToken = default) { try { @@ -77,12 +77,12 @@ await stream.WriteAsync(data.Array, data.Offset, data.Count) } } - public static Task CopyToEndAsync(this IPipeReader input, Stream stream) + public static Task CopyToEndAsync(this PipeReader input, Stream stream) { return input.CopyToEndAsync(stream, 4096, CancellationToken.None); } - public static async Task CopyToEndAsync(this IPipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken) + public static async Task CopyToEndAsync(this PipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken) { try { @@ -98,9 +98,9 @@ public static async Task CopyToEndAsync(this IPipeReader input, Stream stream, i private class PipelineWriterStream : Stream { - private readonly IPipeWriter _writer; + private readonly PipeWriter _writer; - public PipelineWriterStream(IPipeWriter writer) + public PipelineWriterStream(PipeWriter writer) { _writer = writer; } diff --git a/src/System.IO.Pipelines.Extensions/StreamPipeConnection.cs b/src/System.IO.Pipelines.Extensions/StreamPipeConnection.cs index 4fc6b9bc96f..10b4324af0e 100644 --- a/src/System.IO.Pipelines.Extensions/StreamPipeConnection.cs +++ b/src/System.IO.Pipelines.Extensions/StreamPipeConnection.cs @@ -11,9 +11,9 @@ public StreamPipeConnection(PipeOptions options, Stream stream) Output = CreateWriter(options, stream); } - public IPipeReader Input { get; } + public PipeReader Input { get; } - public IPipeWriter Output { get; } + public PipeWriter Output { get; } public void Dispose() { @@ -21,27 +21,27 @@ public void Dispose() Output.Complete(); } - public static IPipeReader CreateReader(PipeOptions options, Stream stream) + public static PipeReader CreateReader(PipeOptions options, Stream stream) { if (!stream.CanRead) { throw new NotSupportedException(); } - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = stream.CopyToEndAsync(pipe.Writer); return pipe.Reader; } - public static IPipeWriter CreateWriter(PipeOptions options, Stream stream) + public static PipeWriter CreateWriter(PipeOptions options, Stream stream) { if (!stream.CanWrite) { throw new NotSupportedException(); } - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var ignore = pipe.Reader.CopyToEndAsync(stream); return pipe.Writer; diff --git a/src/System.IO.Pipelines.File/FileReader.cs b/src/System.IO.Pipelines.File/FileReader.cs index a759ae04071..dd7c1b4379a 100644 --- a/src/System.IO.Pipelines.File/FileReader.cs +++ b/src/System.IO.Pipelines.File/FileReader.cs @@ -11,9 +11,9 @@ namespace System.IO.Pipelines.File { public class FileReader { - private readonly IPipeWriter _writer; + private readonly PipeWriter _writer; - public FileReader(IPipeWriter writer) + public FileReader(PipeWriter writer) { _writer = writer; } @@ -101,7 +101,7 @@ private class ReadOperation public unsafe NativeOverlapped* Overlapped { get; set; } - public IPipeWriter Writer { get; set; } + public PipeWriter Writer { get; set; } public int Offset { get; set; } diff --git a/src/System.IO.Pipelines.File/ReadableFilePipelineFactoryExtensions.cs b/src/System.IO.Pipelines.File/ReadableFilePipelineFactoryExtensions.cs index 80090308212..8b86c3963af 100644 --- a/src/System.IO.Pipelines.File/ReadableFilePipelineFactoryExtensions.cs +++ b/src/System.IO.Pipelines.File/ReadableFilePipelineFactoryExtensions.cs @@ -5,9 +5,9 @@ namespace System.IO.Pipelines.File { public static class ReadableFilePipelineFactoryExtensions { - public static IPipeReader ReadFile(PipeOptions options, string path) + public static PipeReader ReadFile(PipeOptions options, string path) { - var pipe = new Pipe(options); + var pipe = new ResetablePipe(options); var file = new FileReader(pipe.Writer); file.OpenReadFile(path); return pipe.Reader; diff --git a/src/System.IO.Pipelines.Networking.Libuv/UvTcpConnection.cs b/src/System.IO.Pipelines.Networking.Libuv/UvTcpConnection.cs index 27d4c7c1445..b33bf997ebb 100644 --- a/src/System.IO.Pipelines.Networking.Libuv/UvTcpConnection.cs +++ b/src/System.IO.Pipelines.Networking.Libuv/UvTcpConnection.cs @@ -14,8 +14,8 @@ public class UvTcpConnection : IPipeConnection private static readonly Action _readCallback = ReadCallback; private static readonly Func _allocCallback = AllocCallback; - protected readonly IPipe _input; - protected readonly IPipe _output; + protected readonly Pipe _input; + protected readonly Pipe _output; private readonly UvThread _thread; private readonly UvTcpHandle _handle; private volatile bool _stopping; @@ -28,11 +28,11 @@ public UvTcpConnection(UvThread thread, UvTcpHandle handle) _thread = thread; _handle = handle; - _input = new Pipe(new PipeOptions(thread.Pool, + _input = new ResetablePipe(new PipeOptions(thread.Pool, // resume from back pressure on the uv thread writerScheduler: thread)); - _output = new Pipe(new PipeOptions(thread.Pool, + _output = new ResetablePipe(new PipeOptions(thread.Pool, // user code will dispatch back to the uv thread for writes, readerScheduler: thread)); @@ -70,9 +70,9 @@ protected virtual void Dispose(bool disposing) DisposeAsync().GetAwaiter().GetResult(); } - public IPipeWriter Output => _output.Writer; + public PipeWriter Output => _output.Writer; - public IPipeReader Input => _input.Reader; + public PipeReader Input => _input.Reader; private async Task ProcessWrites() { diff --git a/src/System.IO.Pipelines.Networking.Sockets/SocketConnection.cs b/src/System.IO.Pipelines.Networking.Sockets/SocketConnection.cs index 89b9da9bec5..e3bd4e701ca 100644 --- a/src/System.IO.Pipelines.Networking.Sockets/SocketConnection.cs +++ b/src/System.IO.Pipelines.Networking.Sockets/SocketConnection.cs @@ -38,7 +38,7 @@ public class SocketConnection : IPipeConnection private readonly bool _ownsPool; private MemoryPool _pool; - private IPipe _input, _output; + private Pipe _input, _output; private Socket _socket; private Task _receiveTask; private Task _sendTask; @@ -86,8 +86,8 @@ internal SocketConnection(Socket socket, MemoryPool pool) // TODO: Make this configurable // Dispatch to avoid deadlocks - _input = new Pipe(new PipeOptions(pool, Scheduler.ThreadPool, Scheduler.ThreadPool)); - _output = new Pipe(new PipeOptions(pool, Scheduler.ThreadPool, Scheduler.ThreadPool)); + _input = new ResetablePipe(new PipeOptions(pool, Scheduler.ThreadPool, Scheduler.ThreadPool)); + _output = new ResetablePipe(new PipeOptions(pool, Scheduler.ThreadPool, Scheduler.ThreadPool)); _receiveTask = ReceiveFromSocketAndPushToWriterAsync(); _sendTask = ReadFromReaderAndWriteToSocketAsync(); @@ -96,12 +96,12 @@ internal SocketConnection(Socket socket, MemoryPool pool) /// /// Provides access to data received from the socket /// - public IPipeReader Input => _input.Reader; + public PipeReader Input => _input.Reader; /// /// Provides access to write data to the socket /// - public IPipeWriter Output => _output.Writer; + public PipeWriter Output => _output.Writer; private MemoryPool Pool => _pool; diff --git a/src/System.IO.Pipelines.Networking.Windows.RIO/RioTcpConnection.cs b/src/System.IO.Pipelines.Networking.Windows.RIO/RioTcpConnection.cs index 91eb8ddc15d..ab0df8a828f 100644 --- a/src/System.IO.Pipelines.Networking.Windows.RIO/RioTcpConnection.cs +++ b/src/System.IO.Pipelines.Networking.Windows.RIO/RioTcpConnection.cs @@ -27,8 +27,8 @@ public sealed class RioTcpConnection : IPipeConnection private long _previousSendCorrelation = RestartSendCorrelations; - private readonly IPipe _input; - private readonly IPipe _output; + private readonly Pipe _input; + private readonly Pipe _output; private readonly SemaphoreSlim _outgoingSends = new SemaphoreSlim(RioTcpServer.MaxWritesPerSocket); private readonly SemaphoreSlim _previousSendsComplete = new SemaphoreSlim(1); @@ -44,8 +44,8 @@ internal RioTcpConnection(IntPtr socket, long connectionId, IntPtr requestQueue, _rio = rio; _rioThread = rioThread; - _input = new Pipe(new PipeOptions(rioThread.Pool)); - _output = new Pipe(new PipeOptions(rioThread.Pool)); + _input = new ResetablePipe(new PipeOptions(rioThread.Pool)); + _output = new ResetablePipe(new PipeOptions(rioThread.Pool)); _requestQueue = requestQueue; @@ -55,8 +55,8 @@ internal RioTcpConnection(IntPtr socket, long connectionId, IntPtr requestQueue, _sendTask = ProcessSends(); } - public IPipeReader Input => _input.Reader; - public IPipeWriter Output => _output.Writer; + public PipeReader Input => _input.Reader; + public PipeWriter Output => _output.Writer; private void ProcessReceives() { diff --git a/src/System.IO.Pipelines.Text.Primitives/PipelineTextOutput.cs b/src/System.IO.Pipelines.Text.Primitives/PipelineTextOutput.cs index d1302fef9d9..9623d96a8bb 100644 --- a/src/System.IO.Pipelines.Text.Primitives/PipelineTextOutput.cs +++ b/src/System.IO.Pipelines.Text.Primitives/PipelineTextOutput.cs @@ -10,10 +10,10 @@ namespace System.IO.Pipelines.Text.Primitives { public class PipelineTextOutput : ITextOutput { - private readonly IPipeWriter _writer; + private readonly PipeWriter _writer; private bool _needAlloc = true; - public PipelineTextOutput(IPipeWriter writer, SymbolTable symbolTable) + public PipelineTextOutput(PipeWriter writer, SymbolTable symbolTable) { _writer = writer; SymbolTable = symbolTable; diff --git a/src/System.IO.Pipelines.Text.Primitives/PipelineWriterExtensions.cs b/src/System.IO.Pipelines.Text.Primitives/PipelineWriterExtensions.cs index f5e961f8b02..47ebde36c55 100644 --- a/src/System.IO.Pipelines.Text.Primitives/PipelineWriterExtensions.cs +++ b/src/System.IO.Pipelines.Text.Primitives/PipelineWriterExtensions.cs @@ -7,7 +7,7 @@ namespace System.IO.Pipelines.Text.Primitives { public static class PipelineWriterExtensions { - public static PipelineTextOutput AsTextOutput(this IPipeWriter writer, SymbolTable symbolTable) + public static PipelineTextOutput AsTextOutput(this PipeWriter writer, SymbolTable symbolTable) { return new PipelineTextOutput(writer, symbolTable); } diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/FlushResult.cs b/src/System.IO.Pipelines/System/IO/Pipelines/FlushResult.cs index 661b331e7cf..6b3bb2ecc71 100644 --- a/src/System.IO.Pipelines/System/IO/Pipelines/FlushResult.cs +++ b/src/System.IO.Pipelines/System/IO/Pipelines/FlushResult.cs @@ -13,7 +13,7 @@ public struct FlushResult public bool IsCancelled => (ResultFlags & ResultFlags.Cancelled) != 0; /// - /// True if the is complete + /// True if the is complete /// public bool IsCompleted => (ResultFlags & ResultFlags.Completed) != 0; } diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/IPipe.cs b/src/System.IO.Pipelines/System/IO/Pipelines/IPipe.cs deleted file mode 100644 index a131527999f..00000000000 --- a/src/System.IO.Pipelines/System/IO/Pipelines/IPipe.cs +++ /dev/null @@ -1,12 +0,0 @@ -// Licensed to the .NET Foundation under one or more agreements. -// The .NET Foundation licenses this file to you under the MIT license. -// See the LICENSE file in the project root for more information. - -namespace System.IO.Pipelines -{ - public interface IPipe - { - IPipeReader Reader { get; } - IPipeWriter Writer { get; } - } -} diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/IPipeConnection.cs b/src/System.IO.Pipelines/System/IO/Pipelines/IPipeConnection.cs index 66f6ba84593..b7df611a9f2 100644 --- a/src/System.IO.Pipelines/System/IO/Pipelines/IPipeConnection.cs +++ b/src/System.IO.Pipelines/System/IO/Pipelines/IPipeConnection.cs @@ -9,13 +9,13 @@ namespace System.IO.Pipelines public interface IPipeConnection : IDisposable { /// - /// Gets the half of the duplex connection. + /// Gets the half of the duplex connection. /// - IPipeReader Input { get; } + PipeReader Input { get; } /// - /// Gets the half of the duplex connection. + /// Gets the half of the duplex connection. /// - IPipeWriter Output { get; } + PipeWriter Output { get; } } } diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/Pipe.cs b/src/System.IO.Pipelines/System/IO/Pipelines/Pipe.cs index cf966bfa8df..6c2669d3f9d 100644 --- a/src/System.IO.Pipelines/System/IO/Pipelines/Pipe.cs +++ b/src/System.IO.Pipelines/System/IO/Pipelines/Pipe.cs @@ -1,798 +1,12 @@ -// Copyright (c) Microsoft. All rights reserved. -// Licensed under the MIT license. See LICENSE file in the project root for full license information. - -using System.Buffers; -using System.Collections.Sequences; -using System.Diagnostics; -using System.Runtime.CompilerServices; - -using System.Threading; +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. namespace System.IO.Pipelines { - /// - /// Default and implementation. - /// - public class Pipe : IPipe, IPipeReader, IPipeWriter, IAwaiter, IAwaiter + public abstract class Pipe { - private const int SegmentPoolSize = 16; - - private static readonly Action _signalReaderAwaitable = state => ((Pipe)state).ReaderCancellationRequested(); - private static readonly Action _signalWriterAwaitable = state => ((Pipe)state).WriterCancellationRequested(); - private static readonly Action _invokeCompletionCallbacks = state => ((PipeCompletionCallbacks)state).Execute(); - - // This sync objects protects the following state: - // 1. _commitHead & _commitHeadIndex - // 2. _length - // 3. _readerAwaitable & _writerAwaitable - private readonly object _sync = new object(); - - private readonly MemoryPool _pool; - private readonly int _minimumSegmentSize; - private readonly long _maximumSizeHigh; - private readonly long _maximumSizeLow; - - private readonly Scheduler _readerScheduler; - private readonly Scheduler _writerScheduler; - - private long _length; - private long _currentWriteLength; - - private int _pooledSegmentCount; - - private PipeAwaitable _readerAwaitable; - private PipeAwaitable _writerAwaitable; - - private PipeCompletion _writerCompletion; - private PipeCompletion _readerCompletion; - - private BufferSegment[] _bufferSegmentPool; - // The read head which is the extent of the IPipelineReader's consumed bytes - private BufferSegment _readHead; - private int _readHeadIndex; - - // The commit head which is the extent of the bytes available to the IPipelineReader to consume - private BufferSegment _commitHead; - private int _commitHeadIndex; - - // The write head which is the extent of the IPipelineWriter's written bytes - private BufferSegment _writingHead; - - private PipeOperationState _readingState; - private PipeOperationState _writingState; - - private bool _disposed; - - internal long Length => _length; - - /// - /// Initializes the with the specifed . - /// - /// - /// - public Pipe(PipeOptions options) - { - if (options == null) - { - throw new ArgumentNullException(nameof(options)); - } - - if (options.MaximumSizeLow < 0) - { - throw new ArgumentOutOfRangeException(nameof(options.MaximumSizeLow)); - } - - if (options.MaximumSizeHigh < 0) - { - throw new ArgumentOutOfRangeException(nameof(options.MaximumSizeHigh)); - } - - if (options.MaximumSizeLow > options.MaximumSizeHigh) - { - throw new ArgumentException(nameof(options.MaximumSizeHigh) + " should be greater or equal to " + nameof(options.MaximumSizeLow), nameof(options.MaximumSizeHigh)); - } - - _bufferSegmentPool = new BufferSegment[SegmentPoolSize]; - - _pool = options.Pool; - _minimumSegmentSize = options.MinimumSegmentSize; - _maximumSizeHigh = options.MaximumSizeHigh; - _maximumSizeLow = options.MaximumSizeLow; - _readerScheduler = options.ReaderScheduler ?? Scheduler.Inline; - _writerScheduler = options.WriterScheduler ?? Scheduler.Inline; - _readerAwaitable = new PipeAwaitable(completed: false); - _writerAwaitable = new PipeAwaitable(completed: true); - } - - private void ResetState() - { - _readerCompletion.Reset(); - _writerCompletion.Reset(); - _commitHeadIndex = 0; - _currentWriteLength = 0; - _length = 0; - } - - internal Memory Buffer => _writingHead?.AvailableMemory.Slice(_writingHead.End, _writingHead.WritableBytes) ?? Memory.Empty; - - /// - /// Allocates memory from the pipeline to write into. - /// - /// The minimum size buffer to allocate - /// A that can be written to. - Memory IOutput.GetMemory(int minimumSize) - { - if (_writerCompletion.IsCompleted) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.NoWritingAllowed, _writerCompletion.Location); - } - - if (minimumSize < 0) - { - PipelinesThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.minimumSize); - } - - var segment = _writingHead; - if (segment == null) - { - lock (_sync) - { - // CompareExchange not required as its setting to current value if test fails - _writingState.BeginTentative(ExceptionResource.AlreadyWriting); - - try - { - segment = AllocateWriteHeadUnsynchronized(minimumSize); - } - catch (Exception) - { - // Reset producing state if allocation failed - _writingState.End(ExceptionResource.NoWriteToComplete); - throw; - } - - } - } - - var bytesLeftInBuffer = segment.WritableBytes; - - // If inadequate bytes left or if the segment is readonly - if (bytesLeftInBuffer == 0 || bytesLeftInBuffer < minimumSize || segment.ReadOnly) - { - BufferSegment nextSegment; - lock (_sync) - { - nextSegment = CreateSegmentUnsynchronized(); - } - - nextSegment.SetMemory(_pool.Rent(Math.Max(_minimumSegmentSize, minimumSize))); - - segment.SetNext(nextSegment); - - _writingHead = nextSegment; - } - - return Buffer; - } - - Span IOutput.GetSpan(int minimumSize) => ((IOutput)this).GetMemory(minimumSize).Span; - - private BufferSegment AllocateWriteHeadUnsynchronized(int count) - { - BufferSegment segment = null; - - if (_commitHead != null && !_commitHead.ReadOnly) - { - // Try to return the tail so the calling code can append to it - int remaining = _commitHead.WritableBytes; - - if (count <= remaining && remaining > 0) - { - // Free tail space of the right amount, use that - segment = _commitHead; - } - } - - if (segment == null) - { - // No free tail space, allocate a new segment - segment = CreateSegmentUnsynchronized(); - segment.SetMemory(_pool.Rent(Math.Max(_minimumSegmentSize, count))); - } - - if (_commitHead == null) - { - // No previous writes have occurred - _commitHead = segment; - } - else if (segment != _commitHead && _commitHead.Next == null) - { - // Append the segment to the commit head if writes have been committed - // and it isn't the same segment (unused tail space) - _commitHead.SetNext(segment); - } - - // Set write head to assigned segment - _writingHead = segment; - - return segment; - } - - private BufferSegment CreateSegmentUnsynchronized() - { - if (_pooledSegmentCount > 0) - { - _pooledSegmentCount--; - return _bufferSegmentPool[_pooledSegmentCount]; - } - - return new BufferSegment(); - } - - private void ReturnSegmentUnsynchronized(BufferSegment segment) - { - if (_pooledSegmentCount < _bufferSegmentPool.Length) - { - _bufferSegmentPool[_pooledSegmentCount] = segment; - _pooledSegmentCount++; - } - } - - private void EnsureAlloc() - { - if (!_writingState.IsStarted) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.NotWritingNoAlloc); - } - } - - void IPipeWriter.Commit() - { - // Changing commit head shared with Reader - lock (_sync) - { - CommitUnsynchronized(); - } - } - - internal void CommitUnsynchronized() - { - _writingState.End(ExceptionResource.NoWriteToComplete); - - if (_writingHead == null) - { - // Nothing written to commit - return; - } - - if (_readHead == null) - { - // Update the head to point to the head of the buffer. - // This happens if we called alloc(0) then write - _readHead = _commitHead; - _readHeadIndex = 0; - } - - // Always move the commit head to the write head - _commitHead = _writingHead; - _commitHeadIndex = _writingHead.End; - _length += _currentWriteLength; - - // Do not reset if reader is complete - if (_maximumSizeHigh > 0 && - _length >= _maximumSizeHigh && - !_readerCompletion.IsCompleted) - { - _writerAwaitable.Reset(); - } - // Clear the writing state - _writingHead = null; - _currentWriteLength = 0; - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - void IOutput.Advance(int bytesWritten) - { - EnsureAlloc(); - if (bytesWritten > 0) - { - Debug.Assert(!_writingHead.ReadOnly); - Debug.Assert(_writingHead.Next == null); - - var buffer = _writingHead.AvailableMemory; - var bufferIndex = _writingHead.End + bytesWritten; - - if (bufferIndex > buffer.Length) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.AdvancingPastBufferSize); - } - - _writingHead.End = bufferIndex; - _currentWriteLength += bytesWritten; - } - else if (bytesWritten < 0) - { - PipelinesThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.bytesWritten); - } // and if zero, just do nothing; don't need to validate tail etc - } - - ValueAwaiter IPipeWriter.FlushAsync(CancellationToken cancellationToken) - { - Action awaitable; - CancellationTokenRegistration cancellationTokenRegistration; - lock (_sync) - { - if (_writingState.IsStarted) - { - // Commit the data as not already committed - CommitUnsynchronized(); - } - - awaitable = _readerAwaitable.Complete(); - - cancellationTokenRegistration = _writerAwaitable.AttachToken(cancellationToken, _signalWriterAwaitable, this); - } - - cancellationTokenRegistration.Dispose(); - - TrySchedule(_readerScheduler, awaitable); - - return new ValueAwaiter(this); - } - - /// - /// Marks the pipeline as being complete, meaning no more items will be written to it. - /// - /// Optional Exception indicating a failure that's causing the pipeline to complete. - void IPipeWriter.Complete(Exception exception) - { - if (_writingState.IsStarted && _currentWriteLength > 0) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.CompleteWriterActiveWriter, _writingState.Location); - } - - Action awaitable; - PipeCompletionCallbacks completionCallbacks; - bool readerCompleted; - - lock (_sync) - { - completionCallbacks = _writerCompletion.TryComplete(exception); - awaitable = _readerAwaitable.Complete(); - readerCompleted = _readerCompletion.IsCompleted; - } - - if (completionCallbacks != null) - { - TrySchedule(_readerScheduler, _invokeCompletionCallbacks, completionCallbacks); - } - - TrySchedule(_readerScheduler, awaitable); - - if (readerCompleted) - { - CompletePipe(); - } - } - - // Reading - void IPipeReader.Advance(Position consumed) - { - ((IPipeReader)this).Advance(consumed, consumed); - } - - void IPipeReader.Advance(Position consumed, Position examined) - { - BufferSegment returnStart = null; - BufferSegment returnEnd = null; - - // Reading commit head shared with writer - Action continuation = null; - lock (_sync) - { - bool examinedEverything = false; - if (examined.Segment == _commitHead) - { - examinedEverything = _commitHead != null ? examined.Index == _commitHeadIndex - _commitHead.Start : examined.Index == 0; - } - - if (consumed.Segment != null) - { - if (_readHead == null) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.AdvanceToInvalidCursor); - return; - } - - var consumedSegment = (BufferSegment)consumed.Segment; - - returnStart = _readHead; - returnEnd = consumedSegment; - - // Check if we crossed _maximumSizeLow and complete backpressure - var consumedBytes = new ReadOnlyBuffer(returnStart, _readHeadIndex, consumedSegment, consumed.Index).Length; - var oldLength = _length; - _length -= consumedBytes; - - if (oldLength >= _maximumSizeLow && - _length < _maximumSizeLow) - { - continuation = _writerAwaitable.Complete(); - } - - // Check if we consumed entire last segment - // if we are going to return commit head - // we need to check that there is no writing operation that - // might be using tailspace - if (consumed.Index == returnEnd.Length && - !(_commitHead == returnEnd && _writingState.IsStarted)) - { - var nextBlock = returnEnd.NextSegment; - if (_commitHead == returnEnd) - { - _commitHead = nextBlock; - _commitHeadIndex = 0; - } - - _readHead = nextBlock; - _readHeadIndex = 0; - returnEnd = nextBlock; - } - else - { - _readHead = consumedSegment; - _readHeadIndex = consumed.Index; - } - } - - // We reset the awaitable to not completed if we've examined everything the producer produced so far - // but only if writer is not completed yet - if (examinedEverything && !_writerCompletion.IsCompleted) - { - // Prevent deadlock where reader awaits new data and writer await backpressure - if (!_writerAwaitable.IsCompleted) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.BackpressureDeadlock); - } - _readerAwaitable.Reset(); - } - - _readingState.End(ExceptionResource.NoReadToComplete); - - while (returnStart != null && returnStart != returnEnd) - { - returnStart.ResetMemory(); - ReturnSegmentUnsynchronized(returnStart); - returnStart = returnStart.NextSegment; - } - } - - TrySchedule(_writerScheduler, continuation); - } - - /// - /// Signal to the producer that the consumer is done reading. - /// - /// Optional Exception indicating a failure that's causing the pipeline to complete. - void IPipeReader.Complete(Exception exception) - { - if (_readingState.IsActive) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.CompleteReaderActiveReader, _readingState.Location); - } - - PipeCompletionCallbacks completionCallbacks; - Action awaitable; - bool writerCompleted; - - lock (_sync) - { - completionCallbacks = _readerCompletion.TryComplete(exception); - awaitable = _writerAwaitable.Complete(); - writerCompleted = _writerCompletion.IsCompleted; - } - - if (completionCallbacks != null) - { - TrySchedule(_writerScheduler, _invokeCompletionCallbacks, completionCallbacks); - } - - TrySchedule(_writerScheduler, awaitable); - - if (writerCompleted) - { - CompletePipe(); - } - } - - void IPipeReader.OnWriterCompleted(Action callback, object state) - { - if (callback == null) - { - throw new ArgumentNullException(nameof(callback)); - } - - PipeCompletionCallbacks completionCallbacks; - lock (_sync) - { - completionCallbacks = _writerCompletion.AddCallback(callback, state); - } - - if (completionCallbacks != null) - { - TrySchedule(_readerScheduler, _invokeCompletionCallbacks, completionCallbacks); - } - } - - /// - /// Cancel to currently pending call to without completing the . - /// - void IPipeReader.CancelPendingRead() - { - Action awaitable; - lock (_sync) - { - awaitable = _readerAwaitable.Cancel(); - } - TrySchedule(_readerScheduler, awaitable); - } - - /// - /// Cancel to currently pending call to without completing the . - /// - void IPipeWriter.CancelPendingFlush() - { - Action awaitable; - lock (_sync) - { - awaitable = _writerAwaitable.Cancel(); - } - TrySchedule(_writerScheduler, awaitable); - } - - void IPipeWriter.OnReaderCompleted(Action callback, object state) - { - if (callback == null) - { - throw new ArgumentNullException(nameof(callback)); - } - - PipeCompletionCallbacks completionCallbacks; - lock (_sync) - { - completionCallbacks = _readerCompletion.AddCallback(callback, state); - } - - if (completionCallbacks != null) - { - TrySchedule(_writerScheduler, _invokeCompletionCallbacks, completionCallbacks); - } - } - - ValueAwaiter IPipeReader.ReadAsync(CancellationToken token) - { - CancellationTokenRegistration cancellationTokenRegistration; - if (_readerCompletion.IsCompleted) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.NoReadingAllowed, _readerCompletion.Location); - } - lock (_sync) - { - cancellationTokenRegistration = _readerAwaitable.AttachToken(token, _signalReaderAwaitable, this); - } - cancellationTokenRegistration.Dispose(); - return new ValueAwaiter(this); - } - - bool IPipeReader.TryRead(out ReadResult result) - { - lock (_sync) - { - if (_readerCompletion.IsCompleted) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.NoReadingAllowed, _readerCompletion.Location); - } - - result = new ReadResult(); - if (_length > 0 || _readerAwaitable.IsCompleted) - { - GetResult(ref result); - return true; - } - - if (_readerAwaitable.HasContinuation) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.AlreadyReading); - } - return false; - } - } - - private static void TrySchedule(Scheduler scheduler, Action action) - { - if (action != null) - { - scheduler.Schedule(action); - } - } - - private static void TrySchedule(Scheduler scheduler, Action action, object state) - { - if (action != null) - { - scheduler.Schedule(action, state); - } - } - - private void CompletePipe() - { - lock (_sync) - { - if (_disposed) - { - return; - } - - _disposed = true; - // Return all segments - var segment = _readHead; - while (segment != null) - { - var returnSegment = segment; - segment = segment.NextSegment; - - returnSegment.ResetMemory(); - } - - _readHead = null; - _commitHead = null; - } - } - - // IReadableBufferAwaiter members - - bool IAwaiter.IsCompleted => _readerAwaitable.IsCompleted; - - void IAwaiter.OnCompleted(Action continuation) - { - Action awaitable; - bool doubleCompletion; - lock (_sync) - { - awaitable = _readerAwaitable.OnCompleted(continuation, out doubleCompletion); - } - if (doubleCompletion) - { - Writer.Complete(PipelinesThrowHelper.GetInvalidOperationException(ExceptionResource.NoConcurrentOperation)); - } - TrySchedule(_readerScheduler, awaitable); - } - - ReadResult IAwaiter.GetResult() - { - if (!_readerAwaitable.IsCompleted) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.GetResultNotCompleted); - } - - var result = new ReadResult(); - lock (_sync) - { - GetResult(ref result); - } - return result; - } - - private void GetResult(ref ReadResult result) - { - if (_writerCompletion.IsCompletedOrThrow()) - { - result.ResultFlags |= ResultFlags.Completed; - } - - var isCancelled = _readerAwaitable.ObserveCancelation(); - if (isCancelled) - { - result.ResultFlags |= ResultFlags.Cancelled; - } - - // No need to read end if there is no head - var head = _readHead; - - if (head != null) - { - // Reading commit head shared with writer - result.ResultBuffer = new ReadOnlyBuffer(head, _readHeadIndex, _commitHead, _commitHeadIndex - _commitHead.Start); - } - - if (isCancelled) - { - _readingState.BeginTentative(ExceptionResource.AlreadyReading); - } - else - { - _readingState.Begin(ExceptionResource.AlreadyReading); - } - } - - // IWritableBufferAwaiter members - - bool IAwaiter.IsCompleted => _writerAwaitable.IsCompleted; - - FlushResult IAwaiter.GetResult() - { - var result = new FlushResult(); - lock (_sync) - { - if (!_writerAwaitable.IsCompleted) - { - PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.GetResultNotCompleted); - } - - // Change the state from to be cancelled -> observed - if (_writerAwaitable.ObserveCancelation()) - { - result.ResultFlags |= ResultFlags.Cancelled; - } - if (_readerCompletion.IsCompletedOrThrow()) - { - result.ResultFlags |= ResultFlags.Completed; - } - } - - return result; - } - - void IAwaiter.OnCompleted(Action continuation) - { - Action awaitable; - bool doubleCompletion; - lock (_sync) - { - awaitable = _writerAwaitable.OnCompleted(continuation, out doubleCompletion); - } - if (doubleCompletion) - { - Reader.Complete(PipelinesThrowHelper.GetInvalidOperationException(ExceptionResource.NoConcurrentOperation)); - } - TrySchedule(_writerScheduler, awaitable); - } - - private void ReaderCancellationRequested() - { - Action action; - lock (_sync) - { - action = _readerAwaitable.Cancel(); - } - TrySchedule(_readerScheduler, action); - } - - private void WriterCancellationRequested() - { - Action action; - lock (_sync) - { - action = _writerAwaitable.Cancel(); - } - TrySchedule(_writerScheduler, action); - } - - public IPipeReader Reader => this; - public IPipeWriter Writer => this; - - public void Reset() - { - lock (_sync) - { - if (!_disposed) - { - throw new InvalidOperationException("Both reader and writer need to be completed to be able to reset "); - } - - _disposed = false; - ResetState(); - } - } + public abstract PipeReader Reader { get; } + public abstract PipeWriter Writer { get; } } } diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/PipeExtensions.cs b/src/System.IO.Pipelines/System/IO/Pipelines/PipeExtensions.cs index 5d0c2d3c132..1cfc116c2a8 100644 --- a/src/System.IO.Pipelines/System/IO/Pipelines/PipeExtensions.cs +++ b/src/System.IO.Pipelines/System/IO/Pipelines/PipeExtensions.cs @@ -11,7 +11,7 @@ public static class PipelineExtensions { private static readonly Task _completedTask = Task.FromResult(0); - public static Task WriteAsync(this IPipeWriter output, ReadOnlyMemory source) + public static Task WriteAsync(this PipeWriter output, ReadOnlyMemory source) { var writeBuffer = output; writeBuffer.Write(source.Span); diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/IPipeReader.cs b/src/System.IO.Pipelines/System/IO/Pipelines/PipeReader.cs similarity index 78% rename from src/System.IO.Pipelines/System/IO/Pipelines/IPipeReader.cs rename to src/System.IO.Pipelines/System/IO/Pipelines/PipeReader.cs index 856240f0254..3f7efa178e3 100644 --- a/src/System.IO.Pipelines/System/IO/Pipelines/IPipeReader.cs +++ b/src/System.IO.Pipelines/System/IO/Pipelines/PipeReader.cs @@ -10,21 +10,21 @@ namespace System.IO.Pipelines /// /// Defines a class that provides a pipeline from which data can be read. /// - public interface IPipeReader + public abstract class PipeReader { /// - /// Attempt to synchronously read data the . + /// Attempt to synchronously read data the . /// /// The /// True if data was available, or if the call was cancelled or the writer completed with an error. /// If the pipe returns false, there's no need to call Advance. - bool TryRead(out ReadResult result); + public abstract bool TryRead(out ReadResult result); /// - /// Asynchronously reads a sequence of bytes from the current . + /// Asynchronously reads a sequence of bytes from the current . /// /// A representing the asynchronous read operation. - ValueAwaiter ReadAsync(CancellationToken cancellationToken = default); + public abstract ValueAwaiter ReadAsync(CancellationToken cancellationToken = default); /// /// Moves forward the pipeline's read cursor to after the consumed data. @@ -34,7 +34,7 @@ public interface IPipeReader /// The memory for the consumed data will be released and no longer available. /// The examined data communicates to the pipeline when it should signal more data is available. /// - void Advance(Position consumed); + public abstract void Advance(Position consumed); /// /// Moves forward the pipeline's read cursor to after the consumed data. @@ -45,22 +45,22 @@ public interface IPipeReader /// The memory for the consumed data will be released and no longer available. /// The examined data communicates to the pipeline when it should signal more data is available. /// - void Advance(Position consumed, Position examined); + public abstract void Advance(Position consumed, Position examined); /// - /// Cancel to currently pending or next call to if none is pending, without completing the . + /// Cancel to currently pending or next call to if none is pending, without completing the . /// - void CancelPendingRead(); + public abstract void CancelPendingRead(); /// /// Signal to the producer that the consumer is done reading. /// /// Optional Exception indicating a failure that's causing the pipeline to complete. - void Complete(Exception exception = null); + public abstract void Complete(Exception exception = null); /// /// Registers callback that gets executed when writer side of pipe completes. /// - void OnWriterCompleted(Action callback, object state); + public abstract void OnWriterCompleted(Action callback, object state); } } diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/IPipeWriter.cs b/src/System.IO.Pipelines/System/IO/Pipelines/PipeWriter.cs similarity index 60% rename from src/System.IO.Pipelines/System/IO/Pipelines/IPipeWriter.cs rename to src/System.IO.Pipelines/System/IO/Pipelines/PipeWriter.cs index a653cd3f2a5..c730d37b1bf 100644 --- a/src/System.IO.Pipelines/System/IO/Pipelines/IPipeWriter.cs +++ b/src/System.IO.Pipelines/System/IO/Pipelines/PipeWriter.cs @@ -9,26 +9,29 @@ namespace System.IO.Pipelines /// /// Defines a class that provides a pipeline to which data can be written. /// - public interface IPipeWriter: IOutput + public abstract class PipeWriter: IOutput { /// /// Marks the pipeline as being complete, meaning no more items will be written to it. /// /// Optional Exception indicating a failure that's causing the pipeline to complete. - void Complete(Exception exception = null); + public abstract void Complete(Exception exception = null); /// - /// Cancel to currently pending or next call to if none is pending, without completing the . + /// Cancel to currently pending or next call to if none is pending, without completing the . /// - void CancelPendingFlush(); + public abstract void CancelPendingFlush(); /// /// Registers callback that gets executed when reader side of pipe completes. /// - void OnReaderCompleted(Action callback, object state); + public abstract void OnReaderCompleted(Action callback, object state); - ValueAwaiter FlushAsync(CancellationToken cancellationToken = default); + public abstract ValueAwaiter FlushAsync(CancellationToken cancellationToken = default); - void Commit(); + public abstract void Commit(); + public abstract void Advance(int bytes); + public abstract Memory GetMemory(int minimumLength = 0); + public abstract Span GetSpan(int minimumLength = 0); } } diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/ReadResult.cs b/src/System.IO.Pipelines/System/IO/Pipelines/ReadResult.cs index ae76cb45544..11909fca87f 100644 --- a/src/System.IO.Pipelines/System/IO/Pipelines/ReadResult.cs +++ b/src/System.IO.Pipelines/System/IO/Pipelines/ReadResult.cs @@ -6,7 +6,7 @@ namespace System.IO.Pipelines { /// - /// The result of a call. + /// The result of a call. /// public struct ReadResult { @@ -39,7 +39,7 @@ public ReadResult(ReadOnlyBuffer buffer, bool isCancelled, bool isComplete public bool IsCancelled => (ResultFlags & ResultFlags.Cancelled) != 0; /// - /// True if the is complete + /// True if the is complete /// public bool IsCompleted => (ResultFlags & ResultFlags.Completed) != 0; } diff --git a/src/System.IO.Pipelines/System/IO/Pipelines/ResetablePipe.cs b/src/System.IO.Pipelines/System/IO/Pipelines/ResetablePipe.cs new file mode 100644 index 00000000000..aff2273da12 --- /dev/null +++ b/src/System.IO.Pipelines/System/IO/Pipelines/ResetablePipe.cs @@ -0,0 +1,896 @@ +// Copyright (c) Microsoft. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +using System.Buffers; +using System.Collections.Sequences; +using System.Diagnostics; +using System.Runtime.CompilerServices; + +using System.Threading; + +namespace System.IO.Pipelines +{ + /// + /// Default and implementation. + /// + public class ResetablePipe : Pipe, IAwaiter, IAwaiter + { + private const int SegmentPoolSize = 16; + + private static readonly Action _signalReaderAwaitable = state => ((ResetablePipe)state).ReaderCancellationRequested(); + private static readonly Action _signalWriterAwaitable = state => ((ResetablePipe)state).WriterCancellationRequested(); + private static readonly Action _invokeCompletionCallbacks = state => ((PipeCompletionCallbacks)state).Execute(); + + // This sync objects protects the following state: + // 1. _commitHead & _commitHeadIndex + // 2. _length + // 3. _readerAwaitable & _writerAwaitable + private readonly object _sync = new object(); + + private readonly MemoryPool _pool; + private readonly int _minimumSegmentSize; + private readonly long _maximumSizeHigh; + private readonly long _maximumSizeLow; + + private readonly Scheduler _readerScheduler; + private readonly Scheduler _writerScheduler; + + private long _length; + private long _currentWriteLength; + + private int _pooledSegmentCount; + + private PipeAwaitable _readerAwaitable; + private PipeAwaitable _writerAwaitable; + + private PipeCompletion _writerCompletion; + private PipeCompletion _readerCompletion; + + private BufferSegment[] _bufferSegmentPool; + // The read head which is the extent of the IPipelineReader's consumed bytes + private BufferSegment _readHead; + private int _readHeadIndex; + + // The commit head which is the extent of the bytes available to the IPipelineReader to consume + private BufferSegment _commitHead; + private int _commitHeadIndex; + + // The write head which is the extent of the IPipelineWriter's written bytes + private BufferSegment _writingHead; + + private PipeOperationState _readingState; + private PipeOperationState _writingState; + + private bool _disposed; + + internal long Length => _length; + + /// + /// Initializes the with the specifed . + /// + /// + /// + public ResetablePipe(PipeOptions options) + { + if (options == null) + { + throw new ArgumentNullException(nameof(options)); + } + + if (options.MaximumSizeLow < 0) + { + throw new ArgumentOutOfRangeException(nameof(options.MaximumSizeLow)); + } + + if (options.MaximumSizeHigh < 0) + { + throw new ArgumentOutOfRangeException(nameof(options.MaximumSizeHigh)); + } + + if (options.MaximumSizeLow > options.MaximumSizeHigh) + { + throw new ArgumentException(nameof(options.MaximumSizeHigh) + " should be greater or equal to " + nameof(options.MaximumSizeLow), nameof(options.MaximumSizeHigh)); + } + + _bufferSegmentPool = new BufferSegment[SegmentPoolSize]; + + _pool = options.Pool; + _minimumSegmentSize = options.MinimumSegmentSize; + _maximumSizeHigh = options.MaximumSizeHigh; + _maximumSizeLow = options.MaximumSizeLow; + _readerScheduler = options.ReaderScheduler ?? Scheduler.Inline; + _writerScheduler = options.WriterScheduler ?? Scheduler.Inline; + _readerAwaitable = new PipeAwaitable(completed: false); + _writerAwaitable = new PipeAwaitable(completed: true); + Reader = new ResetablePipeReader(this); + Writer = new ResetablePipeWriter(this); + } + + private void ResetState() + { + _readerCompletion.Reset(); + _writerCompletion.Reset(); + _commitHeadIndex = 0; + _currentWriteLength = 0; + _length = 0; + } + + internal Memory Buffer => _writingHead?.AvailableMemory.Slice(_writingHead.End, _writingHead.WritableBytes) ?? Memory.Empty; + + /// + /// Allocates memory from the pipeline to write into. + /// + /// The minimum size buffer to allocate + /// A that can be written to. + internal Memory GetMemory(int minimumSize) + { + if (_writerCompletion.IsCompleted) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.NoWritingAllowed, _writerCompletion.Location); + } + + if (minimumSize < 0) + { + PipelinesThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.minimumSize); + } + + var segment = _writingHead; + if (segment == null) + { + lock (_sync) + { + // CompareExchange not required as its setting to current value if test fails + _writingState.BeginTentative(ExceptionResource.AlreadyWriting); + + try + { + segment = AllocateWriteHeadUnsynchronized(minimumSize); + } + catch (Exception) + { + // Reset producing state if allocation failed + _writingState.End(ExceptionResource.NoWriteToComplete); + throw; + } + + } + } + + var bytesLeftInBuffer = segment.WritableBytes; + + // If inadequate bytes left or if the segment is readonly + if (bytesLeftInBuffer == 0 || bytesLeftInBuffer < minimumSize || segment.ReadOnly) + { + BufferSegment nextSegment; + lock (_sync) + { + nextSegment = CreateSegmentUnsynchronized(); + } + + nextSegment.SetMemory(_pool.Rent(Math.Max(_minimumSegmentSize, minimumSize))); + + segment.SetNext(nextSegment); + + _writingHead = nextSegment; + } + + return Buffer; + } + + internal Span GetSpan(int minimumSize) => GetMemory(minimumSize).Span; + + private BufferSegment AllocateWriteHeadUnsynchronized(int count) + { + BufferSegment segment = null; + + if (_commitHead != null && !_commitHead.ReadOnly) + { + // Try to return the tail so the calling code can append to it + int remaining = _commitHead.WritableBytes; + + if (count <= remaining && remaining > 0) + { + // Free tail space of the right amount, use that + segment = _commitHead; + } + } + + if (segment == null) + { + // No free tail space, allocate a new segment + segment = CreateSegmentUnsynchronized(); + segment.SetMemory(_pool.Rent(Math.Max(_minimumSegmentSize, count))); + } + + if (_commitHead == null) + { + // No previous writes have occurred + _commitHead = segment; + } + else if (segment != _commitHead && _commitHead.Next == null) + { + // Append the segment to the commit head if writes have been committed + // and it isn't the same segment (unused tail space) + _commitHead.SetNext(segment); + } + + // Set write head to assigned segment + _writingHead = segment; + + return segment; + } + + private BufferSegment CreateSegmentUnsynchronized() + { + if (_pooledSegmentCount > 0) + { + _pooledSegmentCount--; + return _bufferSegmentPool[_pooledSegmentCount]; + } + + return new BufferSegment(); + } + + private void ReturnSegmentUnsynchronized(BufferSegment segment) + { + if (_pooledSegmentCount < _bufferSegmentPool.Length) + { + _bufferSegmentPool[_pooledSegmentCount] = segment; + _pooledSegmentCount++; + } + } + + private void EnsureAlloc() + { + if (!_writingState.IsStarted) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.NotWritingNoAlloc); + } + } + + internal void Commit() + { + // Changing commit head shared with Reader + lock (_sync) + { + CommitUnsynchronized(); + } + } + + internal void CommitUnsynchronized() + { + _writingState.End(ExceptionResource.NoWriteToComplete); + + if (_writingHead == null) + { + // Nothing written to commit + return; + } + + if (_readHead == null) + { + // Update the head to point to the head of the buffer. + // This happens if we called alloc(0) then write + _readHead = _commitHead; + _readHeadIndex = 0; + } + + // Always move the commit head to the write head + _commitHead = _writingHead; + _commitHeadIndex = _writingHead.End; + _length += _currentWriteLength; + + // Do not reset if reader is complete + if (_maximumSizeHigh > 0 && + _length >= _maximumSizeHigh && + !_readerCompletion.IsCompleted) + { + _writerAwaitable.Reset(); + } + // Clear the writing state + _writingHead = null; + _currentWriteLength = 0; + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + internal void Advance(int bytesWritten) + { + EnsureAlloc(); + if (bytesWritten > 0) + { + Debug.Assert(!_writingHead.ReadOnly); + Debug.Assert(_writingHead.Next == null); + + var buffer = _writingHead.AvailableMemory; + var bufferIndex = _writingHead.End + bytesWritten; + + if (bufferIndex > buffer.Length) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.AdvancingPastBufferSize); + } + + _writingHead.End = bufferIndex; + _currentWriteLength += bytesWritten; + } + else if (bytesWritten < 0) + { + PipelinesThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.bytesWritten); + } // and if zero, just do nothing; don't need to validate tail etc + } + + internal ValueAwaiter FlushAsync(CancellationToken cancellationToken) + { + Action awaitable; + CancellationTokenRegistration cancellationTokenRegistration; + lock (_sync) + { + if (_writingState.IsStarted) + { + // Commit the data as not already committed + CommitUnsynchronized(); + } + + awaitable = _readerAwaitable.Complete(); + + cancellationTokenRegistration = _writerAwaitable.AttachToken(cancellationToken, _signalWriterAwaitable, this); + } + + cancellationTokenRegistration.Dispose(); + + TrySchedule(_readerScheduler, awaitable); + + return new ValueAwaiter(this); + } + + /// + /// Marks the pipeline as being complete, meaning no more items will be written to it. + /// + /// Optional Exception indicating a failure that's causing the pipeline to complete. + internal void Complete(Exception exception) + { + if (_writingState.IsStarted && _currentWriteLength > 0) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.CompleteWriterActiveWriter, _writingState.Location); + } + + Action awaitable; + PipeCompletionCallbacks completionCallbacks; + bool readerCompleted; + + lock (_sync) + { + completionCallbacks = _writerCompletion.TryComplete(exception); + awaitable = _readerAwaitable.Complete(); + readerCompleted = _readerCompletion.IsCompleted; + } + + if (completionCallbacks != null) + { + TrySchedule(_readerScheduler, _invokeCompletionCallbacks, completionCallbacks); + } + + TrySchedule(_readerScheduler, awaitable); + + if (readerCompleted) + { + CompletePipe(); + } + } + + // Reading + internal void Advance(Position consumed) + { + Advance(consumed, consumed); + } + + internal void Advance(Position consumed, Position examined) + { + BufferSegment returnStart = null; + BufferSegment returnEnd = null; + + // Reading commit head shared with writer + Action continuation = null; + lock (_sync) + { + bool examinedEverything = false; + if (examined.Segment == _commitHead) + { + examinedEverything = _commitHead != null ? examined.Index == _commitHeadIndex - _commitHead.Start : examined.Index == 0; + } + + if (consumed.Segment != null) + { + if (_readHead == null) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.AdvanceToInvalidCursor); + return; + } + + var consumedSegment = (BufferSegment)consumed.Segment; + + returnStart = _readHead; + returnEnd = consumedSegment; + + // Check if we crossed _maximumSizeLow and complete backpressure + var consumedBytes = new ReadOnlyBuffer(returnStart, _readHeadIndex, consumedSegment, consumed.Index).Length; + var oldLength = _length; + _length -= consumedBytes; + + if (oldLength >= _maximumSizeLow && + _length < _maximumSizeLow) + { + continuation = _writerAwaitable.Complete(); + } + + // Check if we consumed entire last segment + // if we are going to return commit head + // we need to check that there is no writing operation that + // might be using tailspace + if (consumed.Index == returnEnd.Length && + !(_commitHead == returnEnd && _writingState.IsStarted)) + { + var nextBlock = returnEnd.NextSegment; + if (_commitHead == returnEnd) + { + _commitHead = nextBlock; + _commitHeadIndex = 0; + } + + _readHead = nextBlock; + _readHeadIndex = 0; + returnEnd = nextBlock; + } + else + { + _readHead = consumedSegment; + _readHeadIndex = consumed.Index; + } + } + + // We reset the awaitable to not completed if we've examined everything the producer produced so far + // but only if writer is not completed yet + if (examinedEverything && !_writerCompletion.IsCompleted) + { + // Prevent deadlock where reader awaits new data and writer await backpressure + if (!_writerAwaitable.IsCompleted) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.BackpressureDeadlock); + } + _readerAwaitable.Reset(); + } + + _readingState.End(ExceptionResource.NoReadToComplete); + + while (returnStart != null && returnStart != returnEnd) + { + returnStart.ResetMemory(); + ReturnSegmentUnsynchronized(returnStart); + returnStart = returnStart.NextSegment; + } + } + + TrySchedule(_writerScheduler, continuation); + } + + /// + /// Signal to the producer that the consumer is done reading. + /// + /// Optional Exception indicating a failure that's causing the pipeline to complete. + internal void CompleteReader(Exception exception) + { + if (_readingState.IsActive) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.CompleteReaderActiveReader, _readingState.Location); + } + + PipeCompletionCallbacks completionCallbacks; + Action awaitable; + bool writerCompleted; + + lock (_sync) + { + completionCallbacks = _readerCompletion.TryComplete(exception); + awaitable = _writerAwaitable.Complete(); + writerCompleted = _writerCompletion.IsCompleted; + } + + if (completionCallbacks != null) + { + TrySchedule(_writerScheduler, _invokeCompletionCallbacks, completionCallbacks); + } + + TrySchedule(_writerScheduler, awaitable); + + if (writerCompleted) + { + CompletePipe(); + } + } + + internal void OnWriterCompleted(Action callback, object state) + { + if (callback == null) + { + throw new ArgumentNullException(nameof(callback)); + } + + PipeCompletionCallbacks completionCallbacks; + lock (_sync) + { + completionCallbacks = _writerCompletion.AddCallback(callback, state); + } + + if (completionCallbacks != null) + { + TrySchedule(_readerScheduler, _invokeCompletionCallbacks, completionCallbacks); + } + } + + /// + /// Cancel to currently pending call to without completing the . + /// + internal void CancelPendingRead() + { + Action awaitable; + lock (_sync) + { + awaitable = _readerAwaitable.Cancel(); + } + TrySchedule(_readerScheduler, awaitable); + } + + /// + /// Cancel to currently pending call to without completing the . + /// + internal void CancelPendingFlush() + { + Action awaitable; + lock (_sync) + { + awaitable = _writerAwaitable.Cancel(); + } + TrySchedule(_writerScheduler, awaitable); + } + + internal void OnReaderCompleted(Action callback, object state) + { + if (callback == null) + { + throw new ArgumentNullException(nameof(callback)); + } + + PipeCompletionCallbacks completionCallbacks; + lock (_sync) + { + completionCallbacks = _readerCompletion.AddCallback(callback, state); + } + + if (completionCallbacks != null) + { + TrySchedule(_writerScheduler, _invokeCompletionCallbacks, completionCallbacks); + } + } + + internal ValueAwaiter ReadAsync(CancellationToken token) + { + CancellationTokenRegistration cancellationTokenRegistration; + if (_readerCompletion.IsCompleted) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.NoReadingAllowed, _readerCompletion.Location); + } + lock (_sync) + { + cancellationTokenRegistration = _readerAwaitable.AttachToken(token, _signalReaderAwaitable, this); + } + cancellationTokenRegistration.Dispose(); + return new ValueAwaiter(this); + } + + internal bool TryRead(out ReadResult result) + { + lock (_sync) + { + if (_readerCompletion.IsCompleted) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.NoReadingAllowed, _readerCompletion.Location); + } + + result = new ReadResult(); + if (_length > 0 || _readerAwaitable.IsCompleted) + { + GetResult(ref result); + return true; + } + + if (_readerAwaitable.HasContinuation) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.AlreadyReading); + } + return false; + } + } + + private static void TrySchedule(Scheduler scheduler, Action action) + { + if (action != null) + { + scheduler.Schedule(action); + } + } + + private static void TrySchedule(Scheduler scheduler, Action action, object state) + { + if (action != null) + { + scheduler.Schedule(action, state); + } + } + + private void CompletePipe() + { + lock (_sync) + { + if (_disposed) + { + return; + } + + _disposed = true; + // Return all segments + var segment = _readHead; + while (segment != null) + { + var returnSegment = segment; + segment = segment.NextSegment; + + returnSegment.ResetMemory(); + } + + _readHead = null; + _commitHead = null; + } + } + + // IReadableBufferAwaiter members + + bool IAwaiter.IsCompleted => _readerAwaitable.IsCompleted; + + void IAwaiter.OnCompleted(Action continuation) + { + Action awaitable; + bool doubleCompletion; + lock (_sync) + { + awaitable = _readerAwaitable.OnCompleted(continuation, out doubleCompletion); + } + if (doubleCompletion) + { + Writer.Complete(PipelinesThrowHelper.GetInvalidOperationException(ExceptionResource.NoConcurrentOperation)); + } + TrySchedule(_readerScheduler, awaitable); + } + + ReadResult IAwaiter.GetResult() + { + if (!_readerAwaitable.IsCompleted) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.GetResultNotCompleted); + } + + var result = new ReadResult(); + lock (_sync) + { + GetResult(ref result); + } + return result; + } + + private void GetResult(ref ReadResult result) + { + if (_writerCompletion.IsCompletedOrThrow()) + { + result.ResultFlags |= ResultFlags.Completed; + } + + var isCancelled = _readerAwaitable.ObserveCancelation(); + if (isCancelled) + { + result.ResultFlags |= ResultFlags.Cancelled; + } + + // No need to read end if there is no head + var head = _readHead; + + if (head != null) + { + // Reading commit head shared with writer + result.ResultBuffer = new ReadOnlyBuffer(head, _readHeadIndex, _commitHead, _commitHeadIndex - _commitHead.Start); + } + + if (isCancelled) + { + _readingState.BeginTentative(ExceptionResource.AlreadyReading); + } + else + { + _readingState.Begin(ExceptionResource.AlreadyReading); + } + } + + // IWritableBufferAwaiter members + + bool IAwaiter.IsCompleted => _writerAwaitable.IsCompleted; + + FlushResult IAwaiter.GetResult() + { + var result = new FlushResult(); + lock (_sync) + { + if (!_writerAwaitable.IsCompleted) + { + PipelinesThrowHelper.ThrowInvalidOperationException(ExceptionResource.GetResultNotCompleted); + } + + // Change the state from to be cancelled -> observed + if (_writerAwaitable.ObserveCancelation()) + { + result.ResultFlags |= ResultFlags.Cancelled; + } + if (_readerCompletion.IsCompletedOrThrow()) + { + result.ResultFlags |= ResultFlags.Completed; + } + } + + return result; + } + + void IAwaiter.OnCompleted(Action continuation) + { + Action awaitable; + bool doubleCompletion; + lock (_sync) + { + awaitable = _writerAwaitable.OnCompleted(continuation, out doubleCompletion); + } + if (doubleCompletion) + { + Reader.Complete(PipelinesThrowHelper.GetInvalidOperationException(ExceptionResource.NoConcurrentOperation)); + } + TrySchedule(_writerScheduler, awaitable); + } + + private void ReaderCancellationRequested() + { + Action action; + lock (_sync) + { + action = _readerAwaitable.Cancel(); + } + TrySchedule(_readerScheduler, action); + } + + private void WriterCancellationRequested() + { + Action action; + lock (_sync) + { + action = _writerAwaitable.Cancel(); + } + TrySchedule(_writerScheduler, action); + } + + public override PipeReader Reader { get; } + + public override PipeWriter Writer { get; } + + public void Reset() + { + lock (_sync) + { + if (!_disposed) + { + throw new InvalidOperationException("Both reader and writer need to be completed to be able to reset "); + } + + _disposed = false; + ResetState(); + } + } + + private class ResetablePipeReader : PipeReader + { + private readonly ResetablePipe _pipe; + + public ResetablePipeReader(ResetablePipe pipe) + { + _pipe = pipe; + } + + public override bool TryRead(out ReadResult result) + { + return _pipe.TryRead(out result); + } + + public override ValueAwaiter ReadAsync(CancellationToken cancellationToken = default) + { + return _pipe.ReadAsync(cancellationToken); + } + + public override void Advance(Position consumed) + { + _pipe.Advance(consumed); + } + + public override void Advance(Position consumed, Position examined) + { + _pipe.Advance(consumed, examined); + } + + public override void CancelPendingRead() + { + _pipe.CancelPendingRead(); + } + + public override void Complete(Exception exception = null) + { + _pipe.CompleteReader(exception); + } + + public override void OnWriterCompleted(Action callback, object state) + { + _pipe.OnWriterCompleted(callback, state); + } + } + + private class ResetablePipeWriter : PipeWriter + { + private readonly ResetablePipe _pipe; + + public ResetablePipeWriter(ResetablePipe pipe) + { + _pipe = pipe; + } + + public override void Complete(Exception exception = null) + { + _pipe.Complete(exception); + } + + public override void CancelPendingFlush() + { + _pipe.CancelPendingFlush(); + } + + public override void OnReaderCompleted(Action callback, object state) + { + _pipe.OnReaderCompleted(callback, state); + } + + public override ValueAwaiter FlushAsync(CancellationToken cancellationToken = default) + { + return _pipe.FlushAsync(cancellationToken); + } + + public override void Commit() + { + _pipe.Commit(); + } + + public override void Advance(int bytes) + { + _pipe.Advance(bytes); + } + + public override Memory GetMemory(int minimumLength = 0) + { + return _pipe.GetMemory(minimumLength); + } + + public override Span GetSpan(int minimumLength = 0) + { + return _pipe.GetSpan(minimumLength); + } + } + } +} diff --git a/tests/Benchmarks/E2EPipelineNoIO.cs b/tests/Benchmarks/E2EPipelineNoIO.cs index 286c0f00bfb..312954c85f2 100644 --- a/tests/Benchmarks/E2EPipelineNoIO.cs +++ b/tests/Benchmarks/E2EPipelineNoIO.cs @@ -32,7 +32,7 @@ private static void TechEmpowerHelloWorldNoIO(int numberOfRequests, int concurre using (iteration.StartMeasurement()) { RawInMemoryHttpServer.Run(numberOfRequests, concurrentConnections, s_genericRequest, (request, response) => { - var formatter = new OutputFormatter(response, SymbolTable.InvariantUtf8); + var formatter = new OutputFormatter(response, SymbolTable.InvariantUtf8); formatter.Append("HTTP/1.1 200 OK"); formatter.Append("\r\nContent-Length: 13"); formatter.Append("\r\nContent-Type: text/plain"); @@ -56,7 +56,7 @@ private static void TechEmpowerJsonNoIO(int numberOfRequests, int concurrentConn using (iteration.StartMeasurement()) { RawInMemoryHttpServer.Run(numberOfRequests, concurrentConnections, s_genericRequest, (request, response) => { - var formatter = new OutputFormatter(response, SymbolTable.InvariantUtf8); + var formatter = new OutputFormatter(response, SymbolTable.InvariantUtf8); formatter.Append("HTTP/1.1 200 OK"); formatter.Append("\r\nContent-Length: 25"); formatter.Append("\r\nContent-Type: application/json"); diff --git a/tests/Benchmarks/Helpers/Connection.cs b/tests/Benchmarks/Helpers/Connection.cs index f8213bf4d6a..2e9c7dd25c1 100644 --- a/tests/Benchmarks/Helpers/Connection.cs +++ b/tests/Benchmarks/Helpers/Connection.cs @@ -8,16 +8,16 @@ public class PipeConnection : IPipeConnection { public PipeConnection(PipeOptions pipeOptions) { - Input = new Pipe(pipeOptions); - Output = new Pipe(pipeOptions); + Input = new ResetablePipe(pipeOptions); + Output = new ResetablePipe(pipeOptions); } - IPipeReader IPipeConnection.Input => Input.Reader; - IPipeWriter IPipeConnection.Output => Output.Writer; + PipeReader IPipeConnection.Input => Input.Reader; + PipeWriter IPipeConnection.Output => Output.Writer; - public IPipe Input { get; } + public Pipe Input { get; } - public IPipe Output { get; } + public Pipe Output { get; } public void Dispose() { diff --git a/tests/Benchmarks/Helpers/Server.cs b/tests/Benchmarks/Helpers/Server.cs index 0692fc57d62..76889bb0b30 100644 --- a/tests/Benchmarks/Helpers/Server.cs +++ b/tests/Benchmarks/Helpers/Server.cs @@ -10,7 +10,7 @@ namespace System.IO.Pipelines.Samples { public static class RawInMemoryHttpServer { - public static void Run(int numberOfRequests, int concurrentConnections, byte[] requestPayload, Action writeResponse) + public static void Run(int numberOfRequests, int concurrentConnections, byte[] requestPayload, Action writeResponse) { var memoryPool = new MemoryPool(); var listener = new FakeListener(memoryPool, concurrentConnections); diff --git a/tests/System.IO.Pipelines.Extensions.Tests/PipelineReaderWriterFacts.cs b/tests/System.IO.Pipelines.Extensions.Tests/PipelineReaderWriterFacts.cs index 856bc32ad2d..27fd6baab19 100644 --- a/tests/System.IO.Pipelines.Extensions.Tests/PipelineReaderWriterFacts.cs +++ b/tests/System.IO.Pipelines.Extensions.Tests/PipelineReaderWriterFacts.cs @@ -13,13 +13,13 @@ namespace System.IO.Pipelines.Tests { public class PipelineReaderWriterFacts : IDisposable { - private IPipe _pipe; + private Pipe _pipe; private MemoryPool _pool; public PipelineReaderWriterFacts() { _pool = new MemoryPool(); - _pipe = new Pipe(new PipeOptions(_pool)); + _pipe = new ResetablePipe(new PipeOptions(_pool)); } public void Dispose() { diff --git a/tests/System.IO.Pipelines.Extensions.Tests/ReadableBufferFacts.cs b/tests/System.IO.Pipelines.Extensions.Tests/ReadableBufferFacts.cs index e2404ff14f3..55430aac692 100644 --- a/tests/System.IO.Pipelines.Extensions.Tests/ReadableBufferFacts.cs +++ b/tests/System.IO.Pipelines.Extensions.Tests/ReadableBufferFacts.cs @@ -24,13 +24,13 @@ public class ReadableBufferFacts: IDisposable { const int BlockSize = 4032; - private IPipe _pipe; + private Pipe _pipe; private MemoryPool _pool; public ReadableBufferFacts() { _pool = new MemoryPool(); - _pipe = new Pipe(new PipeOptions(_pool)); + _pipe = new ResetablePipe(new PipeOptions(_pool)); } public void Dispose() { @@ -472,7 +472,7 @@ public async Task CopyToAsync() { using (var pool = new MemoryPool()) { - var readerWriter = new Pipe(new PipeOptions(pool)); + var readerWriter = new ResetablePipe(new PipeOptions(pool)); var output = readerWriter.Writer; output.Append("Hello World", SymbolTable.InvariantUtf8); await output.FlushAsync(); diff --git a/tests/System.IO.Pipelines.Extensions.Tests/WritableBufferFacts.cs b/tests/System.IO.Pipelines.Extensions.Tests/WritableBufferFacts.cs index 224ef2e945f..54d63f63223 100644 --- a/tests/System.IO.Pipelines.Extensions.Tests/WritableBufferFacts.cs +++ b/tests/System.IO.Pipelines.Extensions.Tests/WritableBufferFacts.cs @@ -27,7 +27,7 @@ public async Task WriteLargeDataTextUtf8(int length) FillRandomStringData(data, length); using (var memoryPool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(memoryPool)); + var pipe = new ResetablePipe(new PipeOptions(memoryPool)); var output = pipe.Writer; output.Append(data, SymbolTable.InvariantUtf8); @@ -64,7 +64,7 @@ public async Task WriteLargeDataTextAscii(int length) FillRandomStringData(data, length); using (var memoryPool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(memoryPool)); + var pipe = new ResetablePipe(new PipeOptions(memoryPool)); var output = pipe.Writer; output.Append(data, SymbolTable.InvariantUtf8); @@ -112,7 +112,7 @@ public async Task CanWriteUInt64ToBuffer(ulong value, string valueAsString) { using (var memoryPool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(memoryPool)); + var pipe = new ResetablePipe(new PipeOptions(memoryPool)); var buffer = pipe.Writer; buffer.Append(value, SymbolTable.InvariantUtf8); await buffer.FlushAsync(); @@ -142,7 +142,7 @@ public async Task WriteHex(int value, string hex) { using (var memoryPool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(memoryPool)); + var pipe = new ResetablePipe(new PipeOptions(memoryPool)); var buffer = pipe.Writer; buffer.Append(value, SymbolTable.InvariantUtf8, 'x'); await buffer.FlushAsync(); diff --git a/tests/System.IO.Pipelines.Performance.Tests/PipeThroughput.cs b/tests/System.IO.Pipelines.Performance.Tests/PipeThroughput.cs index 4d75beb21cd..21d99bdee7e 100644 --- a/tests/System.IO.Pipelines.Performance.Tests/PipeThroughput.cs +++ b/tests/System.IO.Pipelines.Performance.Tests/PipeThroughput.cs @@ -25,14 +25,14 @@ public class PipeThroughput new byte[] { 72, 101, 108, 108, 111, 44, 32, 87, 111, 114, 108, 100, 33}, // Hello, World! }; - private IPipe _pipe; + private Pipe _pipe; private MemoryPool _memoryPool; [GlobalSetup] public void Setup() { _memoryPool = new MemoryPool(); - _pipe = new Pipe(new PipeOptions(_memoryPool)); + _pipe = new ResetablePipe(new PipeOptions(_memoryPool)); } [Benchmark(OperationsPerInvoke = InnerLoopCount)] diff --git a/tests/System.IO.Pipelines.Tests/BackpressureTests.cs b/tests/System.IO.Pipelines.Tests/BackpressureTests.cs index 79efbc1497e..e174d78834b 100644 --- a/tests/System.IO.Pipelines.Tests/BackpressureTests.cs +++ b/tests/System.IO.Pipelines.Tests/BackpressureTests.cs @@ -11,12 +11,12 @@ namespace System.IO.Pipelines.Tests public class BackpressureTests : IDisposable { private MemoryPool _pool; - private Pipe _pipe; + private ResetablePipe _pipe; public BackpressureTests() { _pool = new MemoryPool(); - _pipe = new Pipe(new PipeOptions(_pool, maximumSizeLow: 32, maximumSizeHigh: 64)); + _pipe = new ResetablePipe(new PipeOptions(_pool, maximumSizeLow: 32, maximumSizeHigh: 64)); } public void Dispose() diff --git a/tests/System.IO.Pipelines.Tests/FlushAsyncCancellationTests.cs b/tests/System.IO.Pipelines.Tests/FlushAsyncCancellationTests.cs index 2940a2b8548..d997ddb25da 100644 --- a/tests/System.IO.Pipelines.Tests/FlushAsyncCancellationTests.cs +++ b/tests/System.IO.Pipelines.Tests/FlushAsyncCancellationTests.cs @@ -10,7 +10,7 @@ namespace System.IO.Pipelines.Tests { public static class TestWriterExtensions { - public static IPipeWriter WriteEmpty(this IPipeWriter writer, int count) + public static PipeWriter WriteEmpty(this PipeWriter writer, int count) { writer.GetMemory(count); writer.Advance(count); diff --git a/tests/System.IO.Pipelines.Tests/PipeCompletionCallbacksTests.cs b/tests/System.IO.Pipelines.Tests/PipeCompletionCallbacksTests.cs index 94b4c6d2e0f..fcfd9d5b708 100644 --- a/tests/System.IO.Pipelines.Tests/PipeCompletionCallbacksTests.cs +++ b/tests/System.IO.Pipelines.Tests/PipeCompletionCallbacksTests.cs @@ -26,7 +26,7 @@ public void OnReaderCompletedExecutesOnSchedulerIfCompleted() { var callbackRan = false; var scheduler = new TestScheduler(); - var pipe = new Pipe(new PipeOptions(_pool, writerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(_pool, writerScheduler: scheduler)); pipe.Reader.Complete(); pipe.Writer.OnReaderCompleted((exception, state) => @@ -44,7 +44,7 @@ public void OnWriterCompletedExecutedSchedulerIfCompleted() { var callbackRan = false; var scheduler = new TestScheduler(); - var pipe = new Pipe(new PipeOptions(_pool, readerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(_pool, readerScheduler: scheduler)); pipe.Writer.Complete(); pipe.Reader.OnWriterCompleted((exception, state) => @@ -60,7 +60,7 @@ public void OnWriterCompletedExecutedSchedulerIfCompleted() [Fact] public void OnReaderCompletedThrowsWithNullCallback() { - var pipe = new Pipe(new PipeOptions(_pool)); + var pipe = new ResetablePipe(new PipeOptions(_pool)); Assert.Throws(() => pipe.Writer.OnReaderCompleted(null, null)); } @@ -68,7 +68,7 @@ public void OnReaderCompletedThrowsWithNullCallback() [Fact] public void OnWriterCompletedThrowsWithNullCallback() { - var pipe = new Pipe(new PipeOptions(_pool)); + var pipe = new ResetablePipe(new PipeOptions(_pool)); Assert.Throws(() => pipe.Reader.OnWriterCompleted(null, null)); } @@ -78,7 +78,7 @@ public void OnReaderCompletedUsingWriterScheduler() { var callbackRan = false; var scheduler = new TestScheduler(); - var pipe = new Pipe(new PipeOptions(_pool, writerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(_pool, writerScheduler: scheduler)); pipe.Writer.OnReaderCompleted((exception, state) => { callbackRan = true; @@ -94,7 +94,7 @@ public void OnWriterCompletedUsingReaderScheduler() { var callbackRan = false; var scheduler = new TestScheduler(); - var pipe = new Pipe(new PipeOptions(_pool, readerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(_pool, readerScheduler: scheduler)); pipe.Reader.OnWriterCompleted((exception, state) => { callbackRan = true; @@ -110,7 +110,7 @@ public void OnReaderCompletedExceptionSurfacesToWriterScheduler() { var exception = new Exception(); var scheduler = new TestScheduler(); - var pipe = new Pipe(new PipeOptions(_pool, writerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(_pool, writerScheduler: scheduler)); pipe.Writer.OnReaderCompleted((e, state) => throw exception, null); pipe.Reader.Complete(); @@ -124,7 +124,7 @@ public void OnWriterCompletedExceptionSurfacesToReaderScheduler() { var exception = new Exception(); var scheduler = new TestScheduler(); - var pipe = new Pipe(new PipeOptions(_pool, readerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(_pool, readerScheduler: scheduler)); pipe.Reader.OnWriterCompleted((e, state) => throw exception, null); pipe.Writer.Complete(); @@ -138,7 +138,7 @@ public void OnReaderCompletedIsDetachedDuringReset() { var callbackRan = false; var scheduler = new TestScheduler(); - var pipe = new Pipe(new PipeOptions(_pool, writerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(_pool, writerScheduler: scheduler)); pipe.Writer.OnReaderCompleted((exception, state) => { callbackRan = true; @@ -163,7 +163,7 @@ public void OnWriterCompletedIsDetachedDuringReset() { var callbackRan = false; var scheduler = new TestScheduler(); - var pipe = new Pipe(new PipeOptions(_pool, readerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(_pool, readerScheduler: scheduler)); pipe.Reader.OnWriterCompleted((exception, state) => { callbackRan = true; @@ -187,7 +187,7 @@ public void OnReaderCompletedPassesState() { var callbackRan = false; var callbackState = new object(); - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); pipe.Writer.OnReaderCompleted((exception, state) => { Assert.Equal(callbackState, state); @@ -203,7 +203,7 @@ public void OnWriterCompletedPassesState() { var callbackRan = false; var callbackState = new object(); - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); pipe.Reader.OnWriterCompleted((exception, state) => { Assert.Equal(callbackState, state); @@ -222,7 +222,7 @@ public void OnReaderCompletedRunsInRegistrationOrder() var callbackState3 = new object(); var counter = 0; - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); pipe.Writer.OnReaderCompleted((exception, state) => { Assert.Equal(callbackState1, state); @@ -257,7 +257,7 @@ public void OnWriterCompletedRunsInRegistrationOrder() var callbackState3 = new object(); var counter = 0; - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); pipe.Reader.OnWriterCompleted((exception, state) => { Assert.Equal(callbackState1, state); @@ -294,7 +294,7 @@ public void OnReaderCompletedContinuesOnException() var counter = 0; - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); pipe.Writer.OnReaderCompleted((exception, state) => { Assert.Equal(callbackState1, state); @@ -327,7 +327,7 @@ public void OnWriterCompletedContinuesOnException() var counter = 0; - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); pipe.Reader.OnWriterCompleted((exception, state) => { Assert.Equal(callbackState1, state); @@ -354,7 +354,7 @@ public void OnWriterCompletedContinuesOnException() public void OnWriterCompletedPassesException() { var callbackRan = false; - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); var readerException = new Exception(); pipe.Reader.OnWriterCompleted((exception, state) => @@ -371,7 +371,7 @@ public void OnWriterCompletedPassesException() public void OnReaderCompletedPassesException() { var callbackRan = false; - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); var readerException = new Exception(); pipe.Writer.OnReaderCompleted((exception, state) => @@ -389,7 +389,7 @@ public void OnWriterCompletedRanBeforeReadContinuation() { var callbackRan = false; var continuationRan = false; - var pipe = new Pipe(new PipeOptions(_pool) ); + var pipe = new ResetablePipe(new PipeOptions(_pool) ); pipe.Reader.OnWriterCompleted((exception, state) => { @@ -413,7 +413,7 @@ public void OnReaderCompletedRanBeforeFlushContinuation() { var callbackRan = false; var continuationRan = false; - var pipe = new Pipe(new PipeOptions(_pool, maximumSizeHigh: 5)); + var pipe = new ResetablePipe(new PipeOptions(_pool, maximumSizeHigh: 5)); pipe.Writer.OnReaderCompleted((exception, state) => { @@ -439,7 +439,7 @@ public void OnReaderCompletedRanBeforeFlushContinuation() public void CompletingReaderFromWriterCallbackWorks() { var callbackRan = false; - var pipe = new Pipe(new PipeOptions(_pool, maximumSizeHigh: 5)); + var pipe = new ResetablePipe(new PipeOptions(_pool, maximumSizeHigh: 5)); pipe.Writer.OnReaderCompleted((exception, state) => { @@ -459,7 +459,7 @@ public void CompletingReaderFromWriterCallbackWorks() public void CompletingWriterFromReaderCallbackWorks() { var callbackRan = false; - var pipe = new Pipe(new PipeOptions(_pool, maximumSizeHigh: 5)); + var pipe = new ResetablePipe(new PipeOptions(_pool, maximumSizeHigh: 5)); pipe.Reader.OnWriterCompleted((exception, state) => { diff --git a/tests/System.IO.Pipelines.Tests/PipeLengthTests.cs b/tests/System.IO.Pipelines.Tests/PipeLengthTests.cs index 9b63b2ce109..f13d618cc17 100644 --- a/tests/System.IO.Pipelines.Tests/PipeLengthTests.cs +++ b/tests/System.IO.Pipelines.Tests/PipeLengthTests.cs @@ -11,12 +11,12 @@ namespace System.IO.Pipelines.Tests public class PipeLengthTests : IDisposable { private MemoryPool _pool; - private Pipe _pipe; + private ResetablePipe _pipe; public PipeLengthTests() { _pool = new MemoryPool(); - _pipe = new Pipe(new PipeOptions(_pool)); + _pipe = new ResetablePipe(new PipeOptions(_pool)); } public void Dispose() diff --git a/tests/System.IO.Pipelines.Tests/PipePoolTests.cs b/tests/System.IO.Pipelines.Tests/PipePoolTests.cs index 2518f41e6c7..f64f620d67d 100644 --- a/tests/System.IO.Pipelines.Tests/PipePoolTests.cs +++ b/tests/System.IO.Pipelines.Tests/PipePoolTests.cs @@ -15,7 +15,7 @@ public async Task MultipleCompleteReaderWriterCauseDisposeOnlyOnce() { var pool = new DisposeTrackingBufferPool(); - var readerWriter = new Pipe(new PipeOptions(pool)); + var readerWriter = new ResetablePipe(new PipeOptions(pool)); await readerWriter.Writer.WriteAsync(new byte[] {1}); readerWriter.Writer.Complete(); @@ -34,7 +34,7 @@ public async Task AdvanceToEndReturnsAllBlocks() var writeSize = 512; - var pipe = new Pipe(new PipeOptions(pool)); + var pipe = new ResetablePipe(new PipeOptions(pool)); while (pool.CurrentlyRentedBlocks != 3) { var writableBuffer = pipe.Writer.WriteEmpty(writeSize); @@ -54,7 +54,7 @@ public async Task WriteDuringReadIsNotReturned() var writeSize = 512; - var pipe = new Pipe(new PipeOptions(pool)); + var pipe = new ResetablePipe(new PipeOptions(pool)); await pipe.Writer.WriteAsync(new byte[writeSize]); pipe.Writer.GetMemory(writeSize); @@ -73,7 +73,7 @@ public async Task CanWriteAfterReturningMultipleBlocks() var writeSize = 512; - var pipe = new Pipe(new PipeOptions(pool)); + var pipe = new ResetablePipe(new PipeOptions(pool)); // Write two blocks var buffer = pipe.Writer.GetMemory(writeSize); @@ -98,7 +98,7 @@ public async Task RentsMinimumSegmentSize() var pool = new DisposeTrackingBufferPool(); var writeSize = 512; - var pipe = new Pipe(new PipeOptions(pool, minimumSegmentSize: 2020)); + var pipe = new ResetablePipe(new PipeOptions(pool, minimumSegmentSize: 2020)); var buffer = pipe.Writer.GetMemory(writeSize); var allocatedSize = buffer.Length; diff --git a/tests/System.IO.Pipelines.Tests/PipeResetTests.cs b/tests/System.IO.Pipelines.Tests/PipeResetTests.cs index 6d5d4e461c8..18d1b0bbd3c 100644 --- a/tests/System.IO.Pipelines.Tests/PipeResetTests.cs +++ b/tests/System.IO.Pipelines.Tests/PipeResetTests.cs @@ -10,12 +10,12 @@ namespace System.IO.Pipelines.Tests public class PipeResetTests : IDisposable { private MemoryPool _pool; - private Pipe _pipe; + private ResetablePipe _pipe; public PipeResetTests() { _pool = new MemoryPool(); - _pipe = new Pipe(new PipeOptions(_pool)); + _pipe = new ResetablePipe(new PipeOptions(_pool)); } public void Dispose() diff --git a/tests/System.IO.Pipelines.Tests/PipeTest.cs b/tests/System.IO.Pipelines.Tests/PipeTest.cs index 41c742a3716..5bf9a91333c 100644 --- a/tests/System.IO.Pipelines.Tests/PipeTest.cs +++ b/tests/System.IO.Pipelines.Tests/PipeTest.cs @@ -9,13 +9,13 @@ public abstract class PipeTest : IDisposable { protected const int MaximumSizeHigh = 65; - protected IPipe Pipe; + protected Pipe Pipe; private readonly MemoryPool _pool; protected PipeTest() { _pool = new MemoryPool(); - Pipe = new Pipe(new PipeOptions(_pool, + Pipe = new ResetablePipe(new PipeOptions(_pool, maximumSizeHigh: 65, maximumSizeLow: 6 )); diff --git a/tests/System.IO.Pipelines.Tests/PipelineReaderWriterFacts.cs b/tests/System.IO.Pipelines.Tests/PipelineReaderWriterFacts.cs index a373b748ce6..ca23c97368a 100644 --- a/tests/System.IO.Pipelines.Tests/PipelineReaderWriterFacts.cs +++ b/tests/System.IO.Pipelines.Tests/PipelineReaderWriterFacts.cs @@ -14,13 +14,13 @@ namespace System.IO.Pipelines.Tests { public class PipelineReaderWriterFacts : IDisposable { - private IPipe _pipe; + private Pipe _pipe; private MemoryPool _pool; public PipelineReaderWriterFacts() { _pool = new MemoryPool(); - _pipe = new Pipe(new PipeOptions(_pool)); + _pipe = new ResetablePipe(new PipeOptions(_pool)); } public void Dispose() { @@ -193,7 +193,7 @@ public async Task ReaderShouldNotGetUnflushedBytesWithAppend() // Write Hello to another pipeline and get the buffer var bytes = Encoding.ASCII.GetBytes("Hello"); - var c2 = new Pipe(new PipeOptions(_pool)); + var c2 = new ResetablePipe(new PipeOptions(_pool)); await c2.Writer.WriteAsync(bytes); var result = await c2.Reader.ReadAsync(); var c2Buffer = result.Buffer; diff --git a/tests/System.IO.Pipelines.Tests/SchedulerFacts.cs b/tests/System.IO.Pipelines.Tests/SchedulerFacts.cs index 0d0c13afae9..24d4bb1db15 100644 --- a/tests/System.IO.Pipelines.Tests/SchedulerFacts.cs +++ b/tests/System.IO.Pipelines.Tests/SchedulerFacts.cs @@ -20,7 +20,7 @@ public async Task ReadAsyncCallbackRunsOnReaderScheduler() { using (var scheduler = new ThreadScheduler()) { - var pipe = new Pipe(new PipeOptions(pool, readerScheduler: scheduler)); + var pipe = new ResetablePipe(new PipeOptions(pool, readerScheduler: scheduler)); Func doRead = async () => { @@ -55,7 +55,7 @@ public async Task FlushCallbackRunsOnWriterScheduler() { using (var scheduler = new ThreadScheduler()) { - var pipe = new Pipe(new PipeOptions(pool, + var pipe = new ResetablePipe(new PipeOptions(pool, maximumSizeLow: 32, maximumSizeHigh: 64, writerScheduler: scheduler)); @@ -96,7 +96,7 @@ public async Task DefaultReaderSchedulerRunsInline() { using (var pool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(pool)); + var pipe = new ResetablePipe(new PipeOptions(pool)); var id = 0; @@ -130,7 +130,7 @@ public async Task DefaultWriterSchedulerRunsInline() { using (var pool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(pool, + var pipe = new ResetablePipe(new PipeOptions(pool, maximumSizeLow: 32, maximumSizeHigh: 64 )); diff --git a/tests/System.IO.Pipelines.Tests/WritableBufferFacts.cs b/tests/System.IO.Pipelines.Tests/WritableBufferFacts.cs index 3e63743034a..a50c00d3515 100644 --- a/tests/System.IO.Pipelines.Tests/WritableBufferFacts.cs +++ b/tests/System.IO.Pipelines.Tests/WritableBufferFacts.cs @@ -15,7 +15,7 @@ public async Task CanWriteNothingToBuffer() { using (var memoryPool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(memoryPool)); + var pipe = new ResetablePipe(new PipeOptions(memoryPool)); var buffer = pipe.Writer; buffer.GetMemory(0); buffer.Advance(0); // doing nothing, the hard way @@ -28,7 +28,7 @@ public void ThrowsOnAdvanceWithNoMemory() { using (var memoryPool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(memoryPool)); + var pipe = new ResetablePipe(new PipeOptions(memoryPool)); var buffer = pipe.Writer; var exception = Assert.Throws(() => buffer.Advance(1)); Assert.Equal("No writing operation. Make sure GetMemory() was called.", exception.Message); @@ -40,7 +40,7 @@ public void ThrowsOnAdvanceOverMemorySize() { using (var memoryPool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(memoryPool)); + var pipe = new ResetablePipe(new PipeOptions(memoryPool)); var buffer = pipe.Writer.GetMemory(1); var exception = Assert.Throws(() => pipe.Writer.Advance(buffer.Length + 1)); Assert.Equal("Can't advance past buffer size", exception.Message); @@ -59,7 +59,7 @@ public async Task WriteLargeDataBinary(int length) new Random(length).NextBytes(data); using (var memoryPool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(memoryPool)); + var pipe = new ResetablePipe(new PipeOptions(memoryPool)); var output = pipe.Writer; output.Write(data); @@ -86,7 +86,7 @@ public void EnsureMoreThanPoolBlockSizeThrows() { using (var pool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(pool)); + var pipe = new ResetablePipe(new PipeOptions(pool)); var buffer = pipe.Writer; Assert.Throws(() => buffer.GetMemory(8192)); } @@ -97,7 +97,7 @@ public void EmptyWriteDoesNotThrow() { using (var pool = new MemoryPool()) { - var pipe = new Pipe(new PipeOptions(pool)); + var pipe = new ResetablePipe(new PipeOptions(pool)); var buffer = pipe.Writer; buffer.Write(new byte[0]); } diff --git a/tests/System.IO.Pipelines.Tests/WritableBufferWriterFacts.cs b/tests/System.IO.Pipelines.Tests/WritableBufferWriterFacts.cs index fd5630e3990..a1241d911de 100644 --- a/tests/System.IO.Pipelines.Tests/WritableBufferWriterFacts.cs +++ b/tests/System.IO.Pipelines.Tests/WritableBufferWriterFacts.cs @@ -10,12 +10,12 @@ namespace System.IO.Pipelines.Tests public class WritableBufferWriterFacts : IDisposable { private MemoryPool _pool; - private Pipe _pipe; + private ResetablePipe _pipe; public WritableBufferWriterFacts() { _pool = new MemoryPool(); - _pipe = new Pipe(new PipeOptions(_pool)); + _pipe = new ResetablePipe(new PipeOptions(_pool)); } public void Dispose()