mirrors/Iris
1
0
Fork 0
mirror of https://github.com/VolmitSoftware/Iris.git synced 2026-10-19 23:14:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Retire consumed source plans

Browse Source 
- Count the destinations in reach that asked for each source
- Drop the entry once all of them did, so only the frontier stays cached
This commit is contained in:
Brian Neumann-Fopiano
2026-09-28 11:50:13 -04:00
parent 8878e9aa5f
commit 14f3a263df
4 changed files with 133 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/main/java/art/arcane/iris/generation/mantle/MantleObjectComponent.java
+5 -2
View File
@@ -140,12 +140,14 @@ public class MantleObjectComponent extends IrisMantleComponent {
@Override
public void generateLayer(MantleWriter writer, int x, int z, ChunkContext context) {
ObjectDestinationTransaction transaction = new ObjectDestinationTransaction(writer, x, z);
int radius = getRadius();
int sourceChunkRadius = sourceChunkRadius(radius);
replaySourceChunks(
x,
z,
getRadius(),
radius,
(sourceX, sourceZ) -> {
ObjectSourcePlan plan = sourcePlans.acquire(sourceX, sourceZ, x, z,
ObjectSourcePlan plan = sourcePlans.acquire(sourceX, sourceZ, x, z, sourceChunkRadius,
() -> buildSourcePlan(writer, sourceX, sourceZ, context));
if (plan != null) {
transaction.apply(plan);
@@ -169,6 +171,7 @@ public class MantleObjectComponent extends IrisMantleComponent {
+ " waits=" + stats.waits()
+ " replays=" + stats.replays()
+ " drained=" + stats.drained()
+ " retired=" + stats.retired()
+ " retained=" + stats.retained()
+ " retainedMiB=" + (stats.retainedBytes() >> 20)
+ " budgetMiB=" + (stats.budgetBytes() >> 20));
core/src/main/java/art/arcane/iris/generation/mantle/ObjectSourcePlanCache.java
+67 -15
View File
@@ -23,6 +23,7 @@ final class ObjectSourcePlanCache {
private final LongAdder waits = new LongAdder();
private final LongAdder replays = new LongAdder();
private final LongAdder drained = new LongAdder();
private final LongAdder retired = new LongAdder();
private volatile State state;
ObjectSourcePlanCache() {
@@ -44,18 +45,31 @@ final class ObjectSourcePlanCache {
/**
* Returns the plan whose mutations the destination still has to replay, or null when the source writes nothing
* there. Each destination consumes a retained plan once; a plan whose destinations were all consumed keeps only
* its destination index, so a repeated consumption rebuilds it instead of trusting released state.
* its destination index, so a repeated consumption rebuilds it instead of trusting released state. The entry is
* dropped once every destination within the source chunk radius has asked for it.
*/
ObjectSourcePlan acquire(int sourceChunkX, int sourceChunkZ, int destinationChunkX, int destinationChunkZ,
Supplier<ObjectSourcePlan> builder) {
int sourceChunkRadius, Supplier<ObjectSourcePlan> builder) {
Objects.requireNonNull(builder, "Source plan builder");
lookups.increment();
State current = state;
long source = CacheKey.mix(key(sourceChunkX, sourceChunkZ));
Entry entry = current.plans.getIfPresent(source);
if (entry == null) {
entry = load(current, source, builder);
entry = load(current, source, sourceChunkRadius, builder);
}
boolean lastVisitor = entry.visitors.visit(destinationChunkX - sourceChunkX, destinationChunkZ - sourceChunkZ);
ObjectSourcePlan plan = claim(current, source, entry, destinationChunkX, destinationChunkZ, builder);
if (lastVisitor) {
retired.increment();
Visitors visitors = entry.visitors;
current.plans.asMap().computeIfPresent(source, (Long key, Entry mapped) -> mapped.visitors == visitors ? null : mapped);
}
return plan;
}
private ObjectSourcePlan claim(State current, long source, Entry entry, int destinationChunkX, int destinationChunkZ,
Supplier<ObjectSourcePlan> builder) {
int slot = entry.slot(destinationChunkX, destinationChunkZ);
if (slot < 0) {
return null;
@@ -74,7 +88,7 @@ final class ObjectSourcePlanCache {
private ObjectSourcePlan replay(State current, long source, Entry consumed, int destinationChunkX, int destinationChunkZ,
Supplier<ObjectSourcePlan> builder) {
Entry rebuilt = new Entry(build(builder));
Entry rebuilt = new Entry(build(builder), consumed.visitors);
int slot = rebuilt.slot(destinationChunkX, destinationChunkZ);
Claim claim = slot < 0 ? null : rebuilt.claim(slot);
if (claim == null) {
@@ -86,7 +100,7 @@ final class ObjectSourcePlanCache {
return claim.plan();
}
private Entry load(State current, long source, Supplier<ObjectSourcePlan> builder) {
private Entry load(State current, long source, int sourceChunkRadius, Supplier<ObjectSourcePlan> builder) {
Pending created = new Pending(Thread.currentThread());
Pending pending = current.pending.putIfAbsent(source, created);
if (pending != null) {
@@ -96,7 +110,7 @@ final class ObjectSourcePlanCache {
try {
Entry entry = current.plans.getIfPresent(source);
if (entry == null) {
entry = new Entry(build(builder));
entry = new Entry(build(builder), new Visitors(sourceChunkRadius));
if (entry.weight() <= maximumRetainedBytes) {
current.plans.put(source, entry);
}
@@ -133,11 +147,11 @@ final class ObjectSourcePlanCache {
long retainedBytes = current.plans.policy().eviction()
.map(eviction -> eviction.weightedSize().orElse(0L))
.orElse(0L);
return new Stats(lookups.sum(), builds.sum(), waits.sum(), replays.sum(), drained.sum(),
return new Stats(lookups.sum(), builds.sum(), waits.sum(), replays.sum(), drained.sum(), retired.sum(),
current.plans.estimatedSize(), retainedBytes, maximumRetainedBytes);
}
record Stats(long lookups, long builds, long waits, long replays, long drained, long retained,
record Stats(long lookups, long builds, long waits, long replays, long drained, long retired, long retained,
long retainedBytes, long budgetBytes) {
}
@@ -159,22 +173,25 @@ final class ObjectSourcePlanCache {
private static final class Entry {
private final ObjectSourcePlan index;
private final boolean[] claimed;
private final Visitors visitors;
private final int weight;
private ObjectSourcePlan plan;
private int remaining;
private Entry(ObjectSourcePlan plan) {
private Entry(ObjectSourcePlan plan, Visitors visitors) {
this.index = plan.isEmpty() ? ObjectSourcePlan.EMPTY : plan;
this.plan = plan.isEmpty() ? null : plan;
this.claimed = new boolean[plan.destinationCount()];
this.visitors = visitors;
this.remaining = claimed.length;
this.weight = saturatedWeight(ENTRY_BYTES + (long) claimed.length + index.estimatedRetainedBytes());
this.weight = weigh(index, claimed, visitors);
}
private Entry(ObjectSourcePlan index, boolean[] claimed) {
private Entry(ObjectSourcePlan index, boolean[] claimed, Visitors visitors) {
this.index = index;
this.claimed = claimed;
this.weight = saturatedWeight(ENTRY_BYTES + (long) claimed.length + index.estimatedRetainedBytes());
this.visitors = visitors;
this.weight = weigh(index, claimed, visitors);
}
private int slot(int destinationChunkX, int destinationChunkZ) {
@@ -195,15 +212,50 @@ final class ObjectSourcePlanCache {
}
private Entry released() {
return new Entry(index.destinationIndex(), claimed);
return new Entry(index.destinationIndex(), claimed, visitors);
}
private int weight() {
return weight;
}
private static int saturatedWeight(long weight) {
return (int) Math.min(Integer.MAX_VALUE, weight);
private static int weigh(ObjectSourcePlan index, boolean[] claimed, Visitors visitors) {
return (int) Math.min(Integer.MAX_VALUE,
ENTRY_BYTES + (long) claimed.length + visitors.retainedBytes() + index.estimatedRetainedBytes());
}
}
private static final class Visitors {
private final int radius;
private final int side;
private final long[] seen;
private int remaining;
private Visitors(int radius) {
if (radius < 0) {
throw new IllegalArgumentException("Source chunk radius must not be negative");
}
this.radius = radius;
this.side = 2 * radius + 1;
this.seen = new long[(side * side + 63) >>> 6];
this.remaining = side * side;
}
private synchronized boolean visit(int offsetX, int offsetZ) {
if (Math.abs(offsetX) > radius || Math.abs(offsetZ) > radius) {
return false;
}
int bit = (offsetX + radius) * side + offsetZ + radius;
long mask = 1L << (bit & 63);
if ((seen[bit >>> 6] & mask) != 0L) {
return false;
}
seen[bit >>> 6] |= mask;
return --remaining == 0;
}
private long retainedBytes() {
return 32L + 8L * seen.length;
}
}
core/src/test/java/art/arcane/iris/generation/mantle/MantleObjectComponentCollisionDeterminismTest.java
+1 -1
View File
@@ -141,7 +141,7 @@ public class MantleObjectComponentCollisionDeterminismTest {
ObjectDestinationTransaction transaction = new ObjectDestinationTransaction(writer, destination, 0);
MantleObjectComponent.replaySourceChunks(destination, 0, RADIUS, (sourceX, sourceZ) -> {
ObjectSourcePlan plan = cache.acquire(sourceX, sourceZ, destination, 0,
() -> buildSourcePlan(writer, sourceX, sourceZ, scenario));
MantleObjectComponent.sourceChunkRadius(RADIUS), () -> buildSourcePlan(writer, sourceX, sourceZ, scenario));
if (plan != null) {
transaction.apply(plan);
}
core/src/test/java/art/arcane/iris/generation/mantle/ObjectSourcePlanCacheTest.java
+60 -27
View File
@@ -30,14 +30,14 @@ public class ObjectSourcePlanCacheTest {
if ((x == 0 || x == 1) && z == 0) {
continue;
}
assertNull(cache.acquire(0, 0, x, z, () -> {
assertNull(cache.acquire(0, 0, x, z, 7, () -> {
builds.incrementAndGet();
return plan;
}));
}
}
assertEquals(1, builds.get());
assertSame(plan, cache.acquire(0, 0, 1, 0, () -> {
assertSame(plan, cache.acquire(0, 0, 1, 0, 7, () -> {
builds.incrementAndGet();
return plan;
}));
@@ -54,29 +54,30 @@ public class ObjectSourcePlanCacheTest {
AtomicInteger builds = new AtomicInteger();
ObjectSourcePlan plan = planTouching(new int[]{0, 0}, new int[]{1, 0});
ObjectSourcePlan rebuilt = planTouching(new int[]{0, 0}, new int[]{1, 0});
assertSame(plan, cache.acquire(0, 0, 0, 0, () -> {
assertSame(plan, cache.acquire(0, 0, 0, 0, 7, () -> {
builds.incrementAndGet();
return plan;
}));
assertSame(plan, cache.acquire(0, 0, 1, 0, () -> {
long retainedBeforeDrain = cache.stats().retainedBytes();
assertSame(plan, cache.acquire(0, 0, 1, 0, 7, () -> {
builds.incrementAndGet();
return rebuilt;
}));
assertEquals(1, builds.get());
assertEquals(1L, cache.stats().drained());
assertTrue(cache.stats().retainedBytes() < plan.estimatedRetainedBytes());
assertTrue(cache.stats().retainedBytes() < retainedBeforeDrain);
assertNull(cache.acquire(0, 0, 2, 0, () -> {
assertNull(cache.acquire(0, 0, 2, 0, 7, () -> {
builds.incrementAndGet();
return rebuilt;
}));
assertSame(rebuilt, cache.acquire(0, 0, 0, 0, () -> {
assertSame(rebuilt, cache.acquire(0, 0, 0, 0, 7, () -> {
builds.incrementAndGet();
return rebuilt;
}));
assertEquals(2, builds.get());
assertEquals(1L, cache.stats().replays());
assertSame(rebuilt, cache.acquire(0, 0, 1, 0, () -> {
assertSame(rebuilt, cache.acquire(0, 0, 1, 0, 7, () -> {
builds.incrementAndGet();
return plan;
}));
@@ -84,13 +85,45 @@ public class ObjectSourcePlanCacheTest {
assertEquals(2L, cache.stats().drained());
}
@Test
public void sourcesRetireOnceEveryDestinationInReachAskedForThem() {
ObjectSourcePlanCache cache = new ObjectSourcePlanCache(1024L * 1024L);
AtomicInteger builds = new AtomicInteger();
ObjectSourcePlan plan = planTouching(new int[]{0, 0}, new int[]{1, 0});
int returned = 0;
int visits = 0;
for (int x = 7; x >= -7; x--) {
for (int z = -7; z <= 7; z++) {
if (cache.acquire(0, 0, x, z, 7, () -> {
builds.incrementAndGet();
return plan;
}) != null) {
returned++;
}
if (++visits < 225) {
assertEquals(1L, cache.estimatedSize());
}
}
}
assertEquals(2, returned);
assertEquals(1, builds.get());
assertEquals(0L, cache.estimatedSize());
assertEquals(1L, cache.stats().retired());
assertSame(plan, cache.acquire(0, 0, 1, 0, 7, () -> {
builds.incrementAndGet();
return plan;
}));
assertEquals(2, builds.get());
}
@Test
public void emptyPlansAreRetainedAsIndexesAndNeverRebuilt() {
ObjectSourcePlanCache cache = new ObjectSourcePlanCache(1024L * 1024L);
AtomicInteger builds = new AtomicInteger();
for (int sweep = 0; sweep < 3; sweep++) {
for (int chunk = 0; chunk < 1000; chunk++) {
assertNull(cache.acquire(chunk, 0, chunk, 0, () -> {
assertNull(cache.acquire(chunk, 0, chunk, 0, 7, () -> {
builds.incrementAndGet();
return new ObjectSourcePlan(List.of());
}));
@@ -113,13 +146,13 @@ public class ObjectSourcePlanCacheTest {
try {
assertEquals(Long.hashCode(CacheKey.mix(CacheKey.key(32113, 37))),
Long.hashCode(CacheKey.mix(CacheKey.key(32667, 37))));
Future<ObjectSourcePlan> firstResult = executor.submit(() -> cache.acquire(32113, 37, 32113, 37, () -> {
Future<ObjectSourcePlan> firstResult = executor.submit(() -> cache.acquire(32113, 37, 32113, 37, 7, () -> {
firstEntered.countDown();
await(releaseFirst);
return first;
}));
assertTrue(firstEntered.await(5L, TimeUnit.SECONDS));
Future<ObjectSourcePlan> secondResult = executor.submit(() -> cache.acquire(32667, 37, 32667, 37, () -> second));
Future<ObjectSourcePlan> secondResult = executor.submit(() -> cache.acquire(32667, 37, 32667, 37, 7, () -> second));
assertSame(second, secondResult.get(3L, TimeUnit.SECONDS));
releaseFirst.countDown();
@@ -149,7 +182,7 @@ public class ObjectSourcePlanCacheTest {
futures.add(executor.submit(() -> {
ready.countDown();
assertTrue(start.await(5L, TimeUnit.SECONDS));
return cache.acquire(4, -7, destinationX, -7, () -> {
return cache.acquire(4, -7, destinationX, -7, 7, () -> {
builds.incrementAndGet();
return expected;
});
@@ -174,12 +207,12 @@ public class ObjectSourcePlanCacheTest {
ObjectSourcePlanCache cache = new ObjectSourcePlanCache(large.estimatedRetainedBytes());
AtomicInteger builds = new AtomicInteger();
assertSame(large, cache.acquire(0, 0, 0, 0, () -> {
assertSame(large, cache.acquire(0, 0, 0, 0, 7, () -> {
builds.incrementAndGet();
return large;
}));
assertEquals(0L, cache.estimatedSize());
assertSame(large, cache.acquire(0, 0, 1, 0, () -> {
assertSame(large, cache.acquire(0, 0, 1, 0, 7, () -> {
builds.incrementAndGet();
return large;
}));
@@ -190,10 +223,10 @@ public class ObjectSourcePlanCacheTest {
public void cacheEvictsByRetainedBytesAndCanBeCleared() {
ObjectSourcePlan first = planTouching(new int[]{0, 0}, new int[]{9, 9});
ObjectSourcePlan second = planTouching(new int[]{1, 0}, new int[]{9, 9});
ObjectSourcePlanCache cache = new ObjectSourcePlanCache(first.estimatedRetainedBytes() + 128L);
ObjectSourcePlanCache cache = new ObjectSourcePlanCache(first.estimatedRetainedBytes() + 256L);
cache.acquire(0, 0, 0, 0, () -> first);
cache.acquire(1, 0, 1, 0, () -> second);
cache.acquire(0, 0, 0, 0, 7, () -> first);
cache.acquire(1, 0, 1, 0, 7, () -> second);
assertEquals(1L, cache.estimatedSize());
cache.clear();
@@ -213,7 +246,7 @@ public class ObjectSourcePlanCacheTest {
CountDownLatch entered = new CountDownLatch(1);
CountDownLatch release = new CountDownLatch(1);
try (ExecutorService executor = Executors.newSingleThreadExecutor()) {
Future<ObjectSourcePlan> first = executor.submit(() -> cache.acquire(7, 8, 7, 8, () -> {
Future<ObjectSourcePlan> first = executor.submit(() -> cache.acquire(7, 8, 7, 8, 7, () -> {
entered.countDown();
await(release);
return old;
@@ -221,12 +254,12 @@ public class ObjectSourcePlanCacheTest {
try {
assertTrue(entered.await(5L, TimeUnit.SECONDS));
cache.clear();
assertSame(replacement, cache.acquire(7, 8, 7, 8, () -> replacement));
assertSame(replacement, cache.acquire(7, 8, 7, 8, 7, () -> replacement));
} finally {
release.countDown();
}
assertSame(old, first.get(5L, TimeUnit.SECONDS));
assertSame(old, cache.acquire(7, 8, 7, 8, () -> old));
assertSame(old, cache.acquire(7, 8, 7, 8, 7, () -> old));
}
}
@@ -239,7 +272,7 @@ public class ObjectSourcePlanCacheTest {
CountDownLatch release = new CountDownLatch(1);
try (ExecutorService owner = Executors.newSingleThreadExecutor();
ForkJoinPool pool = new ForkJoinPool(1)) {
Future<ObjectSourcePlan> first = owner.submit(() -> cache.acquire(7, 8, 7, 8, () -> {
Future<ObjectSourcePlan> first = owner.submit(() -> cache.acquire(7, 8, 7, 8, 7, () -> {
entered.countDown();
await(release);
return expected;
@@ -248,7 +281,7 @@ public class ObjectSourcePlanCacheTest {
assertTrue(entered.await(5L, TimeUnit.SECONDS));
Future<ObjectSourcePlan> second = pool.submit(() -> {
waiting.countDown();
return cache.acquire(7, 8, 8, 8, () -> { throw new AssertionError("Duplicate build"); });
return cache.acquire(7, 8, 8, 8, 7, () -> { throw new AssertionError("Duplicate build"); });
});
assertTrue(waiting.await(5L, TimeUnit.SECONDS));
pool.submit(release::countDown).get(5L, TimeUnit.SECONDS);
@@ -265,12 +298,12 @@ public class ObjectSourcePlanCacheTest {
ObjectSourcePlanCache cache = new ObjectSourcePlanCache(4096L);
IllegalArgumentException failure = new IllegalArgumentException("Build failed");
assertSame(failure, assertThrows(IllegalArgumentException.class,
() -> cache.acquire(7, 8, 7, 8, () -> { throw failure; })));
assertThrows(NullPointerException.class, () -> cache.acquire(7, 8, 7, 8, () -> null));
assertThrows(IllegalStateException.class, () -> cache.acquire(7, 8, 7, 8,
() -> cache.acquire(7, 8, 7, 8, () -> planTouching(new int[]{7, 8}))));
() -> cache.acquire(7, 8, 7, 8, 7, () -> { throw failure; })));
assertThrows(NullPointerException.class, () -> cache.acquire(7, 8, 7, 8, 7, () -> null));
assertThrows(IllegalStateException.class, () -> cache.acquire(7, 8, 7, 8, 7,
() -> cache.acquire(7, 8, 7, 8, 7, () -> planTouching(new int[]{7, 8}))));
ObjectSourcePlan expected = planTouching(new int[]{7, 8});
assertSame(expected, cache.acquire(7, 8, 7, 8, () -> expected));
assertSame(expected, cache.acquire(7, 8, 7, 8, 7, () -> expected));
}
@Test