Merge schedulers, cleanup locking and adjust thread counts

Author: ishland

src/main/java/org/yatopiamc/c2me/common/threading/GlobalExecutors.java

@@ -0,0 +1,14 @@
+package org.yatopiamc.c2me.common.threading;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+public class GlobalExecutors {
+
+    public static final ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(
+            1,
+            new ThreadFactoryBuilder().setNameFormat("C2ME scheduler").setDaemon(true).setPriority(Thread.NORM_PRIORITY - 1).build()
+    );
+
+}

src/main/java/org/yatopiamc/c2me/common/threading/chunkio/C2MECachedRegionStorage.java

@@ -16,6 +16,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.jetbrains.annotations.Nullable;
+import org.yatopiamc.c2me.common.threading.GlobalExecutors;
 import org.yatopiamc.c2me.common.util.C2MEForkJoinWorkerThreadFactory;
 import org.yatopiamc.c2me.common.util.SneakyThrow;
 
@@ -36,14 +37,10 @@ public class C2MECachedRegionStorage extends StorageIoWorker {
     private static final CompoundTag EMPTY_VALUE = new CompoundTag();
     private static final Logger LOGGER = LogManager.getLogger();
     private static final ForkJoinPool IOExecutor = new ForkJoinPool(
-            Math.min(8, Runtime.getRuntime().availableProcessors()),
+            Math.min(6, Runtime.getRuntime().availableProcessors()),
             new C2MEForkJoinWorkerThreadFactory("C2ME chunkio io worker #%d", Thread.NORM_PRIORITY - 3),
             null, true
     );
-    private static final ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(
-            1,
-            new ThreadFactoryBuilder().setNameFormat("C2ME chunkio scheduler").setDaemon(true).setPriority(Thread.NORM_PRIORITY - 1).build()
-    );
 
     private final RegionBasedStorage storage;
     private final Cache<ChunkPos, CompoundTag> chunkCache;
@@ -73,7 +70,7 @@ private void tick() {
         long startTime = System.currentTimeMillis();
         chunkCache.cleanUp();
         if (!isClosed.get())
-            scheduler.schedule(this::tick, 1000 - (System.currentTimeMillis() - startTime), TimeUnit.MILLISECONDS);
+            GlobalExecutors.scheduler.schedule(this::tick, 1000 - (System.currentTimeMillis() - startTime), TimeUnit.MILLISECONDS);
     }
 
     private CompletableFuture<RegionFile> getRegionFile(ChunkPos pos) {
@@ -86,7 +83,7 @@ private CompletableFuture<RegionFile> getRegionFile(ChunkPos pos) {
             } finally {
                 lockToken.releaseLock();
             }
-        }, scheduler);
+        }, GlobalExecutors.scheduler);
     }
 
     private void scheduleWrite(ChunkPos pos, CompoundTag chunkData) {
@@ -164,7 +161,7 @@ public CompletableFuture<CompoundTag> getNbtAtAsync(ChunkPos pos) {
             } finally {
                 lockToken.releaseLock();
             }
-        }, scheduler);
+        }, GlobalExecutors.scheduler);
     }
 
     @Nullable

src/main/java/org/yatopiamc/c2me/common/threading/chunkio/ChunkIoThreadingExecutorUtils.java

