jerrypeng commented on code in PR #57286:
URL: https://github.com/apache/spark/pull/57286#discussion_r3600414869
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -76,15 +77,104 @@ class SparkEnv (
val outputCommitCoordinator: OutputCommitCoordinator,
val conf: SparkConf) extends Logging {
- // We initialize the ShuffleManager later in SparkContext and Executor to
allow
- // user jars to define custom ShuffleManagers.
- @volatile private var _shuffleManager: ShuffleManager = _
+ // The two shuffle managers are peers keyed by kind, not a default and an
override: a shuffle is
+ // routed to one or the other by its dependency type via
`shuffleManagerFor`, so neither is ever
+ // installed "behind" the other.
+ //
+ // The blocking manager (spark.shuffle.manager) serves all regular,
materialized shuffles and owns
+ // block-by-id resolution. It is always a BlockingShuffleManager --
initializeShuffleManager
+ // rejects a non-blocking manager in this slot -- so block-resolution has a
single, well-typed
+ // source. We initialize it later in SparkContext and Executor to allow user
jars to define custom
+ // ShuffleManagers.
+ @volatile private var _blockingShuffleManager: BlockingShuffleManager = _
+
+ // The pipelined manager (spark.shuffle.manager.incremental) serves pipelined
+ // (incrementally-readable) shuffle dependencies. Present only when that
config is set, and always
+ // a PipelinedShuffleManager -- initializeShuffleManager rejects a blocking
manager in this slot,
+ // so its output is never expected to be reachable through the block-manager
resolver.
+ @volatile private var _pipelinedShuffleManager:
Option[PipelinedShuffleManager] = None
// Latch to signal when the ShuffleManager has been initialized.
// Used to allow callers to wait for initialization.
private val shuffleManagerInitLatch = new CountDownLatch(1)
- def shuffleManager: ShuffleManager = _shuffleManager
+ /**
+ * The [[BlockingShuffleManager]] (configured by spark.shuffle.manager),
which serves all regular,
+ * materialized shuffle dependencies and owns block-by-id resolution. Use
this when the intent is
+ * specifically the blocking manager -- e.g. inspecting its concrete type.
To serve a specific
+ * shuffle's reads/writes, use `shuffleManagerFor`, which routes by
dependency type; to resolve a
+ * block by id, use `shuffleBlockResolver`.
+ */
+ def blockingShuffleManager: BlockingShuffleManager = _blockingShuffleManager
+
+ /**
+ * The [[PipelinedShuffleManager]] (configured by
spark.shuffle.manager.incremental) that serves
+ * pipelined shuffle dependencies, or `None` when none is configured. A
pipelined shuffle is read
+ * incrementally and served out-of-band, so this manager never provides a
[[ShuffleBlockResolver]]
+ * of its own.
+ */
+ def pipelinedShuffleManager: Option[PipelinedShuffleManager] =
_pipelinedShuffleManager
Review Comment:
This shouldn't be an optional. We have a default implementation of the
PipelinedShuffleManager, i.e. streaming shuffle manager, we should just by
default that. Follow the same logic as blockingShuffleManager.
--
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]
