cloud-fan commented on code in PR #57286:
URL: https://github.com/apache/spark/pull/57286#discussion_r3606191630
##########
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`.
+ */
+ private[spark] def blockingShuffleManager: BlockingShuffleManager =
_blockingShuffleManager
+
+ /**
+ * The `PipelinedShuffleManager` (configured by
spark.shuffle.manager.incremental, defaulting to
+ * the built-in streaming manager) that serves pipelined shuffle
dependencies. A pipelined shuffle
+ * is read incrementally and served out-of-band, so this manager never
provides a
+ * `ShuffleBlockResolver` of its own.
+ */
+ private[spark] def pipelinedShuffleManager: 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`.
+ */
+ private[spark] 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, defaulting to the built-in streaming
manager), 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.
+ *
+ * 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.
+ */
+ private[spark] def shuffleManagerFor(dependency: ShuffleDependency[_, _,
_]): ShuffleManager =
+ dependency match {
+ case _: PipelinedShuffleDependency[_, _, _] =>
+ _pipelinedShuffleManager
+ case _ => _blockingShuffleManager
+ }
+
+ /**
+ * Unregisters the shuffle from every configured manager (default and, if
present, incremental).
+ * Used by the `RemoveShuffle` cleanup path, which holds only a shuffleId
and cannot know which
+ * manager owns it -- and must reach the owning manager on every node,
including one that never
+ * performed this shuffle's I/O. Notifying all managers is safe:
`unregisterShuffle` for an
+ * unknown id is a no-op. Returns true if any manager reports it removed
metadata; false if no
Review Comment:
This broadcast relies on unknown IDs being safe, but
`ShuffleManager.unregisterShuffle` does not require that; a custom manager may
reject or mutate state for an unknown ID. Please make idempotent unknown-ID
handling part of the interface contract, since
`unregisterShuffleFromAllManagers` now calls every configured 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]
