This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-http.git
The following commit(s) were added to refs/heads/main by this push:
new f98599752 perf: find SSE line terminators with ByteString.indexOf
(#1226)
f98599752 is described below
commit f9859975230740340d3f23f4a7e7b0b6f0209375
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 1 11:21:56 2026 +0100
perf: find SSE line terminators with ByteString.indexOf (#1226)
Motivation:
LineParser scanned its buffer one byte at a time with `bs(at)`. The buffer
is
built as `buffer ++ grab(in)`, so it is a multi-fragment ByteString
whenever a
line spans several chunks, and `ByteStrings.apply` walks the fragment list
from
the first fragment on every access. Parsing an SSE line that arrives in many
chunks therefore costs O(bytes * fragments).
Modification:
Locate the next CR or LF with ByteString.indexOf, which is fragment aware
and
scans several bytes at a time, instead of testing every byte. The three
identical line-emitting blocks are factored into a local helper. Line
termination semantics (CR, LF, CRLF, and a CRLF split across chunks) are
unchanged.
Result:
Large SSE lines delivered in small chunks parse dramatically faster; small
lines are unaffected.
lineSize chunkSize before after
1 KB 512 0.265 ms 0.189 ms
1 KB 8192 0.295 ms 0.183 ms
128 KB 512 71.210 ms 1.013 ms
128 KB 8192 3.269 ms 0.342 ms
1 MB 512 9821.281 ms 54.361 ms
Tests:
- sbt "http-tests / Test / testOnly
org.apache.pekko.http.scaladsl.unmarshalling.sse.*" - 48 passed (46 existing
plus 2 new)
- sbt "http-bench-jmh/Jmh/run -f 1 -wi 3 -i 3 -p lineSize=... -p
chunkSize=... .*LineParserBenchmark.*" - numbers above, JDK 21
- scalafmt --mode diff-ref=upstream/main - clean
References:
None - found while auditing ByteString usage across the code base
---
.../unmarshalling/sse/LineParserSpec.scala | 18 +++++
.../scaladsl/unmarshalling/sse/LineParser.scala | 78 ++++++++++------------
2 files changed, 54 insertions(+), 42 deletions(-)
diff --git
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/LineParserSpec.scala
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/LineParserSpec.scala
index 9c83956a9..dd7373232 100644
---
a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/LineParserSpec.scala
+++
b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/LineParserSpec.scala
@@ -63,5 +63,23 @@ final class LineParserSpec extends AsyncWordSpec with
Matchers with BaseUnmarsha
.runWith(Sink.seq)
.map(_ shouldBe Vector("after", "", "before", "middle"))
}
+
+ "parse a CRLF that is split across two chunks" in {
+ Source(("line1\r" :: "\nline2\r\n" :: Nil).map(ByteString(_)))
+ .via(new LineParser(1048576))
+ .runWith(Sink.seq)
+ .map(_ shouldBe Vector("line1", "line2"))
+ }
+
+ "parse lines from a multi-fragment ByteString" in {
+ // a single element that is a concatenation of several chunks, as
produced by the stage's own buffering
+ val input = ByteString("line1\nli") ++ ByteString("ne2\r") ++
ByteString("\nline3\r") ++ ByteString("line4\n")
+ input.isCompact shouldBe false
+ Source
+ .single(input)
+ .via(new LineParser(1048576))
+ .runWith(Sink.seq)
+ .map(_ shouldBe Vector("line1", "line2", "line3", "line4"))
+ }
}
}
diff --git
a/http/src/main/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/LineParser.scala
b/http/src/main/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/LineParser.scala
index e339c0ebc..c29c67250 100644
---
a/http/src/main/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/LineParser.scala
+++
b/http/src/main/scala/org/apache/pekko/http/scaladsl/unmarshalling/sse/LineParser.scala
@@ -73,6 +73,13 @@ private final class LineParser(maxLineSize: Int,
}
}
+ def lineAt(bs: ByteString, from: Int, until: Int): Option[String] = {
+ val lineByteSize = until - from
+ val line = bs.slice(from, until).utf8String
+ if (maxLineSize > 0 && lineByteSize > maxLineSize)
handleLineOversized(lineByteSize, line)
+ else Some(line)
+ }
+
@tailrec
def parseLines(
bs: ByteString,
@@ -82,49 +89,36 @@ private final class LineParser(maxLineSize: Int,
lastCharWasCr: Boolean): (ByteString, Vector[String], Boolean) =
if (at >= bs.length)
(bs.drop(from), parsedLines, lastCharWasCr)
- else
- bs(at) match {
- case CR_BYTE if at < bs.length - 1 && bs(at + 1) == LF_BYTE =>
- // Lookahead for LF after CR
- val lineByteSize = at - from
- val line = bs.slice(from, at).utf8String
- val processedLine = if (maxLineSize > 0 && lineByteSize >
maxLineSize) {
- handleLineOversized(lineByteSize, line)
- } else {
- Some(line)
- }
- val newParsedLines =
processedLine.fold(parsedLines)(parsedLines :+ _)
- parseLines(bs, at + 2, at + 2, newParsedLines, lastCharWasCr =
false)
- case CR_BYTE =>
- // if is a CR but we don't know the next character, slice it
but flag that the last character was a CR so if the next happens to be a LF we
just ignore
- val lineByteSize = at - from
- val line = bs.slice(from, at).utf8String
- val processedLine = if (maxLineSize > 0 && lineByteSize >
maxLineSize) {
- handleLineOversized(lineByteSize, line)
- } else {
- Some(line)
- }
- val newParsedLines =
processedLine.fold(parsedLines)(parsedLines :+ _)
- parseLines(bs, at + 1, at + 1, newParsedLines, lastCharWasCr =
true)
- case LF_BYTE if lastCharWasCr =>
- // if is a LF and we just sliced a CR then we simply advance
- parseLines(bs, at + 1, at + 1, parsedLines, lastCharWasCr =
false)
- case LF_BYTE =>
- // a LF that wasn't preceded by a CR means we found a new slice
- val lineByteSize = at - from
- val line = bs.slice(from, at).utf8String
- val processedLine = if (maxLineSize > 0 && lineByteSize >
maxLineSize) {
- handleLineOversized(lineByteSize, line)
- } else {
- Some(line)
- }
- val newParsedLines =
processedLine.fold(parsedLines)(parsedLines :+ _)
- parseLines(bs, at + 1, at + 1, newParsedLines, lastCharWasCr =
false)
- case _ =>
- // for other input, simply advance
- // Reset lastCharWasCr if we encounter any non-LF character
after CR
- parseLines(bs, from, at + 1, parsedLines, lastCharWasCr =
false)
+ else if (lastCharWasCr && bs(at) == LF_BYTE)
+ // the LF of a CRLF whose CR already ended a line in a previous
chunk, simply advance
+ parseLines(bs, at + 1, at + 1, parsedLines, lastCharWasCr = false)
+ else {
+ // jump straight to the next line terminator instead of testing
every single byte:
+ // ByteString.indexOf scans several bytes at a time and, unlike
indexed access, does not
+ // walk the fragment list of a multi-chunk ByteString on every byte
+ val crIx = bs.indexOf(CR_BYTE, at)
+ val lfIx = bs.indexOf(LF_BYTE, at)
+ val terminator =
+ if (crIx == -1) lfIx
+ else if (lfIx == -1) crIx
+ else math.min(crIx, lfIx)
+
+ if (terminator == -1)
+ // no line terminator in the rest of the buffer
+ (bs.drop(from), parsedLines, false)
+ else {
+ val newParsedLines = lineAt(bs, from,
terminator).fold(parsedLines)(parsedLines :+ _)
+ if (terminator == lfIx)
+ parseLines(bs, terminator + 1, terminator + 1, newParsedLines,
lastCharWasCr = false)
+ else if (terminator < bs.length - 1 && bs(terminator + 1) ==
LF_BYTE)
+ // lookahead for LF after CR
+ parseLines(bs, terminator + 2, terminator + 2, newParsedLines,
lastCharWasCr = false)
+ else
+ // a CR but we don't know the next character yet, flag it so
that a LF starting the
+ // next chunk is ignored
+ parseLines(bs, terminator + 1, terminator + 1, newParsedLines,
lastCharWasCr = true)
}
+ }
// start the search where it ended, prevent iterating over all the
buffer again
val currentBufferStart = math.max(0, buffer.length - 1)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