@@ -7,7 +7,7 @@
 public class ChunkIoThreadingExecutorUtils {
 
     public static final ForkJoinPool serializerExecutor = new ForkJoinPool(
-            Math.min(6, Runtime.getRuntime().availableProcessors()),
+            Math.min(2, Runtime.getRuntime().availableProcessors()),
             new C2MEForkJoinWorkerThreadFactory("C2ME chunkio serializer worker #%d", Thread.NORM_PRIORITY - 1),
             null,
             true

src/main/java/org/yatopiamc/c2me/common/threading/worldgen/ChunkStatusUtils.java

@@ -2,13 +2,16 @@
 
 import com.ibm.asyncutil.locks.AsyncLock;
 import com.ibm.asyncutil.locks.AsyncNamedLock;
+import com.ibm.asyncutil.util.Combinators;
 import net.minecraft.util.math.ChunkPos;
 import net.minecraft.world.chunk.ChunkStatus;
 import org.jetbrains.annotations.NotNull;
+import org.yatopiamc.c2me.common.threading.GlobalExecutors;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.function.Supplier;
 
 public class ChunkStatusUtils {
@@ -32,25 +35,17 @@ public static ChunkStatusThreadingType getThreadingType(final ChunkStatus status
         }
     }
 
-    private static <T> CompletableFuture<T> buildChain0(@NotNull List<CompletableFuture<AsyncLock.LockToken>> list, int index, Supplier<CompletableFuture<T>> code) {
-        if (index < list.size()) {
-            return list.get(index).thenCompose(lockToken -> {
-                final CompletableFuture<T> future = buildChain0(list, index + 1, code);
-                future.thenRun(lockToken::releaseLock);
-                return future;
-            });
-        } else {
-            return code.get();
-        }
-    }
-
     public static <T> CompletableFuture<T> runChunkGenWithLock(ChunkPos target, int radius, AsyncNamedLock<ChunkPos> chunkLock, Supplier<CompletableFuture<T>> action) {
-        List<CompletableFuture<AsyncLock.LockToken>> acquiredLocks = new ArrayList<>((radius + 1) * (radius + 1));
+        List<CompletionStage<AsyncLock.LockToken>> acquiredLocks = new ArrayList<>((radius + 1) * (radius + 1));
         for (int x = target.x - radius; x <= target.x + radius; x++)
             for (int z = target.z - radius; z <= target.z + radius; z++)
-                acquiredLocks.add(chunkLock.acquireLock(new ChunkPos(x, z)).toCompletableFuture());
+                acquiredLocks.add(chunkLock.acquireLock(new ChunkPos(x, z)));
 
-        return buildChain0(acquiredLocks, 0, action);
+        return Combinators.collect(acquiredLocks).toCompletableFuture().thenComposeAsync(lockTokens -> {
+            final CompletableFuture<T> future = action.get();
+            future.thenRun(() -> lockTokens.forEach(AsyncLock.LockToken::releaseLock));
+            return future;
+        }, GlobalExecutors.scheduler);
     }
 
 }

src/main/java/org/yatopiamc/c2me/common/threading/worldgen/WorldGenThreadingExecutorUtils.java

@@ -11,17 +11,10 @@
 public class WorldGenThreadingExecutorUtils {
 
     public static final ForkJoinPool mainExecutor = new ForkJoinPool(
-            Math.min(8, Runtime.getRuntime().availableProcessors()),
+            Math.min(6, Runtime.getRuntime().availableProcessors()),
             new C2MEForkJoinWorkerThreadFactory("C2ME worldgen worker #%d", Thread.NORM_PRIORITY - 1),
             null,
             true
     );
-    public static final ThreadPoolExecutor scheduler = new ThreadPoolExecutor(
-            1,
-            1,
-            0, TimeUnit.SECONDS,
-            new LinkedBlockingQueue<>(),
-            new ThreadFactoryBuilder().setNameFormat("C2ME worldgen scheduler").setDaemon(true).setPriority(Thread.NORM_PRIORITY).build()
-    );
 
 }

src/main/java/org/yatopiamc/c2me/mixin/threading/worldgen/MixinThreadedAnvilChunkStorage.java

@@ -5,6 +5,7 @@
 import org.spongepowered.asm.mixin.Dynamic;
 import org.spongepowered.asm.mixin.Mixin;
 import org.spongepowered.asm.mixin.Overwrite;
+import org.yatopiamc.c2me.common.threading.GlobalExecutors;
 import org.yatopiamc.c2me.common.threading.worldgen.WorldGenThreadingExecutorUtils;
 
 @Mixin(ThreadedAnvilChunkStorage.class)
@@ -18,7 +19,7 @@ public class MixinThreadedAnvilChunkStorage {
     @Dynamic
     @Overwrite
     private void method_17259(ChunkHolder chunkHolder, Runnable runnable) { // synthetic method for worldGenExecutor scheduling
-        WorldGenThreadingExecutorUtils.scheduler.execute(runnable);
+        GlobalExecutors.scheduler.execute(runnable);
     }
 
 }