otterc commented on a change in pull request #32140:
URL: https://github.com/apache/spark/pull/32140#discussion_r648847338
##########
File path:
core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
##########
@@ -767,6 +908,43 @@ final class ShuffleBlockFetcherIterator(
deferredFetchRequests.getOrElseUpdate(address, new
Queue[FetchRequest]())
defReqQueue.enqueue(request)
result = null
+
+ case IgnoreFetchResult(blockId, address, size, isNetworkReqDone) =>
+ if (pushBasedFetchHelper.isNotExecutorOrMergedLocal(address)) {
+ numBlocksInFlightPerAddress(address) =
numBlocksInFlightPerAddress(address) - 1
+ bytesInFlight -= size
+ }
+ if (isNetworkReqDone) {
+ reqsInFlight -= 1
+ logDebug("Number of requests in flight " + reqsInFlight)
+ }
+ numBlocksProcessed +=
pushBasedFetchHelper.initiateFallbackBlockFetchForMergedBlock(
+ blockId, address)
+ // Set result to null to trigger another iteration of the while loop
to get either
+ // a SuccessFetchResult or a FailureFetchResult.
+ result = null
+
+ case MergedBlocksMetaFetchResult(shuffleId, reduceId, blockSize,
numChunks, bitmaps,
+ address, _) =>
+ // The original meta request is processed so we decrease
numBlocksToFetch by 1. We will
+ // collect new chunks request and the count of this is added to
numBlocksToFetch in
+ // collectFetchReqsFromMergedBlocks.
+ numBlocksToFetch -= 1
+ val blocksToRequest =
pushBasedFetchHelper.createChunkBlockInfosFromMetaResponse(
+ shuffleId, reduceId, blockSize, numChunks, bitmaps)
+ val additionalRemoteReqs = new ArrayBuffer[FetchRequest]
+ collectFetchRequests(address, blocksToRequest.toSeq,
additionalRemoteReqs)
+ fetchRequests ++= additionalRemoteReqs
+ // Set result to null to force another iteration.
+ result = null
Review comment:
Actually, this is the existing code which I haven't modified. The while
loop inside iterator.next() is as below, so `fetchUpToMaxBytes` is always
called after a response is matched and processed.
```
while (result == null) {
val startFetchWait = System.nanoTime()
result = results.take()
val fetchWaitTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startFetchWait)
shuffleMetrics.incFetchWaitTime(fetchWaitTime)
result match {...}
// Send fetch requests up to maxBytesInFlight
fetchUpToMaxBytes()
}
```
--
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]
