dongjoon-hyun commented on PR #58097:
URL: https://github.com/apache/spark/pull/58097#issuecomment-5363201382
Thanks for the detailed write-up in the PR description -- the inline
rationale (especially in `ChannelShuffleRendezvous` and the `epoch` design)
made this much easier to follow. I reviewed the full diff and ran the new
suites locally. One confirmed deadlock plus three smaller items below.
## 1. Deadlock: partial-read job whose result RDD reaches the pipelined
shuffle through a non-identity chain
`submitMissingTasks` only stamps `SPARK_PIPELINED_LIVE_REDUCE_PARTITIONS`
when `readsShuffleByIdentity` holds (`DAGScheduler.scala:2736`), i.e. the
result RDD reaches the shuffle through a chain of *single* `OneToOneDependency`
hops. When it does not hold, the property is left unset and the writer reads
that as "every reduce partition is live".
The comment there argues this is safe because "that operator drains every
reduce partition". That is true for a **full** read, but not for a **partial**
one. `executeTake`/LIMIT runs its first job on partition 0 only, so the
remaining reduce partitions have no reduce task at all -- not a reader that
stopped early (which `abandon` covers), but one that never started, so the
abandoned mark never appears and `putUnlessAbandoned`'s `q.offer(..., 100ms)`
loop has no exit.
A join is the easy case to hit: the result RDD is a `ZippedPartitionsRDD2`
with two dependencies, so the match falls through to `case _ => false`.
```scala
// local mode, spark.sql.pipelinedShuffle.enabled=true, channel manager,
shuffle.partitions=4
left.join(right, $"k" === $"k2").limit(10).collect() // hangs
```
I added this to `PipelinedLimitHangSuite` (purely additive; existing helpers
untouched):
```scala
private def joinLimitOverPipelinedCompletesWithin(seconds: Int, aqe:
Boolean): Boolean = {
val pool = Executors.newSingleThreadExecutor()
val fut = pool.submit(new Runnable {
override def run(): Unit = withSession(aqe) { spark =>
import spark.implicits._
// Force a shuffled join: a broadcast side would leave only one
exchange, and the
// AQE rule flips a join's inputs only as a symmetric pair.
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "-1")
// Distinct row counts and column names so the two exchanges cannot
canonicalize
// alike: a ReusedExchangeExec would make EnablePipelinedShuffle
leave the whole plan
// regular and the query would pass for the wrong reason.
val left = spark.range(0, 4000000L, 1, 4).withColumn("k", ($"id" %
100))
val right = spark.range(0, 100000L, 1, 4).withColumn("k2", ($"id" %
100))
val rows = left.join(right, $"k" ===
$"k2").limit(10).collect().length
require(rows == 10, s"expected 10 rows, got $rows")
}
})
try {
fut.get(seconds.toLong, TimeUnit.SECONDS)
true
} catch {
case _: java.util.concurrent.TimeoutException =>
fut.cancel(true)
false
} finally {
pool.shutdownNow()
}
}
test("LIMIT over a pipelined shuffle read through a join completes (AQE
off)") {
assert(joinLimitOverPipelinedCompletesWithin(90, aqe = false),
"LIMIT over a joined pipelined shuffle should complete, but the writer
hung feeding " +
"reduce partitions that executeTake never reads")
}
test("LIMIT over a pipelined shuffle read through a join completes (AQE
on)") {
assert(joinLimitOverPipelinedCompletesWithin(90, aqe = true),
"LIMIT over a joined pipelined shuffle should complete, but the writer
hung feeding " +
"reduce partitions that executeTake never reads")
}
```
Result -- the two existing tests pass, the two new ones burn the full 90s
deadline in both AQE modes:
```
[info] - LIMIT over a pipelined shuffle completes (AQE off) (3 seconds, 136
milliseconds)
[info] - LIMIT over a pipelined shuffle completes (AQE on) (287 milliseconds)
[info] - LIMIT over a pipelined shuffle read through a join completes (AQE
off) *** FAILED *** (1 minute, 30 seconds)
[info] - LIMIT over a pipelined shuffle read through a join completes (AQE
on) *** FAILED *** (1 minute, 30 seconds)
```
Thread dump of the forked test JVM at the 45s mark confirms it is a
deadlock, not slowness -- every live executor thread is parked, each having
burned ~0.1s of CPU over 44.8s elapsed:
```
"Executor task launch worker for task 0.0 in stage 0.0 (TID 0)"
TIMED_WAITING (parking)
at java.util.concurrent.LinkedBlockingQueue.offer(...)
at
...ChannelShuffleWriter.putUnlessAbandoned(ChannelShuffleWriterReader.scala:116)
at ...ChannelShuffleWriter.write(ChannelShuffleWriterReader.scala:164)
"Executor task launch worker for task 0.0 in stage 2.0 (TID 8)"
TIMED_WAITING (parking)
at java.util.concurrent.LinkedBlockingQueue.poll(...)
at
...ChannelShuffleReader$$anon$1.takeItem(ChannelShuffleWriterReader.scala:281)
at
...ChannelShuffleReader$$anon$1.advance(ChannelShuffleWriterReader.scala:298)
```
| thread | state | cpu / elapsed |
|---|---|---|
| stage 0.0 TID 0..3 (producer map tasks) | `putUnlessAbandoned` -> `offer`
| 104-116ms / 44.81s |
| stage 2.0 TID 8 (result stage, partition 0) | `takeItem` -> `poll` | 128ms
/ 44.81s |
Note there is exactly **one** reduce task for 4 shuffle partitions, which is
the crux: partitions 1-3 never get a reader, so they never get abandoned, so
the writers never reach their end-of-stream loop, so the partition-0 reader
never counts `numMaps` markers.
The same reasoning applies to `coalesce(...).limit(n)` and
`union(...).limit(n)`.
On the fix: extending `readsShuffleByIdentity` to cover join/union/coalesce
means re-deriving each operator's partition mapping, which seems fragile. It
may be more robust to invert the fallback. Today "cannot determine the live
set" degrades to "everything is live", but the safe degradation is the
opposite: if the result stage does not read all of its partitions
(`rs.partitions.length != rs.rdd.partitions.length`) and the live set cannot be
determined exactly, reject the job fail-fast (or leave the plan regular) rather
than risk a hang. Happy to be wrong here if you see a cheaper invariant.
## 2. `AQEEnablePipelinedShuffle`: `toFlip` matches structurally, not by
identity
`toFlip` is a `mutable.HashSet[ShuffleExchangeExec]`
(`AQEEnablePipelinedShuffle.scala:79`). `TreeNode` overrides `hashCode` but not
`equals`, so the case-class structural `equals` applies and
`toFlip.contains(s)` at line 91 flips *every* structurally equal exchange, not
just the node the collector selected.
`duplicatedShuffleForms` is the only guard, and it is skipped entirely when
`spark.sql.exchange.reuse=false` (`val shared = if (conf.exchangeReuseEnabled)
... else Set.empty`). In that configuration a twin on a `blocked` path -- one
`collectCandidates` deliberately left regular -- gets flipped too. If it sits
below a regular boundary, `classifyJobShuffleShape` then rejects the whole job
with `SparkException`.
Keying on `SparkPlan.id`, or an `IdentityHashMap`-backed set, removes the
ambiguity regardless of the reuse flag.
## 3. `ContextCleaner`: the new arm is gated on the manager, not on the
shuffle
The `else if (tracklessPipelinedManagerActive)` arm
(`ContextCleaner.scala:264`) fires for *any* shuffle found in neither tracker
once the channel manager is configured -- including regular ones. The comment
at 265-277 anticipates the double-clean problem and gates on the manager to
avoid it for default deployments, but the gate does not distinguish a regular
shuffle from a pipelined one.
That matters here precisely because this feature produces regular shuffles:
the materialized prefix of a mixed job. A prefix shuffle unregistered early
(`RDD.cleanShuffleDependencies`, or `Dataset.rdd` with
`spark.sql.classic.shuffleDependency.fileCleanup.enabled`) will, on later GC,
take the new arm and fire `shuffleCleaned` a second time plus an extra
`RemoveShuffle` RPC. Tracking channel shuffle ids in the manager, or checking
that the dependency was pipelined, would scope it correctly.
## 4. `classifyJobShuffleShape` adds per-boundary graph walks to every job
submission
`rddGraphHasPipelinedDependency(sd.rdd)` (`DAGScheduler.scala:1202`) is
called once per frontier regular boundary, and each call builds its own
`visited` set. The previous `classifyJobShuffleKinds` visited each RDD exactly
once.
For a job whose frontier boundaries share a large ancestor subgraph (a wide
join or union over a common base), this is O(K x |graph|) on the
single-threaded DAGScheduler event loop, before any stage is created -- and it
runs for everyone, including users who never enable this feature, where the
answer is always `false`. Carrying a `belowRegular` flag through the single
existing traversal would restore the original cost.
## Minor notes
- The PR description says "Two registered configs" but there are three --
`spark.shuffle.pipelined.channel.queueCapacity` is missing from the
user-facing-change section. The description also mentions a `clearAbandoned`
test helper that does not exist in the code.
- `PipelinedChannelShuffleManager.stop()`'s comment says the rendezvous is
"keyed only by (shuffleId, reducePartitionId)", which is stale now that `epoch`
is part of the key.
Nothing above touches the core design, which I think is sound -- routing by
dependency type, the `requiresDetachedRecords` gate combined with the
`needToCopyObjectsBeforeShuffle` de-duplication, the per-run `epoch` keying,
and the cooperative `killTaskIfInterrupted` escapes on both sides all look
right to me.
--
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]
