LuciferYang opened a new pull request, #12717: URL: https://github.com/apache/gluten/pull/12717
### What changes were proposed in this pull request? `Spillers.AppendableSpillerList` holds a plain `ArrayList` that `append` mutates and `spill` iterates, with no lock on either side, and the two run on different threads. `NativeMemoryManager` hands the list to `ReservationListeners`, which registers it as a node of the task's memory tree, before anything is appended to it. The shuffle writers then append their own spiller on the first non-empty batch, which is well after `records.next()` started driving the upstream pipeline. On the other side `TreeMemoryTargets#spillTree` walks every consumer of the task by design, per the comment at its root-level caller in `MemoryTargets`: "Spill from root node so other consumers also get spilled". So a spill triggered by any consumer reaches every other consumer's list, across runtime boundaries. The triggering thread need not be the task thread: an allocation on a Velox io thread goes through that runtime's `ReservationListener`, so an async split prefetch can fail to reserve and start a root-level walk while the task thread is back in Java appending. The walk also stays inside the loop across a JNI shrink or reclaim call, so the window is milliseconds rather than one instruction. Five call sites append to such a list, and they are not equally exposed. `NativeMemoryManager.scala:59` and `NativePlanEvaluator.java:94` append while their `NativeMemoryManager` is still being constructed, when no other thread of the task is allocating yet. The three shuffle writers (`ColumnarShuffleWriter.scala:194`, and the Celeborn and Uniffle variants) are the reachable ones, because their list is registered when the writer is built and appended to only on the first non-empty batch. The fix is in `gluten-core`, so it covers all five regardless. This switches the field to `CopyOnWriteArrayList`. The two sides cannot be brought under one lock cheaply, iteration is held open across a JNI spill, and appends are two per list per task while walks are on the reclaim path, so the copy is cheap. Snapshot iteration also stays correct if a spiller ever appends during its own spill, which locking `append` would not cover. The field is declared as the concrete type so a revert to `ArrayList` is a compile error, and the class is marked `@ThreadSafe` because `MemoryTarget` documents the opposite default for its implementations. Worth naming what this does not change: a copy-on-write iterator is a snapshot, so an append landing mid-walk still misses that round, and `SpillersTest` asserts exactly that. `ThrowOnOomMemoryTarget.borrow` retries the reservation and each retry re-enters `spillTree` on a fresh snapshot, so the skip defers one round of reclaim rather than losing it. An index walk over the same list would be equally thread-safe and would pick up mid-walk appends, but it would also let a self-appending spiller extend a single walk without bound; snapshot iteration fixes the work per round instead. GLUTEN-11509 was this race one field over, on `TreeMemoryConsumer#children`, with a production stack trace from the Delta statistics writer thread. Its fix (#11553) switched that map to `ConcurrentHashMap` and left this list alone. That issue noted the main branch had no asynchronous use of the memory tree yet; the per-runtime hooked executor for Velox io threads (#11882, #12302) and the Delta native statistics writer (#11419) both supply one now. `NativeMemoryManager`'s `mutableStats` map has the same publish-then-mutate shape and is still unguarded; it is read only by `Node#stats()` on the OOM-message path, never by `spillTree`, so it does not affect reclaim and is tracked separately. ### How was this patch tested? New `SpillersTest` with three tests. `testAppendFromAnotherThreadDuringSpill` uses two latches to pin the interleaving deterministically: the appending thread runs while the spilling thread sits between two entries. It then asserts the appended spiller was not lost by running a second walk, and asserts the first walk visited the three registered entries in order. `testAppendDuringOwnSpill` covers a spiller that appends during its own spill, the case a lock around `append` would not cover. `testSpillStopsOnceTheRequestIsMet` pins the pre-existing short-circuit and is a boundary assertion, not a guardrail for this change. Each assertion was checked against a mutant, restoring the source after every run: | mutant | caught by | | --- | --- | | `CopyOnWriteArrayList` back to `ArrayList` | both concurrency tests, `ConcurrentModificationException` | | `append` silently drops once the list holds three (keeps COW, no CME) | the appended-spiller assertion, `expected:<1> but was:<0>` | | walk order reversed | the order assertion, `expected:<[first, blocking, third]> but was:<[third, blocking, first]>` | `mvn -Pspark-3.5 -pl gluten-core test` gives 33 Java and 39 Scala tests passing. Cross-version `test-compile` passes on spark-3.3, spark-3.4, spark-4.0 with scala-2.13, and spark-4.1 with scala-2.13. ### Was this patch authored or co-authored using generative AI tooling? No Closes #12716 -- 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]
