cloud-fan commented on code in PR #56559:
URL: https://github.com/apache/spark/pull/56559#discussion_r3499050906
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -105,6 +105,32 @@ private class ShuffleStatus(
*/
private[spark] val checksumMismatchIndices: Set[Int] = Set()
+ /**
+ * Set of stale pushed partition indexes for this shuffle. Each entry is a
partitionId (which
+ * equals mapIndex, not MapStatus.mapId). When task retry or speculation
causes multiple
+ * attempts for the same map output to push, the merger may include data
from a stale attempt.
+ * We record the stale partition indexes here so the reduce side can check
chunkBitmaps and
+ * fallback if stale data is present in a merged block.
+ */
+ private[this] val staleMapIndexes = new java.util.HashSet[Int]()
+
+ /**
+ * Mark a partition as having stale (redundant) push attempts. Called from
TaskSetManager when it
+ * detects that multiple task attempts for the same map output pushed data
to the merger.
+ * @param mapIndex the partition index (== mapIndex) of the stale
(redundant) attempt;
+ * this is NOT MapStatus.mapId
+ */
+ def markStalePushedPartition(mapIndex: Int): Unit = withWriteLock {
Review Comment:
Non-blocking design question. `markStalePushedPartition` records the stale
partition but does not `incrementEpoch()`, and the worker
(`MapOutputTrackerWorker.staleMapIndexes`) caches the stale set per-shuffle,
refreshing it only on a cache miss, an epoch bump, or a fetch failure. So a
reducer that fetched its merge statuses *before* a late attempt's mark keeps
its already-cached (empty) stale set and never sees the mark — meaning layer-3
fallback does not fire for that reducer in exactly the layer-1-failed race this
PR targets (original attempt pushed, then killed by a speculative winner that
also pushed).
The window is narrow and layer 1 (deferred push) is the primary defense, so
this is a question rather than a blocker: is there an ordering guarantee that
the mark always precedes any reducer's first status fetch for the shuffle? If
not, should the mark advance the epoch (the checksum-mismatch path effectively
forces a refresh via stage recompute)? A sentence in the code or PR description
either way would help. Distinct from the "never cleared" thread above, which is
about *removing* marks, not propagating new ones.
##########
core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala:
##########
@@ -288,6 +300,36 @@ 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.
+ *
+ * @param shuffleBlockId ShuffleMergedBlockId to be checked
+ * @param address BlockManagerId of push-based shuffle service
+ * @param chunkBitmaps Chunks bitmap from push-based shuffle service site
Review Comment:
Typo: "site" -> "side" (cf. the surrounding "server-side" / "service side"
usage).
```suggestion
* @param chunkBitmaps Chunks bitmap from push-based shuffle service side
```
--
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]
