OOM issue with zgc

[teads-lionel-ngo]

Hello there,

I am considering switching the gc used in production to zgc for my app but it constantly crashed with a OOM error. I have been actively chasing the root cause of the OOM and I believe there may be some issues when using pekko http with some garbage collectors.
To make it short, ZGC doesn't reclaim off-heap memory because my app has a fairly large heap with very moderate pressure. My app uses decodeRequest & encodeResponse.
For example, decodeRequest uses a GZipCompressor and from my understanding for every incoming request, it creates a new Inflater. The issue for my use case is that there is no explicit call .end() to properly release the resource leaving it to the garbage collector to reclaim it. Same goes for encoreResponse.
One workaround I found is using the flag ZCollectionInterval to force the gc to release off-heap memory but at the expense of CPU. I also tried to do a custom hacky decoderRequest and it also worked.

So here's my question: is it an intended behaviour in pekko http to let the gc handle this?

[He-Pin]

I’ve also run into OOM issues when using ZGC in production; it requires a large memory footprint, and in some cases, we need to be more proactive about explicitly setting references to null. We encountered similar issues in our code previously, but since we started using the Alibaba JDK—which utilizes jemalloc—we haven't run into this problem again.

[He-Pin]

Another problem is that the default glibc is bad, and we should use jemalloc for the JVM. The best we can do on our side is to do some pooling I think, you are welcome to submit a PR.

[jrudolph]

Ultimately, it would be better to move to a Java-native gzip implementation. That would also allow negotiation for the websocket compression params.

[pjfanning]

@teads-lionel-ngo could you provide your workaround, the decodeRequest?

[teads-lionel-ngo]

@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
    )
  }
}

[He-Pin]

I think we can take a look at how netty handle this too

[teads-lionel-ngo]

@He-Pin Good point, during my research, I identified netty to allocate off heap memory but ruled it out because it uses some resources pooling.

[He-Pin]

We don't use Pekko HTTP at work; our gateway uses Netty, and we use zstd-jni for compression, utilizing a DirectByteBufAllocator. I feel there is plenty of room for optimization here.

[teads-lionel-ngo]

Netty probably pools the buffers: https://netty.io/4.0/api/io/netty/buffer/PooledByteBufAllocator.html

[pjfanning]

@teads-lionel-ngo would #1134 help in your case? Copilot suggested that GzipCompressor cleans up its inflater but that GzipDecompressor does not.

[teads-lionel-ngo]

@pjfanning Yes it would help a lot. And I believe there's the same issue with DeflateDecompressor: the inflater returned by examineAndBuildInflater is not ended either.