Heavily increase usage of ForkJoinPool and fix issues on low-core env…

…ironments
AlexProgrammerDE · Sep 30, 2024 · d190730 · d190730
1 parent 54400be
commit d190730
Show file tree

Hide file tree

Showing 4 changed files with 70 additions and 20 deletions.
diff --git a/server/src/main/java/com/soulfiremc/server/InstanceManager.java b/server/src/main/java/com/soulfiremc/server/InstanceManager.java
@@ -311,12 +311,7 @@ private void start() {
             break;
           }
 
-          try {
-            connectSemaphore.acquire();
-          } catch (InterruptedException e) {
-            logger.error("Error while waiting for connection slot", e);
-            break;
-          }
+          TimeUtil.acquireYielding(connectSemaphore);
 
           logger.debug("Scheduling bot {}", factory.minecraftAccount().lastKnownName());
           scheduler.schedule(

diff --git a/server/src/main/java/com/soulfiremc/server/grpc/MCAuthServiceImpl.java b/server/src/main/java/com/soulfiremc/server/grpc/MCAuthServiceImpl.java
@@ -20,6 +20,7 @@
 import com.soulfiremc.grpc.generated.*;
 import com.soulfiremc.server.account.MCAuthService;
 import com.soulfiremc.server.user.Permissions;
+import com.soulfiremc.server.util.TimeUtil;
 import com.soulfiremc.settings.account.MinecraftAccount;
 import com.soulfiremc.settings.proxy.SFProxy;
 import io.grpc.Status;
@@ -53,10 +54,8 @@ public void loginCredentials(CredentialsAuthRequest request, StreamObserver<Cred
         .parallelStream()
         .map(payload -> {
           try {
-            semaphore.acquire();
+            TimeUtil.acquireYielding(semaphore);
             return service.createDataAndLogin(payload, proxy).join().toProto();
-          } catch (InterruptedException e) {
-            throw new RuntimeException(e);
           } finally {
             semaphore.release();
           }

diff --git a/server/src/main/java/com/soulfiremc/server/protocol/bot/state/ChunkHolder.java b/server/src/main/java/com/soulfiremc/server/protocol/bot/state/ChunkHolder.java
@@ -24,6 +24,7 @@
 import com.soulfiremc.server.protocol.bot.model.ChunkKey;
 import com.soulfiremc.server.protocol.bot.utils.SectionUtils;
 import com.soulfiremc.server.util.NoopLock;
+import com.soulfiremc.server.util.TimeUtil;
 import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
 import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import org.cloudburstmc.math.vector.Vector3i;
@@ -64,7 +65,7 @@ public ChunkData getChunk(Vector3i block) {
   }
 
   private ChunkData getChunkFromSection(long sectionIndex) {
-    readLock.lock();
+    TimeUtil.lockYielding(readLock);
     try {
       return chunks.get(sectionIndex);
     } finally {
@@ -73,7 +74,7 @@ private ChunkData getChunkFromSection(long sectionIndex) {
   }
 
   public boolean isChunkLoaded(int x, int z) {
-    readLock.lock();
+    TimeUtil.lockYielding(readLock);
     try {
       return chunks.containsKey(ChunkKey.calculateKey(x, z));
     } finally {
@@ -87,7 +88,7 @@ public boolean isChunkLoaded(Vector3i block) {
   }
 
   public void removeChunk(int x, int z) {
-    writeLock.lock();
+    TimeUtil.lockYielding(writeLock);
     try {
       chunks.remove(ChunkKey.calculateKey(x, z));
     } finally {
@@ -96,7 +97,7 @@ public void removeChunk(int x, int z) {
   }
 
   public ChunkData getOrCreateChunk(int x, int z) {
-    writeLock.lock();
+    TimeUtil.lockYielding(writeLock);
     try {
       return chunks.computeIfAbsent(
         ChunkKey.calculateKey(x, z), (key) -> new ChunkData(levelHeightAccessor, false));

diff --git a/server/src/main/java/com/soulfiremc/server/util/TimeUtil.java b/server/src/main/java/com/soulfiremc/server/util/TimeUtil.java
@@ -17,7 +17,10 @@
  */
 package com.soulfiremc.server.util;
 
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
 import java.util.function.BooleanSupplier;
 
 /**
@@ -31,19 +34,71 @@ private TimeUtil() {}
 
   public static void waitTime(long time, TimeUnit unit) {
     try {
-      unit.sleep(time);
+      ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
+        private final long endTime = System.nanoTime() + unit.toNanos(time);
+
+        public boolean isReleasable() {
+          return System.nanoTime() >= endTime;
+        }
+
+        public boolean block() throws InterruptedException {
+          Thread.sleep(1);
+          return System.nanoTime() >= endTime;
+        }
+      });
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
     }
   }
 
   public static void waitCondition(BooleanSupplier condition) {
-    while (condition.getAsBoolean()) {
-      try {
-        TimeUnit.MILLISECONDS.sleep(100);
-      } catch (InterruptedException e) {
-        Thread.currentThread().interrupt();
-      }
+    try {
+      ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
+        public boolean isReleasable() {
+          return !condition.getAsBoolean();
+        }
+
+        public boolean block() throws InterruptedException {
+          Thread.sleep(1);
+          return !condition.getAsBoolean();
+        }
+      });
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  public static void acquireYielding(Semaphore semaphore) {
+    try {
+      ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
+        public boolean isReleasable() {
+          return semaphore.tryAcquire();
+        }
+
+        public boolean block() throws InterruptedException {
+          semaphore.acquire();
+          return true;
+        }
+      });
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  public static void lockYielding(Lock lock) {
+    try {
+      ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
+        public boolean isReleasable() {
+          return lock.tryLock();
+        }
+
+        public boolean block() {
+          lock.lock();
+          return true;
+        }
+      });
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
     }
   }
 }