attilapiros commented on code in PR #50439:
URL: https://github.com/apache/spark/pull/50439#discussion_r2042748146
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -863,36 +863,36 @@ class BlockManagerMasterEndpoint(
blockId: BlockId,
requesterHost: String): Option[BlockLocationsAndStatus] = {
val allLocations =
Option(blockLocations.get(blockId)).map(_.toSeq).getOrElse(Seq.empty)
- val hostLocalLocations = allLocations.filter(bmId => bmId.host ==
requesterHost)
-
val blockStatusWithBlockManagerId: Option[(BlockStatus, BlockManagerId)] =
(if (externalShuffleServiceRddFetchEnabled) {
- // if fetching RDD is enabled from the external shuffle service then
first try to find
- // the block in the external shuffle service of the same host
- val location = hostLocalLocations.find(_.port ==
externalShuffleServicePort)
- location
- .flatMap(blockStatusByShuffleService.get(_).flatMap(_.get(blockId)))
- .zip(location)
- } else {
- None
- })
- .orElse {
- // if the block is not found via the external shuffle service trying
to find it in the
- // executors running on the same host and persisted on the disk
- // using flatMap on iterators makes the transformation lazy
- hostLocalLocations.iterator
- .flatMap { bmId =>
- blockManagerInfo.get(bmId).flatMap { blockInfo =>
- blockInfo.getStatus(blockId).map((_, bmId))
- }
- }
- .find(_._1.storageLevel.useDisk)
- }
- .orElse {
- // if the block cannot be found in the same host search it in all
the executors
- val location = allLocations.headOption
-
location.flatMap(blockManagerInfo.get(_)).flatMap(_.getStatus(blockId)).zip(location)
+ // If fetching disk persisted RDD from the external shuffle service is
enabled then first
+ // try to find the block in the external shuffle service preferring
the one running on
+ // the same host. This search includes blocks stored on already killed
executors as well.
+ val hostLocalLocations = allLocations.find { bmId =>
+ bmId.host == requesterHost && bmId.port == externalShuffleServicePort
}
+ val location = hostLocalLocations
+ .orElse(allLocations.find(_.port == externalShuffleServicePort))
+ location
+ .flatMap(blockStatusByShuffleService.get(_).flatMap(_.get(blockId)))
+ .zip(location)
+ } else {
Review Comment:
That branch is/was for searching the disk persisted blocks on the executors.
For a disk persisted RDD we are safe to skip it but for a broadcast block
stored on the disk this is good idea.
So what about
```
(if (externalShuffleServiceRddFetchEnabled && blockId.isRDD) {
} else {
}).orElse {
}
```
--
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.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
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
