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
This commit is contained in:
Brian Neumann-Fopiano committed 2026-10-30 20:55:54 -04:00
1 parent 1a914a404a
commit ed83c6fdeb
3 files changed
+94 -21

No files matched your search

@@ -26,6 +26,8 @@ import java.util.Objects;
import java.util.Optional; import java.util.Optional;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Predicate; import java.util.function.Predicate;
public final class GenerationHistory { 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_CACHED_TRANSITION_PLANS = 4;
private static final int MAXIMUM_SEMANTIC_BATCH = 32; private static final int MAXIMUM_SEMANTIC_BATCH = 32;
private static final int MAXIMUM_PENDING_SEMANTIC_CLAIMS = 128; 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 GenerationHistoryPaths paths;
private final GenerationPackRepository packs; private final GenerationPackRepository packs;
@@ -50,6 +53,7 @@ public final class GenerationHistory {
private final Map<Long, TransitionGenerationPlan> transitionPlanCache; private final Map<Long, TransitionGenerationPlan> transitionPlanCache;
private final ArrayBlockingQueue<PendingSemanticClaim> pendingSemanticClaims = private final ArrayBlockingQueue<PendingSemanticClaim> pendingSemanticClaims =
new ArrayBlockingQueue<>(MAXIMUM_PENDING_SEMANTIC_CLAIMS); new ArrayBlockingQueue<>(MAXIMUM_PENDING_SEMANTIC_CLAIMS);
private final AtomicBoolean semanticWriter = new AtomicBoolean();
private GenerationHistory( private GenerationHistory(
GenerationHistoryPaths paths, GenerationHistoryPaths paths,
@@ -776,12 +780,26 @@ public final class GenerationHistory {
persistPendingSemanticClaims(); persistPendingSemanticClaims();
} }
} }
synchronized (this) { while (!pending.settled) {
while (!pending.claim.completed) { if (!semanticWriter.compareAndSet(false, true)) {
persistPendingSemanticClaims(); 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() { private void persistPendingSemanticClaims() {
@@ -814,10 +832,15 @@ public final class GenerationHistory {
} }
while ((current = pendingSemanticClaims.poll()) != null) { while ((current = pendingSemanticClaims.poll()) != null) {
reported |= current.claim.fail(failure); reported |= current.claim.fail(failure);
current.settle();
} }
if (!reported) { if (!reported) {
throw failure; throw failure;
} }
} finally {
for (PendingSemanticClaim pending : batch) {
pending.settle();
}
} }
} }
@@ -1354,11 +1377,21 @@ public final class GenerationHistory {
private static final class PendingSemanticClaim { private static final class PendingSemanticClaim {
private final GenerationStage stage; private final GenerationStage stage;
private final GenerationSemanticIndex.Claim claim; private final GenerationSemanticIndex.Claim claim;
private final Thread owner = Thread.currentThread();
private volatile boolean settled;
private PendingSemanticClaim(GenerationStage stage, ChunkGenerationSemantics update) { private PendingSemanticClaim(GenerationStage stage, ChunkGenerationSemantics update) {
this.stage = stage; this.stage = stage;
this.claim = new GenerationSemanticIndex.Claim(update); 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 { public static final class GenerationStage implements AutoCloseable {
@@ -29,6 +29,8 @@ import java.util.Objects;
import java.util.Optional; import java.util.Optional;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue; 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.Matcher;
import java.util.regex.Pattern; import java.util.regex.Pattern;
import java.util.zip.CRC32; import java.util.zip.CRC32;
@@ -49,6 +51,7 @@ public final class SavedBiomeStore {
private static final int MAXIMUM_CACHED_CHUNKS = 128; private static final int MAXIMUM_CACHED_CHUNKS = 128;
private static final long MAXIMUM_CACHED_BYTES = 64L * 1024L * 1024L; private static final long MAXIMUM_CACHED_BYTES = 64L * 1024L * 1024L;
private static final int MAXIMUM_WRITE_BATCH = 64; private static final int MAXIMUM_WRITE_BATCH = 64;
private static final long WRITE_PARK_NANOS = 10_000_000L;
private static final ThreadLocal<Deflater> DEFLATERS = ThreadLocal.withInitial(() -> new Deflater(Deflater.BEST_SPEED)); private static final ThreadLocal<Deflater> DEFLATERS = ThreadLocal.withInitial(() -> new Deflater(Deflater.BEST_SPEED));
private final Path dimensionRoot; private final Path dimensionRoot;
@@ -132,19 +135,34 @@ public final class SavedBiomeStore {
.putInt(body.length).put(body).putInt(checksum(body)).array(); .putInt(body.length).put(body).putInt(checksum(body)).array();
PendingWrite pending = new PendingWrite(required, record); PendingWrite pending = new PendingWrite(required, record);
lock.pending.add(pending); lock.pending.add(pending);
synchronized (lock) { while (!pending.settled) {
while (!pending.completed) { if (!lock.writer.compareAndSet(false, true)) {
persistPending(lock); LockSupport.parkNanos(pending, WRITE_PARK_NANOS);
continue;
} }
if (pending.failure != null) { try {
throw pending.failure; 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) { private void persistPending(WriteStripe stripe) {
Map<Long, List<PendingWrite>> batches = new LinkedHashMap<>(); Map<Long, List<PendingWrite>> batches = new LinkedHashMap<>();
List<PendingWrite> polled = new ArrayList<>(MAXIMUM_WRITE_BATCH);
PendingWrite grouping = null; PendingWrite grouping = null;
try { try {
for (int count = 0; count < MAXIMUM_WRITE_BATCH; count++) { for (int count = 0; count < MAXIMUM_WRITE_BATCH; count++) {
@@ -152,6 +170,7 @@ public final class SavedBiomeStore {
if (grouping == null) { if (grouping == null) {
break; break;
} }
polled.add(grouping);
long regionKey = key(grouping.chunk.chunkX() >> 5, grouping.chunk.chunkZ() >> 5); long regionKey = key(grouping.chunk.chunkX() >> 5, grouping.chunk.chunkZ() >> 5);
batches.computeIfAbsent(regionKey, ignored -> new ArrayList<>()).add(grouping); batches.computeIfAbsent(regionKey, ignored -> new ArrayList<>()).add(grouping);
grouping = null; grouping = null;
@@ -175,6 +194,10 @@ public final class SavedBiomeStore {
} }
} }
throw failure; throw failure;
} finally {
for (PendingWrite pending : polled) {
pending.settle();
}
} }
} }
@@ -822,19 +845,30 @@ public final class SavedBiomeStore {
private static final class WriteStripe { private static final class WriteStripe {
private final ConcurrentLinkedQueue<PendingWrite> pending = new ConcurrentLinkedQueue<>(); private final ConcurrentLinkedQueue<PendingWrite> pending = new ConcurrentLinkedQueue<>();
private final AtomicBoolean writer = new AtomicBoolean();
} }
private static final class PendingWrite { private static final class PendingWrite {
private final SavedBiomeChunk chunk; private final SavedBiomeChunk chunk;
private final byte[] record; private final byte[] record;
private final Thread owner = Thread.currentThread();
private boolean completed; private boolean completed;
private boolean persisted; private boolean persisted;
private IOException failure; private IOException failure;
private volatile boolean settled;
private PendingWrite(SavedBiomeChunk chunk, byte[] record) { private PendingWrite(SavedBiomeChunk chunk, byte[] record) {
this.chunk = chunk; this.chunk = chunk;
this.record = record; this.record = record;
} }
private void settle() {
if (!completed || settled) {
return;
}
settled = true;
LockSupport.unpark(owner);
}
} }
private static final class DecodeBudget { private static final class DecodeBudget {
@@ -250,7 +250,7 @@ public class GenerationHistoryAdmissionTest {
.addSurfaceBiome("iris:conflict").seal().build() .addSurfaceBiome("iris:conflict").seal().build()
: claim(stage); : claim(stage);
results.add(executor.submit(() -> persistCounting(history, stage, update, appends, forces))); results.add(executor.submit(() -> persistCounting(history, stage, update, appends, forces)));
awaitBlockedOnHistory(history, workers, index + 1); awaitQueuedOnHistory(history, workers, index + 1);
} }
stages.get(5).close(); stages.get(5).close();
} }
@@ -340,7 +340,7 @@ public class GenerationHistoryAdmissionTest {
for (int index = 0; index < stages.size(); index++) { for (int index = 0; index < stages.size(); index++) {
GenerationHistory.GenerationStage stage = stages.get(index); GenerationHistory.GenerationStage stage = stages.get(index);
results.add(executor.submit(() -> persistFailingRegion(history, stage, failure))); 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)); assertTrue(results.getFirst().get(5, TimeUnit.SECONDS));
@@ -404,21 +404,27 @@ public class GenerationHistoryAdmissionTest {
} }
} }
private static void awaitBlockedOnHistory(GenerationHistory history, List<Thread> workers, int count) { private static void awaitQueuedOnHistory(GenerationHistory history, List<Thread> workers, int count) {
Await.until("semantic claim callers to queue behind the History monitor", Duration.ofSeconds(5L), Await.until("semantic claim callers to queue behind the History writer", Duration.ofSeconds(5L),
() -> blockedOnHistory(history, workers) == count); () -> queuedOnHistory(history, workers) == count);
} }
private static int blockedOnHistory(GenerationHistory history, List<Thread> workers) { private static int queuedOnHistory(GenerationHistory history, List<Thread> workers) {
int blocked = 0; int queued = 0;
for (Thread worker : workers) { for (Thread worker : workers) {
ThreadInfo info = ManagementFactory.getThreadMXBean().getThreadInfo(worker.threadId()); ThreadInfo info = ManagementFactory.getThreadMXBean().getThreadInfo(worker.threadId());
if (info != null && info.getThreadState() == Thread.State.BLOCKED if (info == null || info.getLockInfo() == null) {
&& info.getLockInfo().getIdentityHashCode() == System.identityHashCode(history)) { continue;
blocked++; }
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, private static boolean persistPaused(GenerationHistory history, GenerationHistory.GenerationStage stage,