kestrel Server的源码分析

今天这一篇博客讲的是.net core 自带的kestrel server，当你开发微服务k8s部署在linux环境下，一般默认开启这个高性能服务，如果大家之前看过我的owin katana的博客，会发现.net core 的好多实现在之前.net standard 的版本已经实现过了，当时开发的asp.net 程序与IIS紧紧耦合在一起，后来的微软团队意识到这个问题并尝试将asp.net 解耦server，制定了owin标准并启动了一个开源项目katana，这个项目的结果并没有带动社区效应，但是此时微软已经制订了.net core的开发，并在katana文档暗示了.net vnext 版本，这个就是。net core 与owin katana 的故事。强烈建议大家有时间看看owin katana,里面有一些 dependency inject, hash map, raw http 协议等等实现。非常收益。说到这些我们开始步入正题吧。原代码在github上的asp.net core 源码。

 上图大致地描述了一个asp.net core 的请求过程，但是我们会发现appication 依赖了server，所以我们需要一个Host 的去解耦server 和aplication 的实现，只要server符合host标准可以任意更换，解耦之后的代码与下图所示。

 所以我们的代码都是创建一个web host然后使用usekestrel,如下所示。

            var host = new WebHostBuilder()
                .UseKestrel(options =>
                {
                    options.Listen(IPAddress.Loopback, 5001);
                })
                .UseStartup<Startup>();

 首先我们知道一个server 实现需要网络编程，所以我们需要socket库来快速编程，它已经帮你实现了tcp与udp协议，不需要自己重复的造轮子。首先我们需要看UseKestrel的方法做了什么。

   public static IWebHostBuilder UseKestrel(this IWebHostBuilder hostBuilder)
        {
            return hostBuilder.ConfigureServices(services =>
            {
                // Don't override an already-configured transport
                services.TryAddSingleton<IConnectionListenerFactory, SocketTransportFactory>();

                services.AddTransient<IConfigureOptions<KestrelServerOptions>, KestrelServerOptionsSetup>();
                services.AddSingleton<IServer, KestrelServer>();
            });
        }

依赖注入注册了三个对象，一个连接池，一个配置类还有一个是server，会和web host注册了IServer 的实现类,然后我们继续看一下，当你调用run的时候会将控制权从web host 转移给server，如下代码第18行所示。

  public virtual async Task StartAsync(CancellationToken cancellationToken = default)
        {
            HostingEventSource.Log.HostStart(); 6 
            var application = BuildApplication();

            // Fire IHostedService.Start
            await _hostedServiceExecutor.StartAsync(cancellationToken).ConfigureAwait(false);//启动后台服务

            var diagnosticSource = _applicationServices.GetRequiredService<DiagnosticListener>();
            var httpContextFactory = _applicationServices.GetRequiredService<IHttpContextFactory>();
            var hostingApp = new HostingApplication(application, _logger, diagnosticSource, httpContextFactory);
            await Server.StartAsync(hostingApp, cancellationToken).ConfigureAwait(false);//socket 启动
            _startedServer = true;

            // Fire IApplicationLifetime.Started
            _applicationLifetime?.NotifyStarted();


            _logger.Started();

            // Log the fact that we did load hosting startup assemblies.
            if (_logger.IsEnabled(LogLevel.Debug))
            {
                foreach (var assembly in _options.GetFinalHostingStartupAssemblies())
                {
                    _logger.LogDebug("Loaded hosting startup assembly {assemblyName}", assembly);
                }
            }

            if (_hostingStartupErrors != null)
            {
                foreach (var exception in _hostingStartupErrors.InnerExceptions)
                {
                    _logger.HostingStartupAssemblyError(exception);
                }
            }
        }

当我们转进到StartAsync方法时会看到如下代码

      public async Task StartAsync<TContext>(IHttpApplication<TContext> application, CancellationToken cancellationToken)
        {
            try
            {
               // ServiceContext.Heartbeat?.Start();//一个是连接池一个是日期时间

                async Task OnBind(ListenOptions options)
                {
                    // Add the HTTP middleware as the terminal connection middleware
                    options.UseHttpServer(ServiceContext, application, options.Protocols);//注册中间件

                    var connectionDelegate = options.Build();

                    // Add the connection limit middleware
                    if (Options.Limits.MaxConcurrentConnections.HasValue)
                    {
                        connectionDelegate = new ConnectionLimitMiddleware(connectionDelegate, Options.Limits.MaxConcurrentConnections.Value, Trace).OnConnectionAsync;
                    }

                    var connectionDispatcher = new ConnectionDispatcher(ServiceContext, connectionDelegate);
                    var transport = await _transportFactory.BindAsync(options.EndPoint).ConfigureAwait(false);

                    // Update the endpoint
                    options.EndPoint = transport.EndPoint;
                    var acceptLoopTask = connectionDispatcher.StartAcceptingConnections(transport);

                    _transports.Add((transport, acceptLoopTask));
                }

                await AddressBinder.BindAsync(_serverAddresses, Options, Trace, OnBind).ConfigureAwait(false);
            }
            catch (Exception ex)
            {51             }
        }

