Skip to content

Run edit-completion tasks on a dedicated executor #3609

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MattBDev wants to merge 1 commit into
base: main
Choose a base branch
from
+72 −24
Open

Run edit-completion tasks on a dedicated executor #3609

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ...main/java/com/sk89q/worldedit/bukkit/adapter/impl/fawe/v1_21_R1/PaperweightGetBlocks.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -787,11 +787,11 @@ protected <T extends Future<T>> T internalCall(
}
if (callback == null) {
if (finalizer != null) {
queueHandler.async(finalizer, null);
queueHandler.completion(finalizer, null);
}
return null;
Comment on lines 788 to 792
} 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);
Expand Down
7 changes: 5 additions & 2 deletions ...t-bukkit/src/main/java/com/fastasyncworldedit/bukkit/adapter/AbstractBukkitGetBlocks.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,10 +159,13 @@ protected <T extends Future<T>> 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) {
Expand Down
40 changes: 40 additions & 0 deletions ...dit-core/src/main/java/com/fastasyncworldedit/core/queue/implementation/QueueHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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)}).
* <p>
* 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()
);

@dordsor21 dordsor21 Aug 6, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not really sure an unbounded thead pool is a good solution. We can't control what others do, so even if we technically limit the submission via STQE (though this is not a global limit), there is no functional limit. Typical target-size (for submissions) is often 1000s in people's confifs, so that's 1000s.of submissions on potentially 16 threads, each Thread spawned can be 50MB, which would just OOM the application and/or destroy performance.

None of the tasks here have any blocking dependencies on child tasks (at least not on child tasks also submitted to the secondary pool). The issue that saw secondary pool starvation was because FAVS command tasks (which synchronised on the LocalSession to enforce one-task-at-a-time) were being submitted to the secondary pool, which obviously need to have previous edits complete before freeing up their thread usage (fork pools don't switch tasks when it's based on a synchronized block). This meant that no tasks could complete at all. The solution is simply to only submit the correct tasks to the secondary pool.

If there is a remaining exhaustion form FAWE's own code then that should be addressed in place. We also should not have to account for potential API misuse in downstream plugins at the expense of FAWE itself imo. We should definitely correct the javadocs (in this case anything submitted to secondary pool should have no dependents in waits on, if it does, they should be resubmitted as separate futures).

/**
* Queue for tasks to be completed on the main thread. These take priority of tasks submitted to syncWhenFree queue
*/
Expand Down Expand Up @@ -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.
* <p>
* Internal API usage only.
*
* @param run Runnable to run
* @param value Value to return when done
* @param <T> Value type
* @return Future for the submitted task
*/
Comment thread
Copilot marked this conversation as resolved.
@ApiStatus.Internal
public <T> Future<T> 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.
Expand Down
45 changes: 25 additions & 20 deletions ...c/main/java/com/fastasyncworldedit/core/queue/implementation/SingleThreadQueueExtent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,8 +225,11 @@ public <V extends Future<V>> 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;
Expand Down Expand Up @@ -278,7 +281,6 @@ public <V extends Future<V>> V submitTaskUnchecked(Callable<V> 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();
Expand Down Expand Up @@ -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);
}
Expand Down
Loading