otterc commented on a change in pull request #32140:
URL: https://github.com/apache/spark/pull/32140#discussion_r648520952
##########
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:
> Hm..is it possible there's only FetchRequest(hasMergedBlocks) at the
beginning? In that case, it seems to cause the fetching process to hang.
It will not cause the fetch process to hang if there is just one
FetchRequest with merged blocks.
Consider this example that if there is a FetchRequest for a merged block
`ShuffleBlock(0, -1, 0)`,
- the iterator will send out the request to fetch the metadata for this
block in `PushBasedFetchHelper.sendFetchMergedStatusRequest`.
- The iterator will wait for a response in the result queue at
`results.take()`.
- Once it receives a response, which is either `MergedBlocksMetaFetchResult`
or `MergedBlocksMetaFailedFetchResult`, it adds more FetchRequests to the fetch
queue and sets `result = null`.
- `fetchUpToMaxBytes()` is always called after processing the response.
- Since `result = null`, while loop repeats and waits again for a response
in the result queue.
I will also add a UT for this case just to verify this.
--
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
