117 lines
4.1 KiB
Diff
117 lines
4.1 KiB
Diff
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
|
|
|
|
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();
|
|
+ }
|
|
+ }
|
|
+
|
|
+}
|