diff --git a/worldedit-bukkit/adapters/adapter-1_21/src/main/java/com/sk89q/worldedit/bukkit/adapter/impl/fawe/v1_21_R1/PaperweightGetBlocks.java b/worldedit-bukkit/adapters/adapter-1_21/src/main/java/com/sk89q/worldedit/bukkit/adapter/impl/fawe/v1_21_R1/PaperweightGetBlocks.java index a434387799..a9ae641047 100644 --- a/worldedit-bukkit/adapters/adapter-1_21/src/main/java/com/sk89q/worldedit/bukkit/adapter/impl/fawe/v1_21_R1/PaperweightGetBlocks.java +++ b/worldedit-bukkit/adapters/adapter-1_21/src/main/java/com/sk89q/worldedit/bukkit/adapter/impl/fawe/v1_21_R1/PaperweightGetBlocks.java @@ -787,11 +787,11 @@ protected > T internalCall( } if (callback == null) { if (finalizer != null) { - queueHandler.async(finalizer, null); + queueHandler.completion(finalizer, null); } return null; } else { - return queueHandler.async(callback, null); + return queueHandler.completion(callback, null); } } catch (Throwable e) { LOGGER.error("Error performing final chunk calling at {},{}", chunkX, chunkZ, e); diff --git a/worldedit-bukkit/src/main/java/com/fastasyncworldedit/bukkit/adapter/AbstractBukkitGetBlocks.java b/worldedit-bukkit/src/main/java/com/fastasyncworldedit/bukkit/adapter/AbstractBukkitGetBlocks.java index bba60917e8..8602a77c2c 100644 --- a/worldedit-bukkit/src/main/java/com/fastasyncworldedit/bukkit/adapter/AbstractBukkitGetBlocks.java +++ b/worldedit-bukkit/src/main/java/com/fastasyncworldedit/bukkit/adapter/AbstractBukkitGetBlocks.java @@ -159,10 +159,13 @@ protected > T handleCallFinalizer( task.run(); } } + // Completion executor, not the secondary pool: this future is the tail of the chain blocked on in + // SingleThreadQueueExtent flushes. If it required a free secondary-pool worker, edits running *on* the + // secondary pool (e.g. plugins submitting via QueueHandler#async) could starve it and deadlock. if (callback != null) { - return queueHandler.async(callback, null); + return queueHandler.completion(callback, null); } else if (finalizer != null) { - return queueHandler.async(finalizer, null); + return queueHandler.completion(finalizer, null); } return null; } catch (Throwable e) { diff --git a/worldedit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/QueueHandler.java b/worldedit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/QueueHandler.java index e5f43ddbcb..cac2237a8c 100644 --- a/worldedit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/QueueHandler.java +++ b/worldedit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/QueueHandler.java @@ -14,6 +14,7 @@ import com.fastasyncworldedit.core.util.MemUtil; import com.fastasyncworldedit.core.util.TaskManager; import com.fastasyncworldedit.core.util.collection.CleanableThreadLocal; +import com.fastasyncworldedit.core.util.task.FaweBasicThreadFactory; import com.fastasyncworldedit.core.util.task.FaweForkJoinWorkerThreadFactory; import com.fastasyncworldedit.core.wrappers.WorldWrapper; import com.google.common.util.concurrent.Futures; @@ -33,6 +34,7 @@ import java.util.concurrent.ForkJoinTask; import java.util.concurrent.Future; import java.util.concurrent.FutureTask; +import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @@ -76,6 +78,26 @@ public abstract class QueueHandler implements Trimable, Runnable { */ private final ThreadPoolExecutor blockingExecutor = FaweCache.INSTANCE.newBlockingExecutor( "FAWE QueueHandler Blocking Executor - %d"); + /** + * Executor for edit "completion" tasks, e.g. sending updated chunks to players and running the caller-supplied + * finalizer. Completion tasks form the tail of the future chains waited on when flushing a + * {@link com.fastasyncworldedit.core.queue.implementation.SingleThreadQueueExtent}, so they must be guaranteed to run + * even while the primary and secondary pools are saturated with (possibly blocked) edit tasks. This executor grows on + * demand instead of queueing behind a bounded set of workers, preventing thread-starvation deadlocks when edits are + * performed on the secondary pool (e.g. by plugins submitting whole edits via {@link #async(Runnable)}). + *

