teads-lionel-ngo commented on issue #1133:
URL: https://github.com/apache/pekko-http/issues/1133#issuecomment-4892910948
@pjfanning I asked claude to implement a custom hacky `decodeRequest` with a
pool and explicit closure of the inflater. Code is AI generated so I wouldn't
blindly trust it.
```
object PooledDecodeRequest {
private val MaxPoolSize = 128
private val pooledDecoders =
PooledInflaterDecompressors.decoders(MaxPoolSize)
val decodeRequest: Directive0 = decodeRequestWith(pooledDecoders*)
}
final class InflaterPool(nowrap: Boolean, maxPoolSize: Int) {
private[this] val pool = new ArrayBlockingQueue[Inflater](maxPoolSize)
def borrow(): Inflater =
Option(pool.poll()).getOrElse(new Inflater(nowrap))
def release(inflater: Inflater): Unit = {
inflater.reset()
if (!pool.offer(inflater)) inflater.end()
}
}
/** Drop-in replacement for `GzipDecompressor` that borrows/returns its
`Inflater` from `pool`
* (which must be built with `nowrap = true`, as required by the gzip
framing).
*/
// Ported from Pekko HTTP's internal GzipDecompressor; keep the wire-parsing
logic verbatim, which
// uses `==`/`head`/etc. that WartRemover would otherwise reject.
@SuppressWarnings(Array("org.wartremover.warts.All"))
private[coding] final class PooledGzipDecompressor(pool: InflaterPool,
maxBytesPerChunk: Int)
extends DeflateDecompressorBase(maxBytesPerChunk) {
override def createLogic(attr: Attributes) = new ParsingLogic {
private[this] val inflater = pool.borrow()
private[this] val crc32: CRC32 = new CRC32
override def postStop(): Unit = {
pool.release(inflater)
super.postStop()
}
trait Step extends ParseStep[ByteString] {
override def onTruncation(): Unit = failStage(new
ZipException("Truncated GZIP stream"))
}
startWith(ReadHeaders)
/** Reading the header bytes */
case object ReadHeaders extends Step {
override def parse(reader: ByteStringParser.ByteReader):
ParseResult[ByteString] = {
import reader._
if (readByte() != 0x1f || readByte() != 0x8b)
fail("Not in GZIP format") //
check magic header
if (readByte() != 8) fail("Unsupported GZIP compression method") //
check compression method
val flags = readByte()
skip(6) // skip MTIME, XFL
and OS fields
if ((flags & 4) > 0) skip(readShortLE()) // skip optional
extra fields
if ((flags & 8) > 0) skipZeroTerminatedString() // skip optional
file name
if ((flags & 16) > 0) skipZeroTerminatedString() // skip optional
file comment
if ((flags & 2) > 0 && crc16(fromStartToHere) != readShortLE())
fail("Corrupt GZIP header")
inflater.reset()
crc32.reset()
ParseResult(None, GzipDeflate, acceptUpstreamFinish = false)
}
}
case object GzipDeflate
extends Inflate(inflater, noPostProcessing = false, ReadTrailer)
with Step {
protected override def afterBytesRead(buffer: Array[Byte], offset:
Int, length: Int): Unit =
crc32.update(buffer, offset, length)
}
/** Reading the trailer */
case object ReadTrailer extends Step {
override def parse(reader: ByteStringParser.ByteReader):
ParseResult[ByteString] = {
import reader._
if (readIntLE() != crc32.getValue.toInt) fail("Corrupt data (CRC32
checksum error)")
if (readIntLE() != inflater.getBytesWritten.toInt /* truncated to
32bit */ )
fail("Corrupt GZIP trailer ISIZE")
ParseResult(None, ReadHeaders, acceptUpstreamFinish = true)
}
}
private def fail(msg: String) = throw new ZipException(msg)
private def crc16(data: ByteString): Int = {
val crc = new CRC32
crc.update(data.toArrayUnsafe())
crc.getValue.toInt & 0xffff
}
}
}
/** Drop-in replacement for `DeflateDecompressor` that borrows/returns its
`Inflater` from a pool.
*
* The deflate stream can be either zlib-wrapped or raw, which requires a
different `nowrap` mode.
* We probe the first byte (exactly as the stock decompressor does) and
then borrow from the pool
* matching the detected framing.
*/
@SuppressWarnings(Array("org.wartremover.warts.All"))
private[coding] final class PooledDeflateDecompressor(
unwrappedPool: InflaterPool, // nowrap = true (raw deflate)
wrappedPool: InflaterPool, // nowrap = false (zlib-wrapped)
maxBytesPerChunk: Int
) extends DeflateDecompressorBase(maxBytesPerChunk) {
override def createLogic(attr: Attributes) = new ParsingLogic {
// Set once the framing has been probed; released in postStop.
private[this] var borrowed: Inflater = null
private[this] var borrowedFrom: InflaterPool = null
override def postStop(): Unit = {
if (borrowed ne null) borrowedFrom.release(borrowed)
super.postStop()
}
/** Step that probes if the deflate stream contains a zlib wrapper, then
builds an inflater. */
case object ProbeWrapping extends ParseStep[ByteString] {
override def onTruncation(): Unit = completeStage()
override def parse(reader: ByteStringParser.ByteReader):
ParseResult[ByteString] = {
// A concatenated stream loops back here; return the previous
inflater before taking another
// so the pool stays bounded.
if (borrowed ne null) borrowedFrom.release(borrowed)
// Wrapped deflate streams start with the low 4 bits of the first
byte == 0x8 (see rfc1950 /
// rfc1951); otherwise fall back to the raw (no-wrap) inflater.
val wrapped = (reader.remainingData.head & 0x0f) == 0x08
borrowedFrom = if (wrapped) wrappedPool else unwrappedPool
borrowed = borrowedFrom.borrow()
ParseResult(None, new Inflate(borrowed, noPostProcessing = true,
ProbeWrapping))
}
}
startWith(ProbeWrapping)
}
}
/** A gzip request decoder backed by a shared `Inflater` pool. */
final class PooledGzipDecoder private[coding] (pool: InflaterPool, val
maxBytesPerChunk: Int)
extends Decoder {
def encoding: HttpEncoding =
HttpEncodings.gzip
def withMaxBytesPerChunk(newMaxBytesPerChunk: Int): Decoder =
new PooledGzipDecoder(pool, newMaxBytesPerChunk)
def decoderFlow: Flow[ByteString, ByteString, NotUsed] =
Flow.fromGraph(new PooledGzipDecompressor(pool, maxBytesPerChunk))
}
/** A deflate request decoder backed by shared `Inflater` pools (one per
framing mode). */
final class PooledDeflateDecoder private[coding] (
unwrappedPool: InflaterPool,
wrappedPool: InflaterPool,
val maxBytesPerChunk: Int
) extends Decoder {
def encoding: HttpEncoding =
HttpEncodings.deflate
def withMaxBytesPerChunk(newMaxBytesPerChunk: Int): Decoder =
new PooledDeflateDecoder(unwrappedPool, wrappedPool, newMaxBytesPerChunk)
def decoderFlow: Flow[ByteString, ByteString, NotUsed] =
Flow.fromGraph(new PooledDeflateDecompressor(unwrappedPool, wrappedPool,
maxBytesPerChunk))
}
object PooledInflaterDecompressors {
/** Builds the decoder list to hand to `decodeRequestWith`, backed by
fresh pools of the given
* capacity. Call this once and reuse the result: the returned decoders
share long-lived pools,
* so recreating them per request would defeat the pooling.
*
* `maxPoolSize` bounds the number of live native `Inflater`s per framing
mode. Beyond that,
* surplus inflaters are `end()`ed on release rather than pooled,
degrading gracefully to the
* stock allocate-per-request behaviour (but still without the leak).
*/
def decoders(maxPoolSize: Int): immutable.Seq[Decoder] = {
val gzipPool = new InflaterPool(nowrap = true, maxPoolSize) // gzip
always uses no-wrap framing
val deflateUnwrapped = new InflaterPool(nowrap = true, maxPoolSize)
val deflateWrapped = new InflaterPool(nowrap = false, maxPoolSize)
immutable.Seq(
new PooledGzipDecoder(gzipPool, Decoder.MaxBytesPerChunkDefault),
new PooledDeflateDecoder(deflateUnwrapped, deflateWrapped,
Decoder.MaxBytesPerChunkDefault),
Coders.NoCoding
)
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
