render jobs take their world snapshots one per tick (a burst or max-concurrent > 1 no longer stacks several snapshots into one tick), trace threads numbered (all were named -0)
This commit is contained in:
1 parent
0b781c3538
commit
3339c45dda
1 file changed
+41
-20
@@ -5,6 +5,7 @@ import java.util.Set;
|
|||||||
import java.util.UUID;
|
import java.util.UUID;
|
||||||
import java.util.concurrent.CompletableFuture;
|
import java.util.concurrent.CompletableFuture;
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||||
import java.util.concurrent.ExecutionException;
|
import java.util.concurrent.ExecutionException;
|
||||||
import java.util.concurrent.Executors;
|
import java.util.concurrent.Executors;
|
||||||
import java.util.concurrent.ForkJoinPool;
|
import java.util.concurrent.ForkJoinPool;
|
||||||
@@ -24,6 +25,7 @@ import java.util.function.Supplier;
|
|||||||
import org.bukkit.Bukkit;
|
import org.bukkit.Bukkit;
|
||||||
import org.bukkit.plugin.IllegalPluginAccessException;
|
import org.bukkit.plugin.IllegalPluginAccessException;
|
||||||
import org.bukkit.plugin.Plugin;
|
import org.bukkit.plugin.Plugin;
|
||||||
|
import org.bukkit.scheduler.BukkitTask;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Server-friendly scheduling for photo renders. Bounds CPU use and protects the server tick by:
|
* Server-friendly scheduling for photo renders. Bounds CPU use and protects the server tick by:
|
||||||
@@ -33,6 +35,8 @@ import org.bukkit.plugin.Plugin;
|
|||||||
* {@code queueSize} more waiting; further requests are rejected.</li>
|
* {@code queueSize} more waiting; further requests are rejected.</li>
|
||||||
* <li><b>CPU:</b> the ray tracing runs on a dedicated {@link ForkJoinPool} of {@code threads}
|
* <li><b>CPU:</b> the ray tracing runs on a dedicated {@link ForkJoinPool} of {@code threads}
|
||||||
* {@link Thread#MIN_PRIORITY} workers instead of fanning out over the common pool.</li>
|
* {@link Thread#MIN_PRIORITY} workers instead of fanning out over the common pool.</li>
|
||||||
|
* <li><b>Main thread:</b> at most one world snapshot ({@link #onMainThread}) per tick, also when several
|
||||||
|
* jobs start at once.</li>
|
||||||
* </ul>
|
* </ul>
|
||||||
*
|
*
|
||||||
* <p>Reserve a slot with {@link #tryReserve(UUID)}; if accepted, run the heavy work via
|
* <p>Reserve a slot with {@link #tryReserve(UUID)}; if accepted, run the heavy work via
|
||||||
@@ -55,6 +59,10 @@ public final class RenderManager {
|
|||||||
private final ScheduledExecutorService watchdog;
|
private final ScheduledExecutorService watchdog;
|
||||||
private final Set<UUID> activeUsers = ConcurrentHashMap.newKeySet();
|
private final Set<UUID> activeUsers = ConcurrentHashMap.newKeySet();
|
||||||
private final Set<Job<?>> jobs = ConcurrentHashMap.newKeySet();
|
private final Set<Job<?>> jobs = ConcurrentHashMap.newKeySet();
|
||||||
|
/** Main-thread tasks of the jobs ({@link #onMainThread}), run one per tick by {@link #runNextMainTask}. */
|
||||||
|
private final ConcurrentLinkedQueue<MainTask> mainTasks = new ConcurrentLinkedQueue<>();
|
||||||
|
|
||||||
|
private final BukkitTask mainTaskPump;
|
||||||
private volatile boolean shuttingDown = false;
|
private volatile boolean shuttingDown = false;
|
||||||
private final AtomicInteger inFlight = new AtomicInteger();
|
private final AtomicInteger inFlight = new AtomicInteger();
|
||||||
private final int capacity;
|
private final int capacity;
|
||||||
@@ -78,6 +86,7 @@ public final class RenderManager {
|
|||||||
thread.setDaemon(true);
|
thread.setDaemon(true);
|
||||||
return thread;
|
return thread;
|
||||||
});
|
});
|
||||||
|
this.mainTaskPump = Bukkit.getScheduler().runTaskTimer(plugin, this::runNextMainTask, 1, 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
/** The bounded, low-priority pool the ray tracer must run its parallel work on. */
|
/** The bounded, low-priority pool the ray tracer must run its parallel work on. */
|
||||||
@@ -148,28 +157,24 @@ public final class RenderManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* From a dispatched job's {@code work}: runs {@code task} on the main thread (next tick) and waits for its result.
|
* From a dispatched job's {@code work}: runs {@code task} on the main thread and waits for its result. Lets a job
|
||||||
* Lets a job capture its world snapshot only once it is its turn, so queued photos hold no snapshot. Returns null
|
* capture its world snapshot only once it is its turn, so queued photos hold no snapshot. The tasks of all jobs run
|
||||||
* without running {@code task} once {@code cancelled} is set (timeout, shutdown), the plugin is disabled or the
|
* one per tick in request order, so jobs starting together (a burst, or {@code max-concurrent} > 1) never stack
|
||||||
* worker is interrupted.
|
* their snapshots into one long tick. Returns null without running {@code task} once {@code cancelled} is set
|
||||||
|
* (timeout, shutdown), the plugin is disabled or the worker is interrupted.
|
||||||
*/
|
*/
|
||||||
public <T> T onMainThread(Supplier<T> task, AtomicBoolean cancelled) {
|
public <T> T onMainThread(Supplier<T> task, AtomicBoolean cancelled) {
|
||||||
|
if (shuttingDown) return null;
|
||||||
CompletableFuture<T> result = new CompletableFuture<>();
|
CompletableFuture<T> result = new CompletableFuture<>();
|
||||||
try {
|
mainTasks.add(new MainTask(
|
||||||
Bukkit.getScheduler().runTask(plugin, () -> {
|
() -> {
|
||||||
if (cancelled.get()) {
|
try {
|
||||||
result.complete(null);
|
result.complete(task.get());
|
||||||
return;
|
} catch (Throwable t) {
|
||||||
}
|
result.completeExceptionally(t);
|
||||||
try {
|
}
|
||||||
result.complete(task.get());
|
},
|
||||||
} catch (Throwable t) {
|
cancelled));
|
||||||
result.completeExceptionally(t);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
} catch (IllegalPluginAccessException e) {
|
|
||||||
return null; // disabled meanwhile
|
|
||||||
}
|
|
||||||
while (!cancelled.get()) {
|
while (!cancelled.get()) {
|
||||||
try {
|
try {
|
||||||
return result.get(50, TimeUnit.MILLISECONDS);
|
return result.get(50, TimeUnit.MILLISECONDS);
|
||||||
@@ -187,6 +192,18 @@ public final class RenderManager {
|
|||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Every tick: runs the oldest main-thread task whose job is still alive (cancelled ones are dropped). */
|
||||||
|
private void runNextMainTask() {
|
||||||
|
MainTask next;
|
||||||
|
while ((next = mainTasks.poll()) != null) {
|
||||||
|
if (next.cancelled.get()) continue; // its worker returns null on its own
|
||||||
|
next.run.run();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private record MainTask(Runnable run, AtomicBoolean cancelled) {}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Stops all rendering without leaving a job undelivered: running and queued jobs are cancelled, the workers get
|
* Stops all rendering without leaving a job undelivered: running and queued jobs are cancelled, the workers get
|
||||||
* a few seconds to wind down, then every job's callback runs here (on the main thread) — {@code onSuccess} for
|
* a few seconds to wind down, then every job's callback runs here (on the main thread) — {@code onSuccess} for
|
||||||
@@ -194,7 +211,9 @@ public final class RenderManager {
|
|||||||
*/
|
*/
|
||||||
public void shutdown() {
|
public void shutdown() {
|
||||||
shuttingDown = true;
|
shuttingDown = true;
|
||||||
|
mainTaskPump.cancel();
|
||||||
for (Job<?> job : jobs) job.cancelled.set(true);
|
for (Job<?> job : jobs) job.cancelled.set(true);
|
||||||
|
mainTasks.clear();
|
||||||
watchdog.shutdownNow();
|
watchdog.shutdownNow();
|
||||||
dispatcher.shutdownNow();
|
dispatcher.shutdownNow();
|
||||||
try {
|
try {
|
||||||
@@ -237,9 +256,11 @@ public final class RenderManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private static ForkJoinPool.ForkJoinWorkerThreadFactory lowPriorityForkJoinFactory() {
|
private static ForkJoinPool.ForkJoinWorkerThreadFactory lowPriorityForkJoinFactory() {
|
||||||
|
AtomicInteger counter = new AtomicInteger();
|
||||||
return pool -> {
|
return pool -> {
|
||||||
ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
|
ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
|
||||||
thread.setName("PixelPics-trace-" + thread.getPoolIndex());
|
// not getPoolIndex(): it is only assigned once the worker registers, here every name would end in -0
|
||||||
|
thread.setName("PixelPics-trace-" + counter.incrementAndGet());
|
||||||
thread.setPriority(Thread.MIN_PRIORITY);
|
thread.setPriority(Thread.MIN_PRIORITY);
|
||||||
thread.setDaemon(true);
|
thread.setDaemon(true);
|
||||||
return thread;
|
return thread;
|
||||||
|
|||||||
Reference in new issue
Block a user