Merge branch 'feature/fully-threaded-sched' into dev/1.21.5

Author: ishland

FlowSched

@@ -6,6 +6,8 @@
 import net.minecraft.server.world.ChunkLevels;
 import net.minecraft.util.math.ChunkPos;
 
+import java.util.ArrayDeque;
+import java.util.Queue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executor;
@@ -18,11 +20,21 @@ public class SchedulingManager {
 
     public static final int MAX_LEVEL = ChunkLevels.INACCESSIBLE + 1;
     private final ConcurrentMap<Long, FreeableTaskList> pos2Tasks = new ConcurrentHashMap<>();
-    private final Long2IntOpenHashMap prioritiesFromLevel = new Long2IntOpenHashMap();
+    private final Long2IntOpenHashMap prioritiesFromLevel = new Long2IntOpenHashMap() {
+        @Override
+        protected void rehash(int newN) {
+            if (n < newN) {
+                super.rehash(newN);
+            }
+        }
+    };
     private final StampedLock prioritiesLock = new StampedLock();
     private final int id = COUNTER.getAndIncrement();
     private volatile ChunkPos currentSyncLoad = null;
 
+    private boolean consolidatingLevelUpdates = false;
+    private Queue<Runnable> consolidatedLevelUpdates = new ArrayDeque<>();
+
     private final Executor executor;
 
     {
@@ -73,21 +85,52 @@ public Executor positionedExecutor(long pos) {
 
     public void updatePriorityFromLevel(long pos, int level) {
         this.executor.execute(() -> {
-            if (this.getPriorityFromMap(pos) == level) return;
-            final long stamp = this.prioritiesLock.writeLock();
-            try {
-                if (level < MAX_LEVEL) {
-                    this.prioritiesFromLevel.put(pos, level);
-                } else {
-                    this.prioritiesFromLevel.remove(pos);
-                }
-            } finally {
-                this.prioritiesLock.unlockWrite(stamp);
-            }
-            updatePriorityInternal(pos);
+            updatePriorityFromLevel0(pos, level);
         });
     }
 
+    private void updatePriorityFromLevel0(long pos, int level) {
+        if (this.getPriorityFromMap(pos) == level) return;
+        final long stamp = this.prioritiesLock.writeLock();
+        try {
+            if (level < MAX_LEVEL) {
+                this.prioritiesFromLevel.put(pos, level);
+            } else {
+                this.prioritiesFromLevel.remove(pos);
+            }
+        } finally {
+            this.prioritiesLock.unlockWrite(stamp);
+        }
+        updatePriorityInternal(pos);
+    }
+
+    public void updatePriorityFromLevelOnMain(long pos, int level) {
+        if (this.consolidatingLevelUpdates) {
+            this.consolidatedLevelUpdates.add(() -> updatePriorityFromLevel0(pos, level));
+        } else {
+            updatePriorityFromLevel(pos, level);
+        }
+    }
+
+    public void setConsolidatingLevelUpdates(boolean value) {
+        this.consolidatingLevelUpdates = value;
+        if (!value) {
+            if (!this.consolidatedLevelUpdates.isEmpty()) {
+                Queue<Runnable> runnables = this.consolidatedLevelUpdates;
+                this.consolidatedLevelUpdates = new ArrayDeque<>();
+                this.executor.execute(() -> {
+                    for (Runnable runnable : runnables) {
+                        try {
+                            runnable.run();
+                        } catch (Throwable t) {
+                            t.printStackTrace();
+                        }
+                    }
+                });
+            }
+        }
+    }
+
     private void updatePriorityInternal(long pos) {
         final int priority = getPriority(pos);
         final FreeableTaskList locks = this.pos2Tasks.get(pos);

c2me-base/src/main/java/com/ishland/c2me/base/common/scheduler/SchedulingManager.java

@@ -0,0 +1,31 @@
+package com.ishland.c2me.base.mixin.scheduler;
+
+import com.ishland.c2me.base.common.scheduler.IVanillaChunkManager;
+import com.ishland.c2me.base.common.scheduler.SchedulingManager;
+import com.llamalad7.mixinextras.injector.wrapoperation.Operation;
+import com.llamalad7.mixinextras.injector.wrapoperation.WrapOperation;
+import net.minecraft.server.world.ChunkLevelManager;
+import net.minecraft.server.world.ServerChunkLoadingManager;
+import net.minecraft.server.world.ServerChunkManager;
+import org.spongepowered.asm.mixin.Final;
+import org.spongepowered.asm.mixin.Mixin;
+import org.spongepowered.asm.mixin.Shadow;
+import org.spongepowered.asm.mixin.injection.At;
+
+@Mixin(ServerChunkManager.class)
+public class MixinServerChunkManager {
+
+    @Shadow @Final public ServerChunkLoadingManager chunkLoadingManager;
+
+    @WrapOperation(method = "updateChunks", at = @At(value = "INVOKE", target = "Lnet/minecraft/server/world/ChunkLevelManager;update(Lnet/minecraft/server/world/ServerChunkLoadingManager;)Z"))
+    private boolean consolidatePriorityUpdates(ChunkLevelManager instance, ServerChunkLoadingManager chunkLoadingManager, Operation<Boolean> original) {
+        SchedulingManager schedulingManager = ((IVanillaChunkManager) this.chunkLoadingManager).c2me$getSchedulingManager();
+        schedulingManager.setConsolidatingLevelUpdates(true);
+        try {
+            return original.call(instance, chunkLoadingManager);
+        } finally {
+            schedulingManager.setConsolidatingLevelUpdates(false);
+        }
+    }
+
+}

c2me-base/src/main/java/com/ishland/c2me/base/mixin/scheduler/MixinServerChunkManager.java

@@ -22,7 +22,7 @@ public class MixinThreadedAnvilChunkStorage implements IVanillaChunkManager {
 
     @Inject(method = "setLevel", at = @At("RETURN"))
     private void onUpdateLevel(long pos, int level, ChunkHolder holder, int i, CallbackInfoReturnable<ChunkHolder> cir) {
-        this.c2me$schedulingManager.updatePriorityFromLevel(pos, level);
+        this.c2me$schedulingManager.updatePriorityFromLevelOnMain(pos, level);
     }
 
 }

c2me-base/src/main/java/com/ishland/c2me/base/mixin/scheduler/MixinThreadedAnvilChunkStorage.java

@@ -58,6 +58,7 @@
     "access.fapi.IArrayBackedEvent",
     "instrumentation.MixinServerChunkManager",
     "report.MixinDedicatedServerWatchdog",
+    "scheduler.MixinServerChunkManager",
     "scheduler.MixinThreadedAnvilChunkStorage",
     "theinterface.MixinStorageIoWorker",
     "util.log4j2shutdownhookisnomore.MixinMain",

c2me-base/src/main/resources/c2me-base.mixins.json

@@ -80,13 +80,14 @@ Whether to allow POIs (Point of Interest) to be unloaded
                     """)
             .getBoolean(true, false);
 
-    public static final boolean lowMemoryMode = new ConfigSystem.ConfigAccessor()
-            .key("chunkSystem.lowMemoryMode")
+    public static final boolean useLegacyScheduling = new ConfigSystem.ConfigAccessor()
+            .key("chunkSystem.useLegacyScheduling")
             .comment("""
-                    Whether to enable low memory mode
+                    Whether to use legacy scheduling for neighbor chunks
                     
-                    This will attempt to aggressively unload unneeded chunks, saving memory at the cost of additional
-                    overhead when generating new chunks.
+                    Enabling this restores the behavior of always loading in neighbor chunks when a chunk is loaded.
+                    
+                    This is currently deprecated and will be removed in the future.
                     """)
             .getBoolean(false, false);
 

c2me-rewrites-chunk-system/src/main/java/com/ishland/c2me/rewrites/chunksystem/common/Config.java

@@ -182,7 +182,7 @@ public void updateDeferredStatus(NewChunkStatus status) {
     }
 
     private void triggerDeferredLoad(NewChunkStatus requestedStatus) {
-        if (!Config.lowMemoryMode) return;
+        if (Config.useLegacyScheduling) return;
         synchronized (this) {
             if (this.loadedDeferredStatus != null && this.loadedDeferredStatus.ordinal() >= requestedStatus.ordinal()) {
                 return; // nothing to do

c2me-rewrites-chunk-system/src/main/java/com/ishland/c2me/rewrites/chunksystem/common/NewChunkHolderVanillaInterface.java

@@ -1,74 +1,55 @@
 package com.ishland.c2me.rewrites.chunksystem.common;
 
-import com.ishland.c2me.base.common.GlobalExecutors;
 import com.ishland.c2me.base.common.scheduler.IVanillaChunkManager;
 import com.ishland.c2me.base.common.scheduler.SchedulingManager;
 import com.ishland.c2me.base.mixin.access.IThreadedAnvilChunkStorage;
 import com.ishland.c2me.base.mixin.access.IVersionedChunkStorage;
-import com.ishland.flowsched.scheduler.DaemonizedStatusAdvancingScheduler;
+import com.ishland.c2me.rewrites.chunksystem.common.structs.ChunkSystemExecutors;
 import com.ishland.flowsched.scheduler.ExceptionHandlingAction;
 import com.ishland.flowsched.scheduler.ItemHolder;
 import com.ishland.flowsched.scheduler.ItemStatus;
 import com.ishland.flowsched.scheduler.KeyStatusPair;
+import com.ishland.flowsched.scheduler.StatusAdvancingScheduler;
 import com.ishland.flowsched.util.Assertions;
-import io.netty.util.internal.PlatformDependent;
 import io.reactivex.rxjava3.core.Scheduler;
-import io.reactivex.rxjava3.schedulers.Schedulers;
 import it.unimi.dsi.fastutil.longs.Long2IntMap;
 import it.unimi.dsi.fastutil.longs.Long2IntMaps;
 import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
 import net.minecraft.server.MinecraftServer;
 import net.minecraft.server.WorldGenerationProgressListener;
 import net.minecraft.server.world.ChunkHolder;
 import net.minecraft.server.world.ServerChunkLoadingManager;
-import net.minecraft.util.Util;
 import net.minecraft.util.collection.BoundedRegionArray;
-import net.minecraft.util.crash.CrashException;
-import net.minecraft.util.crash.ReportType;
 import net.minecraft.util.math.ChunkPos;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.nio.file.Path;
-import java.util.Queue;
 import java.util.concurrent.Executor;
-import java.util.concurrent.ThreadFactory;
 
-public class TheChunkSystem extends DaemonizedStatusAdvancingScheduler<ChunkPos, ChunkState, ChunkLoadingContext, NewChunkHolderVanillaInterface> {
+public class TheChunkSystem extends StatusAdvancingScheduler<ChunkPos, ChunkState, ChunkLoadingContext, NewChunkHolderVanillaInterface> {
 
     private final Logger LOGGER;
 
     private final Long2IntMap managedTickets = Long2IntMaps.synchronize(new Long2IntOpenHashMap());
     private final SchedulingManager schedulingManager;
-    private final Executor backingBackgroundExecutor = GlobalExecutors.prioritizedScheduler.executor(15);
-    private Queue<Runnable> backgroundTaskQueue = PlatformDependent.newSpscQueue();
-    private final Executor backgroundExecutor = command -> {
-        if (Thread.currentThread() != this.thread) {
-            command.run();
-        } else {
-            backgroundTaskQueue.add(command);
-        }
-    };
-    private final Scheduler backgroundScheduler = Schedulers.from(this.backgroundExecutor);
     private final ServerChunkLoadingManager tacs;
 
-    public TheChunkSystem(ThreadFactory threadFactory, ServerChunkLoadingManager tacs) {
-        super(threadFactory, TheSpeedyObjectFactory.INSTANCE);
+    public TheChunkSystem(ServerChunkLoadingManager tacs) {
+        super(TheSpeedyObjectFactory.INSTANCE);
         this.tacs = tacs;
         this.schedulingManager =  ((IVanillaChunkManager) tacs).c2me$getSchedulingManager();
         this.LOGGER = LoggerFactory.getLogger("Chunk System of %s".formatted(((IThreadedAnvilChunkStorage) tacs).getWorld().getRegistryKey().getValue()));
         managedTickets.defaultReturnValue(NewChunkStatus.vanillaLevelToStatus.length - 1);
-        this.thread.start();
     }
 
     @Override
     protected Executor getBackgroundExecutor() {
-        return this.backgroundExecutor;
+        return ChunkSystemExecutors.consolidatingBackgroundExecutor;
     }
 
     @Override
     protected Scheduler getSchedulerBackedByBackgroundExecutor() {
-        return this.backgroundScheduler;
+        return ChunkSystemExecutors.consolidatingBackgroundScheduler;
     }
 
     @Override
@@ -149,7 +130,7 @@ protected void onItemDowngrade(ItemHolder<ChunkPos, ChunkState, ChunkLoadingCont
     }
 
     public ChunkHolder vanillaIf$setLevel(long pos, int level) {
-        Assertions.assertTrue(!Thread.holdsLock(this.managedTickets));
+        assert !Thread.holdsLock(this.managedTickets);
         synchronized (this.managedTickets) {
             final int oldLevel = this.managedTickets.put(pos, level);
             NewChunkStatus oldStatus = c2me$getDeferredStatusFromVanillaLevel(oldLevel);
@@ -170,7 +151,7 @@ protected void onItemDowngrade(ItemHolder<ChunkPos, ChunkState, ChunkLoadingCont
                 Assertions.assertTrue(holder != null, "Holder should be managed by the vanilla interface");
                 assert holder != null;
                 vanillaHolder = holder.getUserData().get();
-                if (Config.lowMemoryMode) {
+                if (!Config.useLegacyScheduling) {
                     vanillaHolder.updateDeferredStatus(NewChunkStatus.fromVanillaLevel(level));
                 }
 
@@ -187,7 +168,7 @@ protected void onItemDowngrade(ItemHolder<ChunkPos, ChunkState, ChunkLoadingCont
                 } else {
                     vanillaHolder = null;
                 }
-                if (Config.lowMemoryMode && vanillaHolder != null) {
+                if (!Config.useLegacyScheduling && vanillaHolder != null) {
                     vanillaHolder.updateDeferredStatus(NewChunkStatus.fromVanillaLevel(level));
                 }
                 if (newStatus != this.getUnloadedStatus() && vanillaHolder != null) {
@@ -200,7 +181,7 @@ protected void onItemDowngrade(ItemHolder<ChunkPos, ChunkState, ChunkLoadingCont
 
     private static NewChunkStatus c2me$getDeferredStatusFromVanillaLevel(int level) {
         NewChunkStatus status = NewChunkStatus.fromVanillaLevel(level);
-        if (Config.lowMemoryMode) {
+        if (!Config.useLegacyScheduling) {
             if (status == NewChunkStatus.NEW) {
                 return status;
             } else if (status.ordinal() < NewChunkStatus.SERVER_ACCESSIBLE.ordinal()) {
@@ -216,24 +197,4 @@ protected void onItemDowngrade(ItemHolder<ChunkPos, ChunkState, ChunkLoadingCont
     public int vanillaIf$getManagedLevel(long pos) {
         return this.managedTickets.get(pos);
     }
-
-    @Override
-    public boolean tick() {
-        boolean tick = super.tick();
-        if (!this.backgroundTaskQueue.isEmpty()) {
-            Queue<Runnable> queue = this.backgroundTaskQueue;
-            this.backgroundTaskQueue = PlatformDependent.newSpscQueue();
-            this.backingBackgroundExecutor.execute(() -> {
-                Runnable runnable;
-                while ((runnable = queue.poll()) != null) {
-                    try {
-                        runnable.run();
-                    } catch (Throwable t) {
-                        t.printStackTrace();
-                    }
-                }
-            });
-        }
-        return tick;
-    }
 }

c2me-rewrites-chunk-system/src/main/java/com/ishland/c2me/rewrites/chunksystem/common/TheChunkSystem.java

@@ -0,0 +1,60 @@
+package com.ishland.c2me.rewrites.chunksystem.common.structs;
+
+import com.ishland.c2me.base.common.GlobalExecutors;
+import io.reactivex.rxjava3.core.Scheduler;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+
+import java.util.ArrayDeque;
+import java.util.Queue;
+import java.util.concurrent.Executor;
+
+public class ChunkSystemExecutors {
+
+    public static final ThreadLocal<Queue<Runnable>> CONSOLIDATING_QUEUE = new ThreadLocal<>();
+
+    public static final Executor backingBackgroundExecutor = GlobalExecutors.prioritizedScheduler.executor(15);
+    public static final Scheduler backgroundScheduler = Schedulers.from(backingBackgroundExecutor);
+    public static final Executor consolidatingBackgroundExecutor = command -> {
+        Queue<Runnable> runnables = CONSOLIDATING_QUEUE.get();
+        if (runnables == null) { // first entry
+            consolidatingRoot(command);
+            CONSOLIDATING_QUEUE.remove(); // get() initielizes threadlocal
+            return;
+        }
+        runnables.add(command);
+    };
+    public static final Scheduler consolidatingBackgroundScheduler = Schedulers.from(consolidatingBackgroundExecutor);
+
+    private static void consolidatingRoot(Runnable initialCommand) {
+        backingBackgroundExecutor.execute(() -> {
+            Queue<Runnable> runnables = CONSOLIDATING_QUEUE.get();
+            if (runnables != null) {
+                new Throwable("CONSOLIDATING_QUEUE leak").printStackTrace();
+                try {
+                    initialCommand.run();
+                } catch (Throwable t) {
+                    t.printStackTrace();
+                }
+                return;
+            }
+
+            CONSOLIDATING_QUEUE.set(runnables = new ArrayDeque<>());
+            runnables.add(initialCommand);
+            try {
+                while (!runnables.isEmpty()) {
+                    try {
+                        runnables.remove().run();
+                    } catch (Throwable t) {
+                        t.printStackTrace();
+                    }
+                }
+            } finally {
+                if (!runnables.isEmpty()) {
+                    new Throwable("runnable leak").printStackTrace();
+                }
+                CONSOLIDATING_QUEUE.remove();
+            }
+        });
+    }
+
+}