Refactor how we capture the IHttpApplication (#28413)

- Use a similar technique to capture the generic IHttpApplication that we do in IIS and Kestrel. Now the AsyncAcceptContext takes an IRequestContextFactory to delegate RequestContext creation to pull the request from the HTTP.sys queue.

Author: davidfowl

src/Servers/HttpSys/src/ApplicationRequestContextFactory.cs

@@ -0,0 +1,24 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+using Microsoft.AspNetCore.Hosting.Server;
+
+namespace Microsoft.AspNetCore.Server.HttpSys
+{
+    internal class ApplicationRequestContextFactory<TContext> : IRequestContextFactory
+    {
+        private readonly IHttpApplication<TContext> _application;
+        private readonly MessagePump _messagePump;
+
+        public ApplicationRequestContextFactory(IHttpApplication<TContext> application, MessagePump messagePump)
+        {
+            _application = application;
+            _messagePump = messagePump;
+        }
+
+        public RequestContext CreateRequestContext(uint? bufferSize, ulong requestId)
+        {
+            return new RequestContext<TContext>(_application, _messagePump, _messagePump.Listener, bufferSize, requestId);
+        }
+    }
+}

src/Servers/HttpSys/src/AsyncAcceptContext.cs

@@ -24,10 +24,12 @@ internal unsafe class AsyncAcceptContext : IValueTaskSource<RequestContext>, IDi
         };
 
         private RequestContext _requestContext;
+        private readonly IRequestContextFactory _requestContextFactory;
 
-        internal AsyncAcceptContext(HttpSysListener server)
+        internal AsyncAcceptContext(HttpSysListener server, IRequestContextFactory requestContextFactory)
         {
             Server = server;
+            _requestContextFactory = requestContextFactory;
             _preallocatedOverlapped = new(IOCallback, state: this, pinData: null);
         }
 
@@ -193,7 +195,7 @@ private void AllocateNativeRequest(uint? size = null, ulong requestId = 0)
                 boundHandle.FreeNativeOverlapped(_overlapped);
             }
 
-            _requestContext = new RequestContext(Server, size, requestId);
+            _requestContext = _requestContextFactory.CreateRequestContext(size, requestId);
             _overlapped = boundHandle.AllocateNativeOverlapped(_preallocatedOverlapped);
         }
 

src/Servers/HttpSys/src/IRequestContextFactory.cs

@@ -0,0 +1,10 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+namespace Microsoft.AspNetCore.Server.HttpSys
+{
+    internal interface IRequestContextFactory
+    {
+        RequestContext CreateRequestContext(uint? bufferSize, ulong requestId);
+    }
+}

src/Servers/HttpSys/src/MessagePump.cs

@@ -66,7 +66,7 @@ public MessagePump(IOptions<HttpSysOptions> options, ILoggerFactory loggerFactor
 
         internal HttpSysListener Listener { get; }
 
-        internal IHttpApplication<object> Application { get; set; }
+        internal IRequestContextFactory RequestContextFactory { get; set; }
 
         public IFeatureCollection Features { get; }
 
@@ -118,12 +118,12 @@ public Task StartAsync<TContext>(IHttpApplication<TContext> application, Cancell
             }
             // else // Attaching to an existing queue, don't add a default.
 
-            // Can't call Start twice
-            Debug.Assert(Application == null);
+            // Can't start twice
+            Debug.Assert(RequestContextFactory != null);
 
             Debug.Assert(application != null);
 
-            Application = new ApplicationWrapper<TContext>(application);
+            RequestContextFactory = new ApplicationRequestContextFactory<TContext>(application, this);
 
             Listener.Start();
 
@@ -179,7 +179,7 @@ internal void SetShutdownSignal()
         private async Task ProcessRequestsWorker()
         {
             // Allocate and accept context per loop and reuse it for all accepts
-            using var acceptContext = new AsyncAcceptContext(Listener);
+            using var acceptContext = new AsyncAcceptContext(Listener, RequestContextFactory);
 
             int workerIndex = Interlocked.Increment(ref _acceptorCounts);
             while (!Stopping && workerIndex <= _maxAccepts)
@@ -189,8 +189,6 @@ private async Task ProcessRequestsWorker()
                 try
                 {
                     requestContext = await Listener.AcceptAsync(acceptContext);
-                    // Assign the message pump to this request context
-                    requestContext.MessagePump = this;
                 }
                 catch (Exception exception)
                 {
@@ -268,30 +266,5 @@ public void Dispose()
 
             Listener.Dispose();
         }
-
-        private class ApplicationWrapper<TContext> : IHttpApplication<object>
-        {
-            private readonly IHttpApplication<TContext> _application;
-
-            public ApplicationWrapper(IHttpApplication<TContext> application)
-            {
-                _application = application;
-            }
-
-            public object CreateContext(IFeatureCollection contextFeatures)
-            {
-                return _application.CreateContext(contextFeatures);
-            }
-
-            public void DisposeContext(object context, Exception exception)
-            {
-                _application.DisposeContext((TContext)context, exception);
-            }
-
-            public Task ProcessRequestAsync(object context)
-            {
-                return _application.ProcessRequestAsync((TContext)context);
-            }
-        }
     }
 }

