Plug in and switch on the 'HTTP/2 first' server handlers (#1050)

Motivation:

We have all the parts in place to use HTTP/2 concepts by default on the
server, so let's use them!

Modifications:

- Wire in the server error delegate to the 'HTTP2ToRawGRPCServerCodec'
  since this was missing
- Modify the 'HTTPProtocolSwitcher' and relevant pipeline initializer to
  configure the pipeline correctly
- Update various tests which relied on previous types
- Move 'CallPath' out of the routing handler
- Remove the routing handler and tests
- Move the 'MessageEncodingHeaderValidator' out of the HTTP1 to GRPC
  server handler
- Remove the 'HTTP1ToGRPCServerCodec'

Result:

When handling requests on HTTP/2 connections we no longer go translate
to HTTP1 types first.

A ~30-35% increase in QPS in unary benchmarks.

Author: glbrntt

Sources/GRPC/CallHandlers/_BaseCallHandler.swift

@@ -30,8 +30,8 @@ public class _BaseCallHandler<
   public typealias RequestPayload = RequestDeserializer.Output
   public typealias ResponsePayload = ResponseSerializer.Input
 
-  public typealias InboundIn = _RawGRPCServerRequestPart
-  public typealias OutboundOut = _RawGRPCServerResponsePart
+  public typealias InboundIn = GRPCServerRequestPart<ByteBuffer>
+  public typealias OutboundOut = GRPCServerResponsePart<ByteBuffer>
 
   /// An interceptor pipeline.
   private var pipeline: ServerInterceptorPipeline<RequestPayload, ResponsePayload>?
@@ -117,8 +117,8 @@ public class _BaseCallHandler<
     let part = self.unwrapInboundIn(data)
 
     switch part {
-    case let .headers(headers):
-      self.act(on: self.state.channelRead(.headers(headers)))
+    case let .metadata(headers):
+      self.act(on: self.state.channelRead(.metadata(headers)))
     case let .message(buffer):
       do {
         let request = try self.requestDeserializer.deserialize(byteBuffer: buffer)
@@ -466,7 +466,7 @@ extension _BaseCallHandler.State {
   /// Receive a request part from the `Channel`. If we're active we just forward these through the
   /// pipeline. We validate at the other end.
   internal mutating func channelRead(
-    _ requestPart: _GRPCServerRequestPart<_BaseCallHandler.RequestPayload>
+    _ requestPart: GRPCServerRequestPart<_BaseCallHandler.RequestPayload>
   ) -> Action {
     switch self {
     case .idle:
@@ -480,7 +480,7 @@ extension _BaseCallHandler.State {
       let part: GRPCServerRequestPart<_BaseCallHandler.RequestPayload>
 
       switch requestPart {
-      case let .headers(headers):
+      case let .metadata(headers):
         filter = state.channelStreamState.receiveHeaders()
         part = .metadata(headers)
       case let .message(message):
@@ -703,7 +703,7 @@ extension _BaseCallHandler {
       // Only flush if we're streaming responses, if we're not streaming responses then we'll wait
       // for the response and end before emitting the flush.
       flush = self.callType.isStreamingResponses
-      context.write(self.wrapOutboundOut(.headers(headers)), promise: promise)
+      context.write(self.wrapOutboundOut(.metadata(headers)), promise: promise)
 
     case let .message(message, metadata):
       do {
@@ -712,7 +712,7 @@ extension _BaseCallHandler {
           allocator: context.channel.allocator
         )
         context.write(
-          self.wrapOutboundOut(.message(.init(serializedResponse, compressed: metadata.compress))),
+          self.wrapOutboundOut(.message(serializedResponse, metadata)),
           promise: promise
         )
         // Flush if we've been told to flush.
@@ -724,7 +724,7 @@ extension _BaseCallHandler {
       }
 
     case let .end(status, trailers):
-      context.write(self.wrapOutboundOut(.statusAndTrailers(status, trailers)), promise: promise)
+      context.write(self.wrapOutboundOut(.end(status, trailers)), promise: promise)
       // Always flush on end.
       flush = true
     }

Sources/GRPC/GRPCServerRequestRoutingHandler.swift

@@ -47,246 +47,34 @@ public struct CallHandlerContext {
   internal var path: String
 }
 
-/// Attempts to route a request to a user-provided call handler. Also validates that the request has
-/// a suitable 'content-type' for gRPC.
-///
-/// Once the request headers are available, asks the `CallHandlerProvider` corresponding to the request's service name
-/// for a `GRPCCallHandler` object. That object is then forwarded the individual gRPC messages.
-///
-/// After the pipeline has been configured with the `GRPCCallHandler`, this handler removes itself
-/// from the pipeline.
-public final class GRPCServerRequestRoutingHandler {
-  private let logger: Logger
-  private let servicesByName: [Substring: CallHandlerProvider]
-  private let encoding: ServerMessageEncoding
-  private weak var errorDelegate: ServerErrorDelegate?
-
-  private enum State: Equatable {
-    case notConfigured
-    case configuring([InboundOut])
-  }
-
-  private var state: State = .notConfigured
-
-  public init(
-    servicesByName: [Substring: CallHandlerProvider],
-    encoding: ServerMessageEncoding,
-    errorDelegate: ServerErrorDelegate?,
-    logger: Logger
-  ) {
-    self.servicesByName = servicesByName
-    self.encoding = encoding
-    self.errorDelegate = errorDelegate
-    self.logger = logger
-  }
-}
-
-extension GRPCServerRequestRoutingHandler: ChannelInboundHandler, RemovableChannelHandler {
-  public typealias InboundIn = HTTPServerRequestPart
-  public typealias InboundOut = HTTPServerRequestPart
-  public typealias OutboundOut = HTTPServerResponsePart
-
-  public func errorCaught(context: ChannelHandlerContext, error: Error) {
-    let status: GRPCStatus
-    if let errorWithContext = error as? GRPCError.WithContext {
-      self.errorDelegate?.observeLibraryError(errorWithContext.error)
-      status = errorWithContext.error.makeGRPCStatus()
-    } else {
-      self.errorDelegate?.observeLibraryError(error)
-      status = (error as? GRPCStatusTransformable)?.makeGRPCStatus() ?? .processingError
-    }
-
-    switch self.state {
-    case .notConfigured:
-      // We don't know what protocol we're speaking at this point. We'll just have to close the
-      // channel.
-      ()
-
-    case let .configuring(messages):
-      // first! is fine here: we only go from `.notConfigured` to `.configuring` when we receive
-      // and validate the request head.
-      let head = messages.compactMap { part -> HTTPRequestHead? in
-        switch part {
-        case let .head(head):
-          return head
-        default:
-          return nil
-        }
-      }.first!
-
-      let responseHead = self.makeResponseHead(requestHead: head, status: status)
-      context.write(self.wrapOutboundOut(.head(responseHead)), promise: nil)
-      context.write(self.wrapOutboundOut(.end(nil)), promise: nil)
-      context.flush()
-    }
-
-    context.close(mode: .all, promise: nil)
-  }
-
-  public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
-    let requestPart = self.unwrapInboundIn(data)
-    switch requestPart {
-    case let .head(requestHead):
-      precondition(self.state == .notConfigured)
-
-      // Validate the 'content-type' is related to gRPC before proceeding.
-      let maybeContentType = requestHead.headers.first(name: GRPCHeaderName.contentType)
-      guard let contentType = maybeContentType,
-        contentType.starts(with: ContentType.commonPrefix) else {
-        self.logger.warning(
-          "received request whose 'content-type' does not exist or start with '\(ContentType.commonPrefix)'",
-          metadata: ["content-type": "\(String(describing: maybeContentType))"]
-        )
-
-        // From: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
-        //
-        //   If 'content-type' does not begin with "application/grpc", gRPC servers SHOULD respond
-        //   with HTTP status of 415 (Unsupported Media Type). This will prevent other HTTP/2
-        //   clients from interpreting a gRPC error response, which uses status 200 (OK), as
-        //   successful.
-        let responseHead = HTTPResponseHead(
-          version: requestHead.version,
-          status: .unsupportedMediaType
-        )
-
-        // Fail the call. Note: we're not speaking gRPC here, so no status or message.
-        context.write(self.wrapOutboundOut(.head(responseHead)), promise: nil)
-        context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
-        return
-      }
-
-      // Do we know how to handle this RPC?
-      guard let callHandler = self.makeCallHandler(
-        channel: context.channel,
-        requestHead: requestHead
-      ) else {
-        self.logger.warning(
-          "unable to make call handler; the RPC is not implemented on this server",
-          metadata: ["uri": "\(requestHead.uri)"]
-        )
-
-        let status = GRPCError.RPCNotImplemented(rpc: requestHead.uri).makeGRPCStatus()
-        let responseHead = self.makeResponseHead(requestHead: requestHead, status: status)
-
-        // Write back a 'trailers-only' response.
-        context.write(self.wrapOutboundOut(.head(responseHead)), promise: nil)
-        context.writeAndFlush(self.wrapOutboundOut(.end(nil)), promise: nil)
-        return
-      }
-
-      self.logger.debug("received request head, configuring pipeline")
-
-      // Buffer the request head; we'll replay it in the next handler when we're removed from the
-      // pipeline.
-      self.state = .configuring([requestPart])
-
-      // Configure the rest of the pipeline to serve the RPC.
-      let httpToGRPC = HTTP1ToGRPCServerCodec(encoding: self.encoding, logger: self.logger)
-      context.pipeline.addHandlers([httpToGRPC, callHandler], position: .after(self))
-        .whenSuccess {
-          context.pipeline.removeHandler(self, promise: nil)
-        }
-
-    case .body, .end:
-      switch self.state {
-      case .notConfigured:
-        // We can reach this point if we're receiving messages for a method that isn't implemented,
-        // in which case we just drop the messages; our response should already be in-flight.
-        ()
-
-      case var .configuring(buffer):
-        // We received a message while the pipeline was being configured; hold on to it while we
-        // finish configuring the pipeline.
-        buffer.append(requestPart)
-        self.state = .configuring(buffer)
-      }
-    }
-  }
-
-  public func handlerRemoved(context: ChannelHandlerContext) {
-    switch self.state {
-    case .notConfigured:
-      ()
-
-    case let .configuring(messages):
-      for message in messages {
-        context.fireChannelRead(self.wrapInboundOut(message))
-      }
-    }
-  }
-
-  /// A call URI split into components.
-  struct CallPath {
-    /// The name of the service to call.
-    var service: String.UTF8View.SubSequence
-    /// The name of the method to call.
-    var method: String.UTF8View.SubSequence
-
-    /// Charater used to split the path into components.
-    private let pathSplitDelimiter = UInt8(ascii: "/")
-
-    /// Split a path into service and method.
-    /// Split is done in UTF8 as this turns out to be approximately 10x faster than a simple split.
-    /// URI format: "/package.Servicename/MethodName"
-    init?(requestURI: String) {
-      var utf8View = requestURI.utf8[...]
-      // Check and remove the split character at the beginning.
-      guard let prefix = utf8View.trimPrefix(to: self.pathSplitDelimiter), prefix.isEmpty else {
-        return nil
-      }
-      guard let service = utf8View.trimPrefix(to: pathSplitDelimiter) else {
-        return nil
-      }
-      guard let method = utf8View.trimPrefix(to: pathSplitDelimiter) else {
-        return nil
-      }
-
-      self.service = service
-      self.method = method
+/// A call URI split into components.
+struct CallPath {
+  /// The name of the service to call.
+  var service: String.UTF8View.SubSequence
+  /// The name of the method to call.
+  var method: String.UTF8View.SubSequence
+
+  /// Charater used to split the path into components.
+  private let pathSplitDelimiter = UInt8(ascii: "/")
+
+  /// Split a path into service and method.
+  /// Split is done in UTF8 as this turns out to be approximately 10x faster than a simple split.
+  /// URI format: "/package.Servicename/MethodName"
+  init?(requestURI: String) {
+    var utf8View = requestURI.utf8[...]
+    // Check and remove the split character at the beginning.
+    guard let prefix = utf8View.trimPrefix(to: self.pathSplitDelimiter), prefix.isEmpty else {
+      return nil
     }
-  }
-
-  private func makeCallHandler(channel: Channel, requestHead: HTTPRequestHead) -> GRPCCallHandler? {
-    // URI format: "/package.Servicename/MethodName", resulting in the following components separated by a slash:
-    // - uriComponents[0]: empty
-    // - uriComponents[1]: service name (including the package name);
-    //     `CallHandlerProvider`s should provide the service name including the package name.
-    // - uriComponents[2]: method name.
-    self.logger.debug("making call handler", metadata: ["path": "\(requestHead.uri)"])
-    let uriComponents = CallPath(requestURI: requestHead.uri)
-
-    let context = CallHandlerContext(
-      errorDelegate: self.errorDelegate,
-      logger: self.logger,
-      encoding: self.encoding,
-      eventLoop: channel.eventLoop,
-      path: requestHead.uri
-    )
-
-    guard let callPath = uriComponents,
-      let providerForServiceName = servicesByName[String.SubSequence(callPath.service)],
-      let callHandler = providerForServiceName.handleMethod(
-        String.SubSequence(callPath.method),
-        callHandlerContext: context
-      ) else {
-      self.logger.notice("could not create handler", metadata: ["path": "\(requestHead.uri)"])
+    guard let service = utf8View.trimPrefix(to: pathSplitDelimiter) else {
       return nil
     }
-    return callHandler
-  }
-
-  private func makeResponseHead(requestHead: HTTPRequestHead,
-                                status: GRPCStatus) -> HTTPResponseHead {
-    var headers: HTTPHeaders = [
-      GRPCHeaderName.contentType: ContentType.protobuf.canonicalValue,
-      GRPCHeaderName.statusCode: "\(status.code.rawValue)",
-    ]
-
-    if let message = status.message.flatMap(GRPCStatusMessageMarshaller.marshall) {
-      headers.add(name: GRPCHeaderName.statusMessage, value: message)
+    guard let method = utf8View.trimPrefix(to: pathSplitDelimiter) else {
+      return nil
     }
 
-    return HTTPResponseHead(version: requestHead.version, status: .ok, headers: headers)
+    self.service = service
+    self.method = method
   }
 }