mirror of
https://github.com/VolmitSoftware/Iris.git
synced 2026-10-26 05:50:39 +00:00
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
This commit is contained in:
1 parent
0fe2363feb
commit
04f2798c06
4 files changed
+142
-68
No files matched your search
+39
-53
@@ -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<Engine, NativeStructureVolumeIndex> INDEXES =
|
||||
Collections.synchronizedMap(new WeakHashMap<>());
|
||||
|
||||
@@ -223,12 +221,9 @@ public final class NativeStructureVolumeIndex {
|
||||
RuntimeChunkKey key,
|
||||
Supplier<KList<NativeStructureVolume>> loader
|
||||
) {
|
||||
Map<RuntimeChunkKey, KList<NativeStructureVolume>> cache = volumes.stripe(key);
|
||||
synchronized (cache) {
|
||||
KList<NativeStructureVolume> hit = cache.get(key);
|
||||
if (hit != null) {
|
||||
return hit;
|
||||
}
|
||||
KList<NativeStructureVolume> hit = volumes.get(key);
|
||||
if (hit != null) {
|
||||
return hit;
|
||||
}
|
||||
|
||||
CompletableFuture<KList<NativeStructureVolume>> 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<NativeStructureVolume> published = cache.get(key);
|
||||
if (published != null) {
|
||||
future.complete(published);
|
||||
builds.remove(key, future);
|
||||
return published;
|
||||
}
|
||||
KList<NativeStructureVolume> published = volumes.get(key);
|
||||
if (published != null) {
|
||||
future.complete(published);
|
||||
builds.remove(key, future);
|
||||
return published;
|
||||
}
|
||||
|
||||
try {
|
||||
KList<NativeStructureVolume> 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<RuntimeChunkKey, KList<NativeStructureVolume>> lru(int capacity) {
|
||||
return new LinkedHashMap<>(16, 0.75F, true) {
|
||||
@Override
|
||||
protected boolean removeEldestEntry(
|
||||
Map.Entry<RuntimeChunkKey, KList<NativeStructureVolume>> eldest
|
||||
) {
|
||||
return size() > capacity;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private static final class VolumeCache {
|
||||
private final List<Map<RuntimeChunkKey, KList<NativeStructureVolume>>> stripes;
|
||||
private final int capacity;
|
||||
private final ConcurrentHashMap<RuntimeChunkKey, KList<NativeStructureVolume>> entries = new ConcurrentHashMap<>();
|
||||
private final ConcurrentLinkedQueue<RuntimeChunkKey> insertionOrder = new ConcurrentLinkedQueue<>();
|
||||
|
||||
private VolumeCache(int capacity) {
|
||||
int stripeCapacity = Math.max(1, capacity / CACHE_STRIPES);
|
||||
List<Map<RuntimeChunkKey, KList<NativeStructureVolume>>> created = new ArrayList<>(CACHE_STRIPES);
|
||||
for (int stripe = 0; stripe < CACHE_STRIPES; stripe++) {
|
||||
created.add(lru(stripeCapacity));
|
||||
}
|
||||
stripes = List.copyOf(created);
|
||||
}
|
||||
|
||||
private Map<RuntimeChunkKey, KList<NativeStructureVolume>> stripe(RuntimeChunkKey key) {
|
||||
int hash = key.hashCode();
|
||||
return stripes.get((hash ^ (hash >>> 16)) & (CACHE_STRIPES - 1));
|
||||
this.capacity = capacity;
|
||||
}
|
||||
|
||||
private KList<NativeStructureVolume> get(RuntimeChunkKey key) {
|
||||
Map<RuntimeChunkKey, KList<NativeStructureVolume>> cache = stripe(key);
|
||||
synchronized (cache) {
|
||||
return cache.get(key);
|
||||
return entries.get(key);
|
||||
}
|
||||
|
||||
private void publish(RuntimeChunkKey key, KList<NativeStructureVolume> 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<RuntimeChunkKey, KList<NativeStructureVolume>> cache : stripes) {
|
||||
synchronized (cache) {
|
||||
cache.keySet().removeIf(key -> key.runtimeId() == runtimeId);
|
||||
}
|
||||
}
|
||||
entries.keySet().removeIf(key -> key.runtimeId() == runtimeId);
|
||||
insertionOrder.removeIf(key -> key.runtimeId() == runtimeId);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Deflater> 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;
|
||||
|
||||
@@ -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}) {
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user