@@ -130,6 +105,7 @@ final class ComputerExecutor { * Note, if command is not {@code null}, then some command is scheduled to be executed. Otherwise it is not * currently in the queue (or is currently being executed). */ + @GuardedBy("queueLock") private volatile @Nullable StateCommand command; /** @@ -137,43 +113,45 @@ final class ComputerExecutor { *
* Note, this should be empty if this computer is off - it is cleared on shutdown and when turning on again.
*/
+ @GuardedBy("queueLock")
private final Queue
@@ -508,15 +461,15 @@ final class ComputerExecutor {
* @see #command
* @see #eventQueue
*/
- void work() throws InterruptedException {
- if (interruptedEvent && !closed) {
- interruptedEvent = false;
- if (machine != null) {
- resumeMachine(null, null);
- return;
- }
+ @Override
+ public void work() throws InterruptedException {
+ workImpl();
+ synchronized (queueLock) {
+ if (wasPaused || command != null || !eventQueue.isEmpty()) enqueue();
}
+ }
+ private void workImpl() throws InterruptedException {
StateCommand command;
Event event = null;
synchronized (queueLock) {
@@ -524,7 +477,7 @@ final class ComputerExecutor {
this.command = null;
// If we've no command, pull something from the event queue instead.
- if (command == null) {
+ if (command == null && !wasPaused) {
if (!isOn) {
// We're not on and had no command, but we had work queued. This should never happen, so clear
// the event queue just in case.
@@ -537,6 +490,7 @@ final class ComputerExecutor {
}
if (command != null) {
+ wasPaused = false;
switch (command) {
case TURN_ON -> {
if (isOn) return;
@@ -553,23 +507,29 @@ final class ComputerExecutor {
shutdown();
computer.turnOn();
}
- case ABORT -> {
+ case ABORT_WITH_TIMEOUT -> {
if (!isOn) return;
displayFailure("Error running computer", TimeoutState.ABORT_MESSAGE);
shutdown();
}
- case ERROR -> {
+ case ABORT_WITH_ERROR -> {
if (!isOn) return;
displayFailure("Error running computer", "An internal error occurred, see logs.");
shutdown();
}
}
+ } else if (wasPaused) {
+ executor.setRemainingTime(timeRemaining);
+ resumeMachine(null, null);
} else if (event != null) {
+ executor.setRemainingTime(TimeoutState.TIMEOUT);
resumeMachine(event.name, event.args);
}
}
- void printState(StringBuilder out) {
+ @Override
+ @SuppressWarnings("GuardedBy")
+ public void writeState(StringBuilder out) {
out.append("Enqueued command: ").append(command).append('\n');
out.append("Enqueued events: ").append(eventQueue.size()).append('\n');
@@ -600,19 +560,23 @@ final class ComputerExecutor {
private void resumeMachine(@Nullable String event, @Nullable Object[] args) throws InterruptedException {
var result = Nullability.assertNonNull(machine).handleEvent(event, args);
- interruptedEvent = result.isPause();
- if (!result.isError()) return;
-
- displayFailure("Error running computer", result.getMessage());
- shutdown();
+ if (result.isError()) {
+ displayFailure("Error running computer", result.getMessage());
+ shutdown();
+ } else if (result.isPause()) {
+ wasPaused = true;
+ timeRemaining = executor.getRemainingTime();
+ } else {
+ wasPaused = false;
+ }
}
private enum StateCommand {
TURN_ON,
SHUTDOWN,
REBOOT,
- ABORT,
- ERROR,
+ ABORT_WITH_TIMEOUT,
+ ABORT_WITH_ERROR,
}
private record Event(String name, @Nullable Object[] args) {
diff --git a/projects/core/src/main/java/dan200/computercraft/core/computer/TimeoutState.java b/projects/core/src/main/java/dan200/computercraft/core/computer/TimeoutState.java
index c39f23cbd..147c32e61 100644
--- a/projects/core/src/main/java/dan200/computercraft/core/computer/TimeoutState.java
+++ b/projects/core/src/main/java/dan200/computercraft/core/computer/TimeoutState.java
@@ -5,7 +5,8 @@
package dan200.computercraft.core.computer;
import com.google.errorprone.annotations.concurrent.GuardedBy;
-import dan200.computercraft.core.computer.computerthread.ComputerThread;
+import dan200.computercraft.core.computer.computerthread.ComputerScheduler;
+import dan200.computercraft.core.computer.computerthread.ManagedTimeoutState;
import dan200.computercraft.core.lua.ILuaMachine;
import dan200.computercraft.core.lua.MachineResult;
@@ -25,90 +26,54 @@ import java.util.concurrent.TimeUnit;
* (namely, throwing a "Too long without yielding" error).
*
* Now, if a computer still does not stop after that period, they're behaving really badly. 1.5 seconds after a soft
- * abort ({@link #ABORT_TIMEOUT}), we trigger a hard abort (note, this is done from the computer thread manager). This
- * will destroy the entire Lua runtime and shut the computer down.
+ * abort ({@link #ABORT_TIMEOUT}), we trigger a hard abort. This will destroy the entire Lua runtime and shut the
+ * computer down.
*
* The Lua runtime is also allowed to pause execution if there are other computers contesting for work. All computers
- * are allowed to run for {@link ComputerThread#scaledPeriod()} nanoseconds (see {@link #currentDeadline}). After that
- * period, if any computers are waiting to be executed then we'll set the paused flag to true ({@link #isPaused()}.
+ * are guaranteed to run for some time. After that period, if any computers are waiting to be executed then we'll set
+ * the paused flag to true ({@link #isPaused()}.
*
- * @see ComputerThread
+ * @see ComputerScheduler
+ * @see ManagedTimeoutState
* @see ILuaMachine
* @see MachineResult#isPause()
*/
-public final class TimeoutState {
+public abstract class TimeoutState {
/**
- * The total time a task is allowed to run before aborting in nanoseconds.
+ * The time (in nanoseconds) are computer is allowed to run for its long-running tasks, such as startup and
+ * shutdown.
*/
- static final long TIMEOUT = TimeUnit.MILLISECONDS.toNanos(7000);
+ public static final long BASE_TIMEOUT = TimeUnit.SECONDS.toNanos(30);
+
+ /**
+ * The total time the Lua VM is allowed to run before aborting in nanoseconds.
+ */
+ public static final long TIMEOUT = TimeUnit.MILLISECONDS.toNanos(7000);
/**
* The time the task is allowed to run after each abort in nanoseconds.
*/
- static final long ABORT_TIMEOUT = TimeUnit.MILLISECONDS.toNanos(1500);
+ public static final long ABORT_TIMEOUT = TimeUnit.MILLISECONDS.toNanos(1500);
/**
* The error message to display when we trigger an abort.
*/
public static final String ABORT_MESSAGE = "Too long without yielding";
- private final ComputerThread scheduler;
@GuardedBy("this")
private final List
+ * Normally this will be called automatically by the {@link ComputerScheduler}, but it may be useful to call this
+ * manually if the most up-to-date information is needed.
*/
- public synchronized void refresh() {
- // Important: The weird arithmetic here is important, as nanoTime may return negative values, and so we
- // need to handle overflow.
- var now = System.nanoTime();
- var changed = false;
- if (!paused && (paused = currentDeadline - now <= 0 && scheduler.hasPendingWork())) { // now >= currentDeadline
- changed = true;
- }
- if (!softAbort && (softAbort = now - cumulativeStart - TIMEOUT >= 0)) { // now - cumulativeStart >= TIMEOUT
- changed = true;
- }
-
- if (changed) updateListeners();
- }
+ public abstract void refresh();
/**
* Whether we should pause execution of this machine.
@@ -118,7 +83,7 @@ public final class TimeoutState {
*
* @return Whether we should pause execution.
*/
- public boolean isPaused() {
+ public final boolean isPaused() {
return paused;
}
@@ -127,7 +92,7 @@ public final class TimeoutState {
*
* @return {@code true} if we should throw a timeout error.
*/
- public boolean isSoftAborted() {
+ public final boolean isSoftAborted() {
return softAbort;
}
@@ -136,64 +101,22 @@ public final class TimeoutState {
*
* @return {@code true} if the machine should be forcibly shut down.
*/
- public boolean isHardAborted() {
+ public final boolean isHardAborted() {
return hardAbort;
}
- /**
- * If the machine should be forcibly aborted.
- */
- void hardAbort() {
- softAbort = hardAbort = true;
- synchronized (this) {
- updateListeners();
- }
- }
-
- /**
- * Start the current and cumulative timers again.
- */
- void startTimer() {
- var now = System.nanoTime();
- currentStart = now;
- currentDeadline = now + scheduler.scaledPeriod();
- // Compute the "nominal start time".
- cumulativeStart = now - cumulativeElapsed;
- }
-
- /**
- * Pauses the cumulative time, to be resumed by {@link #startTimer()}.
- *
- * @see #nanoCumulative()
- */
- synchronized void pauseTimer() {
- // We set the cumulative time to difference between current time and "nominal start time".
- cumulativeElapsed = System.nanoTime() - cumulativeStart;
- paused = false;
- updateListeners();
- }
-
- /**
- * Resets the cumulative time and resets the abort flags.
- */
- synchronized void stopTimer() {
- cumulativeElapsed = 0;
- paused = softAbort = hardAbort = false;
- updateListeners();
- }
-
@GuardedBy("this")
- private void updateListeners() {
+ protected final void updateListeners() {
for (var listener : listeners) listener.run();
}
- public synchronized void addListener(Runnable listener) {
+ public final synchronized void addListener(Runnable listener) {
Objects.requireNonNull(listener, "listener cannot be null");
listeners.add(listener);
listener.run();
}
- public synchronized void removeListener(Runnable listener) {
+ public final synchronized void removeListener(Runnable listener) {
Objects.requireNonNull(listener, "listener cannot be null");
listeners.remove(listener);
}
diff --git a/projects/core/src/main/java/dan200/computercraft/core/computer/computerthread/ComputerScheduler.java b/projects/core/src/main/java/dan200/computercraft/core/computer/computerthread/ComputerScheduler.java
new file mode 100644
index 000000000..3c5ee827a
--- /dev/null
+++ b/projects/core/src/main/java/dan200/computercraft/core/computer/computerthread/ComputerScheduler.java
@@ -0,0 +1,127 @@
+// SPDX-FileCopyrightText: 2023 The CC: Tweaked Developers
+//
+// SPDX-License-Identifier: MPL-2.0
+
+package dan200.computercraft.core.computer.computerthread;
+
+import dan200.computercraft.core.computer.TimeoutState;
+import dan200.computercraft.core.metrics.MetricsObserver;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * The {@link ComputerScheduler} is responsible for executing computers on the computer thread(s).
+ *
+ * This handles both scheduling the computers for work across multiple threads, as well as {@linkplain TimeoutState timing out}
+ * or pausing the computer if they execute for too long.
+ *
+ * This API is composed of two interfaces, a {@link Worker} and {@link Executor}. The {@link ComputerScheduler}
+ * implementation will supply an {@link Executor}, while consuming classes should implement {@link Worker}.
+ *
+ * In practice, this interface is only implemented by {@link ComputerThread} (and consumed by {@link dan200.computercraft.core.computer.ComputerExecutor}),
+ * however this interface is useful to enforce separation of the two.
+ *
+ * @see ManagedTimeoutState
+ */
+public interface ComputerScheduler {
+ Executor createExecutor(Worker worker, MetricsObserver metrics);
+
+ boolean stop(long timeout, TimeUnit unit) throws InterruptedException;
+
+ /**
+ * The {@link Executor} holds the state of a {@link Worker} within the scheduler.
+ *
+ * This is used to schedule the worker for execution, as well as providing some additional control over the
+ * {@link TimeoutState}.
+ */
+ interface Executor {
+ /**
+ * Submit the executor to the scheduler, marking it as ready {@linkplain Worker#work() to run some work}.
+ *
+ * This function is idempotent - if the executor is already queued, nothing will happen.
+ */
+ void submit();
+
+ /**
+ * Get the executor's {@link TimeoutState}.
+ *
+ * @return The executor's timeout state.
+ */
+ TimeoutState timeoutState();
+
+ /**
+ * Get the amount of time this computer can run for before being interrupted.
+ *
+ * This value starts off as {@link TimeoutState#BASE_TIMEOUT}, but may be reduced by
+ * {@link #setRemainingTime(long)}.
+ *
+ * @return The time this computer can run for being interrupted.
+ * @see #getRemainingTime()
+ */
+ long getRemainingTime();
+
+ /**
+ * Set the amount of this computer can execute for before being interrupted.
+ *
+ * This value will typically be {@link TimeoutState#TIMEOUT}, but may be a previous value of
+ * {@link #getRemainingTime()} if the computer is resuming after {@linkplain TimeoutState#isPaused() being
+ * paused}.
+ *
+ * @param time The time this computer can execute for.
+ * @see #getRemainingTime()
+ */
+ void setRemainingTime(long time);
+ }
+
+ /**
+ * A {@link Worker} is responsible for actually running the computer's code.
+ *
+ * his handles {@linkplain Worker#work() running the actual computer logic}, as well as providing some additional
+ * control methods.
+ *
+ * This should be implemented by the consuming class.
+ */
+ interface Worker {
+ /**
+ * Perform any work that the computer needs to do, for instance turning on, shutting down or actually running
+ * code.
+ *
+ * If the computer needs to run immediately again, it should call {@link Executor#submit()} within this method.
+ *
+ * @throws InterruptedException If the computer has run for too long and must be terminated.
+ */
+ void work() throws InterruptedException;
+
+ /**
+ * Get the ID of this computer, used in error messages.
+ *
+ * @return This computers ID.
+ */
+ int getComputerID();
+
+ /**
+ * Write any useful debugging information computer to the provided buffer. This is used in log messages when the
+ * computer has run for too long.
+ *
+ * @param output The buffer to write to.
+ */
+ void writeState(StringBuilder output);
+
+ /**
+ * Abort this whole computer due to a timeout.
+ */
+ void abortWithTimeout();
+
+ /**
+ * Abort this whole computer due to some internal error.
+ */
+ void abortWithError();
+
+ /**
+ * "Unload" this computer, shutting it down and preventing it from running again.
+ *
+ * This is called by the scheduler when {@linkplain ComputerScheduler#stop(long, TimeUnit) it is stopped.}
+ */
+ void unload();
+ }
+}
diff --git a/projects/core/src/main/java/dan200/computercraft/core/computer/computerthread/ComputerThread.java b/projects/core/src/main/java/dan200/computercraft/core/computer/computerthread/ComputerThread.java
index 17c0654c3..7f7821fd8 100644
--- a/projects/core/src/main/java/dan200/computercraft/core/computer/computerthread/ComputerThread.java
+++ b/projects/core/src/main/java/dan200/computercraft/core/computer/computerthread/ComputerThread.java
@@ -5,25 +5,23 @@
package dan200.computercraft.core.computer.computerthread;
import com.google.common.annotations.VisibleForTesting;
-import com.google.errorprone.annotations.concurrent.GuardedBy;
import dan200.computercraft.core.ComputerContext;
import dan200.computercraft.core.Logging;
-import dan200.computercraft.core.computer.ComputerExecutor;
import dan200.computercraft.core.computer.TimeoutState;
+import dan200.computercraft.core.metrics.Metrics;
+import dan200.computercraft.core.metrics.MetricsObserver;
import dan200.computercraft.core.util.ThreadUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
import java.util.Objects;
import java.util.TreeSet;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;
@@ -31,9 +29,9 @@ import java.util.concurrent.locks.ReentrantLock;
/**
* Runs all scheduled tasks for computers in a {@link ComputerContext}.
*
- * This acts as an over-complicated {@link ThreadPoolExecutor}: It creates several {@link Worker} threads which pull
- * tasks from a shared queue, executing them. It also creates a single {@link Monitor} thread, which updates computer
- * timeouts, killing workers if they have not been terminated by {@link TimeoutState#isSoftAborted()}.
+ * This acts as an over-complicated {@link ThreadPoolExecutor}: It creates several {@linkplain WorkerThread worker
+ * threads} which pull tasks from a shared queue, executing them. It also creates a single {@link Monitor} thread, which
+ * updates computer timeouts, killing workers if they have not been terminated by {@link TimeoutState#isSoftAborted()}.
*
* Computers are executed using a priority system, with those who have spent less time executing having a higher
* priority than those hogging the thread. This, combined with {@link TimeoutState#isPaused()} means we can reduce the
@@ -41,20 +39,16 @@ import java.util.concurrent.locks.ReentrantLock;
*
* This is done using an implementation of Linux's Completely Fair Scheduler. When a computer executes, we compute what
* share of execution time it has used (time executed/number of tasks). We then pick the computer who has the least
- * "virtual execution time" (aka {@link ComputerExecutor#virtualRuntime}).
+ * "virtual execution time" (aka {@link ExecutorImpl#virtualRuntime}).
*
* When adding a computer to the queue, we make sure its "virtual runtime" is at least as big as the smallest runtime.
* This means that adding computers which have slept a lot do not then have massive priority over everyone else. See
- * {@link #queue(ComputerExecutor)} for how this is implemented.
+ * {@link #queue(ExecutorImpl)} for how this is implemented.
*
* In reality, it's unlikely that more than a few computers are waiting to execute at once, so this will not have much
* effect unless you have a computer hogging execution time. However, it is pretty effective in those situations.
- *
- * @see TimeoutState For how hard timeouts are handled.
- * @see ComputerExecutor For how computers actually do execution.
*/
-@SuppressWarnings("GuardedBy") // FIXME: Hard to know what the correct thing to do is.
-public final class ComputerThread {
+public final class ComputerThread implements ComputerScheduler {
private static final Logger LOG = LoggerFactory.getLogger(ComputerThread.class);
/**
@@ -98,7 +92,7 @@ public final class ComputerThread {
/**
* Time difference between reporting crashed threads.
*
- * @see Worker#reportTimeout(ComputerExecutor, long)
+ * @see WorkerThread#reportTimeout(ExecutorImpl, long)
*/
private static final long REPORT_DEBOUNCE = TimeUnit.SECONDS.toNanos(1);
@@ -125,7 +119,7 @@ public final class ComputerThread {
* The array of current workers, and their owning threads.
*/
@GuardedBy("threadLock")
- private final Worker[] workers;
+ private final WorkerThread[] workers;
/**
* The number of workers in {@link #workers}.
@@ -139,29 +133,33 @@ public final class ComputerThread {
private final long minPeriod;
private final ReentrantLock computerLock = new ReentrantLock();
- private final Condition workerWakeup = computerLock.newCondition();
- private final Condition monitorWakeup = computerLock.newCondition();
+ private final @GuardedBy("computerLock") Condition workerWakeup = computerLock.newCondition();
+ private final @GuardedBy("computerLock") Condition monitorWakeup = computerLock.newCondition();
private final AtomicInteger idleWorkers = new AtomicInteger(0);
/**
* Active queues to execute.
*/
- private final TreeSet
- * You must be holding {@link ComputerExecutor}'s {@code queueLock} when calling this method - it should only
+ * You must be holding {@link ExecutorImpl}'s {@code queueLock} when calling this method - it should only
* be called from {@code enqueue}.
*
* @param executor The computer to execute work on.
*/
- void queue(ComputerExecutor executor) {
+ void queue(ExecutorImpl executor) {
computerLock.lock();
try {
if (state.get() != RUNNING) throw new IllegalStateException("ComputerThread is no longer running");
@@ -284,9 +298,6 @@ public final class ComputerThread {
// Ensure we've got a worker running.
ensureRunning();
- if (executor.onComputerQueue) throw new IllegalStateException("Cannot queue already queued executor");
- executor.onComputerQueue = true;
-
updateRuntimes(null);
// We're not currently on the queue, so update its current execution time to
@@ -318,14 +329,15 @@ public final class ComputerThread {
/**
- * Update the {@link ComputerExecutor#virtualRuntime}s of all running tasks, and then update the
+ * Update the {@link ExecutorImpl#virtualRuntime}s of all running tasks, and then update the
* {@link #minimumVirtualRuntime} based on the current tasks.
*
* This is called before queueing tasks, to ensure that {@link #minimumVirtualRuntime} is up-to-date.
*
* @param current The machine which we updating runtimes from.
*/
- private void updateRuntimes(@Nullable ComputerExecutor current) {
+ @GuardedBy("computerLock")
+ private void updateRuntimes(@Nullable ExecutorImpl current) {
var minRuntime = Long.MAX_VALUE;
// If we've a task on the queue, use that as our base time.
@@ -334,7 +346,7 @@ public final class ComputerThread {
// Update all the currently executing tasks
var now = System.nanoTime();
var tasks = 1 + computerQueue.size();
- for (@Nullable var runner : workers) {
+ for (@Nullable var runner : workersReadOnly()) {
if (runner == null) continue;
var executor = runner.currentExecutor.get();
if (executor == null) continue;
@@ -359,22 +371,9 @@ public final class ComputerThread {
* Ensure the "currently working" state of the executor is reset, the timings are updated, and then requeue the
* executor if needed.
*
- * @param runner The runner this task was on.
* @param executor The executor to requeue
*/
- private void afterWork(Worker runner, ComputerExecutor executor) {
- // Clear the executor's thread.
- var currentThread = executor.executingThread.getAndSet(null);
- if (currentThread != runner.owner) {
-
- LOG.error(
- "Expected computer #{} to be running on {}, but already running on {}. This is a SERIOUS bug, please report with your debug.log.",
- executor.getComputer().getID(),
- runner.owner.getName(),
- currentThread == null ? "nothing" : currentThread.getName()
- );
- }
-
+ private void afterWork(ExecutorImpl executor) {
computerLock.lock();
try {
updateRuntimes(executor);
@@ -390,6 +389,13 @@ public final class ComputerThread {
}
}
+ @SuppressWarnings("GuardedBy")
+ private int computerQueueSize() {
+ // FIXME: We access this on other threads (in TimeoutState), so their reads won't be consistent. This isn't
+ // "critical" behaviour, so not clear if it matters too much.
+ return computerQueue.size();
+ }
+
/**
* The scaled period for a single task.
*
@@ -399,11 +405,8 @@ public final class ComputerThread {
* @see #LATENCY_MAX_TASKS
*/
long scaledPeriod() {
- // FIXME: We access this on other threads (in TimeoutState), so their reads won't be consistent. This isn't
- // "critical" behaviour, so not clear if it matters too much.
-
// +1 to include the current task
- var count = 1 + computerQueue.size();
+ var count = 1 + computerQueueSize();
return count < LATENCY_MAX_TASKS ? latency / count : minPeriod;
}
@@ -413,9 +416,8 @@ public final class ComputerThread {
* @return If we have work queued up.
*/
@VisibleForTesting
- public boolean hasPendingWork() {
- // FIXME: See comment in scaledPeriod. Again, we access this in multiple threads but not clear if it matters!
- return !computerQueue.isEmpty();
+ boolean hasPendingWork() {
+ return computerQueueSize() > 0;
}
/**
@@ -424,12 +426,11 @@ public final class ComputerThread {
*
* @return If the computer threads are busy.
*/
- @GuardedBy("computerLock")
private boolean isBusy() {
- return computerQueue.size() > idleWorkers.get();
+ return computerQueueSize() > idleWorkers.get();
}
- private void workerFinished(Worker worker) {
+ private void workerFinished(WorkerThread worker) {
// We should only shut down a worker once! This should only happen if we fail to abort a worker and then the
// worker finishes normally.
if (!worker.running.getAndSet(false)) return;
@@ -444,6 +445,7 @@ public final class ComputerThread {
workerCount--;
if (workers[worker.index] != worker) {
+ assert false : "workerFinished but inconsistent worker";
LOG.error("Worker {} closed, but new runner has been spawned.", worker.index);
} else if (state.get() == RUNNING || (state.get() == STOPPING && hasPendingWork())) {
addWorker(worker.index);
@@ -457,7 +459,7 @@ public final class ComputerThread {
}
/**
- * Observes all currently active {@link Worker}s and terminates their tasks once they have exceeded the hard
+ * Observes all currently active {@link WorkerThread}s and terminates their tasks once they have exceeded the hard
* abort limit.
*
* @see TimeoutState
@@ -493,7 +495,7 @@ public final class ComputerThread {
}
private void checkRunners() {
- for (@Nullable var runner : workers) {
+ for (@Nullable var runner : workersReadOnly()) {
if (runner == null) continue;
// If the worker has no work, skip
@@ -505,25 +507,28 @@ public final class ComputerThread {
// If we're still within normal execution times (TIMEOUT) or soft abort (ABORT_TIMEOUT),
// then we can let the Lua machine do its work.
- var afterStart = executor.timeout.nanoCumulative();
- var afterHardAbort = afterStart - TimeoutState.TIMEOUT - TimeoutState.ABORT_TIMEOUT;
+ var remainingTime = executor.timeout.getRemainingTime();
+ // If remainingTime > 0, then we're executing normally,
+ // If remainingTime > -ABORT_TIMEOUT, then we've soft aborted.
+ // Otherwise, remainingTime <= -ABORT_TIMEOUT, and we've run over by -ABORT_TIMEOUT - remainingTime.
+ var afterHardAbort = -remainingTime - TimeoutState.ABORT_TIMEOUT;
if (afterHardAbort < 0) continue;
// Set the hard abort flag.
executor.timeout.hardAbort();
- executor.abort();
+ executor.worker.abortWithTimeout();
if (afterHardAbort >= TimeoutState.ABORT_TIMEOUT * 2) {
// If we've hard aborted and interrupted, and we're still not dead, then mark the worker
// as dead, finish off the task, and spawn a new runner.
- runner.reportTimeout(executor, afterStart);
+ runner.reportTimeout(executor, remainingTime);
runner.owner.interrupt();
workerFinished(runner);
} else if (afterHardAbort >= TimeoutState.ABORT_TIMEOUT) {
// If we've hard aborted but we're still not dead, dump the stack trace and interrupt
// the task.
- runner.reportTimeout(executor, afterStart);
+ runner.reportTimeout(executor, remainingTime);
runner.owner.interrupt();
}
}
@@ -533,11 +538,11 @@ public final class ComputerThread {
/**
* Pulls tasks from the {@link #computerQueue} queue and runs them.
*
- * This is responsible for running the {@link ComputerExecutor#work()}, {@link ComputerExecutor#beforeWork()} and
- * {@link ComputerExecutor#afterWork()} functions. Everything else is either handled by the executor, timeout
- * state or monitor.
+ * This is responsible for running the {@link ComputerScheduler.Worker#work()}, {@link ExecutorImpl#beforeWork()}
+ * and {@link ExecutorImpl#afterWork()} functions. Everything else is either handled by the executor,
+ * timeout state or monitor.
*/
- private final class Worker implements Runnable {
+ private final class WorkerThread implements Runnable {
/**
* The index into the {@link #workers} array.
*/
@@ -552,21 +557,21 @@ public final class ComputerThread {
* Whether this runner is currently executing. This may be set to false when this worker terminates, or when
* we try to abandon a worker in the monitor
*
- * @see #workerFinished(Worker)
+ * @see #workerFinished(WorkerThread)
*/
final AtomicBoolean running = new AtomicBoolean(true);
/**
* The computer we're currently running.
*/
- final AtomicReference
+ * Executors are either enqueued (have more work to do) or not and working or not. This enum encapsulates the four
+ * combinations of these properties, with the following transitions:
+ *
+ *
+ * While almost all {@link TimeoutState} implementations will be derived from this class, the two are intentionally kept
+ * separate. This class is intended for the {@link ComputerScheduler} (which is responsible for controlling the
+ * timeout), and not for the main computer logic, which only needs to check timeout flags.
+ *
+ * This class tracks the time a computer was started (and thus {@linkplain #getExecutionTime()} how long it has been
+ * running for), as well as the deadline for when a computer should be soft aborted and paused.
+ */
+public abstract class ManagedTimeoutState extends TimeoutState {
+ /**
+ * When execution of this computer started.
+ *
+ * @see #getExecutionTime()
+ */
+ private long startTime;
+
+ /**
+ * The time when this computer should be aborted.
+ *
+ * @see #getRemainingTime()
+ * @see #setRemainingTime(long)
+ */
+ private long abortDeadline;
+
+ /**
+ * The time when this computer should be paused if {@link ComputerThread#hasPendingWork()} is set.
+ */
+ private long pauseDeadline;
+
+ @Override
+ public final synchronized void refresh() {
+ // Important: The weird arithmetic here is important, as nanoTime may return negative values, and so we
+ // need to handle overflow.
+ var now = System.nanoTime();
+ var changed = false;
+ if (!paused && Long.compareUnsigned(now, pauseDeadline) >= 0 && shouldPause()) { // now >= currentDeadline
+ paused = true;
+ changed = true;
+ }
+ if (!softAbort && Long.compareUnsigned(now, abortDeadline) >= 0) { // now >= currentAbort
+ softAbort = true;
+ changed = true;
+ }
+ if (softAbort && !hardAbort && Long.compareUnsigned(now, abortDeadline + ABORT_TIMEOUT) >= 0) { // now >= currentAbort + ABORT_TIMEOUT.
+ hardAbort = true;
+ changed = true;
+ }
+
+ if (changed) updateListeners();
+ }
+
+ /**
+ * Get how long this computer has been executing for.
+ *
+ * @return How long the computer has been running for in nanoseconds.
+ */
+ public final long getExecutionTime() {
+ return System.nanoTime() - startTime;
+ }
+
+ /**
+ * Get how long this computer is permitted to run before being aborted.
+ *
+ * @return The remaining time, in nanoseconds.
+ * @see ComputerScheduler.Executor#getRemainingTime()
+ */
+ public final long getRemainingTime() {
+ return abortDeadline - System.nanoTime();
+ }
+
+ /**
+ * Set how long this computer is permitted to run before being aborted.
+ *
+ * @param time The remaining time, in nanoseconds.
+ * @see ComputerScheduler.Executor#setRemainingTime(long)
+ */
+ public final void setRemainingTime(long time) {
+ abortDeadline = startTime + time;
+ }
+
+ /**
+ * Set the hard-abort flag immediately.
+ */
+ public final void hardAbort() {
+ softAbort = hardAbort = true;
+ synchronized (this) {
+ updateListeners();
+ }
+ }
+
+ /**
+ * Start this timer, recording the current start time, and deadline before a computer may be paused.
+ *
+ * @param pauseTimeout The minimum time this computer can run before potentially being paused.
+ */
+ public final synchronized void startTimer(long pauseTimeout) {
+ var now = System.nanoTime();
+ startTime = now;
+ abortDeadline = now + BASE_TIMEOUT;
+ pauseDeadline = now + pauseTimeout;
+ }
+
+ /**
+ * Clear the paused and abort flags.
+ */
+ public final synchronized void reset() {
+ paused = softAbort = hardAbort = false;
+ updateListeners();
+ }
+
+ /**
+ * Determine if this computer should be paused, as other computers are contending for work.
+ *
+ * @return If this computer should be paused.
+ */
+ protected abstract boolean shouldPause();
+}
diff --git a/projects/core/src/test/java/dan200/computercraft/core/computer/computerthread/ComputerThreadRunner.java b/projects/core/src/test/java/dan200/computercraft/core/computer/computerthread/ComputerThreadRunner.java
new file mode 100644
index 000000000..4106e85e0
--- /dev/null
+++ b/projects/core/src/test/java/dan200/computercraft/core/computer/computerthread/ComputerThreadRunner.java
@@ -0,0 +1,142 @@
+// SPDX-FileCopyrightText: 2023 The CC: Tweaked Developers
+//
+// SPDX-License-Identifier: MPL-2.0
+
+package dan200.computercraft.core.computer.computerthread;
+
+import dan200.computercraft.core.computer.TimeoutState;
+import dan200.computercraft.core.metrics.MetricsObserver;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+
+import javax.annotation.concurrent.GuardedBy;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.BiConsumer;
+
+public class ComputerThreadRunner implements AutoCloseable {
+ private final ComputerThread thread;
+
+ private final Lock errorLock = new ReentrantLock();
+ private final @GuardedBy("errorLock") Condition hasError = errorLock.newCondition();
+ @GuardedBy("errorLock")
+ private @MonotonicNonNull Throwable error = null;
+
+ public ComputerThreadRunner() {
+ this.thread = new ComputerThread(1);
+ }
+
+ public ComputerThread thread() {
+ return thread;
+ }
+
+ @Override
+ public void close() {
+ try {
+ if (!thread.stop(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Failed to shutdown ComputerContext in time.");
+ }
+ } catch (InterruptedException e) {
+ throw new IllegalStateException("Runtime thread was interrupted", e);
+ }
+ }
+
+ public Worker createWorker(BiConsumer
- * This instead just exucutes work as soon as possible via {@link Callbacks#setImmediate(TimerHandler)}. Timeouts are
- * instead handled via polling, see {@link cc.tweaked.web.builder.PatchCobalt}.
+ * Timeouts are instead handled via polling, see {@link cc.tweaked.web.builder.PatchCobalt}.
+ *
+ * @see ComputerThread
*/
-public class TComputerThread {
- private static final ArrayDeque{@code
+ * submit() afterWork()
+ * IDLE ---------> ON_QUEUE <----------- REPEAT
+ * ^ | ^
+ * | | runImpl() |
+ * | V |
+ * +---------------RUNNING----------------+
+ * afterWork() submit()
+ * }
+ */
+ enum ExecutorState {
+ /**
+ * This executor is idle.
+ */
+ IDLE,
+
+ /**
+ * This executor is on the queue but idle.
+ */
+ ON_QUEUE,
+
+ /**
+ * This executor is running and will transition to idle after execution.
+ */
+ RUNNING,
+
+ /**
+ * This executor is running and should run again after this task finishes.
+ */
+ REPEAT;
+
+ ExecutorState enqueue() {
+ return switch (this) {
+ case IDLE, ON_QUEUE -> ON_QUEUE;
+ case RUNNING, REPEAT -> REPEAT;
+ };
+ }
+
+ ExecutorState requeue() {
+ return switch (this) {
+ case IDLE, ON_QUEUE -> {
+ assert false : "Impossible state after executing";
+ LOG.error("Impossible state - calling requeue with {}.", this);
+ yield ExecutorState.ON_QUEUE;
+ }
+ case RUNNING -> ExecutorState.IDLE;
+ case REPEAT -> ExecutorState.ON_QUEUE;
+ };
+ }
+ }
+
+ private final class ExecutorImpl implements Executor {
+ public static final AtomicReferenceFieldUpdater