peterxcli commented on PR #5493: URL: https://github.com/apache/datafusion-comet/pull/5493#issuecomment-5441826645
Thanks for the review @andygrove. Point-by-point: **PR description** — agreed, that was a real gap. The description now has a dedicated section explaining `allocateBlocking`: why removing cross-task force-spill needs a replacement on the shared on-heap pool, how the per-thread accounting drives the two fail-fast checks, and why the off-heap path is untouched. On splitting into a second PR: the registry change alone is a regression for opt-in on-heap mode (@sunchao reproduced a real Spark 3.5.9 shuffle failing on the intermediate commit), so the two halves need to land together to keep every commit releasable. **`wait()` has no timeout** — took your first option in cab20e38f: the wait is now `wait(30_000)` chunks that re-log every 30 s with elapsed time, requested bytes, free bytes, and waiter count, so a stall caused by a holder stuck outside the allocator is visible in executor logs. I kept it unbounded rather than adding a deadline because an earlier revision had exactly that (60 s cap) and @sunchao demonstrated it fails a legitimate workload whose healthy holder simply outlives the cap — any finite default has that problem, and it is also how Spark's own `UnifiedMemoryManager` behaves (its `lock.wait()` loop has no deadline either). **`retainedMemory` Thread references** — I looked at this and I believe it is already safe, so no change: entries are removed eagerly the moment a thread's retained total reaches zero (`free` decrements per page owner), so the map cannot accumulate entries across normally-completing tasks. A lingering entry requires a page that is never freed, and such a page pins its owner thread through `pageOwners` regardless, so a `WeakHashMap` would not reclaim anything — and if the entry did vanish while the leaked page stayed allocated, the deadlock check would lose sight of memory that is genuinely unfreeable and waiters would hang instead of failing. A `TaskCompletionListener` cannot free those pages either (the leak means cleanup did not run). The map walk happens only on the failed-allocation path and its size is bounded by the number of concurrent shuffle tasks. **Interrupt classification** — the exception type is what drives it. Spark's killed-task clause in `Executor.TaskRunner.run` (Spark 3.5.9, `core/src/main/scala/org/apache/spark/executor/Executor.scala`) is `case _: InterruptedException | NonFatal(_) if task != null && task.reasonIfKilled.isDefined => ... TaskKilled`. `reasonIfKilled` is necessary but not sufficient — the guard only matches `InterruptedException` or NonFatal throwables, and `SparkOutOfMemoryError` extends `OutOfMemoryError`, which `NonFatal` rejects, so it falls through to the generic `case t: Throwable` and becomes `ExceptionFailure` even for a killed task. @sunchao verified this end-to-end in [this thread](https://github.com/apache/datafusion-comet/pull/5493#discussion_r3870526317): `killTaskAttempt(id, true)` on a waiter produced `ExceptionFailure` with the old rethrow and `TaskKilled` behaves correctly with the NonFatal wrapper. There is also a unit test asserting the surfaced error is not an `OutOfMemoryEr ror` and carries the `InterruptedException` cause. **Test flakiness** — the suite runs in ~8 s locally with 60 s `failAfter` budgets. Every cross-thread ordering the assertions depend on is enforced by a latch or a state poll (threads observed in `WAITING`/`TIMED_WAITING` before the event that releases them); the one bare `Thread.sleep(500)` is one-sided — if the other thread is late, the scenario degrades to a trivially-passing variant, never a failure. Your instinct was right in one concrete way: switching to a timed wait moved blocked threads from `WAITING` to `TIMED_WAITING` and the state polls caught it immediately; they now accept both states (also in cab20e38f). 10/10 consecutive local runs pass. -- 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]