AddressBinder就是server绑定的ip地址，这个可以在StartUp方法或者环境变量里面配置，里面传了一个回调方法OnBind, 在第24行的UseHttpServer会注册server 内部的中间件去处理这个请求，在第35行socet会绑定地址，用tcp协议，默认使用512个最大pending队列，在接受socket会有多处异步编程和开启线程，建议大家在调试的时候可以修改代码用尽可能少的线程来进行调试。accept 的代码如下图所示

     private void StartAcceptingConnectionsCore(IConnectionListener listener)
        {
            // REVIEW: Multiple accept loops in parallel?
            _ = AcceptConnectionsAsync();

            async Task AcceptConnectionsAsync()
            {
                try
                {
                    while (true)
                    {
                        var connection = await listener.AcceptAsync();
19 
                        // Add the connection to the connection manager before we queue it for execution
                        var id = Interlocked.Increment(ref _lastConnectionId);
                        var kestrelConnection = new KestrelConnection(id, _serviceContext, _connectionDelegate, connection, Log);

                        _serviceContext.ConnectionManager.AddConnection(id, kestrelConnection);27 
                        ThreadPool.UnsafeQueueUserWorkItem(kestrelConnection, preferLocal: false);
                       }
                }
                catch (Exception ex)
                {
                    // REVIEW: If the accept loop ends should this trigger a server shutdown? It will manifest as a hang
                    Log.LogCritical(0, ex, "The connection listener failed to accept any new connections.");
                }
                finally
                {
                    _acceptLoopTcs.TrySetResult(null);
                }
            }
        }

接收到accept socket的时候，会创建一个kestrelconnection 对象，这个对象实现线程方法，然后它会重新去等待一个请求的到来，而用户代码的执行则交给线程池执行。在第14行就是之前kerstrel server 内部的中间件build生成的方法，他的主要功能就是解析socket的携带http信息。

    internal async Task ExecuteAsync()
        {
            var connectionContext = TransportConnection;

            try
            {
                using (BeginConnectionScope(connectionContext))
                {
                    try
                    {
                        await _connectionDelegate(connectionContext);
                    }
                    catch (Exception ex)
                    {
                        Logger.LogError(0, ex, "Unhandled exception while processing {ConnectionId}.", connectionContext.ConnectionId);
                    }
                }
            }
            finally
            {
                _serviceContext.ConnectionManager.RemoveConnection(_id);
            }
        }

由于http协议版本的不一致导致解析方式的不同，如果有兴趣的小伙伴可以具体查看这一块的逻辑。

                switch (SelectProtocol())
                {
                    case HttpProtocols.Http1:
                        // _http1Connection must be initialized before adding the connection to the connection manager
                        requestProcessor = _http1Connection = new Http1Connection<TContext>(_context);
                        _protocolSelectionState = ProtocolSelectionState.Selected;
                        break;
                    case HttpProtocols.Http2:
                        // _http2Connection must be initialized before yielding control to the transport thread,
                        // to prevent a race condition where _http2Connection.Abort() is called just as
                        // _http2Connection is about to be initialized.
                        requestProcessor = new Http2Connection(_context);
                        _protocolSelectionState = ProtocolSelectionState.Selected;
                        break;
                    case HttpProtocols.None:
                        // An error was already logged in SelectProtocol(), but we should close the connection.
                        break;
                    default:
                        // SelectProtocol() only returns Http1, Http2 or None.
                        throw new NotSupportedException($"{nameof(SelectProtocol)} returned something other than Http1, Http2 or None.");
                }

然后server解析完请求之后所做的重要的一步就是创建httpContext,然后server在第40行将控制权转给web host，web host 会自动调用application code 也就是用户代码。

     private async Task ProcessRequests<TContext>(IHttpApplication<TContext> application)
        {
            while (_keepAlive)
            {
                var context = application.CreateContext(this);

                try
                {
                    KestrelEventSource.Log.RequestStart(this);

                    // Run the application code for this request
                    await application.ProcessRequestAsync(context);
 55                 }
                catch (BadHttpRequestException ex)
                {
                    // Capture BadHttpRequestException for further processing
                    // This has to be caught here so StatusCode is set properly before disposing the HttpContext
                    // (DisposeContext logs StatusCode).
                    SetBadRequestState(ex);
                    ReportApplicationError(ex);
                }
                catch (Exception ex)
                {
                    ReportApplicationError(ex);
                }

                KestrelEventSource.Log.RequestStop(this);129             }
        }

到这里server 的工作大部分都结束了，在之前的描述中我们看到web host 怎么将控制权给到server 的， server 创建好httpContext规则后又是如何将控制权给到web host ， web host 又如何去调用application code的， web host 实际上build 的时候将用户的中间件定义为链表结构暴露一个入口供web host调用，其他的有时间我会再写博客描述这一块。谢谢大家今天的阅读了。欢迎大家能够留言一起讨论。如果有任何不懂的地方可以私信我。