[BUG] netty OOM with MULTITHREADED shuffle

[abellina]

I ran NDS @100TB and ran into a netty OOM in query50. This needs further investigation. It is likely that either our limits are not working or there is a leak somewhere:

io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 20999453 byte(s) of direct memory (used: 15263094153, max: 15271460864)
  at io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:802)
  at io.netty.util.internal.PlatformDependent.allocateDirectNoCleaner(PlatformDependent.java:731)
  at io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:632)
  at io.netty.buffer.PoolArena$DirectArena.newUnpooledChunk(PoolArena.java:621)
  at io.netty.buffer.PoolArena.allocateHuge(PoolArena.java:213)
  at io.netty.buffer.PoolArena.allocate(PoolArena.java:141)
  at io.netty.buffer.PoolArena.allocate(PoolArena.java:126)
  at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:395)
  at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
  at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
  at io.netty.buffer.CompositeByteBuf.allocBuffer(CompositeByteBuf.java:1872)
  at io.netty.buffer.CompositeByteBuf.consolidate0(CompositeByteBuf.java:1751)
  at io.netty.buffer.CompositeByteBuf.consolidate(CompositeByteBuf.java:1739)
  at org.apache.spark.network.util.TransportFrameDecoder.decodeNext(TransportFrameDecoder.java:179)
  at org.apache.spark.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:98)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
  at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
  at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)

[abellina]

Note that I also see this with Parquet when the number of cores is high. As a test, I changed spark.executor.cores to 64 (increased from 16). I am not recommending this config yet, but it is definitely pushing things in terms of host memory that is available for shuffle it seems. With this many cores, the shuffle stage would have more tasks in flight, so something is definitely amiss with the limits we have in place.

[abellina]

By setting spark.rapids.shuffle.multiThreaded.maxBytesInFlight I was able to fix this. I calculated a number to stay within what netty is computing as its max size (https://github.com/netty/netty/blob/4.1/common/src/main/java/io/netty/util/internal/PlatformDependent.java#L1231C51-L1231C70). Netty allows passing a java system property to override its maximum (-Dio.netty.maxDirectMemory), but by default it is defaulting to the maximum that the JVM reports is available for off heap.

Normally this should default to Xmx according to what I read in a similar issue in Spark https://issues.apache.org/jira/browse/SPARK-27991. In this case, the value is lower than Xmx, as I started with a 16GB heap, so it's not clear why that would be the case. The Spark strategy to deal with this issue, which we also follow, is to perform retries when something like this happens.

We may need to decrease our default maxBytesInFlight or configure this dynamically to make it easier to use.

[revans2]

Great work @abellina we really need to figure out how to document and figure out how this fits in with our off heap memory limit story.

[abellina]

I added the reliability tag because we would like to document the behavior of the maxBytesInFlight setting relative to performance in addition to coming up with better defaults and also likely providing a way to tie this in with the host allocator interface.

[abellina]

Given the runs I have made I believe setting this config to somewhere between 64MB-128MB is going to be a good enough default. I am running more iterations, but that's the current status.

[abellina]

Here are findings at 3TB running in the cloud (dataproc) with 4 x n1-standard-16 nodes and using 16 threads for the MT shuffle. I see a lot of noise in here, but I do have another datapoint in our performance cluster for query67 specifically (this query is particularly sensitive to MT shuffle). Given the dataproc and perf cluster data, I am going to set the config to 128MB, and we'll need to continue this work as part of the host memory framework #8900