From ed83c6fdeb03f75fddd7ae195a744eb6e525abc2 Mon Sep 17 00:00:00 2001 From: Brian Neumann-Fopiano Date: Fri, 30 Oct 2026 20:55:54 -0400 Subject: [PATCH] Unqueued history claim waits - Semantic claims and saved biome writes elect one writer per batch instead of queueing every caller on the monitor - Waiting callers park until their batch settles and wake the next queued caller when the writer leaves - Admission tests count parked followers as queued behind the history writer --- .../iris/world/history/GenerationHistory.java | 41 +++++++++++++++-- .../iris/world/history/SavedBiomeStore.java | 46 ++++++++++++++++--- .../GenerationHistoryAdmissionTest.java | 28 ++++++----- 3 files changed, 94 insertions(+), 21 deletions(-) diff --git a/core/src/main/java/art/arcane/iris/world/history/GenerationHistory.java b/core/src/main/java/art/arcane/iris/world/history/GenerationHistory.java index 83857db64..7731c1774 100644 --- a/core/src/main/java/art/arcane/iris/world/history/GenerationHistory.java +++ b/core/src/main/java/art/arcane/iris/world/history/GenerationHistory.java @@ -26,6 +26,8 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.LockSupport; import java.util.function.Predicate; public final class GenerationHistory { @@ -34,6 +36,7 @@ public final class GenerationHistory { private static final int MAXIMUM_CACHED_TRANSITION_PLANS = 4; private static final int MAXIMUM_SEMANTIC_BATCH = 32; private static final int MAXIMUM_PENDING_SEMANTIC_CLAIMS = 128; + private static final long SEMANTIC_CLAIM_PARK_NANOS = 10_000_000L; private final GenerationHistoryPaths paths; private final GenerationPackRepository packs; @@ -50,6 +53,7 @@ public final class GenerationHistory { private final Map transitionPlanCache; private final ArrayBlockingQueue pendingSemanticClaims = new ArrayBlockingQueue<>(MAXIMUM_PENDING_SEMANTIC_CLAIMS); + private final AtomicBoolean semanticWriter = new AtomicBoolean(); private GenerationHistory( GenerationHistoryPaths paths, @@ -776,12 +780,26 @@ public final class GenerationHistory { persistPendingSemanticClaims(); } } - synchronized (this) { - while (!pending.claim.completed) { - persistPendingSemanticClaims(); + while (!pending.settled) { + if (!semanticWriter.compareAndSet(false, true)) { + LockSupport.parkNanos(pending, SEMANTIC_CLAIM_PARK_NANOS); + continue; + } + try { + synchronized (this) { + while (!pending.claim.completed) { + persistPendingSemanticClaims(); + } + } + } finally { + semanticWriter.set(false); + PendingSemanticClaim next = pendingSemanticClaims.peek(); + if (next != null) { + LockSupport.unpark(next.owner); + } } - return pending.claim.result(); } + return pending.claim.result(); } private void persistPendingSemanticClaims() { @@ -814,10 +832,15 @@ public final class GenerationHistory { } while ((current = pendingSemanticClaims.poll()) != null) { reported |= current.claim.fail(failure); + current.settle(); } if (!reported) { throw failure; } + } finally { + for (PendingSemanticClaim pending : batch) { + pending.settle(); + } } } @@ -1354,11 +1377,21 @@ public final class GenerationHistory { private static final class PendingSemanticClaim { private final GenerationStage stage; private final GenerationSemanticIndex.Claim claim; + private final Thread owner = Thread.currentThread(); + private volatile boolean settled; private PendingSemanticClaim(GenerationStage stage, ChunkGenerationSemantics update) { this.stage = stage; this.claim = new GenerationSemanticIndex.Claim(update); } + + private void settle() { + if (!claim.completed || settled) { + return; + } + settled = true; + LockSupport.unpark(owner); + } } public static final class GenerationStage implements AutoCloseable { diff --git a/core/src/main/java/art/arcane/iris/world/history/SavedBiomeStore.java b/core/src/main/java/art/arcane/iris/world/history/SavedBiomeStore.java index 22d0240e8..23a6520e1 100644 --- a/core/src/main/java/art/arcane/iris/world/history/SavedBiomeStore.java +++ b/core/src/main/java/art/arcane/iris/world/history/SavedBiomeStore.java @@ -29,6 +29,8 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.LockSupport; import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.zip.CRC32; @@ -49,6 +51,7 @@ public final class SavedBiomeStore { private static final int MAXIMUM_CACHED_CHUNKS = 128; private static final long MAXIMUM_CACHED_BYTES = 64L * 1024L * 1024L; private static final int MAXIMUM_WRITE_BATCH = 64; + private static final long WRITE_PARK_NANOS = 10_000_000L; private static final ThreadLocal DEFLATERS = ThreadLocal.withInitial(() -> new Deflater(Deflater.BEST_SPEED)); private final Path dimensionRoot; @@ -132,19 +135,34 @@ public final class SavedBiomeStore { .putInt(body.length).put(body).putInt(checksum(body)).array(); PendingWrite pending = new PendingWrite(required, record); lock.pending.add(pending); - synchronized (lock) { - while (!pending.completed) { - persistPending(lock); + while (!pending.settled) { + if (!lock.writer.compareAndSet(false, true)) { + LockSupport.parkNanos(pending, WRITE_PARK_NANOS); + continue; } - if (pending.failure != null) { - throw pending.failure; + try { + synchronized (lock) { + while (!pending.completed) { + persistPending(lock); + } + } + } finally { + lock.writer.set(false); + PendingWrite next = lock.pending.peek(); + if (next != null) { + LockSupport.unpark(next.owner); + } } - return pending.persisted; } + if (pending.failure != null) { + throw pending.failure; + } + return pending.persisted; } private void persistPending(WriteStripe stripe) { Map> batches = new LinkedHashMap<>(); + List polled = new ArrayList<>(MAXIMUM_WRITE_BATCH); PendingWrite grouping = null; try { for (int count = 0; count < MAXIMUM_WRITE_BATCH; count++) { @@ -152,6 +170,7 @@ public final class SavedBiomeStore { if (grouping == null) { break; } + polled.add(grouping); long regionKey = key(grouping.chunk.chunkX() >> 5, grouping.chunk.chunkZ() >> 5); batches.computeIfAbsent(regionKey, ignored -> new ArrayList<>()).add(grouping); grouping = null; @@ -175,6 +194,10 @@ public final class SavedBiomeStore { } } throw failure; + } finally { + for (PendingWrite pending : polled) { + pending.settle(); + } } } @@ -822,19 +845,30 @@ public final class SavedBiomeStore { private static final class WriteStripe { private final ConcurrentLinkedQueue pending = new ConcurrentLinkedQueue<>(); + private final AtomicBoolean writer = new AtomicBoolean(); } private static final class PendingWrite { private final SavedBiomeChunk chunk; private final byte[] record; + private final Thread owner = Thread.currentThread(); private boolean completed; private boolean persisted; private IOException failure; + private volatile boolean settled; private PendingWrite(SavedBiomeChunk chunk, byte[] record) { this.chunk = chunk; this.record = record; } + + private void settle() { + if (!completed || settled) { + return; + } + settled = true; + LockSupport.unpark(owner); + } } private static final class DecodeBudget { diff --git a/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryAdmissionTest.java b/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryAdmissionTest.java index 15455769d..1cdbada60 100644 --- a/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryAdmissionTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/GenerationHistoryAdmissionTest.java @@ -250,7 +250,7 @@ public class GenerationHistoryAdmissionTest { .addSurfaceBiome("iris:conflict").seal().build() : claim(stage); results.add(executor.submit(() -> persistCounting(history, stage, update, appends, forces))); - awaitBlockedOnHistory(history, workers, index + 1); + awaitQueuedOnHistory(history, workers, index + 1); } stages.get(5).close(); } @@ -340,7 +340,7 @@ public class GenerationHistoryAdmissionTest { for (int index = 0; index < stages.size(); index++) { GenerationHistory.GenerationStage stage = stages.get(index); results.add(executor.submit(() -> persistFailingRegion(history, stage, failure))); - awaitBlockedOnHistory(history, workers, index + 1); + awaitQueuedOnHistory(history, workers, index + 1); } } assertTrue(results.getFirst().get(5, TimeUnit.SECONDS)); @@ -404,21 +404,27 @@ public class GenerationHistoryAdmissionTest { } } - private static void awaitBlockedOnHistory(GenerationHistory history, List workers, int count) { - Await.until("semantic claim callers to queue behind the History monitor", Duration.ofSeconds(5L), - () -> blockedOnHistory(history, workers) == count); + private static void awaitQueuedOnHistory(GenerationHistory history, List workers, int count) { + Await.until("semantic claim callers to queue behind the History writer", Duration.ofSeconds(5L), + () -> queuedOnHistory(history, workers) == count); } - private static int blockedOnHistory(GenerationHistory history, List workers) { - int blocked = 0; + private static int queuedOnHistory(GenerationHistory history, List workers) { + int queued = 0; for (Thread worker : workers) { ThreadInfo info = ManagementFactory.getThreadMXBean().getThreadInfo(worker.threadId()); - if (info != null && info.getThreadState() == Thread.State.BLOCKED - && info.getLockInfo().getIdentityHashCode() == System.identityHashCode(history)) { - blocked++; + if (info == null || info.getLockInfo() == null) { + continue; + } + boolean writer = info.getThreadState() == Thread.State.BLOCKED + && info.getLockInfo().getIdentityHashCode() == System.identityHashCode(history); + boolean follower = info.getThreadState() == Thread.State.TIMED_WAITING + && info.getLockInfo().getClassName().endsWith("GenerationHistory$PendingSemanticClaim"); + if (writer || follower) { + queued++; } } - return blocked; + return queued; } private static boolean persistPaused(GenerationHistory history, GenerationHistory.GenerationStage stage,