src/Servers/HttpSys/src/RequestProcessing/RequestContext.cs

@@ -14,7 +14,7 @@
 
 namespace Microsoft.AspNetCore.Server.HttpSys
 {
-    internal sealed partial class RequestContext : NativeRequestContext, IThreadPoolWorkItem
+    internal partial class RequestContext : NativeRequestContext, IThreadPoolWorkItem
     {
         private static readonly Action<object> AbortDelegate = Abort;
         private CancellationTokenSource _requestAbortSource;
@@ -29,8 +29,6 @@ public RequestContext(HttpSysListener server, uint? bufferSize, ulong requestId)
             AllowSynchronousIO = server.Options.AllowSynchronousIO;
         }
 
-        internal MessagePump MessagePump { get; set; }
-
         internal HttpSysListener Server { get; }
 
         internal ILogger Logger => Server.Logger;
@@ -243,82 +241,17 @@ internal unsafe void SetResetCode(int errorCode)
             }
         }
 
-        public async void Execute()
+        protected virtual Task ExecuteAsync()
         {
-            var messagePump = MessagePump;
-            var application = messagePump.Application;
-
-            try
-            {
-                if (messagePump.Stopping)
-                {
-                    SetFatalResponse(503);
-                    return;
-                }
-
-                object context = null;
-                messagePump.IncrementOutstandingRequest();
-                try
-                {
-                    context = application.CreateContext(Features);
-                    try
-                    {
-                        await application.ProcessRequestAsync(context);
-                        await CompleteAsync();
-                    }
-                    finally
-                    {
-                        await OnCompleted();
-                    }
-                    application.DisposeContext(context, null);
-                    Dispose();
-                }
-                catch (Exception ex)
-                {
-                    Logger.LogError(LoggerEventIds.RequestProcessError, ex, "ProcessRequestAsync");
-                    application.DisposeContext(context, ex);
-                    if (Response.HasStarted)
-                    {
-                        // HTTP/2 INTERNAL_ERROR = 0x2 https://tools.ietf.org/html/rfc7540#section-7
-                        // Otherwise the default is Cancel = 0x8.
-                        SetResetCode(2);
-                        Abort();
-                    }
-                    else
-                    {
-                        // We haven't sent a response yet, try to send a 500 Internal Server Error
-                        Response.Headers.IsReadOnly = false;
-                        Response.Trailers.IsReadOnly = false;
-                        Response.Headers.Clear();
-                        Response.Trailers.Clear();
-
-                        if (ex is BadHttpRequestException badHttpRequestException)
-                        {
-                            SetFatalResponse(badHttpRequestException.StatusCode);
-                        }
-                        else
-                        {
-                            SetFatalResponse(StatusCodes.Status500InternalServerError);
-                        }
-                    }
-                }
-                finally
-                {
-                    if (messagePump.DecrementOutstandingRequest() == 0 && messagePump.Stopping)
-                    {
-                        Logger.LogInformation(LoggerEventIds.RequestsDrained, "All requests drained.");
-                        messagePump.SetShutdownSignal();
-                    }
-                }
-            }
-            catch (Exception ex)
-            {
-                Logger.LogError(LoggerEventIds.RequestError, ex, "ProcessRequestAsync");
-                Abort();
-            }
+            return Task.CompletedTask;
+        }
+        
+        public void Execute()
+        {
+            _ = ExecuteAsync();
         }
 
