jerrypeng commented on code in PR #57286:
URL: https://github.com/apache/spark/pull/57286#discussion_r3600411105
##########
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
+
+ /**
+ * The [[ShuffleBlockResolver]] used to resolve shuffle blocks by id (reads,
push-merge, and
+ * decommission migration), or `None` before the shuffle manager is
initialized. Block resolution
+ * is only ever needed for regular, materialized shuffles, so it always
comes from the blocking
+ * manager (the only manager that produces block-manager-addressed blocks);
a pipelined shuffle is
+ * served out-of-band and never resolved here. Callers that only need to
resolve a block should
+ * use this rather than reaching through a [[ShuffleManager]].
+ */
+ def shuffleBlockResolver: Option[ShuffleBlockResolver] =
+ Option(_blockingShuffleManager).map(_.shuffleBlockResolver)
+
+ /**
+ * Retained for binary compatibility; returns the blocking manager. Prefer
`shuffleManagerFor`
+ * (route a shuffle by its dependency) or `blockingShuffleManager` (the
blocking manager
+ * explicitly), so the routing decision is explicit at the call site.
+ */
+ @deprecated("use shuffleManagerFor(dependency) to route a shuffle by type,
or " +
+ "blockingShuffleManager for the blocking manager explicitly", "4.3.0")
+ def shuffleManager: ShuffleManager = _blockingShuffleManager
+
+ /**
+ * The [[ShuffleManager]] that serves the given shuffle, chosen by
dependency type: a
+ * [[PipelinedShuffleDependency]] is served by the pipelined manager
+ * (spark.shuffle.manager.incremental) when one is configured, every other
[[ShuffleDependency]]
+ * by the blocking manager (spark.shuffle.manager). This is the single
routing point for shuffle
+ * I/O; the decision is a pure function of the dependency, so the driver (at
`registerShuffle`)
+ * and every executor (at `getWriter` / `getReader`) agree without any
shared routing state or
+ * config re-read -- the dependency (a serialized field, deterministic on
driver and executors)
+ * is always in hand at these sites, so no shuffleId -> manager tracking is
needed.
+ *
+ * When no pipelined manager is configured, a pipelined dependency falls
back to the blocking
+ * manager (served as an ordinary materialized shuffle) -- the same behavior
as before a pipelined
+ * manager is opted into. Whether a job may use a pipelined dependency at
all is a separate,
+ * scheduler-level decision (see the fail-fast checks in `DAGScheduler`);
this method only picks
+ * the implementation.
+ */
+ def shuffleManagerFor(dependency: ShuffleDependency[_, _, _]):
ShuffleManager =
+ dependency match {
+ case _: PipelinedShuffleDependency[_, _, _] =>
+ _pipelinedShuffleManager.getOrElse(_blockingShuffleManager)
Review Comment:
We shouldn't fall back to the _blockingShuffleManager. There is already an
existing / default incremental shuffle manager impl i.e.
StreamingShuffleManager. We should make sure it is created just like the
blocking shuffle manager i.e. the previous shuffle manager.
--
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]
