React to pipeline changes (#2234)

React to pipeline changes

benchmarks/Kestrel.Performance/Http1ConnectionParsingOverheadBenchmark.cs

@@ -16,7 +16,7 @@ public class Http1ConnectionParsingOverheadBenchmark
     {
         private const int InnerLoopCount = 512;
 
-        public ReadableBuffer _buffer;
+        public ReadOnlyBuffer _buffer;
         public Http1Connection _http1Connection;
 
         [IterationSetup]

benchmarks/Kestrel.Performance/HttpParserBenchmark.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Buffers;
 using System.IO.Pipelines;
 using BenchmarkDotNet.Attributes;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
@@ -12,7 +13,7 @@ public class HttpParserBenchmark : IHttpRequestLineHandler, IHttpHeadersHandler
     {
         private readonly HttpParser<Adapter> _parser = new HttpParser<Adapter>();
 
-        private ReadableBuffer _buffer;
+        private ReadOnlyBuffer _buffer;
 
         [Benchmark(Baseline = true, OperationsPerInvoke = RequestParsingData.InnerLoopCount)]
         public void PlaintextTechEmpower()
@@ -46,7 +47,7 @@ public void Unicode()
 
         private void InsertData(byte[] data)
         {
-            _buffer = ReadableBuffer.Create(data);
+            _buffer = new ReadOnlyBuffer(data);
         }
 
         private void ParseData()

benchmarks/Kestrel.Performance/Mocks/NullParser.cs

@@ -2,6 +2,8 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Buffers;
+using System.Collections.Sequences;
 using System.IO.Pipelines;
 using System.Text;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
@@ -21,7 +23,7 @@ public class NullParser<TRequestHandler> : IHttpParser<TRequestHandler> where TR
 
         public static readonly NullParser<Http1ParsingHandler> Instance = new NullParser<Http1ParsingHandler>();
 
-        public bool ParseHeaders(TRequestHandler handler, ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined, out int consumedBytes)
+        public bool ParseHeaders(TRequestHandler handler, ReadOnlyBuffer buffer, out Position consumed, out Position examined, out int consumedBytes)
         {
             handler.OnHeader(new Span<byte>(_hostHeaderName), new Span<byte>(_hostHeaderValue));
             handler.OnHeader(new Span<byte>(_acceptHeaderName), new Span<byte>(_acceptHeaderValue));
@@ -34,7 +36,7 @@ public bool ParseHeaders(TRequestHandler handler, ReadableBuffer buffer, out Rea
             return true;
         }
 
-        public bool ParseRequestLine(TRequestHandler handler, ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined)
+        public bool ParseRequestLine(TRequestHandler handler, ReadOnlyBuffer buffer, out Position consumed, out Position examined)
         {
             handler.OnStartLine(HttpMethod.Get,
                 HttpVersion.Http11,

build/dependencies.props

@@ -25,20 +25,21 @@
     <MicrosoftExtensionsLoggingTestingPackageVersion>2.1.0-preview1-27965</MicrosoftExtensionsLoggingTestingPackageVersion>
     <MicrosoftExtensionsOptionsPackageVersion>2.1.0-preview1-27965</MicrosoftExtensionsOptionsPackageVersion>
     <MicrosoftNETCoreApp20PackageVersion>2.0.0</MicrosoftNETCoreApp20PackageVersion>
-    <MicrosoftNETCoreApp21PackageVersion>2.1.0-preview1-26016-05</MicrosoftNETCoreApp21PackageVersion>
+    <MicrosoftNETCoreApp21PackageVersion>2.1.0-preview1-26102-01</MicrosoftNETCoreApp21PackageVersion>
     <MicrosoftNetHttpHeadersPackageVersion>2.1.0-preview1-27965</MicrosoftNetHttpHeadersPackageVersion>
     <MicrosoftNETTestSdkPackageVersion>15.3.0</MicrosoftNETTestSdkPackageVersion>
     <MoqPackageVersion>4.7.49</MoqPackageVersion>
     <NewtonsoftJsonPackageVersion>10.0.1</NewtonsoftJsonPackageVersion>
-    <SystemBuffersPackageVersion>4.5.0-preview1-26016-05</SystemBuffersPackageVersion>
-    <SystemIOPipelinesPackageVersion>0.1.0-e171215-1</SystemIOPipelinesPackageVersion>
-    <SystemIOPipelinesTestingPackageVersion>0.1.0-e171215-1</SystemIOPipelinesTestingPackageVersion>
-    <SystemMemoryPackageVersion>4.5.0-preview1-26016-05</SystemMemoryPackageVersion>
-    <SystemNumericsVectorsPackageVersion>4.5.0-preview1-26016-05</SystemNumericsVectorsPackageVersion>
-    <SystemRuntimeCompilerServicesUnsafePackageVersion>4.5.0-preview1-26016-05</SystemRuntimeCompilerServicesUnsafePackageVersion>
-    <SystemSecurityCryptographyCngPackageVersion>4.5.0-preview1-26016-05</SystemSecurityCryptographyCngPackageVersion>
-    <SystemTextEncodingsWebUtf8PackageVersion>0.1.0-e171215-1</SystemTextEncodingsWebUtf8PackageVersion>
-    <SystemThreadingTasksExtensionsPackageVersion>4.5.0-preview2-25707-02</SystemThreadingTasksExtensionsPackageVersion>
+    <SystemBuffersPackageVersion>4.5.0-preview1-26102-01</SystemBuffersPackageVersion>
+    <SystemBuffersPrimitivesPackageVersion>0.1.0-e180104-2</SystemBuffersPrimitivesPackageVersion>
+    <SystemIOPipelinesPackageVersion>0.1.0-e180104-2</SystemIOPipelinesPackageVersion>
+    <SystemIOPipelinesTestingPackageVersion>0.1.0-e180104-2</SystemIOPipelinesTestingPackageVersion>
+    <SystemMemoryPackageVersion>4.5.0-preview1-26102-01</SystemMemoryPackageVersion>
+    <SystemNumericsVectorsPackageVersion>4.5.0-preview1-26102-01</SystemNumericsVectorsPackageVersion>
+    <SystemRuntimeCompilerServicesUnsafePackageVersion>4.5.0-preview1-26102-01</SystemRuntimeCompilerServicesUnsafePackageVersion>
+    <SystemSecurityCryptographyCngPackageVersion>4.5.0-preview1-26102-01</SystemSecurityCryptographyCngPackageVersion>
+    <SystemTextEncodingsWebUtf8PackageVersion>0.1.0-e180104-2</SystemTextEncodingsWebUtf8PackageVersion>
+    <SystemThreadingTasksExtensionsPackageVersion>4.5.0-preview1-26102-01</SystemThreadingTasksExtensionsPackageVersion>
     <XunitAnalyzersPackageVersion>0.8.0</XunitAnalyzersPackageVersion>
     <XunitPackageVersion>2.3.1</XunitPackageVersion>
     <XunitRunnerVisualStudioPackageVersion>2.3.1</XunitRunnerVisualStudioPackageVersion>

src/Kestrel.Core/Adapter/Internal/AdaptedPipeline.cs

@@ -6,6 +6,7 @@
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
 using System.IO.Pipelines;
+using System.Runtime.InteropServices;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal
 {

src/Kestrel.Core/Internal/ConnectionHandler.cs

@@ -83,7 +83,7 @@ private IDisposable BeginConnectionScope(ConnectionContext connectionContext)
         }
 
         // Internal for testing
-        internal static PipeOptions GetInputPipeOptions(ServiceContext serviceContext, MemoryPool memoryPool, IScheduler writerScheduler) => new PipeOptions
+        internal static PipeOptions GetInputPipeOptions(ServiceContext serviceContext, MemoryPool memoryPool, Scheduler writerScheduler) => new PipeOptions
         (
             pool: memoryPool,
             readerScheduler: serviceContext.ThreadPool,
@@ -92,7 +92,7 @@ private IDisposable BeginConnectionScope(ConnectionContext connectionContext)
             maximumSizeLow: serviceContext.ServerOptions.Limits.MaxRequestBufferSize ?? 0
         );
 
-        internal static PipeOptions GetOutputPipeOptions(ServiceContext serviceContext, MemoryPool memoryPool, IScheduler readerScheduler) => new PipeOptions
+        internal static PipeOptions GetOutputPipeOptions(ServiceContext serviceContext, MemoryPool memoryPool, Scheduler readerScheduler) => new PipeOptions
         (
             pool: memoryPool,
             readerScheduler: readerScheduler,

src/Kestrel.Core/Internal/Http/Http1Connection.cs

@@ -2,11 +2,14 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Buffers;
+using System.Collections.Sequences;
 using System.Diagnostics;
 using System.IO.Pipelines;
 using System.Runtime.InteropServices;
 using System.Text;
 using System.Text.Encodings.Web.Utf8;
+using System.Threading;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 
@@ -66,7 +69,7 @@ public void SendTimeoutResponse()
             Input.CancelPendingRead();
         }
 
-        public void ParseRequest(ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined)
+        public void ParseRequest(ReadOnlyBuffer buffer, out Position consumed, out Position examined)
         {
             consumed = buffer.Start;
             examined = buffer.End;
@@ -104,7 +107,7 @@ public void ParseRequest(ReadableBuffer buffer, out ReadCursor consumed, out Rea
             }
         }
 
-        public bool TakeStartLine(ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined)
+        public bool TakeStartLine(ReadOnlyBuffer buffer, out Position consumed, out Position examined)
         {
             var overLength = false;
             if (buffer.Length >= ServerOptions.Limits.MaxRequestLineSize)
@@ -122,7 +125,7 @@ public bool TakeStartLine(ReadableBuffer buffer, out ReadCursor consumed, out Re
             return result;
         }
 
-        public bool TakeMessageHeaders(ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined)
+        public bool TakeMessageHeaders(ReadOnlyBuffer buffer, out Position consumed, out Position examined)
         {
             // Make sure the buffer is limited
             bool overLength = false;
@@ -424,7 +427,7 @@ protected override void BeginRequestProcessing()
             TimeoutControl.SetTimeout(_keepAliveTicks, TimeoutAction.StopProcessingNextRequest);
         }
 
-        protected override bool BeginRead(out ReadableBufferAwaitable awaitable)
+        protected override bool BeginRead(out ValueAwaiter<ReadResult> awaitable)
         {
             awaitable = Input.ReadAsync();
             return true;

src/Kestrel.Core/Internal/Http/Http1MessageBody.cs

@@ -2,6 +2,8 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Buffers;
+using System.Collections.Sequences;
 using System.IO;
 using System.IO.Pipelines;
 using System.Threading.Tasks;
@@ -149,7 +151,7 @@ protected override async Task OnConsumeAsync()
             }
         }
 
-        protected void Copy(ReadableBuffer readableBuffer, WritableBuffer writableBuffer)
+        protected void Copy(ReadOnlyBuffer readableBuffer, WritableBuffer writableBuffer)
         {
             _context.TimeoutControl.BytesRead(readableBuffer.Length);
 
@@ -171,7 +173,7 @@ protected override void OnReadStarted()
             _pumpTask = PumpAsync();
         }
 
-        protected virtual bool Read(ReadableBuffer readableBuffer, WritableBuffer writableBuffer, out ReadCursor consumed, out ReadCursor examined)
+        protected virtual bool Read(ReadOnlyBuffer readableBuffer, WritableBuffer writableBuffer, out Position consumed, out Position examined)
         {
             throw new NotImplementedException();
         }
@@ -296,7 +298,7 @@ public ForUpgrade(Http1Connection context)
                 RequestUpgrade = true;
             }
 
-            protected override bool Read(ReadableBuffer readableBuffer, WritableBuffer writableBuffer, out ReadCursor consumed, out ReadCursor examined)
+            protected override bool Read(ReadOnlyBuffer readableBuffer, WritableBuffer writableBuffer, out Position consumed, out Position examined)
             {
                 Copy(readableBuffer, writableBuffer);
                 consumed = readableBuffer.End;
@@ -318,7 +320,7 @@ public ForContentLength(bool keepAlive, long contentLength, Http1Connection cont
                 _inputLength = _contentLength;
             }
 
-            protected override bool Read(ReadableBuffer readableBuffer, WritableBuffer writableBuffer, out ReadCursor consumed, out ReadCursor examined)
+            protected override bool Read(ReadOnlyBuffer readableBuffer, WritableBuffer writableBuffer, out Position consumed, out Position examined)
             {
                 if (_inputLength == 0)
                 {
@@ -366,10 +368,10 @@ public ForChunkedEncoding(bool keepAlive, Http1Connection context)
                 RequestKeepAlive = keepAlive;
             }
 
-            protected override bool Read(ReadableBuffer readableBuffer, WritableBuffer writableBuffer, out ReadCursor consumed, out ReadCursor examined)
+            protected override bool Read(ReadOnlyBuffer readableBuffer, WritableBuffer writableBuffer, out Position consumed, out Position examined)
             {
-                consumed = default(ReadCursor);
-                examined = default(ReadCursor);
+                consumed = default(Position);
+                examined = default(Position);
 
                 while (_mode < Mode.Trailer)
                 {
@@ -457,17 +459,17 @@ private void AddAndCheckConsumedBytes(long consumedBytes)
                 }
             }
 
-            private void ParseChunkedPrefix(ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined)
+            private void ParseChunkedPrefix(ReadOnlyBuffer buffer, out Position consumed, out Position examined)
             {
                 consumed = buffer.Start;
                 examined = buffer.Start;
-                var reader = new ReadableBufferReader(buffer);
+                var reader = BufferReader.Create(buffer);
                 var ch1 = reader.Take();
                 var ch2 = reader.Take();
 
                 if (ch1 == -1 || ch2 == -1)
                 {
-                    examined = reader.Cursor;
+                    examined = reader.Position;
                     return;
                 }
 
@@ -478,8 +480,8 @@ private void ParseChunkedPrefix(ReadableBuffer buffer, out ReadCursor consumed,
                 {
                     if (ch1 == ';')
                     {
-                        consumed = reader.Cursor;
-                        examined = reader.Cursor;
+                        consumed = reader.Position;
+                        examined = reader.Position;
 
                         AddAndCheckConsumedBytes(reader.ConsumedBytes);
                         _inputLength = chunkSize;
@@ -490,14 +492,14 @@ private void ParseChunkedPrefix(ReadableBuffer buffer, out ReadCursor consumed,
                     ch2 = reader.Take();
                     if (ch2 == -1)
                     {
-                        examined = reader.Cursor;
+                        examined = reader.Position;
                         return;
                     }
 
                     if (ch1 == '\r' && ch2 == '\n')
                     {
-                        consumed = reader.Cursor;
-                        examined = reader.Cursor;
+                        consumed = reader.Position;
+                        examined = reader.Position;
 
                         AddAndCheckConsumedBytes(reader.ConsumedBytes);
                         _inputLength = chunkSize;
@@ -513,7 +515,7 @@ private void ParseChunkedPrefix(ReadableBuffer buffer, out ReadCursor consumed,
                 _context.ThrowRequestRejected(RequestRejectionReason.BadChunkSizeData);
             }
 
-            private void ParseExtension(ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined)
+            private void ParseExtension(ReadOnlyBuffer buffer, out Position consumed, out Position examined)
             {
                 // Chunk-extensions not currently parsed
                 // Just drain the data
@@ -522,8 +524,8 @@ private void ParseExtension(ReadableBuffer buffer, out ReadCursor consumed, out
 
                 do
                 {
-                    ReadCursor extensionCursor;
-                    if (ReadCursorOperations.Seek(buffer.Start, buffer.End, out extensionCursor, ByteCR) == -1)
+                    Position extensionCursor;
+                    if (ReadOnlyBuffer.Seek(buffer.Start, buffer.End, out extensionCursor, ByteCR) == -1)
                     {
                         // End marker not found yet
                         consumed = buffer.End;
@@ -565,7 +567,7 @@ private void ParseExtension(ReadableBuffer buffer, out ReadCursor consumed, out
                 } while (_mode == Mode.Extension);
             }
 
-            private void ReadChunkedData(ReadableBuffer buffer, WritableBuffer writableBuffer, out ReadCursor consumed, out ReadCursor examined)
+            private void ReadChunkedData(ReadOnlyBuffer buffer, WritableBuffer writableBuffer, out Position consumed, out Position examined)
             {
                 var actual = Math.Min(buffer.Length, _inputLength);
                 consumed = buffer.Move(buffer.Start, actual);
@@ -582,7 +584,7 @@ private void ReadChunkedData(ReadableBuffer buffer, WritableBuffer writableBuffe
                 }
             }
 
-            private void ParseChunkedSuffix(ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined)
+            private void ParseChunkedSuffix(ReadOnlyBuffer buffer, out Position consumed, out Position examined)
             {
                 consumed = buffer.Start;
                 examined = buffer.Start;
@@ -608,7 +610,7 @@ private void ParseChunkedSuffix(ReadableBuffer buffer, out ReadCursor consumed,
                 }
             }
 
-            private void ParseChunkedTrailer(ReadableBuffer buffer, out ReadCursor consumed, out ReadCursor examined)
+            private void ParseChunkedTrailer(ReadOnlyBuffer buffer, out Position consumed, out Position examined)
             {
                 consumed = buffer.Start;
                 examined = buffer.Start;

src/Kestrel.Core/Internal/Http/Http1OutputProducer.cs

@@ -203,7 +203,7 @@ private Task FlushAsync(WritableBuffer writableBuffer, long bytesWritten, Cancel
             return FlushAsyncAwaited(awaitable, bytesWritten, cancellationToken);
         }
 
-        private async Task FlushAsyncAwaited(WritableBufferAwaitable awaitable, long count, CancellationToken cancellationToken)
+        private async Task FlushAsyncAwaited(ValueAwaiter<FlushResult> awaitable, long count, CancellationToken cancellationToken)
         {
             // https://github.com/dotnet/corefxlab/issues/1334
             // Since the flush awaitable doesn't currently support multiple awaiters