dongjoon-hyun commented on a change in pull request #32287:
URL: https://github.com/apache/spark/pull/32287#discussion_r628805054
##########
File path:
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
##########
@@ -258,17 +281,57 @@ final class ShuffleBlockFetcherIterator(
// This needs to be released after use.
buf.retain()
remainingBlocks -= blockId
+ blockOOMRetryTimes.remove(blockId)
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))
+ ShuffleBlockFetcherIterator.this.synchronized {
+ logError(s"Failed to get block(s) from
${req.address.host}:${req.address.port}", e)
+ e match {
+ // SPARK-27991: Catch the Netty OOM and set the flag
`isNettyOOMOnShuffle` (shared among
+ // tasks) to true as early as possible. The pending fetch requests
won't be sent
+ // afterwards until the flag is set to false on:
+ // 1) the Netty free memory >= maxReqSizeShuffleToMem
+ // - 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.
+ case _: OutOfDirectMemoryError
+ if blockOOMRetryTimes.getOrElseUpdate(blockId, 0) <
maxAttemptsOnNettyOOM =>
+ if (!isZombie) {
+ val failureTimes = blockOOMRetryTimes(blockId)
+ blockOOMRetryTimes(blockId) += 1
+ if (isNettyOOMOnShuffle.compareAndSet(false, true)) {
+ // The fetcher can fail remaining blocks in batch for the
same error. So we only
+ // log the warning once to avoid flooding the logs.
+ logInfo(s"Block $blockId has failed $failureTimes times " +
+ s"due to Netty OOM, will retry")
+ }
+ remainingBlocks -= blockId
+ deferredBlocks += blockId
+ enqueueDeferredFetchRequestIfNecessary()
+ }
Review comment:
Shall we add a debug message with `else` statement because this code
path is changed after this PR? Previously, we do `results.put` always.
--
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:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
