Ngone51 commented on a change in pull request #32287:
URL: https://github.com/apache/spark/pull/32287#discussion_r623652602
##########
File path:
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
##########
@@ -261,14 +283,54 @@ final class ShuffleBlockFetcherIterator(
results.put(new SuccessFetchResult(BlockId(blockId),
infoMap(blockId)._2,
address, infoMap(blockId)._1, buf, remainingBlocks.isEmpty))
logDebug("remainingBlocks: " + remainingBlocks)
+ enqueueDeferredFetchRequestIfNecessary()
}
}
logTrace(s"Got remote block $blockId after
${Utils.getUsedTimeNs(startTimeNs)}")
}
override def onBlockFetchFailure(blockId: String, e: Throwable): Unit = {
logError(s"Failed to get block(s) from
${req.address.host}:${req.address.port}", e)
- results.put(new FailureFetchResult(BlockId(blockId),
infoMap(blockId)._2, address, e))
+ val (size, mapIndex) = infoMap(blockId)
+ e match {
+ // SPARK-27991: Catch the Netty OOM and set the flag
`isNettyOOMOnShuffle` (shared among
+ // tasks) to true as early as possible. Unless there's no in-flight
requests, the pending
+ // fetch requests won't be raised afterwards until the flag is set
to false on:
+ // 1) the Netty free memory >= average remote block size - we'll
check this whenever
+ // there's a fetch request succeeds.
+ // 2) the number of in-flight requests becomes 0 - we'll check this
in `fetchUpToMaxBytes`
+ // whenever it's invoked.
+ // Although Netty memory is shared across multiple modules, e.g.,
shuffle, rpc, the flag
+ // only takes effect for the shuffle due to the implementation
simplicity concern.
+ // And we'll buffer the consecutive block failures caused by the OOM
error until there's
+ // no remaining blocks in the current request. Then, we'll package
these blocks into
+ // a same fetch request for the retry later. In this way, instead of
creating the fetch
+ // request per block, it would help reduce the concurrent
connections and data loads
+ // pressure at remote server.
+ // Note that catching OOM and do something based on it is only a
workaround for
+ // handling the Netty OOM issue, which is not the best way towards
memory management.
+ // We can get rid of it when we find a way to manage Netty's memory
precisely.
+
+ // Ensure the Netty memory is at least enough for serving only one
block to avoid
+ // the endless retry. And since the Netty memory is shared among
multiple modules,
+ // we use the factor "1.5" for the overhead concern.
+ case _: OutOfDirectMemoryError if
PlatformDependent.maxDirectMemory() > ( 1.5 * size) =>
Review comment:
Make sense. I think we should fail fast in this case during the
initialization of `ShuffleBlockFetcherIterator`.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
