cloud-fan commented on code in PR #58008:
URL: https://github.com/apache/spark/pull/58008#discussion_r3823303268
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -82,6 +82,19 @@ private[spark] class TaskSetManager(
private val isShuffleMapTasks = tasks(0).isInstanceOf[ShuffleMapTask]
// shuffleId is only available when isShuffleMapTasks=true
private val shuffleId = taskSet.shuffleId
+ // Scopes stale-push reducer fallback to indeterminate stages (deterministic
stages reproduce
+ // identical output across attempts, so a stale push there is benign).
Defaults to false when the
Review Comment:
**Nit:**
Please phrase this as reproducing the same data set. Spark's `UNORDERED`
level is not indeterminate, but its contract explicitly permits record order to
change across reruns, so `identical output` is stronger than the invariant used
by this guard.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -82,6 +82,19 @@ private[spark] class TaskSetManager(
private val isShuffleMapTasks = tasks(0).isInstanceOf[ShuffleMapTask]
// shuffleId is only available when isShuffleMapTasks=true
private val shuffleId = taskSet.shuffleId
+ // Scopes stale-push reducer fallback to indeterminate stages (deterministic
stages reproduce
+ // identical output across attempts, so a stale push there is benign).
Defaults to false when the
+ // stage isn't registered (e.g. tests). Uses taskSet.stageId (constructor
param) because the
+ // `stageId` field below is not initialized yet at this source position.
+ private val isIndeterminateShuffleMapStage: Boolean = isShuffleMapTasks &&
Review Comment:
**Blocking:**
Keep the `ShuffleMapStage` reference and evaluate its indeterminacy when the
duplicate result arrives, rather than caching this Boolean. `DAGScheduler` can
set `isChecksumMismatched` later in the same stage attempt; a subsequent
duplicate then still sees false here, skips stale marking in the
indeterminate-only mode, and leaves reducers able to consume the stale merged
chunk.
##########
core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala:
##########
@@ -301,33 +289,19 @@ private class PushBasedFetchHelper(
}
/**
- * Check whether a push-merged block contains data from stale (duplicate)
task attempts.
- * When speculation is enabled, multiple attempts for the same map output
may both push data
- * to the merger. The merger may include data from both attempts in the same
merged block,
- * but the driver only tracks one as the canonical MapStatus. We detect this
by checking
- * if any stale pushed map index appears in the server-side chunkBitmaps.
+ * Check whether a shuffle chunk contains any stale pushed map index,
enabling
+ * chunk-granularity fallback: only chunks that actually contain stale data
fall back to their
+ * original blocks, while the remaining chunks of the same merged block are
read normally.
*
- * @param shuffleBlockId ShuffleMergedBlockId to be checked
- * @param address BlockManagerId of push-based shuffle service
- * @param chunkBitmaps Chunks bitmap from push-based shuffle service side
- * @return false if any stale-marked mapIndex is present in this block
(forcing fallback),
- * true otherwise
+ * @param blockId ShuffleBlockChunkId to be checked
+ * @return true if this chunk contains any stale-marked mapIndex (forcing a
fallback for this
+ * chunk only), false otherwise
*/
- private[this] def checkStaleMapIdInMergedBlock(
- shuffleBlockId: ShuffleMergedBlockId,
- address: BlockManagerId,
- chunkBitmaps: Array[RoaringBitmap]): Boolean = {
- val staleMapIndexes =
- mapOutputTracker.getStaleMapIndexes(shuffleBlockId.shuffleId)
- if (staleMapIndexes.isEmpty) return true
- val mergedBlockBitmap = new RoaringBitmap()
- chunkBitmaps.foreach(mergedBlockBitmap.or)
- val hasStale = staleMapIndexes.exists(id => mergedBlockBitmap.contains(id))
- if (hasStale) {
- logWarning(s"Found stale pushed map indexes in merged block
$shuffleBlockId from" +
- s" ${address.host}:${address.port}, falling back to fetch the original
blocks")
+ private[spark] def isStaleChunk(blockId: ShuffleBlockChunkId): Boolean = {
+ val staleMapIndexes =
mapOutputTracker.getStaleMapIndexes(blockId.shuffleId)
Review Comment:
**Non-blocking:**
This runs once per fetched chunk, while
`MapOutputTrackerWorker.getStaleMapIndexes` allocates a new `HashSet` and
copies every stale index on each call. A partition with C chunks and S stale
maps therefore copies O(C * S) entries in the reducer fetch path. Please move
the intersection behind a tracker query that inspects the published snapshot
without returning a defensive copy.
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:
##########
@@ -2921,11 +2921,49 @@ class TaskSetManagerSuite
s"\nCaptured logs:\n${logs.mkString("\n")}")
}
- test("SPARK-57491: late-arriving speculative ShuffleMapTask marks stale
partitionId") {
- sc = new SparkContext("local", "test")
+ test("SPARK-57491: late speculative ShuffleMapTask marks stale only when
fallback enabled") {
+ // (fallbackEnabled, detectAllStagesEnabled, expectStale)
+ Seq(
+ // Both switches enabled: the late attempt is marked stale regardless of
stage determinism
+ // (the test does not register an indeterminate ShuffleMapStage, so
detectAllStages is
+ // required here).
+ (true, true, true),
Review Comment:
**Non-blocking:**
This matrix bypasses both contracts it needs to protect: the disabled row
sets false explicitly, and the enabled row turns on `detectAllStages`. Leave
the fallback setting unset in one case, and add an enabled/detect-all-disabled
case with a registered stage whose runtime indeterminacy changes after manager
construction; those cases catch the current default mismatch and stale
constructor snapshot.
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2973,6 +2973,27 @@ package object config {
.doubleConf
.createWithDefault(1.0)
+ private[spark] val STALE_PUSH_FALLBACK_ENABLED =
+ ConfigBuilder("spark.shuffle.push.stale.fallback.enabled")
+ .doc("When true, mark partitions with stale push data so reducers
fallback to " +
+ "unmerged blocks. Default false to keep detection observational only:
duplicate map " +
+ "attempts are always reported (logged), but reducer fallback must be
explicitly " +
+ "enabled. When enabled, fallback marking applies to indeterminate
stages only unless " +
+ "spark.shuffle.push.stale.detectAllStages.enabled is also true.")
+ .version("4.4.0")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
**Blocking:**
The rollout contract is opt-in: both this entry's documentation and the PR
description say the default is false. Leaving this as true enables fallback for
indeterminate stages without an explicit setting and reintroduces the fetch
amplification this PR is meant to avoid.
```suggestion
.createWithDefault(false)
```
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:
##########
@@ -2921,11 +2921,49 @@ class TaskSetManagerSuite
s"\nCaptured logs:\n${logs.mkString("\n")}")
}
- test("SPARK-57491: late-arriving speculative ShuffleMapTask marks stale
partitionId") {
- sc = new SparkContext("local", "test")
+ test("SPARK-57491: late speculative ShuffleMapTask marks stale only when
fallback enabled") {
+ // (fallbackEnabled, detectAllStagesEnabled, expectStale)
+ Seq(
+ // Both switches enabled: the late attempt is marked stale regardless of
stage determinism
+ // (the test does not register an indeterminate ShuffleMapStage, so
detectAllStages is
+ // required here).
+ (true, true, true),
+ // Fallback disabled (default): the late attempt is reported (logged)
but no map index is
+ // marked stale, so reducers keep reading the merged block.
+ (false, false, false)
+ ).foreach { case (fallbackEnabled, detectAllStagesEnabled, expectStale) =>
+ val staleMapIndexes =
+ runLateSpeculativeShuffleMapAttempt(fallbackEnabled,
detectAllStagesEnabled)
+ if (expectStale) {
+ assert(staleMapIndexes.contains(0),
+ s"Expected staleMapIndexes to contain mapIndex 0 " +
+ s"(fallback=$fallbackEnabled), got $staleMapIndexes")
+ } else {
+ assert(staleMapIndexes.isEmpty,
+ s"Expected no stale map indexes (fallback=$fallbackEnabled), got
$staleMapIndexes")
+ }
+ }
+ }
+
+ /**
+ * Drives a shuffle map stage through a late speculative attempt: task 0 and
task 1 start,
+ * task 1 finishes, task 0 is speculated, then the original task 0 finishes
(killing the
+ * speculative attempt) and finally the speculative attempt's result arrives
late. Returns
+ * the stale pushed map indexes recorded by the MapOutputTracker, which are
non-empty only
+ * when reducer fallback marking is enabled.
+ *
+ * Each invocation gets its own SparkContext via [[withSpark]] so the caller
can loop without
Review Comment:
**Nit:**
The unqualified member does not resolve from this suite.
```suggestion
* Each invocation gets its own SparkContext via
[[LocalSparkContext.withSpark]] so the caller can loop without
```
--
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]
