package org.bukkit.craftbukkit.scheduler; import org.apache.commons.lang.Validate; import org.bukkit.plugin.IllegalPluginAccessException; import org.bukkit.plugin.Plugin; import org.bukkit.scheduler.BukkitScheduler; import org.bukkit.scheduler.BukkitTask; import org.bukkit.scheduler.BukkitWorker; import java.util.ArrayList; import java.util.Comparator; import java.util.Iterator; import java.util.List; import java.util.PriorityQueue; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.logging.Level; /** * The fundamental concepts for this implementation: * <li>Main thread owns {@link #head} and {@link #currentTick}, but it may be read from any thread</li> * <li>Main thread exclusively controls {@link #temp} and {@link #pending}. * They are never to be accessed outside of the main thread; alternatives exist to prevent locking.</li> * <li>{@link #head} to {@link #tail} act as a linked list/queue, with 1 consumer and infinite producers. * Adding to the tail is atomic and very efficient; utility method is {@link #handle(CraftTask, long)} or {@link #addTask(CraftTask)}. </li> * <li>Changing the period on a task is delicate. * Any future task needs to notify waiting threads. * Async tasks must be synchronized to make sure that any thread that's finishing will remove itself from {@link #runners}. * Another utility method is provided for this, {@link #cancelTask(CraftTask)}</li> * <li>{@link #runners} provides a moderately up-to-date view of active tasks. * If the linked head to tail set is read, all remaining tasks that were active at the time execution started will be located in runners.</li> * <li>Async tasks are responsible for removing themselves from runners</li> * <li>Sync tasks are only to be removed from runners on the main thread when coupled with a removal from pending and temp.</li> * <li>Most of the design in this scheduler relies on queuing special tasks to perform any data changes on the main thread. * When executed from inside a synchronous method, the scheduler will be updated before next execution by virtue of the frequent {@link #parsePending()} calls.</li> */ public class CraftScheduler implements BukkitScheduler { /** * Counter for IDs. Order doesn't matter, only uniqueness. */ private final AtomicInteger ids = new AtomicInteger(1); /** * Current head of linked-list. This reference is always stale, {@link CraftTask#next} is the live reference. */ private volatile CraftTask head = new CraftTask(); /** * Tail of a linked-list. AtomicReference only matters when adding to queue */ private final AtomicReference<CraftTask> tail = new AtomicReference<CraftTask>(head); /** * Main thread logic only */ private final PriorityQueue<CraftTask> pending = new PriorityQueue<CraftTask>(10, new Comparator<CraftTask>() { public int compare(final CraftTask o1, final CraftTask o2) { return (int) (o1.getNextRun() - o2.getNextRun()); } }); /** * Main thread logic only */ private final List<CraftTask> temp = new ArrayList<CraftTask>(); /** * These are tasks that are currently active. It's provided for 'viewing' the current state. */ private final ConcurrentHashMap<Integer, CraftTask> runners = new ConcurrentHashMap<Integer, CraftTask>(); private volatile int currentTick = -1; private final Executor executor = Executors.newCachedThreadPool(new com.google.common.util.concurrent.ThreadFactoryBuilder().setNameFormat("Craft Scheduler Thread - %1$d").build()); // Spigot private CraftAsyncDebugger debugHead = new CraftAsyncDebugger(-1, null, null) { @Override StringBuilder debugTo(StringBuilder string) { return string; } }; private CraftAsyncDebugger debugTail = debugHead; private static final int RECENT_TICKS; static { RECENT_TICKS = 30; } public int scheduleSyncDelayedTask(final Plugin plugin, final Runnable task) { return this.scheduleSyncDelayedTask(plugin, task, 0l); } public BukkitTask runTask(Plugin plugin, Runnable runnable) { return runTaskLater(plugin, runnable, 0l); } public int scheduleAsyncDelayedTask(final Plugin plugin, final Runnable task) { return this.scheduleAsyncDelayedTask(plugin, task, 0l); } public BukkitTask runTaskAsynchronously(Plugin plugin, Runnable runnable) { return runTaskLaterAsynchronously(plugin, runnable, 0l); } public int scheduleSyncDelayedTask(final Plugin plugin, final Runnable task, final long delay) { return this.scheduleSyncRepeatingTask(plugin, task, delay, -1l); } public BukkitTask runTaskLater(Plugin plugin, Runnable runnable, long delay) { return runTaskTimer(plugin, runnable, delay, -1l); } public int scheduleAsyncDelayedTask(final Plugin plugin, final Runnable task, final long delay) { return this.scheduleAsyncRepeatingTask(plugin, task, delay, -1l); } public BukkitTask runTaskLaterAsynchronously(Plugin plugin, Runnable runnable, long delay) { return runTaskTimerAsynchronously(plugin, runnable, delay, -1l); } public int scheduleSyncRepeatingTask(final Plugin plugin, final Runnable runnable, long delay, long period) { return runTaskTimer(plugin, runnable, delay, period).getTaskId(); } public BukkitTask runTaskTimer(Plugin plugin, Runnable runnable, long delay, long period) { validate(plugin, runnable); if(delay < 0l) { delay = 0; } if(period == 0l) { period = 1l; } else if(period < -1l) { period = -1l; } return handle(new CraftTask(plugin, runnable, nextId(), period), delay); } public int scheduleAsyncRepeatingTask(final Plugin plugin, final Runnable runnable, long delay, long period) { return runTaskTimerAsynchronously(plugin, runnable, delay, period).getTaskId(); } public BukkitTask runTaskTimerAsynchronously(Plugin plugin, Runnable runnable, long delay, long period) { validate(plugin, runnable); if(delay < 0l) { delay = 0; } if(period == 0l) { period = 1l; } else if(period < -1l) { period = -1l; } return handle(new CraftAsyncTask(runners, plugin, runnable, nextId(), period), delay); } public <T> Future<T> callSyncMethod(final Plugin plugin, final Callable<T> task) { validate(plugin, task); final CraftFuture<T> future = new CraftFuture<T>(task, plugin, nextId()); handle(future, 0l); return future; } public void cancelTask(final int taskId) { if(taskId <= 0) { return; } CraftTask task = runners.get(taskId); if(task != null) { task.cancel0(); } task = new CraftTask( new Runnable() { public void run() { if(!check(CraftScheduler.this.temp)) { check(CraftScheduler.this.pending); } } private boolean check(final Iterable<CraftTask> collection) { final Iterator<CraftTask> tasks = collection.iterator(); while(tasks.hasNext()) { final CraftTask task = tasks.next(); if(task.getTaskId() == taskId) { task.cancel0(); tasks.remove(); if(task.isSync()) { runners.remove(taskId); } return true; } } return false; } }); handle(task, 0l); for(CraftTask taskPending = head.getNext(); taskPending != null; taskPending = taskPending.getNext()) { if(taskPending == task) { return; } if(taskPending.getTaskId() == taskId) { taskPending.cancel0(); } } } public void cancelTasks(final Plugin plugin) { Validate.notNull(plugin, "Cannot cancel tasks of null plugin"); final CraftTask task = new CraftTask( new Runnable() { public void run() { check(CraftScheduler.this.pending); check(CraftScheduler.this.temp); } void check(final Iterable<CraftTask> collection) { final Iterator<CraftTask> tasks = collection.iterator(); while(tasks.hasNext()) { final CraftTask task = tasks.next(); if(task.getOwner().equals(plugin)) { task.cancel0(); tasks.remove(); if(task.isSync()) { runners.remove(task.getTaskId()); } } } } }); handle(task, 0l); for(CraftTask taskPending = head.getNext(); taskPending != null; taskPending = taskPending.getNext()) { if(taskPending == task) { return; } if(taskPending.getTaskId() != -1 && taskPending.getOwner().equals(plugin)) { taskPending.cancel0(); } } for(CraftTask runner : runners.values()) { if(runner.getOwner().equals(plugin)) { runner.cancel0(); } } } public void cancelAllTasks() { final CraftTask task = new CraftTask( new Runnable() { public void run() { Iterator<CraftTask> it = CraftScheduler.this.runners.values().iterator(); while(it.hasNext()) { CraftTask task = it.next(); task.cancel0(); if(task.isSync()) { it.remove(); } } CraftScheduler.this.pending.clear(); CraftScheduler.this.temp.clear(); } }); handle(task, 0l); for(CraftTask taskPending = head.getNext(); taskPending != null; taskPending = taskPending.getNext()) { if(taskPending == task) { break; } taskPending.cancel0(); } for(CraftTask runner : runners.values()) { runner.cancel0(); } } public boolean isCurrentlyRunning(final int taskId) { final CraftTask task = runners.get(taskId); if(task == null || task.isSync()) { return false; } final CraftAsyncTask asyncTask = (CraftAsyncTask) task; synchronized(asyncTask.getWorkers()) { return asyncTask.getWorkers().isEmpty(); } } public boolean isQueued(final int taskId) { if(taskId <= 0) { return false; } for(CraftTask task = head.getNext(); task != null; task = task.getNext()) { if(task.getTaskId() == taskId) { return task.getPeriod() >= -1l; // The task will run } } CraftTask task = runners.get(taskId); return task != null && task.getPeriod() >= -1l; } public List<BukkitWorker> getActiveWorkers() { final ArrayList<BukkitWorker> workers = new ArrayList<BukkitWorker>(); 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 if(taskObj.isSync()) { continue; } final CraftAsyncTask task = (CraftAsyncTask) taskObj; synchronized(task.getWorkers()) { // This will never have an issue with stale threads; it's state-safe workers.addAll(task.getWorkers()); } } return workers; } public List<BukkitTask> getPendingTasks() { final ArrayList<CraftTask> truePending = new ArrayList<CraftTask>(); for(CraftTask task = head.getNext(); task != null; task = task.getNext()) { if(task.getTaskId() != -1) { // -1 is special code truePending.add(task); } } final ArrayList<BukkitTask> pending = new ArrayList<BukkitTask>(); for(CraftTask task : runners.values()) { if(task.getPeriod() >= -1l) { pending.add(task); } } for(final CraftTask task : truePending) { if(task.getPeriod() >= -1l && !pending.contains(task)) { pending.add(task); } } return pending; } /** * This method is designed to never block or wait for locks; an immediate execution of all current tasks. */ public void mainThreadHeartbeat(final int currentTick) { this.currentTick = currentTick; final List<CraftTask> temp = this.temp; parsePending(); while(isReady(currentTick)) { final CraftTask task = pending.remove(); if(task.getPeriod() < -1l) { if(task.isSync()) { runners.remove(task.getTaskId(), task); } parsePending(); continue; } if(task.isSync()) { try { task.run(); } catch(final Throwable throwable) { task.getOwner().getLogger().log( Level.WARNING, String.format( "Task #%s for %s generated an exception", task.getTaskId(), task.getOwner().getDescription().getFullName()), throwable); } parsePending(); } else { debugTail = debugTail.setNext(new CraftAsyncDebugger(currentTick + RECENT_TICKS, task.getOwner(), task.getTaskClass())); executor.execute(task); // 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) } final long period = task.getPeriod(); // State consistency if(period > 0) { task.setNextRun(currentTick + period); temp.add(task); } else if(task.isSync()) { runners.remove(task.getTaskId()); } } pending.addAll(temp); temp.clear(); debugHead = debugHead.getNextHead(currentTick); } private void addTask(final CraftTask task) { final AtomicReference<CraftTask> tail = this.tail; CraftTask tailTask = tail.get(); while(!tail.compareAndSet(tailTask, task)) { tailTask = tail.get(); } tailTask.setNext(task); } private CraftTask handle(final CraftTask task, final long delay) { task.setNextRun(currentTick + delay); addTask(task); return task; } private static void validate(final Plugin plugin, final Object task) { Validate.notNull(plugin, "Plugin cannot be null"); Validate.notNull(task, "Task cannot be null"); if(!plugin.isEnabled()) { throw new IllegalPluginAccessException("Plugin attempted to register task while disabled"); } } private int nextId() { return ids.incrementAndGet(); } private void parsePending() { CraftTask head = this.head; CraftTask task = head.getNext(); CraftTask lastTask = head; for(; task != null; task = (lastTask = task).getNext()) { if(task.getTaskId() == -1) { task.run(); } else if(task.getPeriod() >= -1l) { pending.add(task); runners.put(task.getTaskId(), task); } } // We split this because of the way things are ordered for all of the async calls in CraftScheduler // (it prevents race-conditions) for(task = head; task != lastTask; task = head) { head = task.getNext(); task.setNext(null); } this.head = lastTask; } private boolean isReady(final int currentTick) { return !pending.isEmpty() && pending.peek().getNextRun() <= currentTick; } @Override public String toString() { int debugTick = currentTick; StringBuilder string = new StringBuilder("Recent tasks from ").append(debugTick - RECENT_TICKS).append('-').append(debugTick).append('{'); debugHead.debugTo(string); return string.append('}').toString(); } }