-        private void SetFatalResponse(int status)
+        protected void SetFatalResponse(int status)
         {
             Response.StatusCode = status;
             Response.ContentLength = 0;

src/Servers/HttpSys/src/RequestProcessing/RequestContextOfT.cs

@@ -0,0 +1,97 @@
+using System;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Hosting.Server;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Logging;
+
+namespace Microsoft.AspNetCore.Server.HttpSys
+{
+    internal sealed class RequestContext<TContext> : RequestContext
+    {
+        private readonly IHttpApplication<TContext> _application;
+        private readonly MessagePump _messagePump;
+
+        public RequestContext(IHttpApplication<TContext> application, MessagePump messagePump, HttpSysListener server, uint? bufferSize, ulong requestId)
+            : base(server, bufferSize, requestId)
+        {
+            _application = application;
+            _messagePump = messagePump;
+        }
+
+        protected override async Task ExecuteAsync()
+        {
+            var messagePump = _messagePump;
+            var application = _application;
+
+            try
+            {
+                if (messagePump.Stopping)
+                {
+                    SetFatalResponse(503);
+                    return;
+                }
+
+                TContext context = default;
+                messagePump.IncrementOutstandingRequest();
+                try
+                {
+                    context = application.CreateContext(Features);
+                    try
+                    {
+                        await application.ProcessRequestAsync(context);
+                        await CompleteAsync();
+                    }
+                    finally
+                    {
+                        await OnCompleted();
+                    }
+                    application.DisposeContext(context, null);
+                    Dispose();
+                }
+                catch (Exception ex)
+                {
+                    Logger.LogError(LoggerEventIds.RequestProcessError, ex, "ProcessRequestAsync");
+                    application.DisposeContext(context, ex);
+                    if (Response.HasStarted)
+                    {
+                        // HTTP/2 INTERNAL_ERROR = 0x2 https://tools.ietf.org/html/rfc7540#section-7
+                        // Otherwise the default is Cancel = 0x8.
+                        SetResetCode(2);
+                        Abort();
+                    }
+                    else
+                    {
+                        // We haven't sent a response yet, try to send a 500 Internal Server Error
+                        Response.Headers.IsReadOnly = false;
+                        Response.Trailers.IsReadOnly = false;
+                        Response.Headers.Clear();
+                        Response.Trailers.Clear();
+
+                        if (ex is BadHttpRequestException badHttpRequestException)
+                        {
+                            SetFatalResponse(badHttpRequestException.StatusCode);
+                        }
+                        else
+                        {
+                            SetFatalResponse(StatusCodes.Status500InternalServerError);
+                        }
+                    }
+                }
+                finally
+                {
+                    if (messagePump.DecrementOutstandingRequest() == 0 && messagePump.Stopping)
+                    {
+                        Logger.LogInformation(LoggerEventIds.RequestsDrained, "All requests drained.");
+                        messagePump.SetShutdownSignal();
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                Logger.LogError(LoggerEventIds.RequestError, ex, "ProcessRequestAsync");
+                Abort();
+            }
+        }
+    }
+
+}

src/Servers/HttpSys/test/FunctionalTests/Listener/Utilities.cs

@@ -113,7 +113,8 @@ internal static HttpSysListener CreateServerOnExistingQueue(AuthenticationScheme
         /// </summary>
         internal static async Task<RequestContext> AcceptAsync(this HttpSysListener server, TimeSpan timeout)
         {
-            var acceptContext = new AsyncAcceptContext(server);
+            var factory = new TestRequestContextFactory(server);
+            var acceptContext = new AsyncAcceptContext(server, factory);
             var acceptTask = server.AcceptAsync(acceptContext).AsTask();
             var completedTask = await Task.WhenAny(acceptTask, Task.Delay(timeout));
 
@@ -143,5 +144,20 @@ internal static async Task<RequestContext> Before<T>(this Task<RequestContext> a
                 throw new InvalidOperationException("The response completed prematurely: " + response.ToString());
             }
         }
+
+        private class TestRequestContextFactory : IRequestContextFactory
+        {
+            private readonly HttpSysListener _server;
+
+            public TestRequestContextFactory(HttpSysListener server)
+            {
+                _server = server;
+            }
+
+            public RequestContext CreateRequestContext(uint? bufferSize, ulong requestId)
+            {
+                return new RequestContext(_server, bufferSize, requestId);
+            }
+        }
     }
 }