+ * Submission cannot be rejected: if a worker cannot be started (executor shutdown, or the JVM being unable to create + * further threads) the task runs on the submitting thread rather than throwing, preserving the guarantee above. + */ + private final ThreadPoolExecutor completionExecutor = new ThreadPoolExecutor( + 0, + Integer.MAX_VALUE, + 60L, + TimeUnit.SECONDS, + new SynchronousQueue<>(), + new FaweBasicThreadFactory("FAWE Completion Executor - %d"), + new ThreadPoolExecutor.CallerRunsPolicy() + ); /** * Queue for tasks to be completed on the main thread. These take priority of tasks submitted to syncWhenFree queue */ @@ -238,6 +260,24 @@ public ForkJoinTask submit(Runnable run) { return forkJoinPoolPrimary.submit(run); } + /** + * Run an edit-completion task, e.g. sending an updated chunk to players, on the dedicated completion executor. + * Unlike {@link #async(Runnable)}, tasks submitted here are guaranteed to begin execution even if + * the primary and secondary pools are saturated or blocked, as edit flushes block on the futures returned here. + * Completion tasks must therefore never block on other FAWE futures themselves. + *

+ * Internal API usage only. + * + * @param run Runnable to run + * @param value Value to return when done + * @param Value type + * @return Future for the submitted task + */ + @ApiStatus.Internal + public Future completion(Runnable run, T value) { + return completionExecutor.submit(run, value); + } + /** * Submit a task to be run on the main thread. Does not guarantee to be run on the next tick as FAWE will only operate to * maintain approx. 18 tps. diff --git a/worldedit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/SingleThreadQueueExtent.java b/worldedit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/SingleThreadQueueExtent.java index 6aafff7a03..c32140c42d 100644 --- a/worldedit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/SingleThreadQueueExtent.java +++ b/worldedit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/SingleThreadQueueExtent.java @@ -225,8 +225,11 @@ public > V submit(IQueueChunk chunk) { this.lastChunk.compareAndExchange(chunk, null); final long index = MathMan.pairInt(chunk.getX(), chunk.getZ()); getChunkLock.lock(); - chunks.remove(index, chunk); - getChunkLock.unlock(); + try { + chunks.remove(index, chunk); + } finally { + getChunkLock.unlock(); + } V future = submitUnchecked(chunk); submissions.add(future); return future; @@ -278,7 +281,6 @@ public > V submitTaskUnchecked(Callable callable) { public synchronized boolean trim(boolean aggressive) { cacheGet.trim(aggressive); cacheSet.trim(aggressive); - LOGGER.info("trim"); if (Thread.currentThread() == currentThread) { lastChunk.set(null); return chunks.isEmpty(); @@ -481,27 +483,30 @@ private void iterateSubmissions() { public synchronized void flush() { if (!chunks.isEmpty()) { getChunkLock.lock(); - if (MemUtil.isMemoryLimited()) { - while (!chunks.isEmpty()) { - IQueueChunk chunk = chunks.removeFirst(); - this.lastChunk.compareAndExchange(chunk, null); - final Future future = submitUnchecked(chunk); - if (future != null && !future.isDone()) { - pollSubmissions(Settings.settings().QUEUE.PARALLEL_THREADS, true); - submissions.add(future); + try { + if (MemUtil.isMemoryLimited()) { + while (!chunks.isEmpty()) { + IQueueChunk chunk = chunks.removeFirst(); + this.lastChunk.compareAndExchange(chunk, null); + final Future future = submitUnchecked(chunk); + if (future != null && !future.isDone()) { + pollSubmissions(Settings.settings().QUEUE.PARALLEL_THREADS, true); + submissions.add(future); + } } - } - } else { - while (!chunks.isEmpty()) { - IQueueChunk chunk = chunks.removeFirst(); - this.lastChunk.compareAndExchange(chunk, null); - final Future future = submitUnchecked(chunk); - if (future != null && !future.isDone()) { - submissions.add(future); + } else { + while (!chunks.isEmpty()) { + IQueueChunk chunk = chunks.removeFirst(); + this.lastChunk.compareAndExchange(chunk, null); + final Future future = submitUnchecked(chunk); + if (future != null && !future.isDone()) { + submissions.add(future); + } } } + } finally { + getChunkLock.unlock(); } - getChunkLock.unlock(); } pollSubmissions(0, true); }