Add periodic reconciliation sweep (Phase 7)
Adds a rotating, non-overlapping-batch ReconciliationScheduler (common/, loader-agnostic) and wires it into the 1.12.2 leaf: a new tick timer (reconciliationIntervalTicks, mirroring the existing delta-flush timer) periodically re-reads and resends a bounded slice of currently-loaded chunks, catching mutations that never fire a block event (world-gen, other mods writing blocks directly, /fill, etc.). ChunkAdapter gains loadedChunkKeys(); Forge1122ChunkAdapter implements it via ChunkProviderServer.getLoadedChunks(). MapperConfig gains reconciliationChunksPerSweep to bound sweep cost. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015tKdPZt78zbPUZMXWzKEKt
This commit is contained in:
@@ -21,6 +21,14 @@ public interface ChunkAdapter {
|
|||||||
/** Register the loader-specific hooks (block place/break, chunk load/unload) that feed the dirty buffer. */
|
/** Register the loader-specific hooks (block place/break, chunk load/unload) that feed the dirty buffer. */
|
||||||
void registerEventHooks(DeltaSink sink);
|
void registerEventHooks(DeltaSink sink);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Currently-loaded chunk coordinates, packed as {@code (chunkX << 32) | (chunkZ & 0xFFFFFFFFL)}
|
||||||
|
* — the same packing the leaf modules' own dirty-chunk sets already use. Feeds
|
||||||
|
* {@link ReconciliationScheduler}, which picks a bounded rotating slice of this set to
|
||||||
|
* re-read and resend each periodic sweep (see the plan's "hybrid" change-detection decision).
|
||||||
|
*/
|
||||||
|
java.util.List<Long> loadedChunkKeys();
|
||||||
|
|
||||||
interface DeltaSink {
|
interface DeltaSink {
|
||||||
void onDelta(DeltaEvent event);
|
void onDelta(DeltaEvent event);
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,48 @@
|
|||||||
|
package com.octoturge.mcmapper.common;
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.Collection;
|
||||||
|
import java.util.Collections;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Picks a bounded-size, rotating slice of the currently-loaded chunk set to re-read and resend
|
||||||
|
* on each periodic reconciliation sweep (see the plan's "hybrid" change-detection decision) —
|
||||||
|
* catches mutations that never fire a {@code BlockEvent} (world-gen, other mods writing blocks
|
||||||
|
* directly, {@code /fill}, piston pushes into unloaded-at-the-time chunks, etc.) without paying
|
||||||
|
* the cost of re-reading every loaded chunk on every sweep. Pure/loader-agnostic on purpose —
|
||||||
|
* enumerating "currently loaded chunks" stays in each leaf's {@link ChunkAdapter}.
|
||||||
|
*/
|
||||||
|
public class ReconciliationScheduler {
|
||||||
|
private int cursor = 0;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param loadedChunkKeys all currently-loaded chunks, packed as {@code (chunkX << 32) | (chunkZ & 0xFFFFFFFFL)}
|
||||||
|
* (same packing MCMapperMod already uses for its own dirty-chunk sets)
|
||||||
|
* @param maxPerSweep upper bound on how many chunks to return this call
|
||||||
|
* @return up to {@code maxPerSweep} keys to reconcile this sweep; the internal cursor
|
||||||
|
* advances so the next call continues where this one left off, wrapping around once
|
||||||
|
* every key has been covered. Not a strict guarantee if the loaded set changes
|
||||||
|
* between calls (chunks unloading/loading) — best-effort coverage is the point, a
|
||||||
|
* missed chunk just gets picked up on a later sweep or the next event-driven change.
|
||||||
|
*/
|
||||||
|
public List<Long> next(Collection<Long> loadedChunkKeys, int maxPerSweep) {
|
||||||
|
if (loadedChunkKeys.isEmpty() || maxPerSweep <= 0) {
|
||||||
|
return Collections.emptyList();
|
||||||
|
}
|
||||||
|
|
||||||
|
List<Long> sorted = new ArrayList<>(loadedChunkKeys);
|
||||||
|
Collections.sort(sorted);
|
||||||
|
int size = sorted.size();
|
||||||
|
|
||||||
|
// A batch never wraps mid-call — it stops at the end of the current cycle instead of
|
||||||
|
// splicing the start back in, so no key is ever reconciled twice before every other key
|
||||||
|
// has had a turn. This means a batch can come back smaller than maxPerSweep right at a
|
||||||
|
// cycle boundary; the next call resumes at index 0 for a full-size batch again.
|
||||||
|
int start = cursor % size;
|
||||||
|
int count = Math.min(maxPerSweep, size - start);
|
||||||
|
List<Long> batch = new ArrayList<>(sorted.subList(start, start + count));
|
||||||
|
cursor = (start + count) % size;
|
||||||
|
return batch;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -12,4 +12,5 @@ public class MapperConfig {
|
|||||||
public boolean playerTrackingEnabled = true;
|
public boolean playerTrackingEnabled = true;
|
||||||
public int deltaFlushIntervalTicks = 20;
|
public int deltaFlushIntervalTicks = 20;
|
||||||
public int reconciliationIntervalTicks = 20 * 60 * 5;
|
public int reconciliationIntervalTicks = 20 * 60 * 5;
|
||||||
|
public int reconciliationChunksPerSweep = 50;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,123 @@
|
|||||||
|
package com.octoturge.mcmapper.common;
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.Arrays;
|
||||||
|
import java.util.LinkedHashSet;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Set;
|
||||||
|
|
||||||
|
public class ReconciliationSchedulerTest {
|
||||||
|
private static int passed = 0;
|
||||||
|
private static int failed = 0;
|
||||||
|
|
||||||
|
public static void main(String[] args) {
|
||||||
|
test("empty loaded set returns an empty batch", ReconciliationSchedulerTest::emptySetReturnsEmpty);
|
||||||
|
test("maxPerSweep of zero (or negative) returns an empty batch", ReconciliationSchedulerTest::zeroMaxReturnsEmpty);
|
||||||
|
test("a batch is never larger than maxPerSweep", ReconciliationSchedulerTest::batchNeverExceedsMax);
|
||||||
|
test("maxPerSweep >= set size returns every key exactly once", ReconciliationSchedulerTest::coversWholeSetInOneCallWhenItFits);
|
||||||
|
test("repeated calls eventually cover every key exactly once per cycle, then wrap", ReconciliationSchedulerTest::rotatesThroughWholeSetWithoutRepeatsWithinACycle);
|
||||||
|
test("the cursor wraps around the end of the set back to the start", ReconciliationSchedulerTest::wrapsAroundTheEnd);
|
||||||
|
test("a shrinking loaded set doesn't throw or return stale keys", ReconciliationSchedulerTest::toleratesShrinkingSet);
|
||||||
|
|
||||||
|
System.out.println();
|
||||||
|
System.out.println(passed + " passed, " + failed + " failed");
|
||||||
|
if (failed > 0) {
|
||||||
|
System.exit(1);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void emptySetReturnsEmpty() {
|
||||||
|
ReconciliationScheduler scheduler = new ReconciliationScheduler();
|
||||||
|
List<Long> batch = scheduler.next(new ArrayList<>(), 5);
|
||||||
|
assertEquals(0, batch.size());
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void zeroMaxReturnsEmpty() {
|
||||||
|
ReconciliationScheduler scheduler = new ReconciliationScheduler();
|
||||||
|
Set<Long> loaded = new LinkedHashSet<>(Arrays.asList(1L, 2L, 3L));
|
||||||
|
assertEquals(0, scheduler.next(loaded, 0).size());
|
||||||
|
assertEquals(0, scheduler.next(loaded, -1).size());
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void batchNeverExceedsMax() {
|
||||||
|
ReconciliationScheduler scheduler = new ReconciliationScheduler();
|
||||||
|
Set<Long> loaded = new LinkedHashSet<>();
|
||||||
|
for (long i = 0; i < 50; i++) loaded.add(i);
|
||||||
|
for (int i = 0; i < 10; i++) {
|
||||||
|
List<Long> batch = scheduler.next(loaded, 7);
|
||||||
|
if (batch.size() > 7) {
|
||||||
|
throw new AssertionError("batch size " + batch.size() + " exceeds maxPerSweep 7");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void coversWholeSetInOneCallWhenItFits() {
|
||||||
|
ReconciliationScheduler scheduler = new ReconciliationScheduler();
|
||||||
|
Set<Long> loaded = new LinkedHashSet<>(Arrays.asList(10L, 20L, 30L));
|
||||||
|
List<Long> batch = scheduler.next(loaded, 100);
|
||||||
|
assertEquals(3, batch.size());
|
||||||
|
assertEquals(new LinkedHashSet<>(Arrays.asList(10L, 20L, 30L)), new LinkedHashSet<>(batch));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void rotatesThroughWholeSetWithoutRepeatsWithinACycle() {
|
||||||
|
ReconciliationScheduler scheduler = new ReconciliationScheduler();
|
||||||
|
Set<Long> loaded = new LinkedHashSet<>();
|
||||||
|
for (long i = 0; i < 10; i++) loaded.add(i);
|
||||||
|
|
||||||
|
Set<Long> seenThisCycle = new LinkedHashSet<>();
|
||||||
|
// 10 keys, batches of 3 -> 4 calls (3+3+3+1) to complete exactly one cycle.
|
||||||
|
for (int call = 0; call < 4; call++) {
|
||||||
|
List<Long> batch = scheduler.next(loaded, 3);
|
||||||
|
for (Long key : batch) {
|
||||||
|
if (!seenThisCycle.add(key)) {
|
||||||
|
throw new AssertionError("key " + key + " reconciled twice within a single cycle");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assertEquals(10, seenThisCycle.size());
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void wrapsAroundTheEnd() {
|
||||||
|
ReconciliationScheduler scheduler = new ReconciliationScheduler();
|
||||||
|
Set<Long> loaded = new LinkedHashSet<>(Arrays.asList(1L, 2L, 3L, 4L, 5L));
|
||||||
|
scheduler.next(loaded, 5); // consumes the whole set in one call — cursor wraps to 0
|
||||||
|
List<Long> batch = scheduler.next(loaded, 2); // should start a fresh cycle from the top
|
||||||
|
assertEquals(2, batch.size());
|
||||||
|
if (!batch.contains(1L) || !batch.contains(2L)) {
|
||||||
|
throw new AssertionError("expected the cursor to wrap back to the start of the set, got " + batch);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void toleratesShrinkingSet() {
|
||||||
|
ReconciliationScheduler scheduler = new ReconciliationScheduler();
|
||||||
|
Set<Long> loaded = new LinkedHashSet<>();
|
||||||
|
for (long i = 0; i < 20; i++) loaded.add(i);
|
||||||
|
scheduler.next(loaded, 15);
|
||||||
|
|
||||||
|
Set<Long> shrunk = new LinkedHashSet<>(Arrays.asList(0L, 1L, 2L));
|
||||||
|
List<Long> batch = scheduler.next(shrunk, 15); // must not throw despite cursor now out of range
|
||||||
|
assertEquals(3, batch.size());
|
||||||
|
for (Long key : batch) {
|
||||||
|
if (!shrunk.contains(key)) {
|
||||||
|
throw new AssertionError("batch contained a key no longer in the loaded set: " + key);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void test(String name, Runnable body) {
|
||||||
|
try {
|
||||||
|
body.run();
|
||||||
|
passed++;
|
||||||
|
System.out.println("PASS " + name);
|
||||||
|
} catch (AssertionError e) {
|
||||||
|
failed++;
|
||||||
|
System.out.println("FAIL " + name + " — " + e.getMessage());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void assertEquals(Object expected, Object actual) {
|
||||||
|
if (expected == null ? actual != null : !expected.equals(actual)) {
|
||||||
|
throw new AssertionError("expected <" + expected + "> but got <" + actual + ">");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+10
@@ -10,6 +10,7 @@ import net.minecraft.world.World;
|
|||||||
import net.minecraft.world.WorldServer;
|
import net.minecraft.world.WorldServer;
|
||||||
import net.minecraft.world.chunk.Chunk;
|
import net.minecraft.world.chunk.Chunk;
|
||||||
import net.minecraft.world.chunk.storage.ExtendedBlockStorage;
|
import net.minecraft.world.chunk.storage.ExtendedBlockStorage;
|
||||||
|
import net.minecraft.world.gen.ChunkProviderServer;
|
||||||
import net.minecraftforge.common.MinecraftForge;
|
import net.minecraftforge.common.MinecraftForge;
|
||||||
import net.minecraftforge.common.util.BlockSnapshot;
|
import net.minecraftforge.common.util.BlockSnapshot;
|
||||||
import net.minecraftforge.event.world.BlockEvent;
|
import net.minecraftforge.event.world.BlockEvent;
|
||||||
@@ -101,6 +102,15 @@ public class Forge1122ChunkAdapter implements ChunkAdapter {
|
|||||||
MinecraftForge.EVENT_BUS.register(new EventHooks(sink));
|
MinecraftForge.EVENT_BUS.register(new EventHooks(sink));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public List<Long> loadedChunkKeys() {
|
||||||
|
List<Long> keys = new ArrayList<>();
|
||||||
|
for (Chunk chunk : ((ChunkProviderServer) world.getChunkProvider()).getLoadedChunks()) {
|
||||||
|
keys.add((((long) chunk.x) << 32) | (chunk.z & 0xFFFFFFFFL));
|
||||||
|
}
|
||||||
|
return keys;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* {@code BlockEvent.BreakEvent} fires *before* the block is actually removed (it's
|
* {@code BlockEvent.BreakEvent} fires *before* the block is actually removed (it's
|
||||||
* cancellable), so reading world state synchronously inside that handler would see the
|
* cancellable), so reading world state synchronously inside that handler would see the
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package com.octoturge.mcmapper.forge1122;
|
|||||||
|
|
||||||
import com.octoturge.mcmapper.common.ChunkAdapter;
|
import com.octoturge.mcmapper.common.ChunkAdapter;
|
||||||
import com.octoturge.mcmapper.common.DefaultBackendConnection;
|
import com.octoturge.mcmapper.common.DefaultBackendConnection;
|
||||||
|
import com.octoturge.mcmapper.common.ReconciliationScheduler;
|
||||||
import com.octoturge.mcmapper.common.config.MapperConfig;
|
import com.octoturge.mcmapper.common.config.MapperConfig;
|
||||||
import com.octoturge.mcmapper.common.protocol.DeltaEvent;
|
import com.octoturge.mcmapper.common.protocol.DeltaEvent;
|
||||||
import com.octoturge.mcmapper.common.protocol.SectionData;
|
import com.octoturge.mcmapper.common.protocol.SectionData;
|
||||||
@@ -28,9 +29,11 @@ import java.util.Set;
|
|||||||
/**
|
/**
|
||||||
* Entry point for the 1.12.2 leaf — the primary/first-implemented target (Enigmatica 2).
|
* Entry point for the 1.12.2 leaf — the primary/first-implemented target (Enigmatica 2).
|
||||||
* Phase 1: connects to the backend over WS, backfills the overworld's loaded chunks, and
|
* Phase 1: connects to the backend over WS, backfills the overworld's loaded chunks, and
|
||||||
* flushes event-driven column deltas on a timer (no periodic reconciliation sweep yet — see
|
* flushes event-driven column deltas on a timer. Phase 2 adds full-section backfill/flush for
|
||||||
* plan's Phase 7). Phase 2 adds full-section backfill/flush for 3D mesh rendering. Phase 3 adds
|
* 3D mesh rendering. Phase 3 adds `/mcmapper link` and the two-way chat bridge (see LinkCommand,
|
||||||
* `/mcmapper link` and the two-way chat bridge (see LinkCommand, Forge1122ChatBridge).
|
* Forge1122ChatBridge). Phase 7 adds a periodic reconciliation sweep (see {@link
|
||||||
|
* ReconciliationScheduler}) that catches non-event mutations (world-gen, other mods, `/fill`)
|
||||||
|
* the event-driven hooks in {@link Forge1122ChunkAdapter} never see.
|
||||||
*/
|
*/
|
||||||
@Mod(modid = MCMapperMod.MOD_ID, name = "MCMapper", version = MCMapperMod.VERSION)
|
@Mod(modid = MCMapperMod.MOD_ID, name = "MCMapper", version = MCMapperMod.VERSION)
|
||||||
public class MCMapperMod {
|
public class MCMapperMod {
|
||||||
@@ -45,7 +48,9 @@ public class MCMapperMod {
|
|||||||
private Forge1122ChunkAdapter adapter;
|
private Forge1122ChunkAdapter adapter;
|
||||||
private final List<DeltaEvent> pendingDeltas = Collections.synchronizedList(new ArrayList<>());
|
private final List<DeltaEvent> pendingDeltas = Collections.synchronizedList(new ArrayList<>());
|
||||||
private final Set<Long> pendingSectionChunks = Collections.synchronizedSet(new LinkedHashSet<>());
|
private final Set<Long> pendingSectionChunks = Collections.synchronizedSet(new LinkedHashSet<>());
|
||||||
|
private final ReconciliationScheduler reconciliationScheduler = new ReconciliationScheduler();
|
||||||
private int ticksSinceFlush = 0;
|
private int ticksSinceFlush = 0;
|
||||||
|
private int ticksSinceReconciliation = 0;
|
||||||
|
|
||||||
@Mod.EventHandler
|
@Mod.EventHandler
|
||||||
public void preInit(FMLPreInitializationEvent event) {
|
public void preInit(FMLPreInitializationEvent event) {
|
||||||
@@ -62,6 +67,14 @@ public class MCMapperMod {
|
|||||||
config.deltaFlushIntervalTicks = forgeConfig.getInt("deltaFlushIntervalTicks", "network",
|
config.deltaFlushIntervalTicks = forgeConfig.getInt("deltaFlushIntervalTicks", "network",
|
||||||
config.deltaFlushIntervalTicks, 1, 20 * 60,
|
config.deltaFlushIntervalTicks, 1, 20 * 60,
|
||||||
"How often (in ticks) to batch and flush block-change deltas to the backend");
|
"How often (in ticks) to batch and flush block-change deltas to the backend");
|
||||||
|
config.reconciliationIntervalTicks = forgeConfig.getInt("reconciliationIntervalTicks", "network",
|
||||||
|
config.reconciliationIntervalTicks, 20, 20 * 60 * 60,
|
||||||
|
"How often (in ticks) to run a reconciliation sweep, re-reading and resending a " +
|
||||||
|
"rotating slice of loaded chunks to catch changes event hooks miss " +
|
||||||
|
"(world-gen, other mods, /fill, etc.)");
|
||||||
|
config.reconciliationChunksPerSweep = forgeConfig.getInt("reconciliationChunksPerSweep", "network",
|
||||||
|
config.reconciliationChunksPerSweep, 1, 5000,
|
||||||
|
"Max chunks to re-read and resend per reconciliation sweep");
|
||||||
if (forgeConfig.hasChanged()) forgeConfig.save();
|
if (forgeConfig.hasChanged()) forgeConfig.save();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -107,9 +120,32 @@ public class MCMapperMod {
|
|||||||
@SubscribeEvent
|
@SubscribeEvent
|
||||||
public void onServerTick(TickEvent.ServerTickEvent event) {
|
public void onServerTick(TickEvent.ServerTickEvent event) {
|
||||||
if (event.phase != TickEvent.Phase.END) return;
|
if (event.phase != TickEvent.Phase.END) return;
|
||||||
if (++ticksSinceFlush < config.deltaFlushIntervalTicks) return;
|
|
||||||
ticksSinceFlush = 0;
|
if (++ticksSinceFlush >= config.deltaFlushIntervalTicks) {
|
||||||
flush();
|
ticksSinceFlush = 0;
|
||||||
|
flush();
|
||||||
|
}
|
||||||
|
|
||||||
|
if (++ticksSinceReconciliation >= config.reconciliationIntervalTicks) {
|
||||||
|
ticksSinceReconciliation = 0;
|
||||||
|
reconcile();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void reconcile() {
|
||||||
|
List<Long> chunkKeys = reconciliationScheduler.next(adapter.loadedChunkKeys(),
|
||||||
|
config.reconciliationChunksPerSweep);
|
||||||
|
if (chunkKeys.isEmpty()) return;
|
||||||
|
|
||||||
|
String dimensionId = adapter.getDimensionId();
|
||||||
|
for (long key : chunkKeys) {
|
||||||
|
int chunkX = (int) (key >> 32);
|
||||||
|
int chunkZ = (int) key;
|
||||||
|
List<DeltaEvent> columns = adapter.readChunk(dimensionId, chunkX, chunkZ);
|
||||||
|
if (!columns.isEmpty()) connection.sendDeltas(columns);
|
||||||
|
List<SectionData> sections = adapter.readSections(chunkX, chunkZ);
|
||||||
|
if (!sections.isEmpty()) connection.sendSections(dimensionId, chunkX, chunkZ, sections);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private void flush() {
|
private void flush() {
|
||||||
|
|||||||
Reference in New Issue
Block a user