Ngone51 commented on a change in pull request #32287: URL: https://github.com/apache/spark/pull/32287#discussion_r623530116
########## 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: > What if we simply remove this check? Will we hang? Theoretically, it will hang when the netty max memory is not even enough to serve a single block. -- 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