From 04f2798c06f1924f5405c042a4be7abd00be2ab1 Mon Sep 17 00:00:00 2001 From: Brian Neumann-Fopiano Date: Thu, 22 Oct 2026 07:29:53 -0400 Subject: [PATCH] Lock-free volume reads - Structure volume caches read through concurrent maps and publish only current builds - Natural terrain receipts reuse a per-thread deflater with the platform gzip framing - Tests pin receipt and saved biome bytes across reused compressors --- .../nativegen/NativeStructureVolumeIndex.java | 92 ++++++++----------- .../world/history/NativeTerrainReceipt.java | 65 ++++++++++++- .../history/NativeTerrainReceiptTest.java | 9 ++ .../history/SavedBiomeStoreWriteTest.java | 44 ++++++--- 4 files changed, 142 insertions(+), 68 deletions(-) diff --git a/core/src/main/java/art/arcane/iris/structure/nativegen/NativeStructureVolumeIndex.java b/core/src/main/java/art/arcane/iris/structure/nativegen/NativeStructureVolumeIndex.java index 5a119efd4..618023b76 100644 --- a/core/src/main/java/art/arcane/iris/structure/nativegen/NativeStructureVolumeIndex.java +++ b/core/src/main/java/art/arcane/iris/structure/nativegen/NativeStructureVolumeIndex.java @@ -12,15 +12,14 @@ import art.arcane.volmlib.util.cache.CacheKey; import art.arcane.volmlib.util.collection.KList; import java.io.IOException; -import java.util.ArrayList; import java.util.Collections; -import java.util.LinkedHashMap; -import java.util.List; import java.util.Map; import java.util.Objects; import java.util.WeakHashMap; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.function.BooleanSupplier; import java.util.function.IntConsumer; import java.util.function.Supplier; @@ -36,7 +35,6 @@ public final class NativeStructureVolumeIndex { private static final int ORIGIN_REACH_CHUNKS = StructureOwnershipRecordView.MAX_REFERENCE_DISTANCE_CHUNKS; private static final int MAX_CACHED_ORIGIN_CHUNKS = 16_384; private static final int MAX_CACHED_QUERY_CHUNKS = 4_096; - private static final int CACHE_STRIPES = 32; private static final Map INDEXES = Collections.synchronizedMap(new WeakHashMap<>()); @@ -223,12 +221,9 @@ public final class NativeStructureVolumeIndex { RuntimeChunkKey key, Supplier> loader ) { - Map> cache = volumes.stripe(key); - synchronized (cache) { - KList hit = cache.get(key); - if (hit != null) { - return hit; - } + KList hit = volumes.get(key); + if (hit != null) { + return hit; } CompletableFuture> future = new CompletableFuture<>(); @@ -239,22 +234,16 @@ public final class NativeStructureVolumeIndex { // Close the check-then-claim window: a racer whose cache check missed before the // previous builder cached could claim the build slot after it retired and rebuild. - synchronized (cache) { - KList published = cache.get(key); - if (published != null) { - future.complete(published); - builds.remove(key, future); - return published; - } + KList published = volumes.get(key); + if (published != null) { + future.complete(published); + builds.remove(key, future); + return published; } try { KList built = loader.get(); - synchronized (cache) { - if (builds.get(key) == future) { - cache.put(key, built); - } - } + volumes.publish(key, built, () -> builds.get(key) == future); future.complete(built); return built; } catch (RuntimeException | Error error) { @@ -301,47 +290,44 @@ public final class NativeStructureVolumeIndex { } } - private static Map> lru(int capacity) { - return new LinkedHashMap<>(16, 0.75F, true) { - @Override - protected boolean removeEldestEntry( - Map.Entry> eldest - ) { - return size() > capacity; - } - }; - } - private static final class VolumeCache { - private final List>> stripes; + private final int capacity; + private final ConcurrentHashMap> entries = new ConcurrentHashMap<>(); + private final ConcurrentLinkedQueue insertionOrder = new ConcurrentLinkedQueue<>(); private VolumeCache(int capacity) { - int stripeCapacity = Math.max(1, capacity / CACHE_STRIPES); - List>> created = new ArrayList<>(CACHE_STRIPES); - for (int stripe = 0; stripe < CACHE_STRIPES; stripe++) { - created.add(lru(stripeCapacity)); - } - stripes = List.copyOf(created); - } - - private Map> stripe(RuntimeChunkKey key) { - int hash = key.hashCode(); - return stripes.get((hash ^ (hash >>> 16)) & (CACHE_STRIPES - 1)); + this.capacity = capacity; } private KList get(RuntimeChunkKey key) { - Map> cache = stripe(key); - synchronized (cache) { - return cache.get(key); + return entries.get(key); + } + + private void publish(RuntimeChunkKey key, KList volumes, BooleanSupplier current) { + boolean[] inserted = new boolean[1]; + entries.compute(key, (ignored, previous) -> { + if (!current.getAsBoolean()) { + return previous; + } + inserted[0] = previous == null; + return volumes; + }); + if (!inserted[0]) { + return; + } + insertionOrder.add(key); + while (entries.size() > capacity) { + RuntimeChunkKey eldest = insertionOrder.poll(); + if (eldest == null) { + return; + } + entries.remove(eldest); } } private void evictRuntime(int runtimeId) { - for (Map> cache : stripes) { - synchronized (cache) { - cache.keySet().removeIf(key -> key.runtimeId() == runtimeId); - } - } + entries.keySet().removeIf(key -> key.runtimeId() == runtimeId); + insertionOrder.removeIf(key -> key.runtimeId() == runtimeId); } } diff --git a/core/src/main/java/art/arcane/iris/world/history/NativeTerrainReceipt.java b/core/src/main/java/art/arcane/iris/world/history/NativeTerrainReceipt.java index 45245b9e8..4d84d4791 100644 --- a/core/src/main/java/art/arcane/iris/world/history/NativeTerrainReceipt.java +++ b/core/src/main/java/art/arcane/iris/world/history/NativeTerrainReceipt.java @@ -6,11 +6,16 @@ import java.io.ByteArrayOutputStream; import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; +import java.io.OutputStream; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.OptionalInt; +import java.util.zip.CRC32; +import java.util.zip.Deflater; +import java.util.zip.DeflaterOutputStream; import java.util.zip.GZIPInputStream; import java.util.zip.GZIPOutputStream; @@ -18,6 +23,8 @@ public final class NativeTerrainReceipt { private static final int MAGIC = 0x4952544E; private static final int VERSION = 1; private static final int MAXIMUM_BYTES = 64 * 1024 * 1024; + private static final ThreadLocal DEFLATERS = + ThreadLocal.withInitial(() -> new Deflater(Deflater.DEFAULT_COMPRESSION, true)); private NativeTerrainReceipt() { } @@ -25,7 +32,9 @@ public final class NativeTerrainReceipt { public static byte[] encode(SavedTerrainChunk chunk, long activationId, String epochId) throws IOException { ByteArrayOutputStream bytes = new ByteArrayOutputStream(); UtfCache strings = new UtfCache(); - try (DataOutputStream output = new DataOutputStream(new BufferedOutputStream(new GZIPOutputStream(bytes)))) { + Deflater deflater = DEFLATERS.get(); + deflater.reset(); + try (DataOutputStream output = new DataOutputStream(new BufferedOutputStream(new GzipStream(bytes, deflater)))) { output.writeInt(MAGIC); output.writeInt(VERSION); output.writeInt(chunk.chunkX()); @@ -196,6 +205,60 @@ public final class NativeTerrainReceipt { public record Decoded(SavedTerrainChunk terrain, long activationId, String epochId) { } + private static final class GzipStream extends DeflaterOutputStream { + private static final int BUFFER_BYTES = 512; + private static final int TRAILER_BYTES = 8; + private static final byte[] HEADER = gzipHeader(); + + private final CRC32 checksum = new CRC32(); + + private GzipStream(OutputStream out, Deflater deflater) throws IOException { + super(out, deflater, BUFFER_BYTES); + out.write(HEADER); + } + + @Override + public void write(byte[] buffer, int offset, int length) throws IOException { + super.write(buffer, offset, length); + checksum.update(buffer, offset, length); + } + + @Override + public void finish() throws IOException { + if (def.finished()) { + return; + } + def.finish(); + while (!def.finished()) { + int length = def.deflate(buf, 0, buf.length); + if (length > 0) { + out.write(buf, 0, length); + } + } + byte[] trailer = new byte[TRAILER_BYTES]; + writeLittleEndian(trailer, 0, (int) checksum.getValue()); + writeLittleEndian(trailer, 4, (int) def.getBytesRead()); + out.write(trailer); + } + + private static void writeLittleEndian(byte[] target, int offset, int value) { + target[offset] = (byte) value; + target[offset + 1] = (byte) (value >>> 8); + target[offset + 2] = (byte) (value >>> 16); + target[offset + 3] = (byte) (value >>> 24); + } + + private static byte[] gzipHeader() { + ByteArrayOutputStream empty = new ByteArrayOutputStream(); + try (GZIPOutputStream gzip = new GZIPOutputStream(empty)) { + gzip.finish(); + } catch (IOException failure) { + throw new IllegalStateException("Unable to resolve the platform gzip header", failure); + } + return Arrays.copyOf(empty.toByteArray(), 10); + } + } + private static final class UtfCache { private static final int MAXIMUM_ENTRIES = 256; private static final int MAXIMUM_ENCODED_BYTES = 64 * 1024; diff --git a/core/src/test/java/art/arcane/iris/world/history/NativeTerrainReceiptTest.java b/core/src/test/java/art/arcane/iris/world/history/NativeTerrainReceiptTest.java index 3d6dbf755..1c34615fd 100644 --- a/core/src/test/java/art/arcane/iris/world/history/NativeTerrainReceiptTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/NativeTerrainReceiptTest.java @@ -60,6 +60,15 @@ public class NativeTerrainReceiptTest { } } + @Test + public void repeatedEncodingOnOneThreadKeepsLegacyBytes() throws Exception { + for (int round = 0; round < 6; round++) { + SavedTerrainChunk chunk = chunk(round % 2 == 0); + String epoch = "epoch-" + round; + assertArrayEquals(legacyEncode(chunk, round, epoch), NativeTerrainReceipt.encode(chunk, round, epoch)); + } + } + @Test public void saturatedStringCachesPreserveModifiedUtfBytes() throws Exception { for (int padding : new int[]{0, 300}) { diff --git a/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreWriteTest.java b/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreWriteTest.java index d7eaa570e..15c2d6011 100644 --- a/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreWriteTest.java +++ b/core/src/test/java/art/arcane/iris/world/history/SavedBiomeStoreWriteTest.java @@ -60,6 +60,16 @@ public class SavedBiomeStoreWriteTest { assertArrayEquals(expectedRegion(), Files.readAllBytes(regionPath(root))); } + @Test + public void consecutiveRecordsKeepIndependentlyCompressedBytes() throws Exception { + Path root = temporaryFolder.newFolder().toPath(); + SavedBiomeStore store = SavedBiomeStore.open(root); + for (int x = 0; x < 4; x++) { + store.claimAndPersist(chunk(x, 3L)); + } + assertArrayEquals(expectedRegion(4), Files.readAllBytes(regionPath(root))); + } + @Test public void relaxedDurabilityPreservesRecordsWithoutForcingEachAppend() throws Exception { String previousMode = System.getProperty(Durability.MODE_PROPERTY); @@ -379,6 +389,10 @@ public class SavedBiomeStoreWriteTest { } private static byte[] expectedRegion() throws IOException { + return expectedRegion(1); + } + + private static byte[] expectedRegion(int records) throws IOException { ByteArrayOutputStream raw = new ByteArrayOutputStream(); DataOutputStream data = new DataOutputStream(raw); data.writeLong(3L); @@ -392,17 +406,6 @@ public class SavedBiomeStoreWriteTest { data.writeInt(1); data.write(new byte[]{0, 0, 1, (byte) 0x80, 3, 0}); data.write(new byte[256]); - ByteArrayOutputStream body = new ByteArrayOutputStream(); - DataOutputStream coordinates = new DataOutputStream(body); - coordinates.writeInt(0); - coordinates.writeInt(0); - coordinates.writeInt(raw.size()); - Deflater deflater = new Deflater(Deflater.BEST_SPEED); - try (DeflaterOutputStream compressed = new DeflaterOutputStream(coordinates, deflater)) { - raw.writeTo(compressed); - } finally { - deflater.end(); - } ByteArrayOutputStream region = new ByteArrayOutputStream(); DataOutputStream output = new DataOutputStream(region); output.writeInt(0x4942494F); @@ -410,9 +413,22 @@ public class SavedBiomeStoreWriteTest { output.writeInt(0); output.writeInt(0); output.writeInt(checksum(region.toByteArray())); - output.writeInt(body.size()); - output.write(body.toByteArray()); - output.writeInt(checksum(body.toByteArray())); + for (int x = 0; x < records; x++) { + ByteArrayOutputStream body = new ByteArrayOutputStream(); + DataOutputStream coordinates = new DataOutputStream(body); + coordinates.writeInt(x); + coordinates.writeInt(0); + coordinates.writeInt(raw.size()); + Deflater deflater = new Deflater(Deflater.BEST_SPEED); + try (DeflaterOutputStream compressed = new DeflaterOutputStream(coordinates, deflater)) { + raw.writeTo(compressed); + } finally { + deflater.end(); + } + output.writeInt(body.size()); + output.write(body.toByteArray()); + output.writeInt(checksum(body.toByteArray())); + } return region.toByteArray(); }