jerrypeng commented on code in PR #57286:
URL: https://github.com/apache/spark/pull/57286#discussion_r3604941408
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -76,15 +77,105 @@ 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. It defaults to the built-in
+ // StreamingShuffleManager -- just as the blocking manager defaults to sort
-- and is 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. Like the
+ // blocking manager, it is initialized later in SparkContext and Executor to
allow user jars.
+ @volatile private var _pipelinedShuffleManager: PipelinedShuffleManager = _
// 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
Review Comment:
Yup you are right. These should be private[spark], matching the description
--
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]
