From c8cbb14ffc51c91d0746aa18089280993e7b0a0b Mon Sep 17 00:00:00 2001 From: Aikar Date: Fri, 16 Mar 2018 22:59:43 -0400 Subject: [PATCH] Improved Async Task Scheduler The Craft Scheduler still uses the primary thread for task scheduling. This results in the main thread still having to do work as part of the dispatching of async tasks. If plugins make use of lots of async tasks, such as particle emitters that want to keep the logic off the main thread, the main thread still receives quite a bit of load from processing all of these queued tasks. Additionally, resizing and managing the pending entries for all of these asynchronous tasks takes up time on the main thread too. This commit replaces the implementation of the scheduler when working with asynchronous tasks, by forwarding calls to the new scheduler. The Async Scheduler uses a single thread executor for "management" tasks. The Management Thread is responsible for all adding and dispatching of scheduled tasks. The mainThreadHeartbeat will send a heartbeat task to the management thread with the currentTick value, so that it can find which tasks to execute. Scheduling of an async tasks also dispatches a management task, ensuring that any Queue resizing operation occurs off of the main thread. The async queue uses a complete separate PriorityQueue, ensuring that resize operations are decoupled from the sync tasks queue. diff --git a/src/main/java/org/bukkit/craftbukkit/scheduler/CraftAsyncScheduler.java b/src/main/java/org/bukkit/craftbukkit/scheduler/CraftAsyncScheduler.java new file mode 100644 index 000000000..3c1992e21 --- /dev/null +++ b/src/main/java/org/bukkit/craftbukkit/scheduler/CraftAsyncScheduler.java @@ -0,0 +1,122 @@ +/* + * Copyright (c) 2018 Daniel Ennis (Aikar) MIT License + * + * Permission is hereby granted, free of charge, to any person obtaining + * a copy of this software and associated documentation files (the + * "Software"), to deal in the Software without restriction, including + * without limitation the rights to use, copy, modify, merge, publish, + * distribute, sublicense, and/or sell copies of the Software, and to + * permit persons to whom the Software is furnished to do so, subject to + * the following conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF + * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE + * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION + * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ + +package org.bukkit.craftbukkit.scheduler; + +import com.destroystokyo.paper.ServerSchedulerReportingWrapper; +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import org.bukkit.plugin.Plugin; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +public class CraftAsyncScheduler extends CraftScheduler { + + private final ThreadPoolExecutor executor = new ThreadPoolExecutor( + 4, Integer.MAX_VALUE,30L, TimeUnit.SECONDS, new SynchronousQueue<>(), + new ThreadFactoryBuilder().setNameFormat("Craft Scheduler Thread - %1$d").build()); + private final Executor management = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder() + .setNameFormat("Craft Async Scheduler Management Thread").build()); + private final List temp = new ArrayList<>(); + + CraftAsyncScheduler() { + super(true); + executor.allowCoreThreadTimeOut(true); + executor.prestartAllCoreThreads(); + } + + @Override + public void cancelTask(int taskId) { + this.management.execute(() -> this.removeTask(taskId)); + } + + private synchronized void removeTask(int taskId) { + parsePending(); + this.pending.removeIf((task) -> { + if (task.getTaskId() == taskId) { + task.cancel0(); + return true; + } + return false; + }); + } + + @Override + public void mainThreadHeartbeat(int currentTick) { + this.currentTick = currentTick; + this.management.execute(() -> this.runTasks(currentTick)); + } + + private synchronized void runTasks(int currentTick) { + parsePending(); + while (!this.pending.isEmpty() && this.pending.peek().getNextRun() <= currentTick) { + CraftTask task = this.pending.remove(); + if (executeTask(task)) { + final long period = task.getPeriod(); + if (period > 0) { + task.setNextRun(currentTick + period); + temp.add(task); + } + } + parsePending(); + } + this.pending.addAll(temp); + temp.clear(); + } + + private boolean executeTask(CraftTask task) { + if (isValid(task)) { + this.runners.put(task.getTaskId(), task); + this.executor.execute(new ServerSchedulerReportingWrapper(task)); + return true; + } + return false; + } + + @Override + public synchronized void cancelTasks(Plugin plugin) { + parsePending(); + for (Iterator iterator = this.pending.iterator(); iterator.hasNext(); ) { + CraftTask task = iterator.next(); + if (task.getTaskId() != -1 && (plugin == null || task.getOwner().equals(plugin))) { + task.cancel0(); + iterator.remove(); + } + } + } + + /** + * Task is not cancelled + * @param runningTask + * @return + */ + static boolean isValid(CraftTask runningTask) { + return runningTask.getPeriod() >= CraftTask.NO_REPEATING; + } +} diff --git a/src/main/java/org/bukkit/craftbukkit/scheduler/CraftScheduler.java b/src/main/java/org/bukkit/craftbukkit/scheduler/CraftScheduler.java index 36eecb670..15f165d9b 100644 --- a/src/main/java/org/bukkit/craftbukkit/scheduler/CraftScheduler.java +++ b/src/main/java/org/bukkit/craftbukkit/scheduler/CraftScheduler.java @@ -62,7 +62,7 @@ public class CraftScheduler implements BukkitScheduler { /** * Main thread logic only */ - private final PriorityQueue pending = new PriorityQueue(10, + final PriorityQueue pending = new PriorityQueue(10, // Paper new Comparator() { public int compare(final CraftTask o1, final CraftTask o2) { int value = Long.compare(o1.getNextRun(), o2.getNextRun()); @@ -78,21 +78,38 @@ public class CraftScheduler implements BukkitScheduler { /** * These are tasks that are currently active. It's provided for 'viewing' the current state. */ - private final ConcurrentHashMap runners = new ConcurrentHashMap(); + final ConcurrentHashMap runners = new ConcurrentHashMap(); // Paper /** * The sync task that is currently running on the main thread. */ private volatile CraftTask currentTask = null; - private volatile int currentTick = -1; - private final Executor executor = Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("Craft Scheduler Thread - %d").build()); - private CraftAsyncDebugger debugHead = new CraftAsyncDebugger(-1, null, null) {@Override StringBuilder debugTo(StringBuilder string) {return string;}}; - private CraftAsyncDebugger debugTail = debugHead; + volatile int currentTick = -1; // Paper + //private final Executor executor = Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("Craft Scheduler Thread - %d").build()); // Paper - moved to AsyncScheduler + //private CraftAsyncDebugger debugHead = new CraftAsyncDebugger(-1, null, null) {@Override StringBuilder debugTo(StringBuilder string) {return string;}}; // Paper + //private CraftAsyncDebugger debugTail = debugHead; // Paper private static final int RECENT_TICKS; static { RECENT_TICKS = 30; } + + // Paper start + private final CraftScheduler asyncScheduler; + private final boolean isAsyncScheduler; + public CraftScheduler() { + this(false); + } + + public CraftScheduler(boolean isAsync) { + this.isAsyncScheduler = isAsync; + if (isAsync) { + this.asyncScheduler = this; + } else { + this.asyncScheduler = new CraftAsyncScheduler(); + } + } + // Paper end @Override public int scheduleSyncDelayedTask(final Plugin plugin, final Runnable task) { return this.scheduleSyncDelayedTask(plugin, task, 0L); @@ -209,7 +226,7 @@ public class CraftScheduler implements BukkitScheduler { } else if (period < CraftTask.NO_REPEATING) { period = CraftTask.NO_REPEATING; } - return handle(new CraftAsyncTask(runners, plugin, runnable, nextId(), period), delay); + return handle(new CraftAsyncTask(this.asyncScheduler.runners, plugin, runnable, nextId(), period), delay); // Paper } @Override @@ -225,6 +242,11 @@ public class CraftScheduler implements BukkitScheduler { if (taskId <= 0) { return; } + // Paper start + if (!this.isAsyncScheduler) { + this.asyncScheduler.cancelTask(taskId); + } + // Paper end CraftTask task = runners.get(taskId); if (task != null) { task.cancel0(); @@ -265,6 +287,11 @@ public class CraftScheduler implements BukkitScheduler { @Override public void cancelTasks(final Plugin plugin) { Validate.notNull(plugin, "Cannot cancel tasks of null plugin"); + // Paper start + if (!this.isAsyncScheduler) { + this.asyncScheduler.cancelTasks(plugin); + } + // Paper end final CraftTask task = new CraftTask( new Runnable() { public void run() { @@ -303,6 +330,13 @@ public class CraftScheduler implements BukkitScheduler { @Override public boolean isCurrentlyRunning(final int taskId) { + // Paper start + if (!isAsyncScheduler) { + if (this.asyncScheduler.isCurrentlyRunning(taskId)) { + return true; + } + } + // Paper end final CraftTask task = runners.get(taskId); if (task == null) { return false; @@ -321,6 +355,11 @@ public class CraftScheduler implements BukkitScheduler { if (taskId <= 0) { return false; } + // Paper start + if (!this.isAsyncScheduler && this.asyncScheduler.isQueued(taskId)) { + return true; + } + // Paper end for (CraftTask task = head.getNext(); task != null; task = task.getNext()) { if (task.getTaskId() == taskId) { return task.getPeriod() >= CraftTask.NO_REPEATING; // The task will run @@ -332,6 +371,12 @@ public class CraftScheduler implements BukkitScheduler { @Override public List getActiveWorkers() { + // Paper start + if (!isAsyncScheduler) { + //noinspection TailRecursion + return this.asyncScheduler.getActiveWorkers(); + } + // Paper end final ArrayList workers = new ArrayList(); for (final CraftTask taskObj : runners.values()) { // Iterator will be a best-effort (may fail to grab very new values) if called from an async thread @@ -369,6 +414,11 @@ public class CraftScheduler implements BukkitScheduler { pending.add(task); } } + // Paper start + if (!this.isAsyncScheduler) { + pending.addAll(this.asyncScheduler.getPendingTasks()); + } + // Paper end return pending; } @@ -376,6 +426,11 @@ public class CraftScheduler implements BukkitScheduler { * This method is designed to never block or wait for locks; an immediate execution of all current tasks. */ public void mainThreadHeartbeat(final int currentTick) { + // Paper start + if (!this.isAsyncScheduler) { + this.asyncScheduler.mainThreadHeartbeat(currentTick); + } + // Paper end this.currentTick = currentTick; final List temp = this.temp; parsePending(); @@ -412,7 +467,7 @@ public class CraftScheduler implements BukkitScheduler { parsePending(); } else { //debugTail = debugTail.setNext(new CraftAsyncDebugger(currentTick + RECENT_TICKS, task.getOwner(), task.getTaskClass())); // Paper - executor.execute(new ServerSchedulerReportingWrapper(task)); // Paper + task.getOwner().getLogger().log(Level.SEVERE, "Unexpected Async Task in the Sync Scheduler. Report this to Paper"); // Paper // We don't need to parse pending // (async tasks must live with race-conditions if they attempt to cancel between these few lines of code) } @@ -431,7 +486,7 @@ public class CraftScheduler implements BukkitScheduler { //debugHead = debugHead.getNextHead(currentTick); // Paper } - private void addTask(final CraftTask task) { + protected void addTask(final CraftTask task) { final AtomicReference tail = this.tail; CraftTask tailTask = tail.get(); while (!tail.compareAndSet(tailTask, task)) { @@ -440,7 +495,13 @@ public class CraftScheduler implements BukkitScheduler { tailTask.setNext(task); } - private CraftTask handle(final CraftTask task, final long delay) { + protected CraftTask handle(final CraftTask task, final long delay) { // Paper + // Paper start + if (!this.isAsyncScheduler && !task.isSync()) { + this.asyncScheduler.handle(task, delay); + return task; + } + // Paper end task.setNextRun(currentTick + delay); addTask(task); return task; @@ -459,8 +520,8 @@ public class CraftScheduler implements BukkitScheduler { return ids.incrementAndGet(); } - private void parsePending() { - MinecraftTimings.bukkitSchedulerPendingTimer.startTiming(); + void parsePending() { // Paper + if (!this.isAsyncScheduler) MinecraftTimings.bukkitSchedulerPendingTimer.startTiming(); // Paper CraftTask head = this.head; CraftTask task = head.getNext(); CraftTask lastTask = head; @@ -479,7 +540,7 @@ public class CraftScheduler implements BukkitScheduler { task.setNext(null); } this.head = lastTask; - MinecraftTimings.bukkitSchedulerPendingTimer.stopTiming(); + if (!this.isAsyncScheduler) MinecraftTimings.bukkitSchedulerPendingTimer.stopTiming(); // Paper } private boolean isReady(final int currentTick) { -- 2.21.0