=htc Byte string parser input for request target parsing (#21018)

* Avoid byte array allocation in request target parsing

* Request parsing benchmark added

* Missing copyright added

akka-bench-jmh/src/main/scala/akka/http/HttpRequestParsingBenchmark.scala

@@ -0,0 +1,80 @@
+/**
+ * Copyright (C) 2015-2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.http
+
+import java.util.concurrent.{ CountDownLatch, TimeUnit }
+import javax.net.ssl.SSLContext
+
+import akka.Done
+import akka.actor.ActorSystem
+import akka.http.impl.engine.parsing.ParserOutput.RequestOutput
+import akka.http.impl.engine.parsing.{ HttpHeaderParser, HttpMessageParser, HttpRequestParser }
+import akka.http.scaladsl.settings.ParserSettings
+import akka.stream.TLSProtocol.SessionBytes
+import akka.stream.scaladsl.RunnableGraph
+import akka.stream.{ ActorMaterializer, Attributes }
+import akka.stream.scaladsl.{ Flow, Keep, Sink, Source }
+import akka.util.ByteString
+import org.openjdk.jmh.annotations.{ OperationsPerInvocation, _ }
+import org.openjdk.jmh.infra.Blackhole
+
+import scala.concurrent.{ Await, Future }
+import scala.concurrent.duration._
+
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@BenchmarkMode(Array(Mode.Throughput))
+class HttpRequestParsingBenchmark {
+
+  implicit val system: ActorSystem = ActorSystem("HttpRequestParsingBenchmark")
+  implicit val materializer = ActorMaterializer()
+  val parserSettings = ParserSettings(system)
+  val parser = new HttpRequestParser(parserSettings, false, HttpHeaderParser(parserSettings)())
+  val dummySession = SSLContext.getDefault.createSSLEngine.getSession
+  val requestBytes = SessionBytes(
+    dummySession,
+    ByteString(
+      "GET / HTTP/1.1\r\n" +
+        "Accept: */*\r\n" +
+        "Accept-Encoding: gzip, deflate\r\n" +
+        "Connection: keep-alive\r\n" +
+        "Host: example.com\r\n" +
+        "User-Agent: HTTPie/0.9.3\r\n" +
+        "\r\n"
+    )
+  )
+
+  val httpMessageParser = Flow.fromGraph(parser)
+
+  def flow(n: Int): RunnableGraph[Future[Done]] =
+    Source.repeat(requestBytes).take(n)
+      .via(httpMessageParser)
+      .toMat(Sink.ignore)(Keep.right)
+
+  @Benchmark
+  @OperationsPerInvocation(10000)
+  def parse_10000_single_requests(blackhole: Blackhole): Unit = {
+    val done = flow(10000).run()
+    Await.ready(done, 32.days)
+  }
+
+  @Benchmark
+  @OperationsPerInvocation(1000)
+  def parse_1000_single_requests(blackhole: Blackhole): Unit = {
+    val done = flow(1000).run()
+    Await.ready(done, 32.days)
+  }
+
+  @Benchmark
+  @OperationsPerInvocation(100)
+  def parse_100_single_requests(blackhole: Blackhole): Unit = {
+    val done = flow(100).run()
+    Await.ready(done, 32.days)
+  }
+
+  @TearDown
+  def shutdown(): Unit = {
+    Await.result(system.terminate(), 5.seconds)
+  }
+}

akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpRequestParser.scala

@@ -15,6 +15,7 @@ import akka.http.scaladsl.model._
 import headers._
 import StatusCodes._
 import ParserOutput._
+import akka.http.impl.util.ByteStringParserInput
 import akka.stream.{ Attributes, FlowShape, Inlet, Outlet }
 import akka.stream.TLSProtocol.SessionBytes
 import akka.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler }
