This is an automated email from the ASF dual-hosted git repository.
jackylee-ch pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 413901f9d4 [GLUTEN-12716][CORE] Make AppendableSpillerList safe to
append during a spill (#12717)
413901f9d4 is described below
commit 413901f9d4c581830a498cf46b222ec6d622b366
Author: YangJie <[email protected]>
AuthorDate: Tue Aug 11 09:32:00 2026 +0800
[GLUTEN-12716][CORE] Make AppendableSpillerList safe to append during a
spill (#12717)
---
.../apache/gluten/memory/memtarget/Spillers.java | 10 +-
.../gluten/memory/memtarget/SpillersTest.java | 178 +++++++++++++++++++++
2 files changed, 187 insertions(+), 1 deletion(-)
diff --git
a/gluten-core/src/main/java/org/apache/gluten/memory/memtarget/Spillers.java
b/gluten-core/src/main/java/org/apache/gluten/memory/memtarget/Spillers.java
index 220a9de32b..970a500050 100644
--- a/gluten-core/src/main/java/org/apache/gluten/memory/memtarget/Spillers.java
+++ b/gluten-core/src/main/java/org/apache/gluten/memory/memtarget/Spillers.java
@@ -16,7 +16,10 @@
*/
package org.apache.gluten.memory.memtarget;
+import javax.annotation.concurrent.ThreadSafe;
+
import java.util.*;
+import java.util.concurrent.CopyOnWriteArrayList;
public final class Spillers {
private Spillers() {
@@ -61,8 +64,13 @@ public final class Spillers {
}
}
+ @ThreadSafe
public static class AppendableSpillerList implements Spiller {
- private final List<Spiller> spillers = new ArrayList<>();
+ // Callers keep appending after the list is registered with the task's
memory tree, and the walk
+ // below runs on whichever thread hit the memory limit, holding the
iteration open across a JNI
+ // spill. So an append can land mid-walk, and the two sides share no lock.
Copy-on-write also
+ // stays correct if a spiller ever appends during its own spill, which a
lock would not cover.
+ private final CopyOnWriteArrayList<Spiller> spillers = new
CopyOnWriteArrayList<>();
private AppendableSpillerList() {}
diff --git
a/gluten-core/src/test/java/org/apache/gluten/memory/memtarget/SpillersTest.java
b/gluten-core/src/test/java/org/apache/gluten/memory/memtarget/SpillersTest.java
new file mode 100644
index 0000000000..894f76333c
--- /dev/null
+++
b/gluten-core/src/test/java/org/apache/gluten/memory/memtarget/SpillersTest.java
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.memory.memtarget;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class SpillersTest {
+
+ private static Spiller countingSpiller(AtomicInteger counter) {
+ return new Spiller() {
+ @Override
+ public long spill(MemoryTarget self, Phase phase, long size) {
+ counter.incrementAndGet();
+ // Reclaim nothing, so the caller walks the rest of the list.
+ return 0L;
+ }
+ };
+ }
+
+ private static Spiller recordingSpiller(String name, List<String> order,
AtomicInteger counter) {
+ return new Spiller() {
+ @Override
+ public long spill(MemoryTarget self, Phase phase, long size) {
+ order.add(name);
+ counter.incrementAndGet();
+ return 0L;
+ }
+ };
+ }
+
+ @Test
+ public void testAppendFromAnotherThreadDuringSpill() throws Exception {
+ // Callers append after the list is registered with the task's memory
tree, and a spill runs on
+ // whichever thread hit the limit. The latches pin that interleaving: the
appending thread runs
+ // while the spilling thread sits between two entries.
+ final Spillers.AppendableSpillerList spillers = Spillers.appendable();
+ final MemoryTarget target = new NoopMemoryTarget();
+ final AtomicInteger spillCount = new AtomicInteger(0);
+ final AtomicInteger appendedSpills = new AtomicInteger(0);
+ final List<String> order = new CopyOnWriteArrayList<>();
+ final CountDownLatch reachedMiddle = new CountDownLatch(1);
+ final CountDownLatch appended = new CountDownLatch(1);
+ final AtomicReference<Throwable> appendFailure = new AtomicReference<>();
+
+ spillers.append(recordingSpiller("first", order, spillCount));
+ spillers.append(
+ new Spiller() {
+ @Override
+ public long spill(MemoryTarget self, Phase phase, long size) {
+ order.add("blocking");
+ spillCount.incrementAndGet();
+ reachedMiddle.countDown();
+ try {
+ Assert.assertTrue(
+ "appending thread did not finish within 30s",
+ appended.await(30, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ return 0L;
+ }
+ });
+ spillers.append(recordingSpiller("third", order, spillCount));
+
+ final Thread appender =
+ new Thread(
+ () -> {
+ try {
+ Assert.assertTrue(
+ "spilling thread did not reach the middle spiller within
30s",
+ reachedMiddle.await(30, TimeUnit.SECONDS));
+ spillers.append(countingSpiller(appendedSpills));
+ } catch (Throwable t) {
+ appendFailure.compareAndSet(null, t);
+ } finally {
+ appended.countDown();
+ }
+ },
+ "spiller-appender");
+ appender.setDaemon(true);
+ appender.start();
+ final List<String> firstWalk;
+ try {
+ Assert.assertEquals(0, spillers.spill(target, Spiller.Phase.SPILL, 100));
+ } finally {
+ // Snapshot before the second walk records into the same list.
+ firstWalk = Arrays.asList(order.toArray(new String[0]));
+ appender.join(TimeUnit.SECONDS.toMillis(30));
+ }
+
+ if (appendFailure.get() != null) {
+ Assert.fail("Appending thread failed: " + appendFailure.get());
+ }
+ Assert.assertFalse("appender thread is still running after join",
appender.isAlive());
+ // The walk covers the three entries present when it started, in
registration order.
+ Assert.assertEquals(Arrays.asList("first", "blocking", "third"),
firstWalk);
+ Assert.assertEquals(3, spillCount.get());
+ Assert.assertEquals(0, appendedSpills.get());
+
+ // The spiller appended mid-walk did reach the list: it takes part in the
next walk.
+ Assert.assertEquals(0, spillers.spill(target, Spiller.Phase.SPILL, 100));
+ Assert.assertEquals(1, appendedSpills.get());
+ Assert.assertEquals(6, spillCount.get());
+ }
+
+ @Test
+ public void testAppendDuringOwnSpill() {
+ // No production spiller appends today, but a lock around append would not
cover one that did,
+ // so pin the re-entrant case as well.
+ final Spillers.AppendableSpillerList spillers = Spillers.appendable();
+ final MemoryTarget target = new NoopMemoryTarget();
+ final AtomicInteger appendedSpills = new AtomicInteger(0);
+ final AtomicInteger spillCount = new AtomicInteger(0);
+
+ spillers.append(
+ new Spiller() {
+ @Override
+ public long spill(MemoryTarget self, Phase phase, long size) {
+ spillCount.incrementAndGet();
+ spillers.append(countingSpiller(appendedSpills));
+ return 0L;
+ }
+ });
+ spillers.append(countingSpiller(spillCount));
+
+ Assert.assertEquals(0, spillers.spill(target, Spiller.Phase.SPILL, 100));
+ Assert.assertEquals(2, spillCount.get());
+ Assert.assertEquals(0, appendedSpills.get());
+ // The appended spiller takes part in the next walk.
+ Assert.assertEquals(0, spillers.spill(target, Spiller.Phase.SPILL, 100));
+ Assert.assertEquals(1, appendedSpills.get());
+ }
+
+ @Test
+ public void testSpillStopsOnceTheRequestIsMet() {
+ // Pins the pre-existing short-circuit in AppendableSpillerList#spill, not
this fix: once the
+ // request is met the walk stops, so later entries are never consulted.
+ final Spillers.AppendableSpillerList spillers = Spillers.appendable();
+ final MemoryTarget target = new NoopMemoryTarget();
+ final AtomicInteger laterSpills = new AtomicInteger(0);
+
+ spillers.append(
+ new Spiller() {
+ @Override
+ public long spill(MemoryTarget self, Phase phase, long size) {
+ return size;
+ }
+ });
+ spillers.append(countingSpiller(laterSpills));
+
+ Assert.assertEquals(100, spillers.spill(target, Spiller.Phase.SPILL, 100));
+ Assert.assertEquals(0, laterSpills.get());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
