fix: some performance issues

Author: ishland

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

@@ -1,19 +1,17 @@
 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.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;
@@ -26,7 +24,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Queue;
 import java.util.concurrent.Executor;
 
 public class TheChunkSystem extends StatusAdvancingScheduler<ChunkPos, ChunkState, ChunkLoadingContext, NewChunkHolderVanillaInterface> {
@@ -35,8 +32,6 @@ public class TheChunkSystem extends StatusAdvancingScheduler<ChunkPos, ChunkStat
 
     private final Long2IntMap managedTickets = Long2IntMaps.synchronize(new Long2IntOpenHashMap());
     private final SchedulingManager schedulingManager;
-    private final Executor backingBackgroundExecutor = GlobalExecutors.prioritizedScheduler.executor(15);
-    private final Scheduler backgroundScheduler = Schedulers.from(this.backingBackgroundExecutor);
     private final ServerChunkLoadingManager tacs;
 
     public TheChunkSystem(ServerChunkLoadingManager tacs) {
@@ -49,12 +44,12 @@ public TheChunkSystem(ServerChunkLoadingManager tacs) {
 
     @Override
     protected Executor getBackgroundExecutor() {
-        return this.backingBackgroundExecutor;
+        return ChunkSystemExecutors.consolidatingBackgroundExecutor;
     }
 
     @Override
     protected Scheduler getSchedulerBackedByBackgroundExecutor() {
-        return this.backgroundScheduler;
+        return ChunkSystemExecutors.consolidatingBackgroundScheduler;
     }
 
     @Override

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

@@ -0,0 +1,59 @@
+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 {
+
+    private 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);
+            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();
+            }
+        });
+    }
+
+}