@@ -47,7 +48,7 @@ private[http] final class HttpRequestParser(
 
     private[this] var method: HttpMethod = _
     private[this] var uri: Uri = _
-    private[this] var uriBytes: Array[Byte] = _
+    private[this] var uriBytes: ByteString = _
 
     override def onPush(): Unit = handleParserOutput(parseSessionBytes(grab(in)))
     override def onPull(): Unit = handleParserOutput(doPull())
@@ -134,8 +135,8 @@ private[http] final class HttpRequestParser(
 
       val uriEnd = findUriEnd()
       try {
-        uriBytes = input.slice(uriStart, uriEnd).toArray[Byte] // TODO: can we reduce allocations here?
-        uri = Uri.parseHttpRequestTarget(uriBytes, mode = uriParsingMode) // TODO ByteStringParserInput?
+        uriBytes = input.slice(uriStart, uriEnd)
+        uri = Uri.parseHttpRequestTarget(new ByteStringParserInput(uriBytes), mode = uriParsingMode)
       } catch {
         case IllegalUriException(info) ⇒ throw new ParsingException(BadRequest, info)
       }
@@ -153,7 +154,7 @@ private[http] final class HttpRequestParser(
           createEntity: EntityCreator[RequestOutput, RequestEntity],
           headers:      List[HttpHeader]                            = headers) = {
           val allHeaders0 =
-            if (rawRequestUriHeader) `Raw-Request-URI`(new String(uriBytes, HttpCharsets.`US-ASCII`.nioCharset)) :: headers
+            if (rawRequestUriHeader) `Raw-Request-URI`(uriBytes.decodeString(HttpCharsets.`US-ASCII`.nioCharset)) :: headers
             else headers
 
           val allHeaders =

akka-http-core/src/main/scala/akka/http/impl/model/parser/UriParser.scala

@@ -13,9 +13,12 @@ import Parser.DeliveryScheme.Either
 import Uri._
 
 // format: OFF
-
-// http://tools.ietf.org/html/rfc3986
-private[http] class UriParser(val input: ParserInput,
+/**
+ * INTERNAL API
+ *
+ * http://tools.ietf.org/html/rfc3986
+ */
+private[http] final class UriParser(val input: ParserInput,
                               val uriParsingCharset: Charset,
                               val uriParsingMode: Uri.ParsingMode,
                               val maxValueStackSize: Int) extends Parser(maxValueStackSize)

akka-http-core/src/main/scala/akka/http/impl/util/ByteStringParserInput.scala

@@ -0,0 +1,29 @@
+/**
+ * Copyright (C) 2015-2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.http.impl.util
+
+import java.nio.charset.StandardCharsets
+
+import akka.parboiled2.ParserInput.DefaultParserInput
+import akka.util.ByteString
+
+/**
+ * ParserInput reading directly off a ByteString. (Based on the ByteArrayBasedParserInput)
+ * This avoids a separate decoding step but assumes that each byte represents exactly one character,
+ * which is encoded by ISO-8859-1!
+ * You can therefore use this ParserInput type only if you know that all input will be `ISO-8859-1`-encoded,
+ * or only contains 7-bit ASCII characters (which is a subset of ISO-8859-1)!
+ *
+ * Note that this ParserInput type will NOT work with general `UTF-8`-encoded input as this can contain
+ * character representations spanning multiple bytes. However, if you know that your input will only ever contain
+ * 7-bit ASCII characters (0x00-0x7F) then UTF-8 is fine, since the first 127 UTF-8 characters are
+ * encoded with only one byte that is identical to 7-bit ASCII and ISO-8859-1.
+ */
+final class ByteStringParserInput(bytes: ByteString) extends DefaultParserInput {
+  override def charAt(ix: Int): Char = (bytes(ix) & 0xFF).toChar
+  override def length: Int = bytes.size
+  override def sliceString(start: Int, end: Int): String = bytes.slice(start, end - start).decodeString(StandardCharsets.ISO_8859_1)
+  override def sliceCharArray(start: Int, end: Int): Array[Char] =
+    StandardCharsets.ISO_8859_1.decode(bytes.slice(start, end).asByteBuffer).array()
+}

akka-http-core/src/test/scala/akka/http/impl/util/ByteStringParserInputSpec.scala

@@ -0,0 +1,33 @@
+package akka.http.impl.util
+
+import akka.util.ByteString
+import org.scalatest.{ Matchers, WordSpec }
+
+class ByteStringParserInputSpec extends WordSpec with Matchers {
+
+  "The ByteStringParserInput" should {
+    val parser = new ByteStringParserInput(ByteString("abcde", "ISO-8859-1"))
+    "return the correct character for index" in {
+      parser.charAt(0) should ===('a')
+      parser.charAt(4) should ===('e')
+    }
+
+    "return the correct length" in {
+      parser.length should ===(5)
+    }
+
+    "slice the bytes correctly into a string" in {
+      parser.sliceString(0, 3) should ===("abc")
+    }
+
+    "slice the bytes correctly into a char array" in {
+      val array = parser.sliceCharArray(0, 3)
+      array(0) should ===('a')
+      array(1) should ===('b')
+      array(2) should ===('c')
+      array.length should ===(3)
+    }
+
+  }
+
+}