cloud-fan commented on code in PR #57286:
URL: https://github.com/apache/spark/pull/57286#discussion_r3597592245
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -421,8 +417,12 @@ class BlockManagerMasterEndpoint(
mapStatuses.filter(_ != null).foreach { mapStatus =>
// Check if the executor has been deallocated
if
(!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
- val blocksToDel =
-
shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId,
mapStatus.mapId)
+ // Only a BlockingShuffleManager serves block-manager blocks; a
pipelined shuffle is
+ // served out-of-band and is not in the MapOutputTracker this
loop iterates, so this
+ // resolves only regular shuffles (None when the default manager
is not blocking).
+ val blocksToDel = SparkEnv.get.shuffleBlockResolver
Review Comment:
This answers your own question at the `_shuffleManager` param above: with
this call the param is now unused (its only consumer, the old `shuffleManager`
lazy val, is gone). It can be removed from the constructor — that also touches
`SparkEnv.scala:656` (`_shuffleManager = null`) and the two
`BlockManagerReplicationSuite` call sites. Non-blocking.
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -80,12 +80,88 @@ class SparkEnv (
// user jars to define custom ShuffleManagers.
@volatile private var _shuffleManager: ShuffleManager = _
+ // The manager for pipelined (incrementally-readable) shuffle dependencies,
configured by
+ // spark.shuffle.manager.incremental. Present only when that config is set.
It is a peer of
+ // _shuffleManager, not a wrapper around it: a shuffle is routed to one or
the other by its
+ // dependency type via `shuffleManagerFor`, so neither is ever installed
"behind" the other.
+ @volatile private var _incrementalShuffleManager: Option[ShuffleManager] =
None
+
// Latch to signal when the ShuffleManager has been initialized.
// Used to allow callers to wait for initialization.
private val shuffleManagerInitLatch = new CountDownLatch(1)
+ /**
+ * The default [[ShuffleManager]] (configured by spark.shuffle.manager),
which serves all regular
+ * shuffle dependencies. Use this only when the intent is specifically the
default manager -- e.g.
+ * inspecting its concrete type, or resolving a shuffle block by id (a
pipelined shuffle is served
+ * out-of-band by the incremental manager and produces no block-manager
blocks). To serve a
+ * specific shuffle's reads/writes, use `shuffleManagerFor`, which routes by
dependency type.
+ */
+ def defaultShuffleManager: ShuffleManager = _shuffleManager
Review Comment:
There isn't really a "default" manager to name here: block-resolution paths
always want the blocking manager, pipelined reads always want the pipelined
manager, and routing picks by dependency type. Modeling one slot as the default
(and sourcing the resolver from it at `shuffleBlockResolver`, line 112) hides
exactly the blocking-vs-pipelined distinction the design relies on. Suggest
holding both a blocking and a pipelined manager instance in `SparkEnv` and
having each call site pick explicitly — always-blocking, always-pipelined, or
by-dependency-type. That also removes the sharp edge where
`spark.shuffle.manager.incremental` is set to a blocking manager (the routing
suite sets it to `"sort"`) whose blocks are then unreachable through this
accessor.
##########
core/src/main/scala/org/apache/spark/shuffle/ShuffleManager.scala:
##########
@@ -28,14 +28,19 @@ import org.apache.spark.util.Utils
* and on each executor, based on the spark.shuffle.manager setting. The
driver registers shuffles
* with it, and executors (or tasks running locally in the driver) can ask to
read and write data.
*
+ * This trait is sealed and is not meant to be implemented directly. A shuffle
manager declares its
+ * kind by extending one of its two subtypes: [[BlockingShuffleManager]]
(output is materialized as
+ * block-manager-addressed blocks served through a [[ShuffleBlockResolver]]) or
+ * [[PipelinedShuffleManager]] (output is read incrementally and served
out-of-band).
+ *
* NOTE:
* 1. This will be instantiated by SparkEnv so its constructor can take a
SparkConf and
* boolean isDriver as parameters.
- * 2. This contains a method ShuffleBlockResolver which interacts with
External Shuffle Service
- * when it is enabled. Need to pay attention to that, if implementing a custom
ShuffleManager, to
- * make sure the custom ShuffleManager could co-exist with External Shuffle
Service.
+ * 2. A [[BlockingShuffleManager]] exposes a ShuffleBlockResolver which
interacts with the External
+ * Shuffle Service when it is enabled. Need to pay attention to that, if
implementing a custom
+ * shuffle manager, to make sure it could co-exist with the External Shuffle
Service.
*/
-private[spark] trait ShuffleManager {
+private[spark] sealed trait ShuffleManager {
Review Comment:
Please don't `seal` this. Breaking changes to the interface are fine —
moving `shuffleBlockResolver` onto `BlockingShuffleManager` is a good change —
but `sealed` fully stops third-party `ShuffleManager` implementations, which
SPARK-45762 deliberately enabled. We should keep that extension point.
(Consequence: the SPARK-45762 test in `SparkSubmitSuite` should be updated to
implement `BlockingShuffleManager` rather than removed — see review summary.)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
