Use better executors

2022-12-02 12:02:42 +08:00
parent 4032c984ce
commit bfb566a091
8 changed files with 427 additions and 0 deletions
--- a/patches/server/0023-Add-centralized-AsyncExecutor.patch
+++ b/patches/server/0023-Add-centralized-AsyncExecutor.patch
@@ -0,0 +1,117 @@
+From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
+From: MartijnMuijsers <martijnmuijsers@live.nl>
+Date: Tue, 29 Nov 2022 12:35:35 +0100
+Subject: [PATCH] Add centralized AsyncExecutor
+./gs://www.gnu.org/licenses/agpl-3.0.html)
+
+This patch was taken from Gale.
+
+diff --git a/src/main/java/org/galemc/gale/concurrent/AsyncExecutor.java b/src/main/java/org/galemc/gale/concurrent/AsyncExecutor.java
+new file mode 100644
+index 0000000000000000000000000000000000000000..de182b9473963b95085fa612f70884a56765ae43
+--- /dev/null
+++ b/src/main/java/org/galemc/gale/concurrent/AsyncExecutor.java
+@@ -0,0 +1,103 @@
+// Gale - centralized async execution
+
+package org.galemc.gale.concurrent;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import net.minecraft.Util;
+import org.galemc.gale.util.CPUCoresEstimation;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * An executor for tasks that can run asynchronously. This executor uses a fixed thread pool, and as such
+ * is not appropriate for tasks that block.
+ * <br>
+ * It can be paused when all CPU cores may be needed for something else.
+ *
+ * @author Martijn Muijsers
+ */
+public final class AsyncExecutor extends ThreadPoolExecutor {
+
+    /**
+     * The fixed number of threads that will be used by this {@link AsyncExecutor}.
+     * <br>
+     * By default, we do not use two cores, so that there is always a core for the main thread that we do not use,
+     * and another core that we do not use to run other important threads such as garbage collection on.
+     * <br>
+     * This value is at least 1.
+     */
+    public static final int parallelism;
+    static {
+        int parallelismByEnvironmentVariable = Integer.getInteger("gale.threads.async", -1);
+        parallelism = Math.max(1, parallelismByEnvironmentVariable > 0 ? parallelismByEnvironmentVariable : CPUCoresEstimation.get() - 2);
+    }
+
+    /**
+     * The queue of tasks in the {@link AsyncExecutor} singleton instance.
+     * This queue can be accessed externally to steal work from the executor.
+     */
+    public static final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
+
+    /**
+     * Singleton {@link AsyncExecutor} instance.
+     */
+    public static final AsyncExecutor instance = new AsyncExecutor();
+
+    private static volatile boolean isPaused = false;
+    private static final ReentrantLock pauseLock = new ReentrantLock();
+    private static final Condition pauseCondition = pauseLock.newCondition();
+
+    private AsyncExecutor() {
+        super(parallelism, parallelism, 0L, TimeUnit.MILLISECONDS, queue, new ThreadFactoryBuilder()
+            .setNameFormat("Async Executor Thread - %1$d")
+            .setPriority(Thread.NORM_PRIORITY - 1) // Deprioritize over main
+            .setUncaughtExceptionHandler(Util::onThreadException)
+            .build());
+    }
+
+    @Override
+    protected void beforeExecute(Thread t, Runnable r) {
+        super.beforeExecute(t, r);
+        pauseLock.lock();
+        try {
+            while (isPaused) pauseCondition.await();
+        } catch (InterruptedException ie) {
+            t.interrupt();
+        } finally {
+            pauseLock.unlock();
+        }
+    }
+
+    /**
+     * Pauses the {@link AsyncExecutor} from starting to run any new task until {@link #resume()} is called.
+     * <br>
+     * This does not affect execution of tasks that are already being performed when this method is called.
+     */
+    public static void pause() {
+        pauseLock.lock();
+        try {
+            isPaused = true;
+        } finally {
+            pauseLock.unlock();
+        }
+    }
+
+    /**
+     * Resumes the {@link AsyncExecutor} singleton instance after it has been paused using {@link #pause()}.
+     */
+    public static void resume() {
+        pauseLock.lock();
+        try {
+            isPaused = false;
+            pauseCondition.signalAll();
+        } finally {
+            pauseLock.unlock();
+        }
+    }
+
+}