diff --git a/worldedit-core/build.gradle.kts b/worldedit-core/build.gradle.kts index dba8ca0823..597a69bd93 100644 --- a/worldedit-core/build.gradle.kts +++ b/worldedit-core/build.gradle.kts @@ -62,6 +62,7 @@ dependencies { // Tests testRuntimeOnly(libs.log4j.core) testImplementation(libs.parallelgzip) + testImplementation(libs.lz4Java) } tasks.test { diff --git a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/RollbackOptimizedHistory.java b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/RollbackOptimizedHistory.java index 7f677b0d86..117887ed32 100644 --- a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/RollbackOptimizedHistory.java +++ b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/RollbackOptimizedHistory.java @@ -51,7 +51,12 @@ public RollbackOptimizedHistory( this.maxX = region.getMaximumX(); this.maxY = region.getMaximumY(); this.maxZ = region.getMaximumZ(); - this.blockSize = (int) size; + // NOTE: this truncates `size` to an int before storing it, same as the historic + // `this.blockSize = (int) size;` assignment did. That truncation is a known, separate + // bug tracked elsewhere - it is preserved here intentionally and not fixed as part of + // the blockSize -> LongAdder migration. + this.blockSize.reset(); + this.blockSize.add((int) size); this.command = command; this.closed = true; } diff --git a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSet.java b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSet.java index 8c9da71ae5..19639cf0f4 100644 --- a/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSet.java +++ b/worldedit-core/src/main/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSet.java @@ -37,10 +37,37 @@ import java.util.NoSuchElementException; import java.util.Queue; import java.util.concurrent.Exchanger; +import java.util.concurrent.atomic.LongAdder; import java.util.function.BiConsumer; /** * FAWE stream ChangeSet offering support for extended-height worlds + * + *

Thread-safety / reentrancy warning: a single {@code FaweStreamChangeSet} + * instance is not safe for concurrent read traversals. The decoder state used while + * reading back changes - {@link #posDel}, {@link #idDel}, {@link #originX}, {@link #originZ} and + * {@link #version} - is stored directly on the instance rather than being scoped to a single + * traversal. These fields are (re)initialized by {@link #readHeader(InputStream)} / + * {@link #setupStreamDelegates(int)} and are then mutated as each iterator or + * {@link com.fastasyncworldedit.core.history.change.ChangePopulator} advances (e.g. the + * running {@code lx}/{@code ly}/{@code lz} delta-decoding state captured by the position + * delegate).

+ * + *

As a result:

+ * + * + *

This is documentation only for now: a proper fix that gives each traversal its own, + * independent decoder state is planned as a future, larger refactor. No runtime locking or + * guard has been added here to enforce the above.

*/ public abstract class FaweStreamChangeSet extends AbstractChangeSet { @@ -52,7 +79,7 @@ public abstract class FaweStreamChangeSet extends AbstractChangeSet { private final int compression; private final int minY; - protected long blockSize; + protected final LongAdder blockSize = new LongAdder(); private int originX; private int originZ; private int version; @@ -292,21 +319,21 @@ public FaweOutputStream getCompressedOS(OutputStream os) throws IOException { @Override public boolean isEmpty() { - if (blockSize > 0) { + if (blockSize.sum() > 0) { return false; } if (!super.isEmpty()) { return false; } flush(); - return blockSize == 0; + return blockSize.sum() == 0; } @Override public long longSize() { // Flush so we can accurately get the size flush(); - return blockSize; + return blockSize.sum(); } @Override @@ -361,7 +388,7 @@ public int getOriginZ() { @Override public void add(int x, int y, int z, int combinedFrom, int combinedTo) { - blockSize++; + blockSize.increment(); try { FaweOutputStream stream = getBlockOS(x, y, z); //x @@ -374,7 +401,7 @@ public void add(int x, int y, int z, int combinedFrom, int combinedTo) { @Override public void addBiomeChange(int bx, int by, int bz, BiomeType from, BiomeType to) { - blockSize++; + blockSize.increment(); try { int x = bx >> 2; int y = by >> 2; @@ -400,7 +427,7 @@ public void addBiomeChange(int bx, int by, int bz, BiomeType from, BiomeType to) @Override public void addTileCreate(final FaweCompoundTag tag) { - blockSize++; + blockSize.increment(); try { NBTOutputStream nbtos = getTileCreateOS(); nbtos.writeTag(new CompoundTag(tag.linTag())); @@ -411,7 +438,7 @@ public void addTileCreate(final FaweCompoundTag tag) { @Override public void addTileRemove(final FaweCompoundTag tag) { - blockSize++; + blockSize.increment(); try { NBTOutputStream nbtos = getTileRemoveOS(); nbtos.writeTag(new CompoundTag(tag.linTag())); @@ -422,7 +449,7 @@ public void addTileRemove(final FaweCompoundTag tag) { @Override public void addEntityRemove(final FaweCompoundTag tag) { - blockSize++; + blockSize.increment(); try { NBTOutputStream nbtos = getEntityRemoveOS(); nbtos.writeTag(new CompoundTag(tag.linTag())); @@ -433,7 +460,7 @@ public void addEntityRemove(final FaweCompoundTag tag) { @Override public void addEntityCreate(final FaweCompoundTag tag) { - blockSize++; + blockSize.increment(); try { NBTOutputStream nbtos = getEntityCreateOS(); nbtos.writeTag(new CompoundTag(tag.linTag())); diff --git a/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java new file mode 100644 index 0000000000..d49ea91a8e --- /dev/null +++ b/worldedit-core/src/test/java/com/fastasyncworldedit/core/history/changeset/FaweStreamChangeSetBlockSizeTest.java @@ -0,0 +1,104 @@ +package com.fastasyncworldedit.core.history.changeset; + +import com.fastasyncworldedit.core.configuration.Settings; +import com.fastasyncworldedit.core.history.MemoryOptimizedHistory; +import com.sk89q.worldedit.world.World; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Isolated; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Regression test for the {@code blockSize} counter on {@link FaweStreamChangeSet}. Historically + * this was a plain, unsynchronized {@code long} incremented from multiple pipeline worker + * threads via {@code blockSize++}, which is not atomic and can silently lose increments under + * real contention. It is now a {@link java.util.concurrent.atomic.LongAdder}, which this test + * verifies is accurate under many concurrent writers. + * + *

Forces {@code Settings.settings().HISTORY.COMPRESSION_LEVEL} to 0, which bypasses the + * compression backend entirely (see {@code MainUtil#getCompressedOS}), so this test only measures + * counter accuracy and isn't coupled to compression behavior. Because {@link Settings} is + * process-global mutable state, this class is marked {@link Isolated} so no other test running + * concurrently in the same JVM observes the temporarily-changed level.

+ */ +@Isolated +class FaweStreamChangeSetBlockSizeTest { + + private static final int THREADS = 16; + private static final int CALLS_PER_THREAD = 1000; + + @Test + void addIsAccurateUnderConcurrentWriters() throws InterruptedException, java.io.IOException { + int previousCompressionLevel = Settings.settings().HISTORY.COMPRESSION_LEVEL; + // Compression level 0 skips the LZ4/Zstd codecs entirely (see MainUtil#getCompressedOS), + // which keeps this test independent of the compileOnly lz4-java dependency. + Settings.settings().HISTORY.COMPRESSION_LEVEL = 0; + + World world = mock(World.class); + when(world.getMinY()).thenReturn(-64); + when(world.getMaxY()).thenReturn(319); + + MemoryOptimizedHistory changeSet = new MemoryOptimizedHistory(world); + ExecutorService executor = Executors.newFixedThreadPool(THREADS); + try { + // Pre-initialize the lazy block-output stream single-threaded, before any concurrent + // add() calls. getBlockOS()'s own double-checked-locking race is fixed independently + // in a sibling PR, not this one - on this branch it's still present, so without this + // warm-up the first wave of concurrent add() calls below would race on that unrelated + // lazy-init path too, rather than exercising only the blockSize counter this test + // targets. getBlockOS() itself doesn't touch blockSize, so this doesn't affect the + // expected total asserted below. + changeSet.getBlockOS(0, 0, 0); + + CountDownLatch startLatch = new CountDownLatch(1); + + List> futures = new ArrayList<>(THREADS); + for (int t = 0; t < THREADS; t++) { + final int threadIndex = t; + Callable task = () -> { + startLatch.await(); + for (int i = 0; i < CALLS_PER_THREAD; i++) { + changeSet.add(threadIndex, 0, i, 0, 1); + } + return null; + }; + futures.add(executor.submit(task)); + } + + startLatch.countDown(); + for (Future future : futures) { + try { + future.get(30, TimeUnit.SECONDS); + } catch (ExecutionException e) { + throw new AssertionError("worker thread failed while calling add()", e.getCause()); + } catch (java.util.concurrent.TimeoutException e) { + throw new AssertionError("worker thread did not finish in time", e); + } + } + executor.shutdown(); + assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS), "executor did not terminate in time"); + + assertEquals((long) THREADS * CALLS_PER_THREAD, changeSet.longSize()); + } finally { + // shutdownNow() runs even on the failure paths above (a worker throwing or timing + // out), so a stuck/failed worker never leaks non-daemon threads into the rest of the + // suite. It's a no-op once the graceful shutdown() above has already succeeded. + executor.shutdownNow(); + Settings.settings().HISTORY.COMPRESSION_LEVEL = previousCompressionLevel; + } + } + +}